Streaming: Fix optional delay and choking (tickets #1046, 1939)

- Don't always send optional delay
- Don't overwrite choking delay with non-choking delay
- Don't send optional delay of 0 every 8 packets
- Don't set options both in CDR.buildPacket() and Conn.sendPacket()
- Set or clear optional delay in packet when retransmitting
- Move choking state variables from ConnectionOptions to Connection
- Move updateAcks() call from PacketLocal to PacketQueue
- Fully implement choking and un-choking
- Reduce periods for some stats
- Comment out some debug logging
- Cleanups
- Fix javadoc HTML broken in previous checkin

apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java

@@ -69,6 +69,11 @@ class Connection {
     private int _lastCongestionSeenAt;
     private long _lastCongestionTime;
     private volatile long _lastCongestionHighestUnacked;
+    /** has the other side choked us? */
+    private volatile boolean _isChoked;
+    /** are we choking the other side? */
+    private volatile boolean _isChoking;
+    private final AtomicInteger _unchokesToSend = new AtomicInteger();
     private final AtomicBoolean _ackSinceCongestion;
     /** Notify this on connection (or connection failure) */
     private final Object _connectLock;
@@ -102,6 +107,7 @@ class Connection {
     private static final long MAX_CONNECT_TIMEOUT = 2*60*1000;
 
     public static final int MAX_WINDOW_SIZE = 128;
+    private static final int UNCHOKES_TO_SEND = 8;
     
 /****
     public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser,
@@ -187,7 +193,7 @@ class Connection {
             long timeLeft = writeExpire - _context.clock().now();
             synchronized (_outboundPackets) {
                 if (!started)
-                    _context.statManager().addRateData("stream.chokeSizeBegin", _outboundPackets.size(), timeoutMs);
+                    _context.statManager().addRateData("stream.chokeSizeBegin", _outboundPackets.size());
                 if (start + 5*60*1000 < _context.clock().now()) // ok, 5 minutes blocking?  I dont think so
                     return false;
                 
@@ -205,20 +211,20 @@ class Connection {
                 // Limit (highest-lowest) to twice the window (if far end doesn't like it, it can send a choke)
                 int unacked = _outboundPackets.size();
                 int wsz = _options.getWindowSize();
-                if (unacked >= wsz ||
+                if (_isChoked || unacked >= wsz ||
                     _activeResends.get() >= (wsz + 1) / 2 ||
                     _lastSendId.get() - _highestAckedThrough >= Math.max(MAX_WINDOW_SIZE, 2 * wsz)) {
                     if (timeoutMs > 0) {
                         if (timeLeft <= 0) {
                             if (_log.shouldLog(Log.INFO))
-                                _log.info("Outbound window is full " + unacked
+                                _log.info("Outbound window is full (choked? " + _isChoked + ' ' + unacked
                                           + " unacked with " + _activeResends + " active resends"
                                           + " and we've waited too long (" + (0-(timeLeft - timeoutMs)) + "ms): " 
                                           + toString());
                             return false;
                         }
                         if (_log.shouldLog(Log.DEBUG))
-                            _log.debug("Outbound window is full (" + unacked + "/" + wsz + "/" 
+                            _log.debug("Outbound window is full (choked? " + _isChoked + ' ' + unacked + '/' + wsz + '/' 
                                        + _activeResends + "), waiting " + timeLeft);
                         try {
                             _outboundPackets.wait(Math.min(timeLeft,250l));
@@ -240,7 +246,7 @@ class Connection {
                         } //10*1000
                     }
                 } else {
-                    _context.statManager().addRateData("stream.chokeSizeEnd", _outboundPackets.size(), _context.clock().now() - start);
+                    _context.statManager().addRateData("stream.chokeSizeEnd", _outboundPackets.size());
                     return true;
                 }
             }
@@ -257,7 +263,7 @@ class Connection {
     }
     
     void ackImmediately() {
-        PacketLocal packet = null;
+        PacketLocal packet;
 /*** why would we do this?
      was it to force a congestion indication at the other end?
      an expensive way to do that...
@@ -343,6 +349,12 @@ class Connection {
         }
     }
     
+    /**
+     *  This sends all 'normal' packets (acks and data) for the first time.
+     *  Retransmits are done in ResendPacketEvent below.
+     *  Resets, pings, and pongs are done elsewhere in this class,
+     *  or in ConnectionManager or ConnectionHandler.
+     */
     void sendPacket(PacketLocal packet) {
         if (packet == null) return;
         
@@ -353,8 +365,15 @@ class Connection {
         }
         
         if ( (packet.getSequenceNum() == 0) && (!packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) ) {
-            //if (_log.shouldLog(Log.DEBUG))
+            // ACK-only
-            //    _log.debug("No resend for " + packet);
+            if (_isChoking) {
+                packet.setOptionalDelay(Packet.SEND_DELAY_CHOKE);
+                packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
+            } else if (_unchokesToSend.decrementAndGet() > 0) {
+                // don't worry about wrapping around
+                packet.setOptionalDelay(0);
+                packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
+            }
         } else {
             int windowSize;
             int remaining;
@@ -364,13 +383,18 @@ class Connection {
                 remaining = windowSize - _outboundPackets.size() ;
                 _outboundPackets.notifyAll();
             }
-            // the other end has no idea what our window size is, so
+
-            // help him out by requesting acks below the 1/3 point,
+            if (_isChoking) {
-            // if remaining < 3, and every 8 minimum.
+                packet.setOptionalDelay(Packet.SEND_DELAY_CHOKE);
-            if (packet.isFlagSet(Packet.FLAG_CLOSE) ||
+                packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
-                (remaining < (windowSize + 2) / 3) ||
+            } else if (packet.isFlagSet(Packet.FLAG_CLOSE) ||
+                _unchokesToSend.decrementAndGet() > 0 ||
+                // the other end has no idea what our window size is, so
+                // help him out by requesting acks below the 1/3 point,
+                // if remaining < 3, and every 8 minimum.
                 (remaining < 3) ||
-                (packet.getSequenceNum() % 8 == 0)) {
+                (remaining < (windowSize + 2) / 3) /* ||
+                (packet.getSequenceNum() % 8 == 0) */ ) {
                 packet.setOptionalDelay(0);
                 packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
                 //if (_log.shouldLog(Log.DEBUG))
@@ -380,15 +404,15 @@ class Connection {
                 // since the other end limits it to getSendAckDelay()
                 // which is always 2000, but it's good for diagnostics to see what the other end thinks
                 // the RTT is.
+/**
                 int delay = _options.getRTT() / 2;
                 packet.setOptionalDelay(delay);
                 if (delay > 0)
                     packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
                 if (_log.shouldLog(Log.DEBUG))
                     _log.debug("Requesting ack delay of " + delay + "ms for packet " + packet);
+**/
             }
-            // WHY always set?
-            //packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
             
             long timeout = _options.getRTO();
             if (timeout > MAX_RESEND_DELAY)
@@ -984,6 +1008,55 @@ class Connection {
         }
     }
     
+    /**
+     *  Set or clear if we are choking the other side.
+     *  If on is true or the value has changed, this will call ackImmediately().
+     *  @param on true for choking
+     *  @since 0.9.29
+     */
+    public void setChoking(boolean on) {
+        if (on != _isChoking) {
+            _isChoking = on;
+           if (!on)
+               _unchokesToSend.set(UNCHOKES_TO_SEND);
+           ackImmediately();
+        } else if (on) {
+           ackImmediately();
+        }
+    }
+    
+    /**
+     *  Set or clear if we are being choked by the other side.
+     *  @param on true for choked
+     *  @since 0.9.29
+     */
+    public void setChoked(boolean on) {
+        _isChoked = on;
+        if (on) {
+            congestionOccurred();
+            // https://en.wikipedia.org/wiki/Transmission_Control_Protocol
+            // When a receiver advertises a window size of 0, the sender stops sending data and starts the persist timer.
+            // The persist timer is used to protect TCP from a deadlock situation that could arise
+            // if a subsequent window size update from the receiver is lost,
+            // and the sender cannot send more data until receiving a new window size update from the receiver.
+            // When the persist timer expires, the TCP sender attempts recovery by sending a small packet
+            // so that the receiver responds by sending another acknowledgement containing the new window size.
+            // ...
+            // We don't do any of that, but we set the window size to 1, and let the retransmission
+            // of packets do the "attempted recovery".
+            getOptions().setWindowSize(1);
+        }
+    }
+    
+    /**
+     *  Is the other side choking us?
+     *  @return if choked
+     *  @since 0.9.29
+     */
+    public boolean isChoked() {
+        return _isChoked;
+    }
+
     /** how many packets have we sent and the other side has ACKed?
      * @return Count of how many packets ACKed.
      */
@@ -1381,10 +1454,18 @@ class Connection {
                 // revamp various fields, in case we need to ack more, etc
                 // updateAcks done in enqueue()
                 //_inputStream.updateAcks(_packet);
-                int choke = getOptions().getChoke();
+                if (_isChoking) {
-                _packet.setOptionalDelay(choke);
+                    _packet.setOptionalDelay(Packet.SEND_DELAY_CHOKE);
-                if (choke > 0)
                     _packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
+                } else if (_unchokesToSend.decrementAndGet() > 0) {
+                    // don't worry about wrapping around
+                    _packet.setOptionalDelay(0);
+                    _packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
+                } else {
+                    // clear flag
+                    _packet.setFlag(Packet.FLAG_DELAY_REQUESTED, false);
+                }
+
                 // this seems unnecessary to send the MSS again:
                 //_packet.setOptionalMaxSize(getOptions().getMaxMessageSize());
                 // bugfix release 0.7.8, we weren't dividing by 1000
@@ -1396,7 +1477,10 @@ class Connection {
                 
                 int newWindowSize = getOptions().getWindowSize();
 
-                if (_ackSinceCongestion.get()) {
+                if (_isChoked) {
+                    congestionOccurred();
+                    getOptions().setWindowSize(1);
+                } else if (_ackSinceCongestion.get()) {
                     // only shrink the window once per window
                     if (_packet.getSequenceNum() > _lastCongestionHighestUnacked) {
                         congestionOccurred();

apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java

@@ -156,6 +156,11 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
     }
     
     /** 
+     * Compose a packet.
+     * Most flags are set here; however, some are set in Connection.sendPacket()
+     * and Connection.ResendPacketEvent.retransmit().
+     * Take care not to set the same options both here and in Connection.
+     *
      * @param buf data to be sent - may be null
      * @param off offset into the buffer to start writing from
      * @param size how many bytes of the buffer to write (may be 0)
@@ -164,12 +169,11 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
      * @return the packet to be sent
      */
     private PacketLocal buildPacket(byte buf[], int off, int size, boolean forceIncrement) {
-        Connection con = _connection;
         if (size > Packet.MAX_PAYLOAD_SIZE) throw new IllegalArgumentException("size is too large (" + size + ")");
-        boolean ackOnly = isAckOnly(con, size);
+        boolean ackOnly = isAckOnly(_connection, size);
-        boolean isFirst = (con.getAckedPackets() <= 0) && (con.getUnackedPacketsSent() <= 0);
+        boolean isFirst = (_connection.getAckedPackets() <= 0) && (_connection.getUnackedPacketsSent() <= 0);
         
-        PacketLocal packet = new PacketLocal(_context, con.getRemotePeer(), con);
+        PacketLocal packet = new PacketLocal(_context, _connection.getRemotePeer(), _connection);
         //ByteArray data = packet.acquirePayload();
         ByteArray data = new ByteArray(new byte[size]);
         if (size > 0)
@@ -180,36 +184,32 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
         if ( (ackOnly && !forceIncrement) && (!isFirst) )
             packet.setSequenceNum(0);
         else
-            packet.setSequenceNum(con.getNextOutboundPacketNum());
+            packet.setSequenceNum(_connection.getNextOutboundPacketNum());
-        packet.setSendStreamId(con.getSendStreamId());
+        packet.setSendStreamId(_connection.getSendStreamId());
-        packet.setReceiveStreamId(con.getReceiveStreamId());
+        packet.setReceiveStreamId(_connection.getReceiveStreamId());
         
         // not needed here, handled in PacketQueue.enqueue()
         //con.getInputStream().updateAcks(packet);
-        // note that the optional delay is usually rewritten in Connection.sendPacket()
+
-        int choke = con.getOptions().getChoke();
+        // Do not set optional delay here, set in Connection.sendPacket()
-        packet.setOptionalDelay(choke);
+
-        if (choke > 0)
-            packet.setFlag(Packet.FLAG_DELAY_REQUESTED);
         // bugfix release 0.7.8, we weren't dividing by 1000
-        packet.setResendDelay(con.getOptions().getResendDelay() / 1000);
+        packet.setResendDelay(_connection.getOptions().getResendDelay() / 1000);
         
-        if (con.getOptions().getProfile() == ConnectionOptions.PROFILE_INTERACTIVE)
+        if (_connection.getOptions().getProfile() == ConnectionOptions.PROFILE_INTERACTIVE)
             packet.setFlag(Packet.FLAG_PROFILE_INTERACTIVE, true);
         else
             packet.setFlag(Packet.FLAG_PROFILE_INTERACTIVE, false);
         
-        packet.setFlag(Packet.FLAG_SIGNATURE_REQUESTED, con.getOptions().getRequireFullySigned());
-        
         //if ( (!ackOnly) && (packet.getSequenceNum() <= 0) ) {
         if (isFirst) {
             packet.setFlag(Packet.FLAG_SYNCHRONIZE);
             packet.setOptionalFrom();
-            packet.setOptionalMaxSize(con.getOptions().getMaxMessageSize());
+            packet.setOptionalMaxSize(_connection.getOptions().getMaxMessageSize());
         }
-        packet.setLocalPort(con.getLocalPort());
+        packet.setLocalPort(_connection.getLocalPort());
-        packet.setRemotePort(con.getPort());
+        packet.setRemotePort(_connection.getPort());
-        if (con.getSendStreamId() == Packet.STREAM_ID_UNKNOWN) {
+        if (_connection.getSendStreamId() == Packet.STREAM_ID_UNKNOWN) {
             packet.setFlag(Packet.FLAG_NO_ACK);
         }
         
@@ -221,10 +221,10 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
         // FIXME Implement better half-close by sending CLOSE whenever. Needs 0.9.9 bug fixes
         // throughout network?
         //
-        if (con.getOutputStream().getClosed() && 
+        if (_connection.getOutputStream().getClosed() && 
-            ( (size > 0) || (con.getUnackedPacketsSent() <= 0) || (packet.getSequenceNum() > 0) ) ) {
+            ( (size > 0) || (_connection.getUnackedPacketsSent() <= 0) || (packet.getSequenceNum() > 0) ) ) {
             packet.setFlag(Packet.FLAG_CLOSE);
-            con.notifyCloseSent();
+            _connection.notifyCloseSent();
         }
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("New OB pkt (acks not yet filled in): " + packet + " on " + _connection);

apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java

@@ -105,13 +105,13 @@ class ConnectionManager {
         _context.statManager().createRateStat("stream.con.lifetimeSendWindowSize", "What is the final send window size when a stream closes?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
         _context.statManager().createRateStat("stream.receiveActive", "How many streams are active when a new one is received (period being not yet dropped)", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
         // Stats for Connection
-        _context.statManager().createRateStat("stream.con.windowSizeAtCongestion", "How large was our send window when we send a dup?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.con.windowSizeAtCongestion", "How large was our send window when we send a dup?", "Stream", new long[] { 60*60*1000 });
-        _context.statManager().createRateStat("stream.chokeSizeBegin", "How many messages were outstanding when we started to choke?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.chokeSizeBegin", "How many messages were outstanding when we started to choke?", "Stream", new long[] { 60*60*1000 });
-        _context.statManager().createRateStat("stream.chokeSizeEnd", "How many messages were outstanding when we stopped being choked?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.chokeSizeEnd", "How many messages were outstanding when we stopped being choked?", "Stream", new long[] { 60*60*1000 });
-        _context.statManager().createRateStat("stream.fastRetransmit", "How long a packet has been around for if it has been resent per the fast retransmit timer?", "Stream", new long[] { 60*1000, 10*60*1000 });
+        _context.statManager().createRateStat("stream.fastRetransmit", "How long a packet has been around for if it has been resent per the fast retransmit timer?", "Stream", new long[] { 10*60*1000 });
         // Stats for PacketQueue
-        _context.statManager().createRateStat("stream.con.sendMessageSize", "Size of a message sent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.con.sendMessageSize", "Size of a message sent on a connection", "Stream", new long[] { 10*60*1000, 60*60*1000 });
-        _context.statManager().createRateStat("stream.con.sendDuplicateSize", "Size of a message resent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.con.sendDuplicateSize", "Size of a message resent on a connection", "Stream", new long[] { 10*60*1000, 60*60*1000 });
     }
     
     Connection getConnectionByInboundId(long id) {

apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionOptions.java

@@ -33,7 +33,6 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
     private int _resendDelay;
     private int _sendAckDelay;
     private int _maxMessageSize;
-    private int _choke;
     private int _maxResends;
     private int _inactivityTimeout;
     private int _inactivityAction;
@@ -327,7 +326,6 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
             setWindowSize(opts.getWindowSize());
             setResendDelay(opts.getResendDelay());
             setMaxMessageSize(opts.getMaxMessageSize());
-            setChoke(opts.getChoke());
             setMaxResends(opts.getMaxResends());
             setInactivityTimeout(opts.getInactivityTimeout());
             setInactivityAction(opts.getInactivityAction());
@@ -677,15 +675,6 @@ class ConnectionOptions extends I2PSocketOptionsImpl {
     public int getMaxMessageSize() { return _maxMessageSize; }
     public void setMaxMessageSize(int bytes) { _maxMessageSize = Math.max(bytes, MIN_MESSAGE_SIZE); }
     
-    /** 
-     * how long we want to wait before any data is transferred on the
-     * connection in either direction
-     *
-     * @return how long to wait before any data is transferred in either direction in ms
-     */
-    public int getChoke() { return _choke; }
-    public void setChoke(int ms) { _choke = ms; }
-
     /**
      * What profile do we want to use for this connection?
      * TODO: Only bulk is supported so far.

apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionPacketHandler.java

@@ -96,23 +96,27 @@ class ConnectionPacketHandler {
         
         boolean choke = false;
         if (packet.isFlagSet(Packet.FLAG_DELAY_REQUESTED)) {
-            if (packet.getOptionalDelay() > 60000) {
+            if (packet.getOptionalDelay() >= Packet.MIN_DELAY_CHOKE) {
                 // requested choke 
                 choke = true;
+                if (_log.shouldWarn())
+                    _log.warn("Got a choke on connection " + con + ": " + packet);
                 //con.getOptions().setRTT(con.getOptions().getRTT() + 10*1000);
             }
+            // Only call this if the flag is set
+            con.setChoked(choke);
         }
         
         if (!con.getInputStream().canAccept(seqNum, packet.getPayloadSize())) {
             if (_log.shouldWarn())
                 _log.warn("Inbound buffer exceeded on connection " + con +
-                          ", dropping " + packet);
+                          ", choking and dropping " + packet);
-            con.getOptions().setChoke(61*1000);
+            // this will call ackImmediately()
+            con.setChoking(true);
+            // TODO we could still process the acks for this packet before discarding
             packet.releasePayload();
-            con.ackImmediately();
             return;
-        }
+        } // else we will call setChoking(false) below
-        con.getOptions().setChoke(0);
 
         _context.statManager().addRateData("stream.con.receiveMessageSize", packet.getPayloadSize());
         
@@ -132,12 +136,20 @@ class ConnectionPacketHandler {
         // MessageInputStream will know the last sequence number.
         // But not ack-only packets!
         boolean isNew;
-        if (seqNum > 0 || isSYN)
+        if (seqNum > 0 || isSYN) {
-            isNew = con.getInputStream().messageReceived(seqNum, packet.getPayload());
+            isNew = con.getInputStream().messageReceived(seqNum, packet.getPayload()) &&
-        else
+                    !allowAck;
-            isNew = false;
+        } else {
-        if (!allowAck)
             isNew = false;
+        }
+
+        if (isNew && packet.getPayloadSize() > 1500) {
+            // don't clear choking unless it was new, and a big packet
+            // this will call ackImmediately() if changed
+            // TODO if this filled in a hole, we shouldn't unchoke
+            // TODO a bunch of small packets should unchoke also
+            con.setChoking(false);
+        }
         
         //if ( (packet.getSequenceNum() == 0) && (packet.getPayloadSize() > 0) ) {
         //    if (_log.shouldLog(Log.DEBUG))
@@ -158,7 +170,6 @@ class ConnectionPacketHandler {
             _log.debug(type + " IB pkt: " + packet + " on " + con);
         }
 
-        boolean fastAck = false;
         boolean ackOnly = false;
         
         if (isNew) {
@@ -170,6 +181,9 @@ class ConnectionPacketHandler {
                     _log.debug("Scheduling immediate ack for " + packet);
                 //con.setNextSendTime(_context.clock().now() + con.getOptions().getSendAckDelay());
                 // honor request "almost" immediately
+                // TODO the 250 below _may_ be a big limiter in how fast local "loopback" connections
+                // can go, however if it goes too fast then we start choking which causes
+                // frequent stalls anyway.
                 con.setNextSendTime(_context.clock().now() + 250);
             } else {
                 int delay = con.getOptions().getSendAckDelay();
@@ -222,14 +236,16 @@ class ConnectionPacketHandler {
             }
         }
 
+        boolean fastAck;
         if (isSYN && (packet.getSendStreamId() <= 0) ) {
             // don't honor the ACK 0 in SYN packets received when the other side
             // has obviously not seen our messages
+            fastAck = false;
         } else {
             fastAck = ack(con, packet.getAckThrough(), packet.getNacks(), packet, isNew, choke);
         }
         con.eventOccurred();
-        if (fastAck) {
+        if (fastAck && !choke) {
             if (!isNew) {
                 // if we're congested (fastAck) but this is also a new packet, 
                 // we've already scheduled an ack above, so there is no need to schedule 
@@ -266,6 +282,8 @@ class ConnectionPacketHandler {
     
     /**
      * Process the acks in a received packet, and adjust our window and RTT
+     * @param isNew was it a new packet? false for ack-only
+     * @param choke did we get a choke in the packet?
      * @return are we congested?
      */
     private boolean ack(Connection con, long ackThrough, long nacks[], Packet packet, boolean isNew, boolean choke) {
@@ -354,16 +372,25 @@ class ConnectionPacketHandler {
         return rv;
     }
     
-    /** @return are we congested? */
+    /**
+     * This either does nothing or increases the window, it never decreases it.
+     * Decreasing is done in Connection.ResendPacketEvent.retransmit()
+     *
+     * @param isNew was it a new packet? false for ack-only
+     * @param sequenceNum 0 for ack-only
+     * @param choke did we get a choke in the packet?
+     * @return are we congested?
+     */
     private boolean adjustWindow(Connection con, boolean isNew, long sequenceNum, int numResends, int acked, boolean choke) {
-        boolean congested = false;
+        boolean congested;
-        if ( (!isNew) && (sequenceNum > 0) ) {
+        if (choke || (!isNew && sequenceNum > 0) || con.isChoked()) {
             if (_log.shouldLog(Log.DEBUG))
                 _log.debug("Congestion occurred on the sending side. Not adjusting window "+con);
-
             congested = true;
-        } 
+        } else {
-        
+            congested = false;
+        }
+
         long lowest = con.getHighestAckedThrough();
         // RFC 2581
         // Why wait until we get a whole cwin to start updating the window?

apps/streaming/java/src/net/i2p/client/streaming/impl/MessageHandler.java

@@ -14,7 +14,7 @@ import net.i2p.util.Log;
 /**
  * Receive raw information from the I2PSession and turn it into
  * Packets, if we can.
- *<p>
+ *<p>
  * I2PSession -&gt; MessageHandler -&gt; PacketHandler -&gt; ConnectionPacketHandler -&gt; MessageInputStream
  */
 class MessageHandler implements I2PSessionMuxedListener {

apps/streaming/java/src/net/i2p/client/streaming/impl/MessageInputStream.java

@@ -16,9 +16,9 @@ import net.i2p.util.Log;
 /**
  * Stream that can be given messages out of order 
  * yet present them in order.
- *<p>
+ *<p>
  * I2PSession -&gt; MessageHandler -&gt; PacketHandler -&gt; ConnectionPacketHandler -&gt; MessageInputStream
- *&lt;p&gt;
+ *<p>
  * This buffers unlimited data via messageReceived() -
  * limiting / blocking is done in ConnectionPacketHandler.receivePacket().
  *
@@ -102,6 +102,7 @@ class MessageInputStream extends InputStream {
     
     /**
      *  Determine if this packet will fit in our buffering limits.
+     *  Always returns true for zero payloadSize.
      *
      *  @return true if we have room. If false, do not call messageReceived()
      *  @since 0.9.20 moved from ConnectionPacketHandler.receivePacket() so it can all be under one lock,

apps/streaming/java/src/net/i2p/client/streaming/impl/MessageOutputStream.java

@@ -16,7 +16,7 @@ import net.i2p.util.SimpleTimer2;
  * A stream that we can shove data into that fires off those bytes
  * on flush or when the buffer is full.  It also blocks according
  * to the data receiver's needs.
- *<p>
+ *<p>
  * MessageOutputStream -&gt; ConnectionDataReceiver -&gt; Connection -&gt; PacketQueue -&gt; I2PSession
  */
 class MessageOutputStream extends OutputStream {

apps/streaming/java/src/net/i2p/client/streaming/impl/Packet.java

@@ -164,6 +164,8 @@ class Packet {
 
     public static final int DEFAULT_MAX_SIZE = 32*1024;
     protected static final int MAX_DELAY_REQUEST = 65535;
+    public static final int MIN_DELAY_CHOKE = 60001;
+    public static final int SEND_DELAY_CHOKE = 61000;
 
     /**
      *  Does no initialization.

apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java

@@ -11,7 +11,7 @@ import net.i2p.util.Log;
 /**
  * receive a packet and dispatch it correctly to the connection specified,
  * the server socket, or queue a reply RST packet.
- *<p>
+ *<p>
  * I2PSession -&gt; MessageHandler -&gt; PacketHandler -&gt; ConnectionPacketHandler -&gt; MessageInputStream
  */
 class PacketHandler {

apps/streaming/java/src/net/i2p/client/streaming/impl/PacketLocal.java

@@ -111,18 +111,6 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus {
                          FLAG_ECHO);
     }
     
-    /** last minute update of ack fields, just before write/sign  */
-    public void prepare() {
-        if (_connection != null)
-            _connection.getInputStream().updateAcks(this);
-        int numSends = _numSends.get();
-        if (numSends > 0) {
-            // so we can debug to differentiate resends
-            setOptionalDelay(numSends * 1000);
-            setFlag(FLAG_DELAY_REQUESTED);
-        }
-    }
-    
     public long getCreatedOn() { return _createdOn; }
     public long getLifetime() { return _context.clock().now() - _createdOn; }
     public void incrementSends() { 

apps/streaming/java/src/net/i2p/client/streaming/impl/PacketQueue.java

@@ -23,7 +23,7 @@ import net.i2p.util.SimpleTimer2;
  * Well, thats the theory at least... in practice we just
  * send them immediately with no blocking, since the 
  * mode=bestEffort doesnt block in the SDK.
- *<p>
+ *<p>
  * MessageOutputStream -&gt; ConnectionDataReceiver -&gt; Connection -&gt; PacketQueue -&gt; I2PSession
  */
 class PacketQueue implements SendMessageStatusListener, Closeable {
@@ -64,7 +64,8 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
     }
     
     /**
-     * Add a new packet to be sent out ASAP
+     * Add a new packet to be sent out ASAP.
+     * This updates the acks.
      *
      * keys and tags disabled since dropped in I2PSession
      * @return true if sent
@@ -72,8 +73,6 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
     public boolean enqueue(PacketLocal packet) {
         if (_dead)
             return false;
-        // this updates the ack/nack field
-        packet.prepare();
         
         //SessionKey keyUsed = packet.getKeyUsed();
         //if (keyUsed == null)
@@ -87,6 +86,12 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
                 _log.debug("Not resending " + packet);
             return false;
         }
+
+        Connection con = packet.getConnection();
+        if (con != null) {
+            // this updates the ack/nack fields
+            con.getInputStream().updateAcks(packet);
+        }
     
         ByteArray ba = _cache.acquire();
         byte buf[] = ba.getData();
@@ -96,14 +101,14 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
         boolean sent = false;
         try {
             int size = 0;
-            long beforeWrite = System.currentTimeMillis();
+            //long beforeWrite = System.currentTimeMillis();
             if (packet.shouldSign())
                 size = packet.writeSignedPacket(buf, 0);
             else
                 size = packet.writePacket(buf, 0);
-            long writeTime = System.currentTimeMillis() - beforeWrite;
+            //long writeTime = System.currentTimeMillis() - beforeWrite;
-            if ( (writeTime > 1000) && (_log.shouldLog(Log.WARN)) )
+            //if ( (writeTime > 1000) && (_log.shouldLog(Log.WARN)) )
-                _log.warn("took " + writeTime + "ms to write the packet: " + packet);
+            //    _log.warn("took " + writeTime + "ms to write the packet: " + packet);
 
             // last chance to short circuit...
             if (packet.getAckTime() > 0) return false;
@@ -121,7 +126,6 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
                 options.setDate(expires);
             boolean listenForStatus = false;
             if (packet.isFlagSet(FLAGS_INITIAL_TAGS)) {
-                Connection con = packet.getConnection();
                 if (con != null) {
                     if (con.isInbound())
                         options.setSendLeaseSet(false);
@@ -141,7 +145,6 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
                 options.setTagsToSend(FINAL_TAGS_TO_SEND);
                 options.setTagThreshold(FINAL_TAG_THRESHOLD);
             } else {
-                Connection con = packet.getConnection();
                 if (con != null) {
                     if (con.isInbound() && con.getLifetime() < 2*60*1000)
                         options.setSendLeaseSet(false);
@@ -157,7 +160,7 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
                 long id = session.sendMessage(packet.getTo(), buf, 0, size,
                                  I2PSession.PROTO_STREAMING, packet.getLocalPort(), packet.getRemotePort(),
                                  options, this);
-                _messageStatusMap.put(Long.valueOf(id), packet.getConnection());
+                _messageStatusMap.put(Long.valueOf(id), con);
                 sent = true;
             } else {
                 sent = session.sendMessage(packet.getTo(), buf, 0, size,
@@ -173,7 +176,6 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
             if (packet.getNumSends() > 1)
                 _context.statManager().addRateData("stream.con.sendDuplicateSize", size, packet.getLifetime());
             
-            Connection con = packet.getConnection();
             if (con != null) {
                 con.incrementBytesSent(size);
                 if (packet.getNumSends() > 1)
@@ -189,17 +191,15 @@ class PacketQueue implements SendMessageStatusListener, Closeable {
         if (!sent) {
             if (_log.shouldLog(Log.WARN))
                 _log.warn("Send failed for " + packet);
-            Connection c = packet.getConnection();
+            if (con != null) // handle race on b0rk
-            if (c != null) // handle race on b0rk
+                con.disconnect(false);
-                c.disconnect(false);
         } else {
             //packet.setKeyUsed(keyUsed);
             //packet.setTagsSent(tagsSent);
             packet.incrementSends();
-            Connection c = packet.getConnection();
+            if (con != null && _log.shouldDebug()) {
-            if (c != null && _log.shouldDebug()) {
+                String suffix = "wsize " + con.getOptions().getWindowSize() + " rto " + con.getOptions().getRTO();
-                String suffix = "wsize " + c.getOptions().getWindowSize() + " rto " + c.getOptions().getRTO();
+                con.getConnectionManager().getPacketHandler().displayPacket(packet, "SEND", suffix);
-                c.getConnectionManager().getPacketHandler().displayPacket(packet, "SEND", suffix);
             }
             if (I2PSocketManagerFull.pcapWriter != null &&
                 _context.getBooleanProperty(I2PSocketManagerFull.PROP_PCAP))

history.txt

@@ -1,4 +1,21 @@
+2017-02-09 zzz
+ * Streaming: Fix optional delay and choking (tickets #1046, 1939)
+
+2017-02-08 zzz
+ * I2CP: Return local delivery failure on queue overflow (ticket #1939)
+
+2017-02-05 zzz
+ * Console: Consolidate timer threads (ticket #1068)
+ * NTCP: Don't write to an inbound connection before
+   fully established, causing NPE (ticket #996)
+ * Streaming:
+   - Don't always send optional delay (ticket #1046)
+   - Don't hard fail on expired message error (ticket #1748)
+
 2017-02-04 zzz
+ * HTTP proxies:
+   - Pass through relative referer URIs, convert same-origin
+     absolute referer URIs to relative (ticket #1862)
  * NTP: Enable IPv6 support (ticket #1896)
 
 2017-01-30 zzz

router/java/src/net/i2p/router/RouterVersion.java

@@ -18,7 +18,7 @@ public class RouterVersion {
     /** deprecated */
     public final static String ID = "Monotone";
     public final static String VERSION = CoreVersion.VERSION;
-    public final static long BUILD = 5;
+    public final static long BUILD = 6;
 
     /** for example "-test" */
     public final static String EXTRA = "";