Streaming:

- Make I2PSocketFull.close() nonblocking; it will now cause any user-side
     writes blocked in I/O (Connection.packetSendChoke()) to throw
     an exception (tickets #629, #1041)
   - Don't ignore InterruptedExceptions; throw InterruptedIOException
   - Back out static disconnect exception
   - MessageInputStream locking fixes
   - Cleanups
 I2PSnark:
   - Close socket before closing output stream to avoid blocking in
     Peer.disconnect(), and prevent Peer.disconnect() loop

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

@@ -168,7 +168,7 @@ class Connection {
      * @return true if the packet should be sent, false for a fatal error
      *         will return false after 5 minutes even if timeoutMs is <= 0.
      */
-    boolean packetSendChoke(long timeoutMs) {
+    public boolean packetSendChoke(long timeoutMs) throws IOException, InterruptedException {
         long start = _context.clock().now();
         long writeExpire = start + timeoutMs;  // only used if timeoutMs > 0
         boolean started = false;
@@ -183,8 +183,10 @@ class Connection {
                 // no need to wait until the other side has ACKed us before sending the first few wsize
                 // packets through
 		// Incorrect assumption, the constructor defaults _connected to true --Sponge
-                    if (!_connected.get())
-                       return false;
+                if (!_connected.get())
+                    throw new IOException("disconnected");
+                if (_outputStream.getClosed())
+                    throw new IOException("output stream closed");
                 started = true;
                 // Try to keep things moving even during NACKs and retransmissions...
                 // Limit unacked packets to the window
@@ -207,12 +209,24 @@ class Connection {
                         if (_log.shouldLog(Log.DEBUG))
                             _log.debug("Outbound window is full (" + unacked + "/" + wsz + "/" 
                                        + _activeResends + "), waiting " + timeLeft);
-                        try { _outboundPackets.wait(Math.min(timeLeft,250l)); } catch (InterruptedException ie) { if (_log.shouldLog(Log.DEBUG)) _log.debug("InterruptedException while Outbound window is full (" + _outboundPackets.size() + "/" + _activeResends +")"); return false;}
+                        try {
+                            _outboundPackets.wait(Math.min(timeLeft,250l));
+                        } catch (InterruptedException ie) {
+                            if (_log.shouldLog(Log.DEBUG))
+                                _log.debug("InterruptedException while Outbound window is full (" + _outboundPackets.size() + "/" + _activeResends +")");
+                            throw ie;
+                        }
                     } else {
                         //if (_log.shouldLog(Log.DEBUG))
                         //    _log.debug("Outbound window is full (" + _outboundPackets.size() + "/" + _activeResends 
                         //               + "), waiting indefinitely");
-                        try { _outboundPackets.wait(250); } catch (InterruptedException ie) {if (_log.shouldLog(Log.DEBUG)) _log.debug("InterruptedException while Outbound window is full (" + _outboundPackets.size() + "/" + _activeResends + ")"); return false;} //10*1000
+                        try {
+                            _outboundPackets.wait(250);
+                        } catch (InterruptedException ie) {
+                            if (_log.shouldLog(Log.DEBUG))
+                                _log.debug("InterruptedException while Outbound window is full (" + _outboundPackets.size() + "/" + _activeResends + ")");
+                            throw ie;
+                        } //10*1000
                     }
                 } else {
                     _context.statManager().addRateData("stream.chokeSizeEnd", _outboundPackets.size(), _context.clock().now() - start);
@@ -222,6 +236,9 @@ class Connection {
         }
     }
 
+    /**
+     *  Notify all threads waiting in packetSendChoke()
+     */
     void windowAdjusted() {
         synchronized (_outboundPackets) {
             _outboundPackets.notifyAll();
@@ -710,8 +727,6 @@ class Connection {
         }
     }
     
-    private static final IOException DISCON_IOE = new IOException("disconnected!");
-
     /**
      *  Must be called when we are done with this connection.
      *  Final disconnect. Remove from conn manager.
@@ -729,7 +744,7 @@ class Connection {
         _outputStream.destroy();
         _receiver.destroy();
         _activityTimer.cancel();
-        _inputStream.streamErrorOccurred(DISCON_IOE);
+        _inputStream.streamErrorOccurred(new IOException("disconnected"));
         
         if (_log.shouldLog(Log.INFO))
             _log.info("Connection disconnect complete: "

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

@@ -4,6 +4,7 @@ import java.util.List;
 
 import net.i2p.I2PAppContext;
 import net.i2p.I2PException;
+import net.i2p.data.DataHelper;
 import net.i2p.data.Destination;
 import net.i2p.util.Log;
 import net.i2p.util.SimpleScheduler;
@@ -200,8 +201,8 @@ class ConnectionPacketHandler {
                 final long lastSendTime = con.getLastSendTime();
                 
                 if (_log.shouldLog(Log.WARN))
-                    _log.warn(String.format("%s congestion.. dup packet %s now %d ackDelay %d lastSend %d",
-                                    con, packet, now, ackDelay, lastSendTime));
+                    _log.warn(String.format("%s congestion.. dup packet %s ackDelay %d lastSend %d ago",
+                                    con, packet, now, ackDelay, DataHelper.formatDuration(now - lastSendTime)));
                 
                 final long nextSendTime = lastSendTime + ackDelay;
                 if (nextSendTime <= now) {

apps/streaming/java/src/net/i2p/client/streaming/I2PSocketFull.java

@@ -33,6 +33,15 @@ class I2PSocketFull implements I2PSocket {
             _remotePeer = _localPeer = null;
     }
     
+    /**
+     *  Closes this socket.
+     *
+     *  Nonblocking as of 0.9.9:
+     *  Any thread currently blocked in an I/O operation upon this socket will throw an IOException.
+     *  Once a socket has been closed, it is not available for further networking use
+     *  (i.e. can't be reconnected or rebound). A new socket needs to be created.
+     *  Closing this socket will also close the socket's InputStream and OutputStream.
+     */
     public void close() throws IOException {
         if (!_closed.compareAndSet(false,true)) {
             // log a trace to find out why
@@ -42,15 +51,13 @@ class I2PSocketFull implements I2PSocket {
         Connection c = _connection;
         if (c == null) return;
         if (c.getIsConnected()) {
-            OutputStream out = c.getOutputStream();
-            try {
-                out.close();
-            } catch (IOException ioe) {
-                // ignore any write error, as we want to keep on and kill the
-                // con (thanks Complication!)
-            }
             MessageInputStream in = c.getInputStream();
             in.close();
+            MessageOutputStream out = c.getOutputStream();
+            out.closeInternal();
+            // this will cause any thread waiting in Connection.packetSendChoke()
+            // to throw an IOE
+            c.windowAdjusted();
         } else {
             //throw new IOException("Not connected");
         }

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

@@ -2,6 +2,7 @@ package net.i2p.client.streaming;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -287,12 +288,12 @@ class MessageInputStream extends InputStream {
     
     @Override
     public int read(byte target[], int offset, int length) throws IOException {
-        if (_locallyClosed) throw new IOException("Already locally closed");
-        throwAnyError();
         long expiration = -1;
         if (_readTimeout > 0)
             expiration = _readTimeout + System.currentTimeMillis();
         synchronized (_dataLock) {
+            if (_locallyClosed) throw new IOException("Already locally closed");
+            throwAnyError();
             for (int i = 0; i < length; i++) {
                 if ( (_readyDataBlocks.isEmpty()) && (i == 0) ) {
                     // ok, we havent found anything, so lets block until we get 
@@ -312,7 +313,13 @@ class MessageInputStream extends InputStream {
                                 if (_log.shouldLog(Log.DEBUG))
                                     _log.debug("read(...," + offset+", " + length+ ")[" + i 
                                                + ") with no timeout: " + toString());
-                                try { _dataLock.wait(); } catch (InterruptedException ie) { }
+                                try {
+                                    _dataLock.wait();
+                                } catch (InterruptedException ie) {
+                                    IOException ioe2 = new InterruptedIOException("Interrupted read");
+                                    ioe2.initCause(ie);
+                                    throw ioe2;
+                                }
                                 if (_log.shouldLog(Log.DEBUG))
                                     _log.debug("read(...," + offset+", " + length+ ")[" + i 
                                                + ") with no timeout complete: " + toString());
@@ -321,7 +328,13 @@ class MessageInputStream extends InputStream {
                                 if (_log.shouldLog(Log.DEBUG))
                                     _log.debug("read(...," + offset+", " + length+ ")[" + i 
                                                + ") with timeout: " + _readTimeout + ": " + toString());
-                                try { _dataLock.wait(_readTimeout); } catch (InterruptedException ie) { }
+                                try {
+                                    _dataLock.wait(_readTimeout);
+                                } catch (InterruptedException ie) {
+                                    IOException ioe2 = new InterruptedIOException("Interrupted read");
+                                    ioe2.initCause(ie);
+                                    throw ioe2;
+                                }
                                 if (_log.shouldLog(Log.DEBUG))
                                     _log.debug("read(...," + offset+", " + length+ ")[" + i 
                                                + ") with timeout complete: " + _readTimeout + ": " + toString());
@@ -382,10 +395,10 @@ class MessageInputStream extends InputStream {
     
     @Override
     public int available() throws IOException {
-        if (_locallyClosed) throw new IOException("Already closed");
-        throwAnyError();
         int numBytes = 0;
         synchronized (_dataLock) {
+            if (_locallyClosed) throw new IOException("Already closed");
+            throwAnyError();
             for (int i = 0; i < _readyDataBlocks.size(); i++) {
                 ByteArray cur = _readyDataBlocks.get(i);
                 if (i == 0)
@@ -467,14 +480,15 @@ class MessageInputStream extends InputStream {
      *
      */
     void streamErrorOccurred(IOException ioe) {
-        if (_streamError == null)
-            _streamError = ioe;
-        _locallyClosed = true;
         synchronized (_dataLock) {
+            if (_streamError == null)
+                _streamError = ioe;
+            _locallyClosed = true;
             _dataLock.notifyAll();
         }
     }
     
+    /** Caller must lock _dataLock */
     private void throwAnyError() throws IOException {
         IOException ioe = _streamError;
         if (ioe != null) {

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

@@ -89,15 +89,17 @@ class MessageOutputStream extends OutputStream {
 
         _writeTimeout = ms; 
     }
+
     public int getWriteTimeout() { return _writeTimeout; }
+
     public void setBufferSize(int size) { _nextBufferSize = size; }
     
-	@Override
+    @Override
     public void write(byte b[]) throws IOException {
         write(b, 0, b.length);
     }
     
-	@Override
+    @Override
     public void write(byte b[], int off, int len) throws IOException {
         if (_closed.get()) throw new IOException("Already closed");
         if (_log.shouldLog(Log.DEBUG))
@@ -150,7 +152,13 @@ class MessageOutputStream extends OutputStream {
                 // ok, we've actually added a new packet - lets wait until
                 // its accepted into the queue before moving on (so that we 
                 // dont fill our buffer instantly)
-                ws.waitForAccept(_writeTimeout);
+                try {
+                    ws.waitForAccept(_writeTimeout);
+                } catch (InterruptedException ie) {
+                    IOException ioe2 = new InterruptedIOException("Interrupted write");
+                    ioe2.initCause(ie);
+                    throw ioe2;
+                }
                 if (!ws.writeAccepted()) {
                     if (_log.shouldLog(Log.WARN))
                         _log.warn("Write not accepted of " + ws);
@@ -296,7 +304,7 @@ class MessageOutputStream extends OutputStream {
      *
      * @throws IOException if the write fails
      */
-	@Override
+    @Override
     public void flush() throws IOException {
      /* @throws InterruptedIOException if the write times out
       * Documented here, but doesn't belong in the javadoc. 
@@ -343,14 +351,20 @@ class MessageOutputStream extends OutputStream {
         // Wait a loooooong time, until we have the ACK
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("before waiting " + _writeTimeout + "ms for completion of " + ws);
-        if (_closed.get() && 
-            ( (_writeTimeout > Connection.DISCONNECT_TIMEOUT) ||
-              (_writeTimeout <= 0) ) )
-            ws.waitForCompletion(Connection.DISCONNECT_TIMEOUT);
-        else if ( (_writeTimeout <= 0) || (_writeTimeout > Connection.DISCONNECT_TIMEOUT) )
-            ws.waitForCompletion(Connection.DISCONNECT_TIMEOUT);
-        else
-            ws.waitForCompletion(_writeTimeout);
+        try {
+            if (_closed.get() && 
+                ( (_writeTimeout > Connection.DISCONNECT_TIMEOUT) ||
+                  (_writeTimeout <= 0) ) )
+                ws.waitForCompletion(Connection.DISCONNECT_TIMEOUT);
+            else if ( (_writeTimeout <= 0) || (_writeTimeout > Connection.DISCONNECT_TIMEOUT) )
+                ws.waitForCompletion(Connection.DISCONNECT_TIMEOUT);
+            else
+                ws.waitForCompletion(_writeTimeout);
+        } catch (InterruptedException ie) {
+            IOException ioe2 = new InterruptedIOException("Interrupted flush");
+            ioe2.initCause(ie);
+            throw ioe2;
+        }
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("after waiting " + _writeTimeout + "ms for completion of " + ws);
         if (ws.writeFailed() && (_writeTimeout > 0) )
@@ -466,6 +480,7 @@ class MessageOutputStream extends OutputStream {
     void flushAvailable(DataReceiver target) throws IOException {
         flushAvailable(target, true);
     }
+
     void flushAvailable(DataReceiver target, boolean blocking) throws IOException {
         WriteStatus ws = null;
         long before = System.currentTimeMillis();
@@ -487,7 +502,13 @@ class MessageOutputStream extends OutputStream {
             _log.debug("Took " + (afterBuild-before) + "ms to build a packet?  " + ws);
         
         if (blocking && ws != null) {
-            ws.waitForAccept(_writeTimeout);
+            try {
+                ws.waitForAccept(_writeTimeout);
+            } catch (InterruptedException ie) {
+                IOException ioe2 = new InterruptedIOException("Interrupted flush");
+                ioe2.initCause(ie);
+                throw ioe2;
+            }
             if (ws.writeFailed())
                 throw new IOException("Flush available failed");
             else if (!ws.writeAccepted())
@@ -526,7 +547,7 @@ class MessageOutputStream extends OutputStream {
          * Success means an ACK FROM THE FAR END.
          * @param maxWaitMs -1 = forever
          */
-        public void waitForCompletion(int maxWaitMs);
+        public void waitForCompletion(int maxWaitMs) throws IOException, InterruptedException;
 
         /** 
          * Wait until the data written is accepted into the outbound pool,
@@ -534,9 +555,9 @@ class MessageOutputStream extends OutputStream {
          * which we throttle rather than accept arbitrary data and queue 
          * @param maxWaitMs -1 = forever
          */
-        public void waitForAccept(int maxWaitMs);
+        public void waitForAccept(int maxWaitMs) throws IOException, InterruptedException;
 
-        /** the write was accepted.  aka did the socket not close? */
+        /** Was the write was accepted.  aka did the socket not close? */
         public boolean writeAccepted();
         /** did the write fail?  */
         public boolean writeFailed();

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

@@ -254,6 +254,7 @@ class Packet {
      * @return Delay before resending a packet in seconds.
      */
     public int getResendDelay() { return _resendDelay; }
+
     /**
      *  Unused.
      *  Broken before release 0.7.8
@@ -267,17 +268,22 @@ class Packet {
      * @return the payload of the message, null if none.
      */
     public ByteArray getPayload() { return _payload; }
+
     public void setPayload(ByteArray payload) { 
         _payload = payload; 
         if ( (payload != null) && (payload.getValid() > MAX_PAYLOAD_SIZE) )
             throw new IllegalArgumentException("Too large payload: " + payload.getValid());
     }
+
     public int getPayloadSize() {
         return (_payload == null ? 0 : _payload.getValid());
     }
+
+    /** does nothing right now */
     public void releasePayload() {
         //_payload = null;
     }
+
     public ByteArray acquirePayload() {
         _payload = new ByteArray(new byte[Packet.MAX_PAYLOAD_SIZE]);
         return _payload;
@@ -288,13 +294,16 @@ class Packet {
      * @return true if set, false if not.
      */
     public boolean isFlagSet(int flag) { return 0 != (_flags & flag); }
+
     public void setFlag(int flag) { _flags |= flag; }
+
     public void setFlag(int flag, boolean set) { 
         if (set)
             _flags |= flag; 
         else
             _flags &= ~flag;
     }
+
     public void setFlags(int flags) { _flags = flags; } 
 
     /** the signature on the packet (only included if the flag for it is set)

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

@@ -140,6 +140,7 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus {
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("Cancelled! " + toString(), new Exception("cancelled"));
     }
+
     public SimpleTimer2.TimedEvent getResendEvent() { return _resendEvent; }
     
     /** how long after packet creation was it acked?
@@ -230,59 +231,71 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus {
         return buf;
     }
     
+    ////// begin WriteStatus methods
+
     /**
      * Blocks until outbound window is not full. See Connection.packetSendChoke().
      * @param maxWaitMs MessageOutputStream is the only caller, generally with -1
      */
-    public void waitForAccept(int maxWaitMs) {
+    public void waitForAccept(int maxWaitMs) throws IOException, InterruptedException {
         long before = _context.clock().now();
-        int queued = _connection.getUnackedPacketsSent();
-        int window = _connection.getOptions().getWindowSize();
-        boolean accepted = _connection.packetSendChoke(maxWaitMs);
-        long after = _context.clock().now();
-        if (accepted) {
-            _acceptedOn = after;
-        } else {
-            _acceptedOn = -1;
-            releasePayload();
+        boolean accepted = false;
+        try {
+            // throws IOE or IE
+            accepted = _connection.packetSendChoke(maxWaitMs);
+        } finally {
+            if (accepted) {
+                _acceptedOn = _context.clock().now();
+            } else {
+                _acceptedOn = -1;
+                releasePayload();
+            }
+            if ( (_acceptedOn - before > 1000) && (_log.shouldLog(Log.DEBUG)) )  {
+                int queued = _connection.getUnackedPacketsSent();
+                int window = _connection.getOptions().getWindowSize();
+                int afterQueued = _connection.getUnackedPacketsSent();
+                _log.debug("Took " + (_acceptedOn - before) + "ms to get " 
+                           + (accepted ? "accepted" : "rejected")
+                           + (_cancelledOn > 0 ? " and CANCELLED" : "")
+                           + ", queued behind " + queued +" with a window size of " + window 
+                           + ", finally accepted with " + afterQueued + " queued: " 
+                           + toString());
+            }
         }
-        int afterQueued = _connection.getUnackedPacketsSent();
-        if ( (after - before > 1000) && (_log.shouldLog(Log.DEBUG)) )
-            _log.debug("Took " + (after-before) + "ms to get " 
-                       + (accepted ? "accepted" : "rejected")
-                       + (_cancelledOn > 0 ? " and CANCELLED" : "")
-                       + ", queued behind " + queued +" with a window size of " + window 
-                       + ", finally accepted with " + afterQueued + " queued: " 
-                       + toString());
     }
     
     /** block until the packet is acked from the far end */
-    public void waitForCompletion(int maxWaitMs) {
+    public void waitForCompletion(int maxWaitMs) throws IOException, InterruptedException {
         long expiration = _context.clock().now()+maxWaitMs;
-        while (true) {
-            long timeRemaining = expiration - _context.clock().now();
-            if ( (timeRemaining <= 0) && (maxWaitMs > 0) ) break;
-            try {
+        try {
+            while (true) {
+                long timeRemaining = expiration - _context.clock().now();
+                if ( (timeRemaining <= 0) && (maxWaitMs > 0) ) break;
                 synchronized (this) {
                     if (_ackOn > 0) break;
-                    if (_cancelledOn > 0) break;
-                    if (!_connection.getIsConnected()) break;
+                    if (!_connection.getIsConnected())
+                        throw new IOException("disconnected");
+                    if (_cancelledOn > 0)
+                        throw new IOException("cancelled");
                     if (timeRemaining > 60*1000)
                         timeRemaining = 60*1000;
                     else if (timeRemaining <= 0)
                         timeRemaining = 10*1000;
                     wait(timeRemaining);
                 }
-            } catch (InterruptedException ie) { }//{ break; }
+            }
+        } finally {
+            if (!writeSuccessful())
+                releasePayload();
         }
-        if (!writeSuccessful())
-            releasePayload();
     }
     
     public synchronized boolean writeAccepted() { return _acceptedOn > 0 && _cancelledOn <= 0; }
     public synchronized boolean writeFailed() { return _cancelledOn > 0; }
     public synchronized boolean writeSuccessful() { return _ackOn > 0 && _cancelledOn <= 0; }
 
+    ////// end WriteStatus methods
+
     /** Generate a pcap/tcpdump-compatible format,
      *  so we can use standard debugging tools.
      */