Changeset 16a46b3


Ignore:
Timestamp:
Aug 13, 2012 3:12:33 PM (8 years ago)
Author:
zzz <zzz@…>
Branches:
master
Children:
44edf70
Parents:
e9cc851
Message:
  • SSU EstablishmentManager?:
    • Remove use of outbound timers in EstablishmentManager?; drive all events in Establisher thread
    • Don't change nonces when retransmitting intro packets
    • More synchronization in EstablishmentManager?
    • Increase establishment timeouts and implement timeouts for individual phases (ticket #669)
    • Fix bug where InboundEstablishState?.createdPacketSent() wasn't being called, so SessionCreated? packets weren't retransmitted
    • Increase retransmission timeout for SessionCreated? and implement backoff
    • Send destroy if establishment times out in the middle
    • Fix code that pulls outbound states off a deferred queue
    • Improve UDPPacket.toString() for debugging
    • More logging of packets dropped in EstablishmentManager?
    • Change establish states to enums
Files:
9 edited

Legend:

Unmodified
Added
Removed
  • history.txt

    re9cc851 r16a46b3  
     12012-08-13 zzz
     2 * SSU EstablishmentManager:
     3   - Remove use of outbound timers in EstablishmentManager; drive all events in Establisher thread
     4   - Don't change nonces when retransmitting intro packets
     5   - More synchronization in EstablishmentManager
     6   - Increase establishment timeouts and implement timeouts for individual phases (ticket #669)
     7   - Fix bug where InboundEstablishState.createdPacketSent() wasn't being called,
     8     so SessionCreated packets weren't retransmitted
     9   - Increase retransmission timeout for SessionCreated and implement backoff
     10   - Send destroy if establishment times out in the middle
     11   - Fix code that pulls outbound states off a deferred queue
     12   - Improve UDPPacket.toString() for debugging
     13   - More logging of packets dropped in EstablishmentManager
     14   - Change establish states to enums
     15
    1162012-08-12 zzz
    217 * Jetty: Don't use direct byte buffers that may be leaking (ticket #679)
  • router/java/src/net/i2p/router/RouterVersion.java

    re9cc851 r16a46b3  
    1919    public final static String ID = "Monotone";
    2020    public final static String VERSION = CoreVersion.VERSION;
    21     public final static long BUILD = 7;
     21    public final static long BUILD = 8;
    2222
    2323    /** for example "-test" */
  • router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java

    re9cc851 r16a46b3  
    66import java.util.Iterator;
    77import java.util.List;
     8import java.util.Map;
    89import java.util.concurrent.ConcurrentHashMap;
    910
     
    2122import net.i2p.router.networkdb.kademlia.FloodfillNetworkDatabaseFacade;
    2223import net.i2p.router.transport.crypto.DHSessionKeyBuilder;
     24import static net.i2p.router.transport.udp.InboundEstablishState.InboundState.*;
     25import static net.i2p.router.transport.udp.OutboundEstablishState.OutboundState.*;
    2326import net.i2p.util.Addresses;
    2427import net.i2p.util.I2PThread;
    2528import net.i2p.util.Log;
    26 import net.i2p.util.SimpleScheduler;
    27 import net.i2p.util.SimpleTimer;
    2829
    2930/**
     
    5051    private int _activity;
    5152   
    52     /** max outbound in progress */
    53     private static final int DEFAULT_MAX_CONCURRENT_ESTABLISH = 20;
     53    /** max outbound in progress - max inbound is half of this */
     54    private static final int DEFAULT_MAX_CONCURRENT_ESTABLISH = 30;
    5455    private static final String PROP_MAX_CONCURRENT_ESTABLISH = "i2np.udp.maxConcurrentEstablish";
    5556
     
    5960    /** max queued msgs per peer while the peer connection is queued */
    6061    private static final int MAX_QUEUED_PER_PEER = 3;
    61  
     62   
     63    private static final long MAX_NONCE = 0xFFFFFFFFl;
     64
     65    /**
     66     * Kill any outbound that takes more than this.
     67     * Two round trips (Req-Created-Confirmed-Data) for direct;
     68     * 3 1/2 round trips (RReq-RResp+Intro-HolePunch-Req-Created-Confirmed-Data) for indirect.
     69     * Note that this is way too long for us to be able to fall back to NTCP
     70     * for individual messages unless the message timer fires first.
     71     * But SSU probably isn't higher priority than NTCP.
     72     * And it's important to not fail an establishment too soon and waste it.
     73     */
     74    private static final int MAX_OB_ESTABLISH_TIME = 35*1000;
     75
     76    /**
     77     * Kill any inbound that takes more than this
     78     * One round trip (Created-Confirmed)
     79     */
     80    private static final int MAX_IB_ESTABLISH_TIME = 20*1000;
     81
     82    /** max before receiving a response to a single message during outbound establishment */
     83    private static final int OB_MESSAGE_TIMEOUT = 15*1000;
     84
     85    /** for the DSM and or netdb store */
     86    private static final int DATA_MESSAGE_TIMEOUT = 10*1000;
    6287   
    6388    public EstablishmentManager(RouterContext ctx, UDPTransport transport) {
     
    7398        _context.statManager().createRateStat("udp.inboundEstablishTime", "How long it takes for a new inbound session to be established", "udp", UDPTransport.RATES);
    7499        _context.statManager().createRateStat("udp.outboundEstablishTime", "How long it takes for a new outbound session to be established", "udp", UDPTransport.RATES);
    75         _context.statManager().createRateStat("udp.inboundEstablishFailedState", "What state a failed inbound establishment request fails in", "udp", UDPTransport.RATES);
    76         _context.statManager().createRateStat("udp.outboundEstablishFailedState", "What state a failed outbound establishment request fails in", "udp", UDPTransport.RATES);
     100        //_context.statManager().createRateStat("udp.inboundEstablishFailedState", "What state a failed inbound establishment request fails in", "udp", UDPTransport.RATES);
     101        //_context.statManager().createRateStat("udp.outboundEstablishFailedState", "What state a failed outbound establishment request fails in", "udp", UDPTransport.RATES);
    77102        _context.statManager().createRateStat("udp.sendIntroRelayRequest", "How often we send a relay request to reach a peer", "udp", UDPTransport.RATES);
    78103        _context.statManager().createRateStat("udp.sendIntroRelayTimeout", "How often a relay request times out before getting a response (due to the target or intro peer being offline)", "udp", UDPTransport.RATES);
     
    101126        t.start();
    102127    }
     128
    103129    public void shutdown() {
    104130        _alive = false;
     
    108134    /**
    109135     * Grab the active establishing state
     136     * @return null if none
    110137     */
    111138    InboundEstablishState getInboundState(RemoteHostId from) {
     
    116143    }
    117144   
     145    /**
     146     * Grab the active establishing state
     147     * @return null if none
     148     */
    118149    OutboundEstablishState getOutboundState(RemoteHostId from) {
    119150            OutboundEstablishState state = _outboundStates.get(from);
     
    123154    }
    124155   
     156    /**
     157     * How many concurrent outbound sessions to deal with
     158     */
    125159    private int getMaxConcurrentEstablish() {
    126160        return _context.getProperty(PROP_MAX_CONCURRENT_ESTABLISH, DEFAULT_MAX_CONCURRENT_ESTABLISH);
     
    175209            if (state == null) {
    176210                if (_outboundStates.size() >= getMaxConcurrentEstablish()) {
    177                     if (_queuedOutbound.size() > MAX_QUEUED_OUTBOUND) {
     211                    if (_queuedOutbound.size() >= MAX_QUEUED_OUTBOUND) {
    178212                        rejected = true;
    179213                    } else {
     214                        if (_log.shouldLog(Log.WARN))
     215                            _log.warn("Queueing outbound establish, increase " + PROP_MAX_CONCURRENT_ESTABLISH);
    180216                        List<OutNetMessage> newQueued = new ArrayList(MAX_QUEUED_PER_PEER);
    181217                        List<OutNetMessage> queued = _queuedOutbound.putIfAbsent(to, newQueued);
     
    211247                        return;
    212248                    }
    213                     state = new OutboundEstablishState(_context, remAddr, port, 
     249                    state = new OutboundEstablishState(_context, remAddr, port,
    214250                                                       msg.getTarget().getIdentity(),
    215251                                                       sessionKey, addr, _transport.getDHBuilder());
     
    219255                        // whoops, somebody beat us to it, throw out the state we just created
    220256                        state = oldState;
    221                     else
    222                         _context.simpleScheduler().addEvent(new Expire(to, state), 10*1000);
    223257                }
    224258            }
     
    250284       
    251285        if (deferred > 0)
    252             msg.timestamp("too many deferred establishers: " + deferred);
     286            msg.timestamp("too many deferred establishers");
    253287        else if (state != null)
    254             msg.timestamp("establish state already waiting " + state.getLifetime());
     288            msg.timestamp("establish state already waiting");
    255289        notifyActivity();
    256     }
    257    
    258     private class Expire implements SimpleTimer.TimedEvent {
    259         private final RemoteHostId _to;
    260         private final OutboundEstablishState _state;
    261 
    262         public Expire(RemoteHostId to, OutboundEstablishState state) {
    263             _to = to;
    264             _state = state;
    265         }
    266 
    267         public void timeReached() {
    268             // remove only if value == state
    269             boolean removed = _outboundStates.remove(_to, _state);
    270             if (removed) {
    271                 _context.statManager().addRateData("udp.outboundEstablishFailedState", _state.getState(), _state.getLifetime());
    272                 if (_log.shouldLog(Log.WARN))
    273                     _log.warn("Timing out expired outbound: " + _state);
    274                 processExpired(_state);
    275             }
    276         }
    277290    }
    278291   
     
    289302     */
    290303    void receiveSessionRequest(RemoteHostId from, UDPPacketReader reader) {
    291         if (!_transport.isValid(from.getIP()))
     304        if (!_transport.isValid(from.getIP())) {
     305            if (_log.shouldLog(Log.WARN))
     306                _log.warn("Receive session request from invalid IP: " + from);
    292307            return;
     308        }
    293309       
    294310        int maxInbound = getMaxInboundEstablishers();
     
    296312        boolean isNew = false;
    297313
    298             if (_inboundStates.size() >= maxInbound) {
    299                 if (_log.shouldLog(Log.WARN))
    300                     _log.warn("Dropping inbound establish, increase " + PROP_MAX_CONCURRENT_ESTABLISH);
    301                 _context.statManager().addRateData("udp.establishDropped", 1);
    302                 return; // drop the packet
    303             }
    304            
    305314            InboundEstablishState state = _inboundStates.get(from);
    306315            if (state == null) {
     316                // TODO this is insufficient to prevent DoSing, especially if
     317                // IP spoofing is used. For further study.
     318                if (_inboundStates.size() >= maxInbound) {
     319                    if (_log.shouldLog(Log.WARN))
     320                        _log.warn("Dropping inbound establish, increase " + PROP_MAX_CONCURRENT_ESTABLISH);
     321                    _context.statManager().addRateData("udp.establishDropped", 1);
     322                    return; // drop the packet
     323                }
     324           
    307325                if (_context.blocklist().isBlocklisted(from.getIP())) {
    308326                    if (_log.shouldLog(Log.WARN))
     
    333351                state.setSentRelayTag(tag);
    334352                if (_log.shouldLog(Log.INFO))
    335                     _log.info("Received session request from " + from + ", sending relay tag " + tag);
     353                    _log.info("Received NEW session request from " + from + ", sending relay tag " + tag);
    336354            } else {
    337355                if (_log.shouldLog(Log.INFO))
    338356                    _log.info("Received session request, but our status is " + _transport.getReachabilityStatus());
    339357            }
    340         }
    341        
    342         if (_log.shouldLog(Log.DEBUG))
    343             _log.debug("Receive session request from: " + state.getRemoteHostId().toString());
     358        } else {
     359            if (_log.shouldLog(Log.DEBUG))
     360                _log.debug("Receive DUP session request from: " + state.getRemoteHostId());
     361        }
    344362       
    345363        notifyActivity();
     
    357375            if (_log.shouldLog(Log.DEBUG))
    358376                _log.debug("Receive session confirmed from: " + state.getRemoteHostId().toString());
     377        } else {
     378            if (_log.shouldLog(Log.WARN))
     379                _log.warn("Receive (DUP?) session confirmed from: " + from);
    359380        }
    360381    }
     
    371392            if (_log.shouldLog(Log.DEBUG))
    372393                _log.debug("Receive session created from: " + state.getRemoteHostId().toString());
     394        } else {
     395            if (_log.shouldLog(Log.WARN))
     396                _log.warn("Receive (DUP?) session created from: " + from);
    373397        }
    374398    }
     
    393417        _outboundStates.remove(from);
    394418        Hash peer = state.getRemoteIdentity().calculateHash();
    395         _transport.dropPeer(peer, false, "received destroy message");
    396     }
    397 
    398     /**
    399      * Got a SessionDestroy - maybe after an inbound establish?
     419        _transport.dropPeer(peer, false, "received destroy message during OB establish");
     420    }
     421
     422    /**
     423     * Got a SessionDestroy - maybe during an inbound establish?
     424     * TODO - PacketHandler won't look up inbound establishes
    400425     * As this packet was essentially unauthenticated (i.e. intro key, not session key)
    401426     * we just log it as it could be spoofed.
     
    404429    void receiveSessionDestroy(RemoteHostId from) {
    405430        if (_log.shouldLog(Log.WARN))
    406             _log.warn("Receive session destroy (IB) from: " + from);
     431            _log.warn("Receive session destroy (none) from: " + from);
    407432        //InboundEstablishState state = _inboundStates.remove(from);
    408433        //if (state != null) {
     
    437462                }
    438463               
    439                 //admitted = locked_admitQueued();
     464                locked_admitQueued();
    440465            //remaining = _queuedOutbound.size();
    441466
     
    450475    }
    451476
    452 /********
     477    /**
     478     *  Move pending OB messages from _queuedOutbound to _outboundStates.
     479     *  This isn't so great because _queuedOutbound is not a FIFO.
     480     */
    453481    private int locked_admitQueued() {
     482        if (_queuedOutbound.isEmpty())
     483            return 0;
    454484        int admitted = 0;
    455         while ( (!_queuedOutbound.isEmpty()) && (_outboundStates.size() < getMaxConcurrentEstablish()) ) {
    456             // ok, active shrunk, lets let some queued in.  duplicate the synchronized
    457             // section from the add(
    458 
    459             RemoteHostId to = (RemoteHostId)_queuedOutbound.keySet().iterator().next();
    460             List queued = (List)_queuedOutbound.remove(to);
    461 
     485        int max = getMaxConcurrentEstablish();
     486        for (Iterator<Map.Entry<RemoteHostId, List<OutNetMessage>>> iter = _queuedOutbound.entrySet().iterator();
     487             iter.hasNext() && _outboundStates.size() < max; ) {
     488            // ok, active shrunk, lets let some queued in.
     489
     490            Map.Entry<RemoteHostId, List<OutNetMessage>> entry = iter.next();
     491            iter.remove();
     492            RemoteHostId to = entry.getKey();
     493            List<OutNetMessage> allQueued = entry.getValue();
     494            List<OutNetMessage> queued = new ArrayList();
     495            long now = _context.clock().now();
     496            synchronized (allQueued) {
     497                for (OutNetMessage msg : allQueued) {
     498                    if (now - Router.CLOCK_FUDGE_FACTOR > msg.getExpiration()) {
     499                        _transport.failed(msg, "Took too long in est. mgr OB queue");
     500                    } else {
     501                        queued.add(msg);
     502                    }
     503
     504                }
     505            }
    462506            if (queued.isEmpty())
    463507                continue;
    464508           
    465             OutNetMessage msg = (OutNetMessage)queued.get(0);
     509            OutNetMessage msg = queued.get(0);
    466510            RouterAddress ra = msg.getTarget().getTargetAddress(_transport.getStyle());
    467511            if (ra == null) {
    468512                for (int i = 0; i < queued.size(); i++)
    469                     _transport.failed((OutNetMessage)queued.get(i), "Cannot admit to the queue, as it has no address");
     513                    _transport.failed(queued.get(i), "Cannot admit to the queue, as it has no address");
    470514                continue;
    471515            }
     
    476520            OutboundEstablishState qstate = new OutboundEstablishState(_context, remAddr, port,
    477521                                               msg.getTarget().getIdentity(),
    478                                                new SessionKey(addr.getIntroKey()), addr);
    479             _outboundStates.put(to, qstate);
    480             _context.simpleScheduler().addEvent(new Expire(to, qstate), 10*1000);
     522                                               new SessionKey(addr.getIntroKey()), addr,
     523                                               _transport.getDHBuilder());
     524            OutboundEstablishState old = _outboundStates.putIfAbsent(to, qstate);
     525            if (old != null)
     526                qstate = old;
    481527
    482528            for (int i = 0; i < queued.size(); i++) {
    483                 OutNetMessage m = (OutNetMessage)queued.get(i);
     529                OutNetMessage m = queued.get(i);
    484530                m.timestamp("no longer deferred... establishing");
    485531                qstate.addMessage(m);
     
    489535        return admitted;
    490536    }
    491 *******/
    492537   
    493538    private void notifyActivity() {
     
    497542        }
    498543    }
    499    
    500     /** kill any inbound or outbound that takes more than 30s */
    501     private static final int MAX_ESTABLISH_TIME = 30*1000;
    502544   
    503545    /**
     
    554596                                           // This causes huge values in the inNetPool.droppedDeliveryStatusDelay stat
    555597                                           // so it needs to be caught in InNetMessagePool.
    556         dsm.setMessageExpiration(_context.clock().now()+10*1000);
     598        dsm.setMessageExpiration(_context.clock().now() + DATA_MESSAGE_TIMEOUT);
    557599        dsm.setMessageId(_context.random().nextLong(I2NPMessage.MAX_ID_VALUE));
    558600        _transport.send(dsm, peer);
    559         _context.simpleScheduler().addEvent(new PublishToNewInbound(peer), 0);
    560     }
    561 
    562     private class PublishToNewInbound implements SimpleTimer.TimedEvent {
    563         private final PeerState _peer;
    564 
    565         public PublishToNewInbound(PeerState peer) { _peer = peer; }
    566 
    567         public void timeReached() {
    568             Hash peer = _peer.getRemotePeer();
    569             if ((peer != null) && (!_context.shitlist().isShitlisted(peer)) && (!_transport.isUnreachable(peer))) {
     601
     602        // just do this inline
     603        //_context.simpleScheduler().addEvent(new PublishToNewInbound(peer), 0);
     604
     605            Hash hash = peer.getRemotePeer();
     606            if ((hash != null) && (!_context.shitlist().isShitlisted(hash)) && (!_transport.isUnreachable(hash))) {
    570607                // ok, we are fine with them, send them our latest info
    571                 if (_log.shouldLog(Log.INFO))
    572                     _log.info("Publishing to the peer after confirm plus delay (without shitlist): " + peer);
    573                 sendOurInfo(_peer, true);
     608                //if (_log.shouldLog(Log.INFO))
     609                //    _log.info("Publishing to the peer after confirm plus delay (without shitlist): " + peer);
     610                sendOurInfo(peer, true);
    574611            } else {
    575612                // nuh uh.
    576613                if (_log.shouldLog(Log.WARN))
    577                     _log.warn("NOT publishing to the peer after confirm plus delay (WITH shitlist): " + (peer != null ? peer.toString() : "unknown"));
    578             }
    579         }
     614                    _log.warn("NOT publishing to the peer after confirm plus delay (WITH shitlist): " + (hash != null ? hash.toString() : "unknown"));
     615            }
    580616    }
    581617   
     
    613649        sendOurInfo(peer, false);
    614650       
    615         int i = 0;
    616         while (true) {
    617             OutNetMessage msg = state.getNextQueuedMessage();
    618             if (msg == null)
    619                 break;
     651        OutNetMessage msg;
     652        while ((msg = state.getNextQueuedMessage()) != null) {
    620653            if (now - Router.CLOCK_FUDGE_FACTOR > msg.getExpiration()) {
    621654                msg.timestamp("took too long but established...");
    622655                _transport.failed(msg, "Took too long to establish, but it was established");
    623656            } else {
    624                 msg.timestamp("session fully established and sent " + i);
     657                msg.timestamp("session fully established and sent");
    625658                _transport.send(msg);
    626659            }
    627             i++;
    628660        }
    629661        return peer;
     
    637669        DatabaseStoreMessage m = new DatabaseStoreMessage(_context);
    638670        m.setEntry(_context.router().getRouterInfo());
    639         m.setMessageExpiration(_context.clock().now() + 10*1000);
     671        m.setMessageExpiration(_context.clock().now() + DATA_MESSAGE_TIMEOUT);
    640672        _transport.send(m, peer);
    641673    }
     
    679711        }
    680712        _transport.send(_builder.buildSessionCreatedPacket(state, _transport.getExternalPort(), _transport.getIntroKey()));
    681         // if they haven't advanced to sending us confirmed packets in 1s,
    682         // repeat
    683         state.setNextSendTime(now + 1000);
    684     }
    685 
     713        state.createdPacketSent();
     714    }
     715
     716    /**
     717     *  Caller should probably synch on outboundState
     718     */
    686719    private void sendRequest(OutboundEstablishState state) {
    687720        if (_log.shouldLog(Log.DEBUG))
     
    697730    }
    698731   
    699     private static final long MAX_NONCE = 0xFFFFFFFFl;
    700     /** if we don't get a relayResponse in 3 seconds, try again */
    701     private static final int INTRO_ATTEMPT_TIMEOUT = 3*1000;
    702    
     732    /**
     733     *  Send RelayRequests to multiple introducers.
     734     *  This may be called multiple times, it sets the nonce the first time only
     735     *  Caller should probably synch on state.
     736     */
    703737    private void handlePendingIntro(OutboundEstablishState state) {
    704         long nonce = _context.random().nextLong(MAX_NONCE);
    705         while (true) {
    706                 OutboundEstablishState old = _liveIntroductions.putIfAbsent(Long.valueOf(nonce), state);
    707                 if (old != null) {
    708                     nonce = _context.random().nextLong(MAX_NONCE);
    709                 } else {
    710                     break;
    711                 }
    712         }
    713         _context.simpleScheduler().addEvent(new FailIntroduction(state, nonce), INTRO_ATTEMPT_TIMEOUT);
    714         state.setIntroNonce(nonce);
     738        long nonce = state.getIntroNonce();
     739        if (nonce < 0) {
     740            OutboundEstablishState old;
     741            do {
     742                nonce = _context.random().nextLong(MAX_NONCE);
     743                old = _liveIntroductions.putIfAbsent(Long.valueOf(nonce), state);
     744            } while (old != null);
     745            state.setIntroNonce(nonce);
     746        }
    715747        _context.statManager().addRateData("udp.sendIntroRelayRequest", 1, 0);
    716748        UDPPacket requests[] = _builder.buildRelayRequest(_transport, state, _transport.getIntroKey());
     
    720752        }
    721753        if (_log.shouldLog(Log.DEBUG))
    722             _log.debug("Send intro for " + state.getRemoteHostId().toString() + " with our intro key as " + _transport.getIntroKey());
     754            _log.debug("Send intro for " + state.getRemoteHostId() + " with our intro key as " + _transport.getIntroKey());
    723755        state.introSent();
    724756    }
    725757
    726     private class FailIntroduction implements SimpleTimer.TimedEvent {
    727         private final long _nonce;
    728         private final OutboundEstablishState _state;
    729 
    730         public FailIntroduction(OutboundEstablishState state, long nonce) {
    731             _nonce = nonce;
    732             _state = state;
    733         }
    734 
    735         public void timeReached() {
    736             // remove only if value equal to state
    737             boolean removed = _liveIntroductions.remove(Long.valueOf(_nonce), _state);
    738             if (removed) {
    739                 if (_log.shouldLog(Log.DEBUG))
    740                     _log.debug("Send intro for " + _state.getRemoteHostId().toString() + " timed out");
    741                 _context.statManager().addRateData("udp.sendIntroRelayTimeout", 1, 0);
    742                 notifyActivity();
    743             }
    744         }
    745     }
    746    
    747758    void receiveRelayResponse(RemoteHostId bob, UDPPacketReader reader) {
    748759        long nonce = reader.getRelayResponseReader().readNonce();
     
    777788    }
    778789   
     790    /**
     791     *  Note that while a SessionConfirmed could in theory be fragmented,
     792     *  in practice a RouterIdentity is 387 bytes and a single fragment is 512 bytes max,
     793     *  so it will never be fragmented.
     794     *  Caller should probably synch on state.
     795     */
    779796    private void sendConfirmation(OutboundEstablishState state) {
    780797        boolean valid = state.validateSessionCreated();
     
    797814       
    798815        if (_log.shouldLog(Log.DEBUG))
    799             _log.debug("Send confirm to: " + state.getRemoteHostId().toString());
     816            _log.debug("Send confirm to: " + state);
    800817       
    801818        for (int i = 0; i < packets.length; i++)
     
    805822    }
    806823   
     824    /**
     825     *  Tell the other side never mind.
     826     *  This is only useful after we have received SessionCreated,
     827     *  and sent SessionConfirmed, but not yet gotten a data packet as an
     828     *  ack to the SessionConfirmed - otherwise we haven't generated the keys.
     829     *  Caller should probably synch on state.
     830     *
     831     *  @since 0.9.2
     832     */
     833    private void sendDestroy(OutboundEstablishState state) {
     834        UDPPacket packet = _builder.buildSessionDestroyPacket(state);
     835        if (packet != null) {
     836            if (_log.shouldLog(Log.DEBUG))
     837                _log.debug("Send destroy to: " + state);
     838            _transport.send(packet);
     839        }
     840    }
     841   
     842    /**
     843     *  Tell the other side never mind.
     844     *  This is only useful after we have sent SessionCreated,
     845     *  but not received SessionConfirmed
     846     *  Otherwise we haven't generated the keys.
     847     *  Caller should probably synch on state.
     848     *
     849     *  @since 0.9.2
     850     */
     851    private void sendDestroy(InboundEstablishState state) {
     852        UDPPacket packet = _builder.buildSessionDestroyPacket(state);
     853        if (packet != null) {
     854            if (_log.shouldLog(Log.DEBUG))
     855                _log.debug("Send destroy to: " + state);
     856            _transport.send(packet);
     857        }
     858    }
    807859   
    808860    /**
     
    815867        long nextSendTime = -1;
    816868        InboundEstablishState inboundState = null;
    817 
    818             //int active = _inboundStates.size();
    819             //if (active > 0 && _log.shouldLog(Log.DEBUG))
    820             //    _log.debug("# inbound states: " + active);
     869        boolean expired = false;
     870
    821871            for (Iterator<InboundEstablishState> iter = _inboundStates.values().iterator(); iter.hasNext(); ) {
    822872                InboundEstablishState cur = iter.next();
    823                 if (cur.getState() == InboundEstablishState.STATE_CONFIRMED_COMPLETELY) {
     873                if (cur.getState() == IB_STATE_CONFIRMED_COMPLETELY) {
    824874                    // completely received (though the signature may be invalid)
    825875                    iter.remove();
     
    828878                        _log.debug("Removing completely confirmed inbound state");
    829879                    break;
    830                 } else if (cur.getLifetime() > MAX_ESTABLISH_TIME) {
     880                } else if (cur.getLifetime() > MAX_IB_ESTABLISH_TIME) {
    831881                    // took too long
    832882                    iter.remove();
    833                     _context.statManager().addRateData("udp.inboundEstablishFailedState", cur.getState(), cur.getLifetime());
     883                    inboundState = cur;
     884                    //_context.statManager().addRateData("udp.inboundEstablishFailedState", cur.getState(), cur.getLifetime());
    834885                    if (_log.shouldLog(Log.DEBUG))
    835886                        _log.debug("Removing expired inbound state");
    836                 } else if (cur.getState() == InboundEstablishState.STATE_FAILED) {
     887                    expired = true;
     888                    break;
     889                } else if (cur.getState() == IB_STATE_FAILED) {
    837890                    iter.remove();
    838                     _context.statManager().addRateData("udp.inboundEstablishFailedState", cur.getState(), cur.getLifetime());
     891                    //_context.statManager().addRateData("udp.inboundEstablishFailedState", cur.getState(), cur.getLifetime());
    839892                } else {
    840893                    if (cur.getNextSendTime() <= now) {
     
    850903                        long when = -1;
    851904                        if (cur.getNextSendTime() <= 0) {
    852                             when = cur.getEstablishBeginTime() + MAX_ESTABLISH_TIME;
     905                            when = cur.getEstablishBeginTime() + MAX_IB_ESTABLISH_TIME;
    853906                        } else {
    854907                            when = cur.getNextSendTime();
     
    863916            //if (_log.shouldLog(Log.DEBUG))
    864917            //    _log.debug("Processing for inbound: " + inboundState);
    865             switch (inboundState.getState()) {
    866                 case InboundEstablishState.STATE_REQUEST_RECEIVED:
    867                     sendCreated(inboundState);
    868                     break;
    869                 case InboundEstablishState.STATE_CREATED_SENT: // fallthrough
    870                 case InboundEstablishState.STATE_CONFIRMED_PARTIALLY:
    871                     // if its been 5s since we sent the SessionCreated, resend
    872                     if (inboundState.getNextSendTime() <= now)
     918            synchronized (inboundState) {
     919                switch (inboundState.getState()) {
     920                  case IB_STATE_REQUEST_RECEIVED:
     921                    if (!expired)
    873922                        sendCreated(inboundState);
    874923                    break;
    875                 case InboundEstablishState.STATE_CONFIRMED_COMPLETELY:
     924
     925                  case IB_STATE_CREATED_SENT: // fallthrough
     926                  case IB_STATE_CONFIRMED_PARTIALLY:
     927                    if (expired) {
     928                        sendDestroy(inboundState);
     929                    } else if (inboundState.getNextSendTime() <= now) {
     930                        sendCreated(inboundState);
     931                    }
     932                    break;
     933
     934                  case IB_STATE_CONFIRMED_COMPLETELY:
    876935                    RouterIdentity remote = inboundState.getConfirmedIdentity();
    877936                    if (remote != null) {
     
    882941                            _context.blocklist().add(inboundState.getSentIP());
    883942                            inboundState.fail();
    884                             break;
     943                        } else {
     944                            handleCompletelyEstablished(inboundState);
    885945                        }
    886                         handleCompletelyEstablished(inboundState);
    887                         break;
    888946                    } else {
    889947                        if (_log.shouldLog(Log.WARN))
    890948                            _log.warn("confirmed with invalid? " + inboundState);
    891949                        inboundState.fail();
    892                         break;
    893950                    }
    894                 case InboundEstablishState.STATE_FAILED:
     951                    break;
     952
     953                  case IB_STATE_FAILED:
    895954                    break; // already removed;
    896                 case InboundEstablishState.STATE_UNKNOWN: // fallthrough
    897                 default:
    898                     // wtf
     955
     956                  case IB_STATE_UNKNOWN:
     957                    // Can't happen, always call receiveSessionRequest() before putting in map
    899958                    if (_log.shouldLog(Log.ERROR))
    900959                        _log.error("hrm, state is unknown for " + inboundState);
     960                }
    901961            }
    902962
     
    922982        //int active = 0;
    923983
    924             //int active = _outboundStates.size();
    925             //if (active > 0 && _log.shouldLog(Log.DEBUG))
    926             //    _log.debug("# outbound states: " + active);
    927984            for (Iterator<OutboundEstablishState> iter = _outboundStates.values().iterator(); iter.hasNext(); ) {
    928985                OutboundEstablishState cur = iter.next();
    929                 if (cur.getState() == OutboundEstablishState.STATE_CONFIRMED_COMPLETELY) {
     986                if (cur.getState() == OB_STATE_CONFIRMED_COMPLETELY) {
    930987                    // completely received
    931988                    iter.remove();
     
    934991                        _log.debug("Removing confirmed outbound: " + cur);
    935992                    break;
    936                 } else if (cur.getLifetime() > MAX_ESTABLISH_TIME) {
     993                } else if (cur.getLifetime() > MAX_OB_ESTABLISH_TIME) {
    937994                    // took too long
    938995                    iter.remove();
    939996                    outboundState = cur;
    940                     _context.statManager().addRateData("udp.outboundEstablishFailedState", cur.getState(), cur.getLifetime());
     997                    //_context.statManager().addRateData("udp.outboundEstablishFailedState", cur.getState(), cur.getLifetime());
    941998                    if (_log.shouldLog(Log.DEBUG))
    942999                        _log.debug("Removing expired outbound: " + cur);
     
    9551012                        long when = -1;
    9561013                        if (cur.getNextSendTime() <= 0) {
    957                             when = cur.getEstablishBeginTime() + MAX_ESTABLISH_TIME;
     1014                            when = cur.getEstablishBeginTime() + MAX_OB_ESTABLISH_TIME;
    9581015                        } else {
    9591016                            when = cur.getNextSendTime();
     
    9761033            //if (_log.shouldLog(Log.DEBUG))
    9771034            //    _log.debug("Processing for outbound: " + outboundState);
    978             if (outboundState.getLifetime() > MAX_ESTABLISH_TIME) {
    979                 processExpired(outboundState);
    980             } else {
     1035            synchronized (outboundState) {
     1036                boolean expired = outboundState.getLifetime() > MAX_OB_ESTABLISH_TIME;
    9811037                switch (outboundState.getState()) {
    982                     case OutboundEstablishState.STATE_UNKNOWN:
    983                         sendRequest(outboundState);
    984                         break;
    985                     case OutboundEstablishState.STATE_REQUEST_SENT:
    986                         // no response yet (or it was invalid), lets retry
    987                         if (outboundState.getNextSendTime() <= now)
     1038                    case OB_STATE_UNKNOWN:
     1039                        if (expired)
     1040                            processExpired(outboundState);
     1041                        else
    9881042                            sendRequest(outboundState);
    9891043                        break;
    990                     case OutboundEstablishState.STATE_CREATED_RECEIVED: // fallthrough
    991                     case OutboundEstablishState.STATE_CONFIRMED_PARTIALLY:
    992                         if (outboundState.getNextSendTime() <= now)
     1044
     1045                    case OB_STATE_REQUEST_SENT:
     1046                        // no response yet (or it was invalid), lets retry
     1047                        long rtime = outboundState.getRequestSentTime();
     1048                        if (expired || (rtime > 0 && rtime + OB_MESSAGE_TIMEOUT < now))
     1049                            processExpired(outboundState);
     1050                        else if (outboundState.getNextSendTime() <= now)
     1051                            sendRequest(outboundState);
     1052                        break;
     1053
     1054                    case OB_STATE_CREATED_RECEIVED:
     1055                        if (expired)
     1056                            processExpired(outboundState);
     1057                        else if (outboundState.getNextSendTime() <= now)
    9931058                            sendConfirmation(outboundState);
    9941059                        break;
    995                     case OutboundEstablishState.STATE_CONFIRMED_COMPLETELY:
    996                         handleCompletelyEstablished(outboundState);
     1060
     1061                    case OB_STATE_CONFIRMED_PARTIALLY:
     1062                        long ctime = outboundState.getConfirmedSentTime();
     1063                        if (expired || (ctime > 0 && ctime + OB_MESSAGE_TIMEOUT < now)) {
     1064                            sendDestroy(outboundState);
     1065                            processExpired(outboundState);
     1066                        } else if (outboundState.getNextSendTime() <= now) {
     1067                            sendConfirmation(outboundState);
     1068                        }
    9971069                        break;
    998                     case OutboundEstablishState.STATE_PENDING_INTRO:
    999                         handlePendingIntro(outboundState);
     1070
     1071                    case OB_STATE_CONFIRMED_COMPLETELY:
     1072                        if (expired)
     1073                            processExpired(outboundState);
     1074                        else
     1075                            handleCompletelyEstablished(outboundState);
    10001076                        break;
    1001                     default:
    1002                         // wtf
     1077
     1078                    case OB_STATE_PENDING_INTRO:
     1079                        long itime = outboundState.getIntroSentTime();
     1080                        if (expired || (itime > 0 && itime + OB_MESSAGE_TIMEOUT < now))
     1081                            processExpired(outboundState);
     1082                        else if (outboundState.getNextSendTime() <= now)
     1083                            handlePendingIntro(outboundState);
     1084                        break;
    10031085                }
    10041086            }
    10051087           
    1006             //if (_log.shouldLog(Log.DEBUG))
    1007             //    _log.debug("Since something happened outbound, next=now");
    10081088            // ok, since there was something to do, we want to loop again
    10091089            nextSendTime = now;
    1010         } else {
    1011             //if (_log.shouldLog(Log.DEBUG))
    1012             //    _log.debug("Nothing happened outbound, next is in " + (nextSendTime-now));
    10131090        }
    10141091       
     
    10161093    }
    10171094   
     1095    /**
     1096     *  Caller should probably synch on outboundState
     1097     */
    10181098    private void processExpired(OutboundEstablishState outboundState) {
    1019         if (outboundState.getState() != OutboundEstablishState.STATE_CONFIRMED_COMPLETELY) {
     1099        long nonce = outboundState.getIntroNonce();
     1100        if (nonce >= 0) {
     1101            // remove only if value == state
     1102            boolean removed = _liveIntroductions.remove(Long.valueOf(nonce), outboundState);
     1103            if (removed) {
     1104                if (_log.shouldLog(Log.DEBUG))
     1105                    _log.debug("Send intro for " + outboundState.getRemoteHostId() + " timed out");
     1106                _context.statManager().addRateData("udp.sendIntroRelayTimeout", 1, 0);
     1107            }
     1108        }
     1109        // should have already been removed in handleOutbound() above
     1110        // remove only if value == state
     1111        boolean removed = _outboundStates.remove(outboundState.getRemoteHostId(), outboundState);
     1112        if (outboundState.getState() != OB_STATE_CONFIRMED_COMPLETELY) {
    10201113            if (_log.shouldLog(Log.INFO))
    10211114                _log.info("Lifetime of expired outbound establish: " + outboundState.getLifetime());
     
    10261119                _transport.failed(msg, "Expired during failed establish");
    10271120            }
    1028             String err = null;
    1029             switch (outboundState.getState()) {
    1030                 case OutboundEstablishState.STATE_CONFIRMED_PARTIALLY:
    1031                     err = "Took too long to establish remote connection (confirmed partially)";
    1032                     break;
    1033                 case OutboundEstablishState.STATE_CREATED_RECEIVED:
    1034                     err = "Took too long to establish remote connection (created received)";
    1035                     break;
    1036                 case OutboundEstablishState.STATE_REQUEST_SENT:
    1037                     err = "Took too long to establish remote connection (request sent)";
    1038                     break;
    1039                 case OutboundEstablishState.STATE_PENDING_INTRO:
    1040                     err = "Took too long to establish remote connection (intro failed)";
    1041                     break;
    1042                 case OutboundEstablishState.STATE_UNKNOWN: // fallthrough
    1043                 default:
    1044                     err = "Took too long to establish remote connection (unknown state)";
    1045             }
    1046 
     1121            String err = "Took too long to establish OB connection, state = " + outboundState.getState();
    10471122            Hash peer = outboundState.getRemoteIdentity().calculateHash();
    10481123            //_context.shitlist().shitlistRouter(peer, err, UDPTransport.STYLE);
     
    10751150                }
    10761151            }
    1077         }
    1078     }
     1152            _inboundStates.clear();
     1153            _outboundStates.clear();
     1154            _queuedOutbound.clear();
     1155            _liveIntroductions.clear();
     1156        }
     1157    }
     1158
     1159    // Debugging
     1160    private long _lastPrinted;
     1161    private static final long PRINT_INTERVAL = 5*1000;
    10791162   
    10801163    private void doPass() {
     1164        if (_log.shouldLog(Log.DEBUG) && _lastPrinted + PRINT_INTERVAL < _context.clock().now()) {
     1165            _lastPrinted = _context.clock().now();
     1166            int iactive = _inboundStates.size();
     1167            int oactive = _outboundStates.size();
     1168            if (iactive > 0 || oactive > 0) {
     1169                int queued = _queuedOutbound.size();
     1170                int live = _liveIntroductions.size();
     1171                _log.debug("OB states: " + oactive + " IB states: " + iactive +
     1172                           " OB queued: " + queued + " intros: " + live);
     1173            }
     1174        }
    10811175        _activity = 0;
    10821176        long now = _context.clock().now();
  • router/java/src/net/i2p/router/transport/udp/InboundEstablishState.java

    re9cc851 r16a46b3  
    4040    private SessionKey _macKey;
    4141    private Signature _sentSignature;
    42     // SessionConfirmed messages
     42    // SessionConfirmed messages - fragmented in theory but not in practice - see below
    4343    private byte _receivedIdentity[][];
    4444    private long _receivedSignedOnTime;
     
    4949    private final long _establishBegin;
    5050    //private long _lastReceive;
    51     // private long _lastSend;
     51    private long _lastSend;
    5252    private long _nextSend;
    5353    private final RemoteHostId _remoteHostId;
    54     private int _currentState;
     54    private InboundState _currentState;
    5555    private boolean _complete;
    56    
    57     /** nothin known yet */
    58     public static final int STATE_UNKNOWN = 0;
    59     /** we have received an initial request */
    60     public static final int STATE_REQUEST_RECEIVED = 1;
    61     /** we have sent a signed creation packet */
    62     public static final int STATE_CREATED_SENT = 2;
    63     /** we have received one or more confirmation packets */
    64     public static final int STATE_CONFIRMED_PARTIALLY = 3;
    65     /** we have completely received all of the confirmation packets */
    66     public static final int STATE_CONFIRMED_COMPLETELY = 4;
    67     /** we are explicitly failing it */
    68     public static final int STATE_FAILED = 5;
    69    
     56    // count for backoff
     57    private int _createdSentCount;
     58   
     59    public enum InboundState {
     60        /** nothin known yet */
     61        IB_STATE_UNKNOWN,
     62        /** we have received an initial request */
     63        IB_STATE_REQUEST_RECEIVED,
     64        /** we have sent a signed creation packet */
     65        IB_STATE_CREATED_SENT,
     66        /** we have received one but not all the confirmation packets
     67          * This never happens in practice - see below. */
     68        IB_STATE_CONFIRMED_PARTIALLY,
     69        /** we have all the confirmation packets */
     70        IB_STATE_CONFIRMED_COMPLETELY,
     71        /** we are explicitly failing it */
     72        IB_STATE_FAILED
     73    }
     74   
     75    /** basic delay before backoff */
     76    private static final long RETRANSMIT_DELAY = 1500;
     77
     78    /** max delay including backoff */
     79    private static final long MAX_DELAY = 15*1000;
     80
    7081    public InboundEstablishState(RouterContext ctx, byte remoteIP[], int remotePort, int localPort,
    7182                                 DHSessionKeyBuilder dh) {
     
    7687        _remoteHostId = new RemoteHostId(_aliceIP, _alicePort);
    7788        _bobPort = localPort;
    78         _currentState = STATE_UNKNOWN;
     89        _currentState = InboundState.IB_STATE_UNKNOWN;
    7990        _establishBegin = ctx.clock().now();
    8091        _keyBuilder = dh;
    8192    }
    8293   
    83     public synchronized int getState() { return _currentState; }
     94    public synchronized InboundState getState() { return _currentState; }
     95
     96    /** @return if previously complete */
    8497    public synchronized boolean complete() {
    8598        boolean already = _complete;
     
    97110        if (_log.shouldLog(Log.DEBUG))
    98111            _log.debug("Receive sessionRequest, BobIP = " + Addresses.toString(_bobIP));
    99         if (_currentState == STATE_UNKNOWN)
    100             _currentState = STATE_REQUEST_RECEIVED;
     112        if (_currentState == InboundState.IB_STATE_UNKNOWN)
     113            _currentState = InboundState.IB_STATE_REQUEST_RECEIVED;
    101114        packetReceived();
    102115    }
     
    106119    public synchronized byte[] getReceivedOurIP() { return _bobIP; }
    107120   
     121    /**
     122     *  Generates session key and mac key.
     123     */
    108124    public synchronized void generateSessionKey() throws DHSessionKeyBuilder.InvalidPublicParameterException {
    109125        if (_sessionKey != null) return;
     
    136152   
    137153    public synchronized void fail() {
    138         _currentState = STATE_FAILED;
     154        _currentState = InboundState.IB_STATE_FAILED;
    139155    }
    140156   
     
    198214    /** note that we just sent a SessionCreated packet */
    199215    public synchronized void createdPacketSent() {
    200         // _lastSend = _context.clock().now();
    201         if ( (_currentState == STATE_UNKNOWN) || (_currentState == STATE_REQUEST_RECEIVED) )
    202             _currentState = STATE_CREATED_SENT;
    203     }
    204    
     216        _lastSend = _context.clock().now();
     217        long delay;
     218        if (_createdSentCount == 0) {
     219            delay = RETRANSMIT_DELAY;
     220        } else {
     221            delay = Math.min(RETRANSMIT_DELAY << _createdSentCount, MAX_DELAY);
     222        }
     223        _createdSentCount++;
     224        _nextSend = _lastSend + delay;
     225        if ( (_currentState == InboundState.IB_STATE_UNKNOWN) || (_currentState == InboundState.IB_STATE_REQUEST_RECEIVED) )
     226            _currentState = InboundState.IB_STATE_CREATED_SENT;
     227    }
     228
    205229    /** how long have we been trying to establish this session? */
    206230    public long getLifetime() { return _context.clock().now() - _establishBegin; }
    207231    public long getEstablishBeginTime() { return _establishBegin; }
    208232    public synchronized long getNextSendTime() { return _nextSend; }
    209     public synchronized void setNextSendTime(long when) { _nextSend = when; }
    210233
    211234    /** RemoteHostId, uniquely identifies an attempt */
    212235    RemoteHostId getRemoteHostId() { return _remoteHostId; }
    213236
     237    /**
     238     *  Note that while a SessionConfirmed could in theory be fragmented,
     239     *  in practice a RouterIdentity is 387 bytes and a single fragment is 512 bytes max,
     240     *  so it will never be fragmented.
     241     */
    214242    public synchronized void receiveSessionConfirmed(UDPPacketReader.SessionConfirmedReader conf) {
    215243        if (_receivedIdentity == null)
     
    236264        }
    237265       
    238         if ( (_currentState == STATE_UNKNOWN) ||
    239              (_currentState == STATE_REQUEST_RECEIVED) ||
    240              (_currentState == STATE_CREATED_SENT) ) {
     266        if ( (_currentState == InboundState.IB_STATE_UNKNOWN) ||
     267             (_currentState == InboundState.IB_STATE_REQUEST_RECEIVED) ||
     268             (_currentState == InboundState.IB_STATE_CREATED_SENT) ) {
    241269            if (confirmedFullyReceived())
    242                 _currentState = STATE_CONFIRMED_COMPLETELY;
     270                _currentState = InboundState.IB_STATE_CONFIRMED_COMPLETELY;
    243271            else
    244                 _currentState = STATE_CONFIRMED_PARTIALLY;
     272                _currentState = InboundState.IB_STATE_CONFIRMED_PARTIALLY;
    245273        }
    246274       
     
    248276    }
    249277   
    250     /** have we fully received the SessionConfirmed messages from Alice? */
    251     public synchronized boolean confirmedFullyReceived() {
     278    /**
     279     *  Have we fully received the SessionConfirmed messages from Alice?
     280     *  Caller must synch on this.
     281     */
     282    private boolean confirmedFullyReceived() {
    252283        if (_receivedIdentity != null) {
    253284            for (int i = 0; i < _receivedIdentity.length; i++)
  • router/java/src/net/i2p/router/transport/udp/OutboundEstablishState.java

    re9cc851 r16a46b3  
    5454    private final SessionKey _introKey;
    5555    private final Queue<OutNetMessage> _queuedMessages;
    56     private int _currentState;
     56    private OutboundState _currentState;
    5757    private long _introductionNonce;
    5858    // intro
     
    6363    private int _requestSentCount;
    6464    private int _introSentCount;
    65    
    66     /** nothin sent yet */
    67     public static final int STATE_UNKNOWN = 0;
    68     /** we have sent an initial request */
    69     public static final int STATE_REQUEST_SENT = 1;
    70     /** we have received a signed creation packet */
    71     public static final int STATE_CREATED_RECEIVED = 2;
    72     /** we have sent one or more confirmation packets */
    73     public static final int STATE_CONFIRMED_PARTIALLY = 3;
    74     /** we have received a data packet */
    75     public static final int STATE_CONFIRMED_COMPLETELY = 4;
    76     /** we need to have someone introduce us to the peer, but haven't received a RelayResponse yet */
    77     public static final int STATE_PENDING_INTRO = 5;
     65    // Times for timeout
     66    private long _confirmedSentTime;
     67    private long _requestSentTime;
     68    private long _introSentTime;
     69   
     70    public enum OutboundState {
     71        /** nothin sent yet */
     72        OB_STATE_UNKNOWN,
     73        /** we have sent an initial request */
     74        OB_STATE_REQUEST_SENT,
     75        /** we have received a signed creation packet */
     76        OB_STATE_CREATED_RECEIVED,
     77        /** we have sent one or more confirmation packets */
     78        OB_STATE_CONFIRMED_PARTIALLY,
     79        /** we have received a data packet */
     80        OB_STATE_CONFIRMED_COMPLETELY,
     81        /** we need to have someone introduce us to the peer, but haven't received a RelayResponse yet */
     82        OB_STATE_PENDING_INTRO
     83    }
    7884   
    7985    /** basic delay before backoff */
     
    100106        _introKey = introKey;
    101107        _queuedMessages = new LinkedBlockingQueue();
    102         _currentState = STATE_UNKNOWN;
     108        _currentState = OutboundState.OB_STATE_UNKNOWN;
    103109        _establishBegin = ctx.clock().now();
    104110        _remoteAddress = addr;
     
    110116            if (_log.shouldLog(Log.DEBUG))
    111117                _log.debug("new outbound establish to " + remotePeer.calculateHash() + ", with address: " + addr);
    112             _currentState = STATE_PENDING_INTRO;
    113         }
    114     }
    115    
    116     public synchronized int getState() { return _currentState; }
     118            _currentState = OutboundState.OB_STATE_PENDING_INTRO;
     119        }
     120    }
     121   
     122    public synchronized OutboundState getState() { return _currentState; }
     123
     124    /** @return if previously complete */
    117125    public synchronized boolean complete() {
    118126        boolean already = _complete;
     
    123131    public UDPAddress getRemoteAddress() { return _remoteAddress; }
    124132    public void setIntroNonce(long nonce) { _introductionNonce = nonce; }
     133
     134    /** @return -1 if unset */
    125135    public long getIntroNonce() { return _introductionNonce; }
    126136   
     
    133143    }
    134144
     145    /** @return null if none */
    135146    public OutNetMessage getNextQueuedMessage() {
    136147        return _queuedMessages.poll();
     
    152163
    153164    public byte[] getSentX() { return _sentX; }
     165    /** the remote side (Bob) */
    154166    public synchronized byte[] getSentIP() { return _bobIP; }
     167    /** the remote side (Bob) */
    155168    public synchronized int getSentPort() { return _bobPort; }
    156169
     
    182195                       + "\nthis: " + this.toString());
    183196       
    184         if ( (_currentState == STATE_UNKNOWN) || (_currentState == STATE_REQUEST_SENT) )
    185             _currentState = STATE_CREATED_RECEIVED;
     197        if ( (_currentState == OutboundState.OB_STATE_UNKNOWN) || (_currentState == OutboundState.OB_STATE_REQUEST_SENT) )
     198            _currentState = OutboundState.OB_STATE_CREATED_RECEIVED;
    186199        packetReceived();
    187200    }
     
    192205     * remnants are dropped (perhaps they were spoofed, etc) so that we can
    193206     * receive another one
     207     *
     208     *  Generates session key and mac key.
    194209     */
    195210    public synchronized boolean validateSessionCreated() {
     
    232247        _receivedSignature = null;
    233248
    234         if ( (_currentState == STATE_UNKNOWN) ||
    235              (_currentState == STATE_REQUEST_SENT) ||
    236              (_currentState == STATE_CREATED_RECEIVED) )
    237             _currentState = STATE_REQUEST_SENT;
     249        if ( (_currentState == OutboundState.OB_STATE_UNKNOWN) ||
     250             (_currentState == OutboundState.OB_STATE_REQUEST_SENT) ||
     251             (_currentState == OutboundState.OB_STATE_CREATED_RECEIVED) )
     252            _currentState = OutboundState.OB_STATE_REQUEST_SENT;
    238253
    239254        _nextSend = _context.clock().now();
    240255    }
    241256   
     257    /**
     258     *  Generates session key and mac key.
     259     *  Caller must synch on this.
     260     */
    242261    private void generateSessionKey() throws DHSessionKeyBuilder.InvalidPublicParameterException {
    243262        if (_sessionKey != null) return;
     
    248267        System.arraycopy(extra.getData(), 0, _macKey.getData(), 0, SessionKey.KEYSIZE_BYTES);
    249268        if (_log.shouldLog(Log.DEBUG))
    250             _log.debug("Established outbound keys.  cipher: " + Base64.encode(_sessionKey.getData())
    251                        + " mac: " + Base64.encode(_macKey.getData()));
     269            _log.debug("Established outbound keys.  cipher: " + _sessionKey
     270                       + " mac: " + _macKey);
    252271    }
    253272   
     
    256275     * additional layer of encryption using the negotiated key along side
    257276     * the packet's IV
     277     *  Caller must synch on this.
    258278     */
    259279    private void decryptSignature() {
     
    273293     * Verify: Alice's IP + Alice's port + Bob's IP + Bob's port + Alice's
    274294     *         new relay tag + Bob's signed on time
     295     *  Caller must synch on this.
    275296     */
    276297    private boolean verifySessionCreated() {
     
    325346   
    326347    /**
    327      * Lets sign everything so we can fragment properly
     348     *  Let's sign everything so we can fragment properly.
    328349     *
     350     *  Note that while a SessionConfirmed could in theory be fragmented,
     351     *  in practice a RouterIdentity is 387 bytes and a single fragment is 512 bytes max,
     352     *  so it will never be fragmented.
    329353     */
    330354    public synchronized void prepareSessionConfirmed() {
     
    366390    public synchronized void confirmedPacketsSent() {
    367391        _lastSend = _context.clock().now();
    368         long delay = Math.min(RETRANSMIT_DELAY << (_confirmedSentCount++), MAX_DELAY);
     392        long delay;
     393        if (_confirmedSentCount == 0) {
     394            delay = RETRANSMIT_DELAY;
     395            _confirmedSentTime = _lastSend;
     396        } else {
     397            delay = Math.min(RETRANSMIT_DELAY << _confirmedSentCount, MAX_DELAY);
     398        }
     399        _confirmedSentCount++;
    369400        _nextSend = _lastSend + delay;
    370401        if (_log.shouldLog(Log.DEBUG))
    371402            _log.debug("Send confirm packets, nextSend in " + delay);
    372         if ( (_currentState == STATE_UNKNOWN) ||
    373              (_currentState == STATE_REQUEST_SENT) ||
    374              (_currentState == STATE_CREATED_RECEIVED) )
    375             _currentState = STATE_CONFIRMED_PARTIALLY;
    376     }
     403        if ( (_currentState == OutboundState.OB_STATE_UNKNOWN) ||
     404             (_currentState == OutboundState.OB_STATE_REQUEST_SENT) ||
     405             (_currentState == OutboundState.OB_STATE_CREATED_RECEIVED) )
     406            _currentState = OutboundState.OB_STATE_CONFIRMED_PARTIALLY;
     407    }
     408
     409    /**
     410     *  @return when we sent the first SessionConfirmed packet, or 0
     411     *  @since 0.9.2
     412     */
     413    public long getConfirmedSentTime() { return _confirmedSentTime; }
    377414
    378415    /** note that we just sent the SessionRequest packet */
    379416    public synchronized void requestSent() {
    380417        _lastSend = _context.clock().now();
    381         long delay = Math.min(RETRANSMIT_DELAY << (_requestSentCount++), MAX_DELAY);
     418        long delay;
     419        if (_requestSentCount == 0) {
     420            delay = RETRANSMIT_DELAY;
     421            _requestSentTime = _lastSend;
     422        } else {
     423            delay = Math.min(RETRANSMIT_DELAY << _requestSentCount, MAX_DELAY);
     424        }
     425        _requestSentCount++;
    382426        _nextSend = _lastSend + delay;
    383427        if (_log.shouldLog(Log.DEBUG))
    384428            _log.debug("Send a request packet, nextSend in " + delay);
    385         if (_currentState == STATE_UNKNOWN)
    386             _currentState = STATE_REQUEST_SENT;
    387     }
     429        if (_currentState == OutboundState.OB_STATE_UNKNOWN)
     430            _currentState = OutboundState.OB_STATE_REQUEST_SENT;
     431    }
     432
     433
     434    /**
     435     *  @return when we sent the first SessionRequest packet, or 0
     436     *  @since 0.9.2
     437     */
     438    public long getRequestSentTime() { return _requestSentTime; }
    388439
    389440    /** note that we just sent the RelayRequest packet */
    390441    public synchronized void introSent() {
    391442        _lastSend = _context.clock().now();
    392         long delay = Math.min(RETRANSMIT_DELAY << (_introSentCount++), MAX_DELAY);
     443        long delay;
     444        if (_introSentCount == 0) {
     445            delay = RETRANSMIT_DELAY;
     446            _introSentTime = _lastSend;
     447        } else {
     448            delay = Math.min(RETRANSMIT_DELAY << _introSentCount, MAX_DELAY);
     449        }
     450        _introSentCount++;
    393451        _nextSend = _lastSend + delay;
    394         if (_currentState == STATE_UNKNOWN)
    395             _currentState = STATE_PENDING_INTRO;
    396     }
     452        if (_currentState == OutboundState.OB_STATE_UNKNOWN)
     453            _currentState = OutboundState.OB_STATE_PENDING_INTRO;
     454    }
     455
     456    /**
     457     *  @return when we sent the first RelayRequest packet, or 0
     458     *  @since 0.9.2
     459     */
     460    public long getIntroSentTime() { return _introSentTime; }
    397461
    398462    public synchronized void introductionFailed() {
    399463        _nextSend = _context.clock().now();
    400         // keep the state as STATE_PENDING_INTRO, so next time the EstablishmentManager asks us
     464        // keep the state as OB_STATE_PENDING_INTRO, so next time the EstablishmentManager asks us
    401465        // whats up, it'll try a new random intro peer
    402466    }
    403467   
     468    /**
     469     *  This changes the remoteHostId from a hash-based one to a IP/Port one
     470     */
    404471    public synchronized void introduced(InetAddress bob, byte bobIP[], int bobPort) {
    405         if (_currentState != STATE_PENDING_INTRO)
     472        if (_currentState != OutboundState.OB_STATE_PENDING_INTRO)
    406473            return; // we've already successfully been introduced, so don't overwrite old settings
    407474        _nextSend = _context.clock().now() + 500; // wait briefly for the hole punching
    408         if (_currentState == STATE_PENDING_INTRO) {
    409             // STATE_UNKNOWN will probe the EstablishmentManager to send a new
     475        if (_currentState == OutboundState.OB_STATE_PENDING_INTRO) {
     476            // OB_STATE_UNKNOWN will probe the EstablishmentManager to send a new
    410477            // session request to this newly known address
    411             _currentState = STATE_UNKNOWN;
     478            _currentState = OutboundState.OB_STATE_UNKNOWN;
    412479        }
    413480        _bobIP = bobIP;
     
    422489    public long getEstablishBeginTime() { return _establishBegin; }
    423490    public synchronized long getNextSendTime() { return _nextSend; }
    424     public synchronized void setNextSendTime(long when) {
    425         _nextSend = when;
    426         if (_log.shouldLog(Log.DEBUG))
    427             _log.debug("Explicit nextSend=" + (_nextSend-_context.clock().now()), new Exception("Set by"));
    428     }
    429491
    430492    /** uniquely identifies an attempt */
     
    434496    public synchronized void dataReceived() {
    435497        packetReceived();
    436         _currentState = STATE_CONFIRMED_COMPLETELY;
     498        _currentState = OutboundState.OB_STATE_CONFIRMED_COMPLETELY;
    437499    }
    438500   
  • router/java/src/net/i2p/router/transport/udp/OutboundMessageState.java

    re9cc851 r16a46b3  
    6565    /**
    6666     *  Called from UDPTransport
     67     *  @return success
    6768     */
    6869    public boolean initialize(I2NPMessage msg, PeerState peer) {
     
    8283    /**
    8384     *  Called from OutboundMessageFragments
     85     *  @return success
    8486     */
    8587    public boolean initialize(OutNetMessage m, I2NPMessage msg) {
     
    9799    }
    98100   
     101    /**
     102     *  Called from OutboundMessageFragments
     103     *  @return success
     104     */
    99105    private boolean initialize(OutNetMessage m, I2NPMessage msg, PeerState peer) {
    100106        _message = m;
  • router/java/src/net/i2p/router/transport/udp/PacketBuilder.java

    re9cc851 r16a46b3  
    686686     * Build a new series of SessionConfirmed packets for the given peer,
    687687     * encrypting it as necessary.
     688     *
     689     * Note that while a SessionConfirmed could in theory be fragmented,
     690     * in practice a RouterIdentity is 387 bytes and a single fragment is 512 bytes max,
     691     * so it will never be fragmented.
    688692     *
    689693     * @return ready to send packets, or null if there was a problem
     
    691695     * TODO: doesn't really return null, and caller doesn't handle null return
    692696     * (null SigningPrivateKey should cause this?)
    693      * Should probably return null if buildSessionConfirmedPacket() turns null for any fragment
     697     * Should probably return null if buildSessionConfirmedPacket() returns null for any fragment
    694698     */
    695699    public UDPPacket[] buildSessionConfirmedPackets(OutboundEstablishState state, RouterIdentity ourIdentity) {
     
    794798     */
    795799    public UDPPacket buildSessionDestroyPacket(PeerState peer) {
     800        if (_log.shouldLog(Log.DEBUG)) {
     801            _log.debug("building session destroy packet to " + peer.getRemotePeer());
     802        }
     803        return buildSessionDestroyPacket(peer.getCurrentCipherKey(), peer.getCurrentMACKey(),
     804                                         peer.getRemoteIPAddress(), peer.getRemotePort());
     805    }
     806
     807    /**
     808     *  Build a destroy packet, which contains a header but no body.
     809     *  If the keys and ip/port are not yet set, this will return null.
     810     *
     811     *  @return packet or null
     812     *  @since 0.9.2
     813     */
     814    public UDPPacket buildSessionDestroyPacket(OutboundEstablishState peer) {
     815        SessionKey cipherKey = peer.getCipherKey();
     816        SessionKey macKey = peer.getMACKey();
     817        byte[] ip = peer.getSentIP();
     818        int port = peer.getSentPort();
     819        if (cipherKey == null || macKey == null || ip == null || port <= 0) {
     820            if (_log.shouldLog(Log.DEBUG))
     821                _log.debug("Cannot send destroy, incomplete " + peer);
     822            return null;
     823        }
     824        InetAddress addr;
     825        try {
     826            addr = InetAddress.getByAddress(ip);
     827        } catch (UnknownHostException uhe) {
     828            return null;
     829        }
     830        if (_log.shouldLog(Log.DEBUG))
     831            _log.debug("building session destroy packet to " + peer);
     832        return buildSessionDestroyPacket(cipherKey, macKey, addr, port);
     833    }
     834
     835
     836    /**
     837     *  Build a destroy packet, which contains a header but no body.
     838     *  If the keys and ip/port are not yet set, this will return null.
     839     *
     840     *  @return packet or null
     841     *  @since 0.9.2
     842     */
     843    public UDPPacket buildSessionDestroyPacket(InboundEstablishState peer) {
     844        SessionKey cipherKey = peer.getCipherKey();
     845        SessionKey macKey = peer.getMACKey();
     846        byte[] ip = peer.getSentIP();
     847        int port = peer.getSentPort();
     848        if (cipherKey == null || macKey == null || ip == null || port <= 0) {
     849            if (_log.shouldLog(Log.DEBUG))
     850                _log.debug("Cannot send destroy, incomplete " + peer);
     851            return null;
     852        }
     853        InetAddress addr;
     854        try {
     855            addr = InetAddress.getByAddress(ip);
     856        } catch (UnknownHostException uhe) {
     857            return null;
     858        }
     859        if (_log.shouldLog(Log.DEBUG))
     860            _log.debug("building session destroy packet to " + peer);
     861        return buildSessionDestroyPacket(cipherKey, macKey, addr, port);
     862    }
     863
     864    /**
     865     *  Build a destroy packet, which contains a header but no body.
     866     *  @param cipherKey non-null
     867     *  @param macKey non-null
     868     *  @since 0.9.2
     869     */
     870    private UDPPacket buildSessionDestroyPacket(SessionKey cipherKey, SessionKey macKey, InetAddress addr, int port) {
    796871        UDPPacket packet = buildPacketHeader((byte)(UDPPacket.PAYLOAD_TYPE_SESSION_DESTROY << 4));
    797872        int off = HEADER_SIZE;
    798873       
    799         StringBuilder msg = null;
    800         if (_log.shouldLog(Log.DEBUG)) {
    801             msg = new StringBuilder(128);
    802             msg.append("building session destroy packet to ").append(peer.getRemotePeer().toBase64().substring(0,6));
    803         }
    804 
    805874        // no body in this message
    806 
    807         if (msg != null)
    808             _log.debug(msg.toString());
    809875       
    810876        // pad up so we're on the encryption boundary
     
    812878            off += 16 - (off % 16);
    813879        packet.getPacket().setLength(off);
    814         authenticate(packet, peer.getCurrentCipherKey(), peer.getCurrentMACKey());
    815         setTo(packet, peer.getRemoteIPAddress(), peer.getRemotePort());
     880        authenticate(packet, cipherKey, macKey);
     881        setTo(packet, addr, port);
    816882        return packet;
    817883    }
  • router/java/src/net/i2p/router/transport/udp/PacketHandler.java

    re9cc851 r16a46b3  
    329329                // packets for which we don't have the PeerState (i.e. SessionKey)
    330330                if (_log.shouldLog(Log.WARN))
    331                     _log.warn("Invalid introduction packet received: " + packet, new Exception("path"));
     331                    _log.warn("Cannot validate rcvd pkt (path): " + packet);
    332332                _context.statManager().addRateData("udp.droppedInvalidEstablish", packet.getLifetime(), packet.getExpiration());
    333333                switch (peerType) {
     
    466466
    467467        /**
    468          * Parse out the interesting bits and honor what it says
     468         * The last step. The packet was decrypted with some key. Now get the message type
     469         * and send it to one of four places: The EstablishmentManager, IntroductionManager,
     470         * PeerTestManager, or InboundMessageFragments.
    469471         *
    470472         * @param state non-null if fully established
    471473         * @param outState non-null if outbound establishing in process
    472          * @param inState unused always null
     474         * @param inState unused always null, TODO use for 48-byte destroys during inbound establishment
    473475         * @param isAuthenticated true if a state key was used, false if our own intro key was used
    474476         */
  • router/java/src/net/i2p/router/transport/udp/UDPPacket.java

    re9cc851 r16a46b3  
    126126        _remoteHost = null;
    127127        _released = false;
     128        // clear out some values to make debugging easier via toString()
     129        _messageType = -1;
     130        _enqueueTime = 0;
     131        _receivedTime = 0;
     132        _fragmentCount = 0;
    128133    }
    129134   
     
    285290        buf.append(_packet.getPort());
    286291        //buf.append(" id=").append(System.identityHashCode(this));
    287         buf.append(" msgType=").append(_messageType);
    288         buf.append(" markType=").append(_markedType);
    289         buf.append(" fragCount=").append(_fragmentCount);
    290 
    291         buf.append(" sinceEnqueued=").append((_enqueueTime > 0 ? _context.clock().now()-_enqueueTime : -1));
    292         buf.append(" sinceReceived=").append((_receivedTime > 0 ? _context.clock().now()-_receivedTime : -1));
     292        if (_messageType >= 0)
     293            buf.append(" msgType=").append(_messageType);
     294        if (_markedType >= 0)
     295            buf.append(" markType=").append(_markedType);
     296        if (_fragmentCount > 0)
     297            buf.append(" fragCount=").append(_fragmentCount);
     298
     299        if (_enqueueTime >= 0)
     300            buf.append(" sinceEnqueued=").append(_context.clock().now() - _enqueueTime);
     301        if (_receivedTime >= 0)
     302            buf.append(" sinceReceived=").append(_context.clock().now() - _receivedTime);
    293303        //buf.append(" beforeReceiveFragments=").append((_beforeReceiveFragments > 0 ? _context.clock().now()-_beforeReceiveFragments : -1));
    294304        //buf.append(" sinceHandled=").append((_afterHandlingTime > 0 ? _context.clock().now()-_afterHandlingTime : -1));
Note: See TracChangeset for help on using the changeset viewer.