Correct / speedup / cleanup EstablishmentManager.java

[jogger]

General remark: Choosing -1 as constant for undefined time made i2p code overly clumsy everywhere and slow in inner loops since lowest time usually wins in comparisons. Long.MAX_VALUE provides shorter code and better readability. But even with using -1 there are many places where 4 or 5 statements within inner loops can be rolled into one.

From 0.9.37-20-rc:

Patch 1: Could also be a BLOCKER because 2 code sections that might need be identical are different.

Completely useless or incorrect code: If I read handleInbound() correctly the following snippet does plain nothing because the final if clause can never strike (line 1173):

else {

long when = -1;

if (cur.getNextSendTime() ⇐ 0) {

when = cur.getEstablishBeginTime() + MAX_IB_ESTABLISH_TIME;

} else {

when = cur.getNextSendTime();

}
if (when < nextSendTime)

nextSendTime = when;

}

If the corresponding code in handleoutbound() is correct then the much more efficient solution from line 1280 would be:

} else {

long when = cur.getNextSendTime();
if (when ⇐ now) {

inboundState = cur;
break;

} else {

if (when ⇐ 0)

when = cur.getEstablishBeginTime() + MAX_IB_ESTABLISH_TIME;

if ( (nextSendTime ⇐ 0)

(when < nextSendTime) )

nextSendTime = when;

}

}

—-

Patch 2: forgotten to substitute now() in lines 1480/1481

—-

Patch 3: In line 1484 initialize nextSendTime with handleInbound(). Saves 3 lines.

—-

Patch 4 : There is an unused "RemoteHostId? to" within "locked_admitQueued"

[jogger]

Addition: Just looked again - whatever is correct for Patch 1:

the "if (when ⇐ 0)" or before "if (cur.getNextSendTime() ⇐ 0)" is always false because of the preceding (when ⇐ now). Four more lines saved.

[zzz]

I haven't looked at this part of the code in a while, and it took a while to follow the OP, but I believe the following captures the intent of the OP "patches 1-4". Please review and test.

#
# old_revision [865ef1bc9a0c75a8a9ae6edb56d9c07e3f5b8ff0]
#
# patch "router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java"
#  from [dcd2ab849c2726ab11974fd80011da7b2934aca5]
#    to [7a892873d89d81ddbe1afd37a02af126e19ed7b1]
# 
# patch "router/java/src/net/i2p/router/transport/udp/InboundEstablishState.java"
#  from [9b09d18daae7d86e7767fad0e62a324a61eae6f8]
#    to [c746bfee971f5ea0f6bb944120177c0e2df7afb9]
# 
# patch "router/java/src/net/i2p/router/transport/udp/OutboundEstablishState.java"
#  from [1e311ff2bc06c95843ad9651476757004ec1c423]
#    to [876f0ffc4d4cf09c0a54d0c2941fea3bcc144b71]
#
============================================================
--- router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java	dcd2ab849c2726ab11974fd80011da7b2934aca5
+++ router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java	7a892873d89d81ddbe1afd37a02af126e19ed7b1
@@ -477,8 +477,8 @@ class EstablishmentManager {
                 byte[] fromIP = from.getIP();
                 state = new InboundEstablishState(_context, fromIP, from.getPort(),
                                                   _transport.getExternalPort(fromIP.length == 16),
-                                                  _transport.getDHBuilder());
-                state.receiveSessionRequest(reader.getSessionRequestReader());
+                                                  _transport.getDHBuilder(),
+                                                  reader.getSessionRequestReader());
 
                 if (_replayFilter.add(state.getReceivedX(), 0, 8)) {
                     if (_log.shouldLog(Log.WARN))
@@ -652,7 +652,6 @@ class EstablishmentManager {
             } catch (IllegalStateException ise) {
                 continue;
             }
-            RemoteHostId to = entry.getKey();
             List<OutNetMessage> allQueued = entry.getValue();
             List<OutNetMessage> queued = new ArrayList<OutNetMessage>();
             long now = _context.clock().now();
@@ -1134,11 +1133,11 @@ class EstablishmentManager {
     /**
      * Drive through the inbound establishment states, adjusting one of them
      * as necessary. Called from Establisher thread only.
-     * @return next requested time or -1
+     * @return next requested time or Long.MAX_VALUE
      */
     private long handleInbound() {
         long now = _context.clock().now();
-        long nextSendTime = -1;
+        long nextSendTime = Long.MAX_VALUE;
         InboundEstablishState inboundState = null;
         boolean expired = false;
 
@@ -1164,7 +1163,9 @@ class EstablishmentManager {
                     iter.remove();
                     //_context.statManager().addRateData("udp.inboundEstablishFailedState", cur.getState(), cur.getLifetime());
                 } else {
-                    if (cur.getNextSendTime() <= now) {
+                    // this will always be > 0
+                    long next = cur.getNextSendTime();
+                    if (next <= now) {
                         // our turn...
                         inboundState = cur;
                         // if (_log.shouldLog(Log.DEBUG))
@@ -1174,14 +1175,8 @@ class EstablishmentManager {
                         // nothin to do but wait for them to send us
                         // stuff, so lets move on to the next one being
                         // established
-                        long when = -1;
-                        if (cur.getNextSendTime() <= 0) {
-                            when = cur.getEstablishBeginTime() + MAX_IB_ESTABLISH_TIME;
-                        } else {
-                            when = cur.getNextSendTime();
-                        }
-                        if (when < nextSendTime)
-                            nextSendTime = when;
+                        if (next < nextSendTime)
+                            nextSendTime = next;
                     }
                 }
             }
@@ -1251,11 +1246,11 @@ class EstablishmentManager {
     /**
      * Drive through the outbound establishment states, adjusting one of them
      * as necessary. Called from Establisher thread only.
-     * @return next requested time or -1
+     * @return next requested time or Long.MAX_VALUE
      */
     private long handleOutbound() {
         long now = _context.clock().now();
-        long nextSendTime = -1;
+        long nextSendTime = Long.MAX_VALUE;
         OutboundEstablishState outboundState = null;
         //int admitted = 0;
         //int remaining = 0;
@@ -1278,7 +1273,9 @@ class EstablishmentManager {
                     //    _log.debug("Removing expired outbound: " + cur);
                     break;
                 } else {
-                    if (cur.getNextSendTime() <= now) {
+                    // this will be 0 for a new OES that needs sending, > 0 for others
+                    long next = cur.getNextSendTime();
+                    if (next <= now) {
                         // our turn...
                         outboundState = cur;
                         // if (_log.shouldLog(Log.DEBUG))
@@ -1288,14 +1285,8 @@ class EstablishmentManager {
                         // nothin to do but wait for them to send us
                         // stuff, so lets move on to the next one being
                         // established
-                        long when = -1;
-                        if (cur.getNextSendTime() <= 0) {
-                            when = cur.getEstablishBeginTime() + MAX_OB_ESTABLISH_TIME;
-                        } else {
-                            when = cur.getNextSendTime();
-                        }
-                        if ( (nextSendTime <= 0) || (when < nextSendTime) )
-                            nextSendTime = when;
+                        if (next < nextSendTime)
+                            nextSendTime = next;
                         // if (_log.shouldLog(Log.DEBUG))
                         //     _log.debug("Outbound doesn't want activity: " + cur + " (next=" + (when-now) + ")");
                     }
@@ -1461,8 +1452,9 @@ class EstablishmentManager {
         private static final long PRINT_INTERVAL = 5*1000;
     
         private void doPass() {
-            if (_log.shouldLog(Log.DEBUG) && _lastPrinted + PRINT_INTERVAL < _context.clock().now()) {
-                _lastPrinted = _context.clock().now();
+            long now = _context.clock().now();
+            if (_log.shouldLog(Log.DEBUG) && _lastPrinted + PRINT_INTERVAL < now) {
+                _lastPrinted = now;
                 int iactive = _inboundStates.size();
                 int oactive = _outboundStates.size();
                 if (iactive > 0 || oactive > 0) {
@@ -1476,31 +1468,21 @@ class EstablishmentManager {
                 }
             } 
             _activity = 0;
-            long now = _context.clock().now();
-            if (_lastFailsafe + FAILSAFE_INTERVAL < _context.clock().now()) {
-                _lastFailsafe = _context.clock().now();
+            if (_lastFailsafe + FAILSAFE_INTERVAL < now) {
+                _lastFailsafe = now;
                 doFailsafe();
             }
-            long nextSendTime = -1;
-            long nextSendInbound = handleInbound();
-            long nextSendOutbound = handleOutbound();
-            if (nextSendInbound > 0)
-                nextSendTime = nextSendInbound;
-            if ( (nextSendTime < 0) || (nextSendOutbound < nextSendTime) )
-                nextSendTime = nextSendOutbound;
 
+            long nextSendTime = Math.min(handleInbound(), handleOutbound());
             long delay = nextSendTime - now;
-            if ( (nextSendTime == -1) || (delay > 0) ) {
+            if (delay > 0) {
                 if (delay > 1000)
                     delay = 1000;
                 try {
                     synchronized (_activityLock) {
                         if (_activity > 0)
                             return;
-                        if (nextSendTime == -1)
-                           _activityLock.wait(1000);
-                        else
-                            _activityLock.wait(delay);
+                        _activityLock.wait(delay);
                     }
                 } catch (InterruptedException ie) {
                 }
============================================================
--- router/java/src/net/i2p/router/transport/udp/InboundEstablishState.java	9b09d18daae7d86e7767fad0e62a324a61eae6f8
+++ router/java/src/net/i2p/router/transport/udp/InboundEstablishState.java	c746bfee971f5ea0f6bb944120177c0e2df7afb9
@@ -97,7 +97,7 @@ class InboundEstablishState {
      *                   SessionCreated message will be bad if the external port != the internal port.
      */
     public InboundEstablishState(RouterContext ctx, byte remoteIP[], int remotePort, int localPort,
-                                 DHSessionKeyBuilder dh) {
+                                 DHSessionKeyBuilder dh, UDPPacketReader.SessionRequestReader req) {
         _context = ctx;
         _log = ctx.logManager().getLog(InboundEstablishState.class);
         _aliceIP = remoteIP;
@@ -108,6 +108,7 @@ class InboundEstablishState {
         _establishBegin = ctx.clock().now();
         _keyBuilder = dh;
         _queuedMessages = new LinkedBlockingQueue<OutNetMessage>();
+        receiveSessionRequest(req);
     }
     
     public synchronized InboundState getState() { return _currentState; }
@@ -288,6 +289,11 @@ class InboundEstablishState {
     /** how long have we been trying to establish this session? */
     public long getLifetime() { return _context.clock().now() - _establishBegin; }
     public long getEstablishBeginTime() { return _establishBegin; }
+
+    /**
+     *  @return rcv time after receiving a packet (including after constructor),
+     *          send time + delay after sending a packet
+     */
     public synchronized long getNextSendTime() { return _nextSend; }
 
     /** RemoteHostId, uniquely identifies an attempt */
@@ -478,6 +484,9 @@ class InboundEstablishState {
             }
     }
     
+    /**
+     *  Call from synchronized method only
+     */
     private void packetReceived() {
         _nextSend = _context.clock().now();
     }
============================================================
--- router/java/src/net/i2p/router/transport/udp/OutboundEstablishState.java	1e311ff2bc06c95843ad9651476757004ec1c423
+++ router/java/src/net/i2p/router/transport/udp/OutboundEstablishState.java	876f0ffc4d4cf09c0a54d0c2941fea3bcc144b71
@@ -653,6 +653,12 @@ class OutboundEstablishState {
     /** how long have we been trying to establish this session? */
     public long getLifetime() { return _context.clock().now() - _establishBegin; }
     public long getEstablishBeginTime() { return _establishBegin; }
+
+    /**
+     *  @return 0 at initialization (to force sending session request),
+     *          rcv time after receiving a packet,
+     *          send time + delay after sending a packet (including session request)
+     */
     public synchronized long getNextSendTime() { return _nextSend; }
 
     /**
@@ -678,10 +684,13 @@ class OutboundEstablishState {
         _currentState = OutboundState.OB_STATE_CONFIRMED_COMPLETELY;
     }
     
+    /**
+     *  Call from synchronized method only
+     */
     private void packetReceived() {
         _nextSend = _context.clock().now();
-        if (_log.shouldLog(Log.DEBUG))
-            _log.debug("Got a packet, nextSend == now");
+        //if (_log.shouldLog(Log.DEBUG))
+        //    _log.debug("Got a packet, nextSend == now");
     }
 
     /** @since 0.8.9 */

[Zlatin Balevsky]

This could explain why newly established SSU connections always have an RTT > 1s as reported by /peers. Only after several messages have been sent back and forth does the RTT go down to more reasonable levels.

[zzz]

Not sure about that. I didn't see anything wrong with the previous code but wasn't looking for the Outbound sessions waiting one second. I assume it was an outbound thing only? Does the above patch fix it?

[Zlatin Balevsky]

Well the previous code for inbound was wrong because it never executed the final "else" clause. So we would not respond to an inbound establishment up to a second later. I believe the patch fixes it, but the effect will be seen in /peers only after most of the network upgrades.

[zzz]

ok, I'll look at it again. In the meantime, awaiting acks and/or test results for my patch from OP and zab.

[Zlatin Balevsky]

I've been running with the patch for a few days. No visible issues; plenty of SSU peers. Can't think of anything else that would be noticeable from the console or the logs.

[jogger]

Running with it for more than a day. Can confirm zab´s comment.

[zzz]

We discussed comment 6. The old code would always respond immediately to a session request. However for retransmissions during the establishment phase, the calculated wait time could be too long.

Patch in comment 3 checked in as-is
in 25c2daedd2027b4ad72c6ab742be0c8209db3efb to be 0.9.38-8

New ticket #2433 opened to investigate related improvements in EstablishmentManager?. Closing this one as the change completely covers the 4 items in the OP.