From 4092f618985cb895a3877966907cf3188a9d5fe5 Mon Sep 17 00:00:00 2001 From: zzz Date: Fri, 29 Jun 2012 14:53:53 +0000 Subject: [PATCH] * Streaming: - Allow at least 3 packets and up to half the window to be active resends instead of just 1, to reduce stall time after a packet drop - Increase fast retransmit threshold back to 3 to reduce retransmissions - Don't fast retransmit if we recently retransmitted it already - Allow double the window as long as gaps are less than the window - Don't set the MSS in a resent packet (saves 2 bytes) - Remove redundant calls to updateAcks() - Update activity timer when resending a packet - Reset unackedPacketsReceived counter at all places where acks are sent so it isn't wrong - Fix some places where the activeResends count could become wrong - Prevent storm of CLOSE packets - Never resend the whole packet in ackImmediately(), just send an ack - Cancel flusher timer in MessageOutputStream when closed - Move some createRateStats to ConnectionManager to reduce repeated calls - Cleanups, javadocs, logging, volatile, finals --- .../net/i2p/client/streaming/Connection.java | 254 ++++++++++-------- .../streaming/ConnectionDataReceiver.java | 47 ++-- .../client/streaming/ConnectionManager.java | 11 +- .../client/streaming/ConnectionOptions.java | 13 +- .../streaming/ConnectionPacketHandler.java | 45 ++-- .../i2p/client/streaming/MessageHandler.java | 3 +- .../client/streaming/MessageInputStream.java | 28 +- .../client/streaming/MessageOutputStream.java | 81 +++--- .../src/net/i2p/client/streaming/Packet.java | 13 +- .../i2p/client/streaming/PacketHandler.java | 13 +- .../net/i2p/client/streaming/PacketLocal.java | 25 +- .../net/i2p/client/streaming/PacketQueue.java | 15 +- .../client/streaming/SchedulerChooser.java | 10 +- .../client/streaming/SchedulerClosing.java | 17 +- .../i2p/client/streaming/SchedulerImpl.java | 5 + .../net/i2p/client/streaming/TCBShare.java | 4 +- history.txt | 21 ++ .../src/net/i2p/router/RouterVersion.java | 2 +- 18 files changed, 363 insertions(+), 244 deletions(-) diff --git a/apps/streaming/java/src/net/i2p/client/streaming/Connection.java b/apps/streaming/java/src/net/i2p/client/streaming/Connection.java index 8db219c60..298a1c21f 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/Connection.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/Connection.java @@ -34,19 +34,19 @@ class Connection { private boolean _resetReceived; private boolean _resetSent; private long _resetSentOn; - private boolean _connected; + private volatile boolean _connected; private boolean _hardDisconnected; private final MessageInputStream _inputStream; private final MessageOutputStream _outputStream; private final SchedulerChooser _chooser; - private long _nextSendTime; + private volatile long _nextSendTime; private long _ackedPackets; private final long _createdOn; private long _closeSentOn; private long _closeReceivedOn; private int _unackedPacketsReceived; private long _congestionWindowEnd; - private long _highestAckedThrough; + private volatile long _highestAckedThrough; private boolean _isInbound; private boolean _updatedShareOpts; /** Packet ID (Long) to PacketLocal for sent but unacked packets */ @@ -60,11 +60,11 @@ class Connection { private String _connectionError; private long _disconnectScheduledOn; private long _lastReceivedOn; - private ActivityTimer _activityTimer; + private final ActivityTimer _activityTimer; /** window size when we last saw congestion */ private int _lastCongestionSeenAt; private long _lastCongestionTime; - private long _lastCongestionHighestUnacked; + private volatile long _lastCongestionHighestUnacked; private boolean _ackSinceCongestion; /** Notify this on connection (or connection failure) */ private final Object _connectLock; @@ -96,7 +96,9 @@ class Connection { } ****/ - /** */ + /** + * @param opts may be null + */ public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, SimpleTimer2 timer, PacketQueue queue, ConnectionPacketHandler handler, ConnectionOptions opts) { @@ -138,10 +140,7 @@ class Connection { _resetSentOn = -1; _connectionEvent = new ConEvent(); _randomWait = _context.random().nextInt(10*1000); // just do this once to reduce usage - _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.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.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.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 }); + // all createRateStats in ConnectionManager if (_log.shouldLog(Log.INFO)) _log.info("New connection created with options: " + _options); } @@ -169,7 +168,6 @@ class Connection { * will return false after 5 minutes even if timeoutMs is <= 0. */ boolean packetSendChoke(long timeoutMs) { - // if (false) return true; // <--- what the fuck?? long start = _context.clock().now(); long writeExpire = start + timeoutMs; // only used if timeoutMs > 0 boolean started = false; @@ -187,19 +185,26 @@ class Connection { if (!_connected) return false; started = true; - if ( (_outboundPackets.size() >= _options.getWindowSize()) || (_activeResends > 0) || - (_lastSendId.get() - _highestAckedThrough > _options.getWindowSize()) ) { + // Try to keep things moving even during NACKs and retransmissions... + // Limit unacked packets to the window + // Limit active resends to half the window + // 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 || + _activeResends >= (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 of " + _outboundPackets.size() - + " with " + _activeResends + " active resends" + _log.info("Outbound window is full " + 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 (" + _outboundPackets.size() + "/" + _options.getWindowSize() + "/" + _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;} } else { @@ -223,6 +228,12 @@ class Connection { void ackImmediately() { PacketLocal packet = null; +/*** why would we do this? + was it to force a congestion indication at the other end? + an expensive way to do that... + One big user was via SchedulerClosing to resend a CLOSE packet, + but why do that either... + synchronized (_outboundPackets) { if (!_outboundPackets.isEmpty()) { // ordered, so pick the lowest to retransmit @@ -239,6 +250,7 @@ class Connection { } ResendPacketEvent evt = (ResendPacketEvent)packet.getResendEvent(); if (evt != null) { + // fixme should we set a flag and reschedule instead? or synch? boolean sent = evt.retransmit(false); if (sent) { if (_log.shouldLog(Log.DEBUG)) @@ -251,7 +263,9 @@ class Connection { } } } +***/ // if we don't have anything to retransmit, send a small ack + // this calls sendPacket() below packet = _receiver.send(null, 0, 0); if (_log.shouldLog(Log.DEBUG)) _log.debug("sending new ack: " + packet); @@ -281,11 +295,15 @@ class Connection { reply.setReceiveStreamId(_receiveStreamId); reply.setOptionalFrom(_connectionManager.getSession().getMyDestination()); // this just sends the packet - no retries or whatnot - _outboundQueue.enqueue(reply); + if (_outboundQueue.enqueue(reply)) { + _unackedPacketsReceived = 0; + _lastSendTime = _context.clock().now(); + resetActivityTimer(); + } } /** - * Flush any data that we can + * Flush any data that we can. Non-blocking. */ void sendAvailable() { // this grabs the data, builds a packet, and queues it up via sendPacket @@ -301,7 +319,6 @@ class Connection { if (packet == null) return; setNextSendTime(-1); - _unackedPacketsReceived = 0; if (_options.getRequireFullySigned()) { packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED); packet.setFlag(Packet.FLAG_SIGNATURE_REQUESTED); @@ -328,8 +345,8 @@ class Connection { (packet.getSequenceNum() % 8 == 0)) { packet.setOptionalDelay(0); packet.setFlag(Packet.FLAG_DELAY_REQUESTED); - if (_log.shouldLog(Log.DEBUG)) - _log.debug("Requesting no ack delay for packet " + packet); + //if (_log.shouldLog(Log.DEBUG)) + // _log.debug("Requesting no ack delay for packet " + packet); } else { // This is somewhat of a waste of time, unless the RTT < 4000, // since the other end limits it to getSendAckDelay() @@ -358,10 +375,12 @@ class Connection { // warning, getStatLog() can be null //_context.statManager().getStatLog().addData(Packet.toId(_sendStreamId), "stream.rtt", _options.getRTT(), _options.getWindowSize()); - _lastSendTime = _context.clock().now(); - _outboundQueue.enqueue(packet); - resetActivityTimer(); - + if (_outboundQueue.enqueue(packet)) { + _unackedPacketsReceived = 0; + _lastSendTime = _context.clock().now(); + resetActivityTimer(); + } + /* if (ackOnly) { // ACK only, don't schedule this packet for retries @@ -397,6 +416,7 @@ class Connection { * @return List of packets acked or null */ List ackPackets(long ackThrough, long nacks[]) { + // FIXME synch this part too? if (ackThrough < _highestAckedThrough) { // dupack which won't tell us anything } else { @@ -415,16 +435,17 @@ class Connection { List acked = null; synchronized (_outboundPackets) { - for (Iterator iter = _outboundPackets.keySet().iterator(); iter.hasNext(); ) { - Long id = iter.next(); - if (id.longValue() <= ackThrough) { + for (Map.Entry e : _outboundPackets.entrySet()) { + long id = e.getKey().longValue(); + if (id <= ackThrough) { boolean nacked = false; if (nacks != null) { // linear search since its probably really tiny for (int i = 0; i < nacks.length; i++) { - if (nacks[i] == id.longValue()) { + if (nacks[i] == id) { nacked = true; - PacketLocal nackedPacket = _outboundPackets.get(id); + PacketLocal nackedPacket = e.getValue(); + // this will do a fast retransmit if appropriate nackedPacket.incrementNACKs(); break; // NACKed } @@ -433,11 +454,27 @@ class Connection { if (!nacked) { // aka ACKed if (acked == null) acked = new ArrayList(1); - PacketLocal ackedPacket = _outboundPackets.get(id); + PacketLocal ackedPacket = e.getValue(); ackedPacket.ackReceived(); acked.add(ackedPacket); } } else { + // TODO + // we do not currently do an "implicit nack" of the packets higher + // than ackThrough, so those will not be fast retransmitted + // we could incrementNACK them here... but we may need to set the fastRettransmit + // threshold back to 3 for that. + // this will do a fast retransmit if appropriate + // This doesn't work because every packet has an ACK in it, so we hit the + // FAST_TRANSMIT threshold in a heartbeat and retransmit everything, + // even with the threshold at 3. (we never set the NO_ACK field in the header) + // Also, we may need to track that we + // have the same ackThrough for 3 or 4 consecutive times. + // See https://secure.wikimedia.org/wikipedia/en/wiki/Fast_retransmit + //if (_log.shouldLog(Log.INFO)) + // _log.info("ACK thru " + ackThrough + " implicitly NACKs " + id); + //PacketLocal nackedPacket = e.getValue(); + //nackedPacket.incrementNACKs(); break; // _outboundPackets is ordered } } @@ -465,31 +502,33 @@ class Connection { return acked; } - private long _occurredTime; - private long _occurredEventCount; + //private long _occurredTime; + //private long _occurredEventCount; + void eventOccurred() { - long now = System.currentTimeMillis(); + //long now = System.currentTimeMillis(); TaskScheduler sched = _chooser.getScheduler(this); - now = now - now % 1000; - if (_occurredTime == now) { - _occurredEventCount++; - } else { - _occurredTime = now; - if ( (_occurredEventCount > 1000) && (_log.shouldLog(Log.WARN)) ) { - _log.warn("More than 1000 events (" + _occurredEventCount + ") in a second on " - + toString() + ": scheduler = " + sched); - } - _occurredEventCount = 0; - } + //now = now - now % 1000; + //if (_occurredTime == now) { + // _occurredEventCount++; + //} else { + // _occurredTime = now; + // if ( (_occurredEventCount > 1000) && (_log.shouldLog(Log.WARN)) ) { + // _log.warn("More than 1000 events (" + _occurredEventCount + ") in a second on " + // + toString() + ": scheduler = " + sched); + // } + // _occurredEventCount = 0; + //} long before = System.currentTimeMillis(); sched.eventOccurred(this); long elapsed = System.currentTimeMillis() - before; - if ( (elapsed > 1000) && (_log.shouldLog(Log.WARN)) ) - _log.warn("Took " + elapsed + "ms to pump through " + sched); + // 250 and warn for debugging + if ( (elapsed > 250) && (_log.shouldLog(Log.WARN)) ) + _log.warn("Took " + elapsed + "ms to pump through " + sched + " on " + toString()); } void resetReceived() { @@ -498,12 +537,8 @@ class Connection { SimpleScheduler.getInstance().addEvent(new DisconnectEvent(), DISCONNECT_TIMEOUT); } _resetReceived = true; - MessageOutputStream mos = _outputStream; - MessageInputStream mis = _inputStream; - if (mos != null) - mos.streamErrorOccurred(new IOException("Reset received")); - if (mis != null) - mis.streamErrorOccurred(new IOException("Reset received")); + _outputStream.streamErrorOccurred(new IOException("Reset received")); + _inputStream.streamErrorOccurred(new IOException("Reset received")); _connectionError = "Connection reset"; synchronized (_connectLock) { _connectLock.notifyAll(); } } @@ -556,15 +591,10 @@ class Connection { s.destroy2(); _socket = null; } - if (_outputStream != null) - _outputStream.destroy(); - if (_receiver != null) - _receiver.destroy(); - if (_activityTimer != null) - _activityTimer.cancel(); - //_activityTimer = null; - if (_inputStream != null) - _inputStream.streamErrorOccurred(new IOException("disconnected!")); + _outputStream.destroy(); + _receiver.destroy(); + _activityTimer.cancel(); + _inputStream.streamErrorOccurred(new IOException("disconnected!")); if (_disconnectScheduledOn < 0) { _disconnectScheduledOn = _context.clock().now(); @@ -656,11 +686,7 @@ class Connection { * @return Last time we sent data */ public long getLastSendTime() { return _lastSendTime; } - /** Set the time we sent data. - * @param when The time we sent data - */ - public void setLastSendTime(long when) { _lastSendTime = when; } - + /** What was the last packet Id sent to the peer? * @return The last sent packet ID */ @@ -795,10 +821,9 @@ class Connection { public long getCongestionWindowEnd() { return _congestionWindowEnd; } public void setCongestionWindowEnd(long endMsg) { _congestionWindowEnd = endMsg; } + /** @return the highest outbound packet we have recieved an ack for */ public long getHighestAckedThrough() { return _highestAckedThrough; } - /** @deprecated unused */ - public void setHighestAckedThrough(long msgNum) { _highestAckedThrough = msgNum; } public long getLastActivityOn() { return (_lastSendTime > _lastReceivedOn ? _lastSendTime : _lastReceivedOn); @@ -878,17 +903,12 @@ class Connection { } private void resetActivityTimer() { - if (_options.getInactivityTimeout() <= 0) { - if (_log.shouldLog(Log.DEBUG)) - _log.debug("Resetting the inactivity timer, but its gone!", new Exception("where did it go?")); - return; - } - if (_activityTimer == null) { - if (_log.shouldLog(Log.DEBUG)) - _log.debug("Resetting the inactivity timer, but its gone!", new Exception("where did it go?")); - return; - } long howLong = _options.getInactivityTimeout(); + if (howLong <= 0) { + if (_log.shouldLog(Log.DEBUG)) + _log.debug("Resetting the inactivity timer, but its gone!", new Exception("where did it go?")); + return; + } howLong += _randomWait; // randomize it a bit, so both sides don't do it at once //if (_log.shouldLog(Log.DEBUG)) // _log.debug("Resetting the inactivity timer to " + howLong); @@ -983,12 +1003,12 @@ class Connection { } /** stream that the local peer receives data on - * @return the inbound message stream + * @return the inbound message stream, non-null */ public MessageInputStream getInputStream() { return _inputStream; } /** stream that the local peer sends data to the remote peer on - * @return the outbound message stream + * @return the outbound message stream, non-null */ public MessageOutputStream getOutputStream() { return _outputStream; } @@ -1032,12 +1052,10 @@ class Connection { */ buf.append("unacked in: ").append(getUnackedPacketsReceived()); int missing = 0; - if (_inputStream != null) { - long nacks[] = _inputStream.getNacks(); - if (nacks != null) { - missing = nacks.length; - buf.append(" [").append(missing).append(" missing]"); - } + long nacks[] = _inputStream.getNacks(); + if (nacks != null) { + missing = nacks.length; + buf.append(" [").append(missing).append(" missing]"); } if (getResetSent()) @@ -1053,8 +1071,7 @@ class Connection { if (getCloseReceivedOn() > 0) buf.append(" close received ").append(DataHelper.formatDuration(_context.clock().now() - getCloseReceivedOn())).append(" ago"); buf.append(" sent: ").append(1 + _lastSendId.get()); - if (_inputStream != null) - buf.append(" rcvd: ").append(1 + _inputStream.getHighestBlockId() - missing); + buf.append(" rcvd: ").append(1 + _inputStream.getHighestBlockId() - missing); buf.append(" maxWin ").append(getOptions().getMaxWindowSize()); buf.append(" MTU ").append(getOptions().getMaxMessageSize()); @@ -1086,14 +1103,15 @@ class Connection { * there are other packets in flight. 3 takes forever, let's try 2. * */ - static final int FAST_RETRANSMIT_THRESHOLD = 2; + static final int FAST_RETRANSMIT_THRESHOLD = 3; /** * Coordinate the resends of a given packet */ class ResendPacketEvent extends SimpleTimer2.TimedEvent { - private PacketLocal _packet; + private final PacketLocal _packet; private long _nextSendTime; + public ResendPacketEvent(PacketLocal packet, long delay) { super(_timer); _packet = packet; @@ -1111,6 +1129,8 @@ class Connection { * we have to use forceReschedule() instead of schedule() below, * to prevent duplicates in the timer queue. * + * don't synchronize this, deadlock with ackPackets->ackReceived->SimpleTimer2.cancel + * * @param penalize true if this retransmission is caused by a timeout, false if we * are just sending this packet instead of an ACK * @return true if the packet was sent, false if it was not @@ -1131,7 +1151,12 @@ class Connection { boolean resend = false; boolean isLowest = false; synchronized (_outboundPackets) { - if (_packet.getSequenceNum() == _highestAckedThrough + 1) + // allow appx. half the window to be "lowest" and be active resends, minimum of 3 + // Note: we should really pick the N lowest, not the lowest one + N more who + // happen to get here next, as the timers get out-of-order esp. after fast retx + if (_packet.getSequenceNum() == _highestAckedThrough + 1 || + _packet.getNumSends() > 1 || + _activeResends < Math.max(3, (_options.getWindowSize() + 1) / 2)) isLowest = true; if (_outboundPackets.containsKey(Long.valueOf(_packet.getSequenceNum()))) resend = true; @@ -1145,24 +1170,28 @@ class Connection { // BUG? seq# = 0, activeResends = 0, loop forever - why? // also seen with seq# > 0. Is the _activeResends count reliable? if (_log.shouldLog(Log.INFO)) - _log.info("Delaying resend of " + _packet + " as there are " - + _activeResends + " active resends already in play"); - forceReschedule(1000); - _nextSendTime = 1000 + _context.clock().now(); + _log.info("Delaying resend of " + _packet + " with " + + _activeResends + " active resend, " + + _outboundPackets.size() + " unacked, window size = " + _options.getWindowSize()); + forceReschedule(1333); + _nextSendTime = 1333 + _context.clock().now(); return false; } + // It's the lowest, or it's fast retransmit time. Resend the packet. + if (fastRetransmit) _context.statManager().addRateData("stream.fastRetransmit", _packet.getLifetime(), _packet.getLifetime()); // revamp various fields, in case we need to ack more, etc - _inputStream.updateAcks(_packet); + // updateAcks done in enqueue() + //_inputStream.updateAcks(_packet); int choke = getOptions().getChoke(); _packet.setOptionalDelay(choke); if (choke > 0) _packet.setFlag(Packet.FLAG_DELAY_REQUESTED); // this seems unnecessary to send the MSS again: - _packet.setOptionalMaxSize(getOptions().getMaxMessageSize()); + //_packet.setOptionalMaxSize(getOptions().getMaxMessageSize()); // bugfix release 0.7.8, we weren't dividing by 1000 _packet.setResendDelay(getOptions().getResendDelay() / 1000); if (_packet.getReceiveStreamId() <= 0) @@ -1186,7 +1215,7 @@ class Connection { getOptions().setWindowSize(newWindowSize); if (_log.shouldLog(Log.WARN)) - _log.warn("Congestion resending packet " + _packet.getSequenceNum() + ": new windowSize " + newWindowSize + _log.warn("Congestion, resending packet " + _packet.getSequenceNum() + " (new windowSize " + newWindowSize + "/" + getOptions().getWindowSize() + ") for " + Connection.this.toString()); windowAdjusted(); @@ -1195,10 +1224,6 @@ class Connection { int numSends = _packet.getNumSends() + 1; - if (numSends == 2) { - // first resend for this packet - _activeResends++; - } // in case things really suck, the other side may have lost thier // session tags (e.g. they restarted), so jump back to ElGamal. @@ -1225,27 +1250,34 @@ class Connection { // set this before enqueue() as it passes it on to the router _nextSendTime = timeout + _context.clock().now(); - if (_log.shouldLog(Log.INFO)) - _log.info("Resend packet " + _packet + " time " + numSends + + if (_outboundQueue.enqueue(_packet)) { + // first resend for this packet ? + if (numSends == 2) + _activeResends++; + if (_log.shouldLog(Log.INFO)) + _log.info("Resent packet " + + (fastRetransmit ? "(fast) " : "(timeout) ") + + _packet + + " next resend in " + timeout + "ms" + " activeResends: " + _activeResends + " (wsize " + newWindowSize + " lifetime " + (_context.clock().now() - _packet.getCreatedOn()) + "ms)"); - _outboundQueue.enqueue(_packet); - _lastSendTime = _context.clock().now(); + _unackedPacketsReceived = 0; + _lastSendTime = _context.clock().now(); + // timer reset added 0.9.1 + resetActivityTimer(); + } - if (_log.shouldLog(Log.DEBUG)) - _log.debug("Scheduling resend in " + timeout + "ms for " + _packet); forceReschedule(timeout); } - // acked during resending (... or somethin') + // acked during resending (... or somethin') ???????????? if ( (_packet.getAckTime() > 0) && (_packet.getNumSends() > 1) ) { _activeResends--; synchronized (_outboundPackets) { _outboundPackets.notifyAll(); } - return true; } return true; diff --git a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionDataReceiver.java b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionDataReceiver.java index 900fb9626..8f2a40f46 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionDataReceiver.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionDataReceiver.java @@ -10,7 +10,12 @@ import net.i2p.util.Log; * do NOT block, but they also do not necessary imply immediate * delivery, or even the generation of a new packet. This class * is the only one that builds useful outbound Packet objects. - * + *

+ * MessageOutputStream -> ConnectionDataReceiver -> Connection -> PacketQueue -> I2PSession + *

+ * There's one of these per MessageOutputStream. + * It stores no state. It sends everything to the Connection unless + * the Connection is closed, */ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { private final I2PAppContext _context; @@ -82,7 +87,7 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { if (_log.shouldLog(Log.INFO) && !doSend) _log.info("writeData called: size="+size + " doSend=" + doSend + " unackedReceived: " + con.getUnackedPacketsReceived() - + " con: " + con, new Exception("write called by")); + + " con: " + con /* , new Exception("write called by") */ ); if (doSend) { PacketLocal packet = send(buf, off, size); @@ -111,6 +116,7 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { public PacketLocal send(byte buf[], int off, int size) { return send(buf, off, size, false); } + /** * @param buf data to be sent - may be null * @param off offset into the buffer to start writing from @@ -120,22 +126,20 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { * @return the packet sent */ public PacketLocal send(byte buf[], int off, int size, boolean forceIncrement) { - Connection con = _connection; - //if (con == null) return null; - long before = System.currentTimeMillis(); - PacketLocal packet = buildPacket(con, buf, off, size, forceIncrement); - long built = System.currentTimeMillis(); - con.sendPacket(packet); - long sent = System.currentTimeMillis(); + //long before = System.currentTimeMillis(); + PacketLocal packet = buildPacket(buf, off, size, forceIncrement); + //long built = System.currentTimeMillis(); + _connection.sendPacket(packet); + //long sent = System.currentTimeMillis(); - if ( (built-before > 5*1000) && (_log.shouldLog(Log.WARN)) ) - _log.warn("wtf, took " + (built-before) + "ms to build a packet: " + packet); - if ( (sent-built> 5*1000) && (_log.shouldLog(Log.WARN)) ) - _log.warn("wtf, took " + (sent-built) + "ms to send a packet: " + packet); + //if ( (built-before > 5*1000) && (_log.shouldLog(Log.WARN)) ) + // _log.warn("wtf, took " + (built-before) + "ms to build a packet: " + packet); + //if ( (sent-built> 5*1000) && (_log.shouldLog(Log.WARN)) ) + // _log.warn("wtf, took " + (sent-built) + "ms to send a packet: " + packet); return packet; } - private boolean isAckOnly(Connection con, int size) { + private static boolean isAckOnly(Connection con, int size) { boolean ackOnly = ( (size <= 0) && // no data (con.getLastSendId() >= 0) && // not a SYN ( (!con.getOutputStream().getClosed()) || // not a CLOSE @@ -144,7 +148,16 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { return ackOnly; } - private PacketLocal buildPacket(Connection con, byte buf[], int off, int size, boolean forceIncrement) { + /** + * @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) + * @param forceIncrement even if the buffer is empty, increment the packetId + * so we get an ACK back + * @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 isFirst = (con.getAckedPackets() <= 0) && (con.getUnackedPacketsSent() <= 0); @@ -164,7 +177,8 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { packet.setSendStreamId(con.getSendStreamId()); packet.setReceiveStreamId(con.getReceiveStreamId()); - con.getInputStream().updateAcks(packet); + // 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(); packet.setOptionalDelay(choke); @@ -195,6 +209,7 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver { // don't set the closed flag if this is a plain ACK and there are outstanding // packets sent, otherwise the other side could receive the CLOSE prematurely, // since this ACK could arrive before the unacked payload message. + // TODO if the only unacked packet is the CLOSE packet and it didn't have any data... if (con.getOutputStream().getClosed() && ( (size > 0) || (con.getUnackedPacketsSent() <= 0) || (packet.getSequenceNum() > 0) ) ) { packet.setFlag(Packet.FLAG_CLOSE); diff --git a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionManager.java b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionManager.java index b2d9a201a..739a42017 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionManager.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionManager.java @@ -75,6 +75,7 @@ class ConnectionManager { /** Socket timeout for accept() */ _soTimeout = -1; + // Stats for this class _context.statManager().createRateStat("stream.con.lifetimeMessagesSent", "How many messages do we send on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 }); _context.statManager().createRateStat("stream.con.lifetimeMessagesReceived", "How many messages do we receive on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 }); _context.statManager().createRateStat("stream.con.lifetimeBytesSent", "How many bytes do we send on a stream?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 }); @@ -85,6 +86,14 @@ class ConnectionManager { _context.statManager().createRateStat("stream.con.lifetimeCongestionSeenAt", "When was the last congestion seen at when a stream closes?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 }); _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.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.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.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 }); + // 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.sendDuplicateSize", "Size of a message resent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 }); } Connection getConnectionByInboundId(long id) { @@ -420,13 +429,11 @@ class ConnectionManager { if (removed) { _context.statManager().addRateData("stream.con.lifetimeMessagesSent", 1+con.getLastSendId(), con.getLifetime()); MessageInputStream stream = con.getInputStream(); - if (stream != null) { long rcvd = 1 + stream.getHighestBlockId(); long nacks[] = stream.getNacks(); if (nacks != null) rcvd -= nacks.length; _context.statManager().addRateData("stream.con.lifetimeMessagesReceived", rcvd, con.getLifetime()); - } _context.statManager().addRateData("stream.con.lifetimeBytesSent", con.getLifetimeBytesSent(), con.getLifetime()); _context.statManager().addRateData("stream.con.lifetimeBytesReceived", con.getLifetimeBytesReceived(), con.getLifetime()); _context.statManager().addRateData("stream.con.lifetimeDupMessagesSent", con.getLifetimeDupMessagesSent(), con.getLifetime()); diff --git a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java index ff3c545df..42a2fc95a 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java @@ -98,6 +98,10 @@ class ConnectionOptions extends I2PSocketOptionsImpl { public static final int DEFAULT_INITIAL_ACK_DELAY = 2*1000; static final int MIN_WINDOW_SIZE = 1; private static final boolean DEFAULT_ANSWER_PINGS = true; + private static final int DEFAULT_INACTIVITY_TIMEOUT = 90*1000; + private static final int DEFAULT_INACTIVITY_ACTION = INACTIVITY_ACTION_SEND; + + /** * If PROTO is enforced, we cannot communicate with destinations earlier than version 0.7.1. * @since 0.9.1 @@ -302,6 +306,7 @@ class ConnectionOptions extends I2PSocketOptionsImpl { _maxTotalConnsPerDay = opts.getMaxTotalConnsPerDay(); } + /** called by super's constructor */ @Override protected void init(Properties opts) { super.init(opts); @@ -318,8 +323,8 @@ class ConnectionOptions extends I2PSocketOptionsImpl { setMaxResends(getInt(opts, PROP_MAX_RESENDS, DEFAULT_MAX_SENDS)); // handled in super() //setWriteTimeout(getInt(opts, PROP_WRITE_TIMEOUT, -1)); - setInactivityTimeout(getInt(opts, PROP_INACTIVITY_TIMEOUT, 90*1000)); - setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, INACTIVITY_ACTION_SEND)); + setInactivityTimeout(getInt(opts, PROP_INACTIVITY_TIMEOUT, DEFAULT_INACTIVITY_TIMEOUT)); + setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, DEFAULT_INACTIVITY_ACTION)); setInboundBufferSize(getMaxMessageSize() * (Connection.MAX_WINDOW_SIZE + 2)); setCongestionAvoidanceGrowthRateFactor(getInt(opts, PROP_CONGESTION_AVOIDANCE_GROWTH_RATE_FACTOR, 1)); setSlowStartGrowthRateFactor(getInt(opts, PROP_SLOW_START_GROWTH_RATE_FACTOR, 1)); @@ -367,9 +372,9 @@ class ConnectionOptions extends I2PSocketOptionsImpl { //if (opts.containsKey(PROP_WRITE_TIMEOUT)) // setWriteTimeout(getInt(opts, PROP_WRITE_TIMEOUT, -1)); if (opts.containsKey(PROP_INACTIVITY_TIMEOUT)) - setInactivityTimeout(getInt(opts, PROP_INACTIVITY_TIMEOUT, 90*1000)); + setInactivityTimeout(getInt(opts, PROP_INACTIVITY_TIMEOUT, DEFAULT_INACTIVITY_TIMEOUT)); if (opts.containsKey(PROP_INACTIVITY_ACTION)) - setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, INACTIVITY_ACTION_SEND)); + setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, DEFAULT_INACTIVITY_ACTION)); setInboundBufferSize(getMaxMessageSize() * (Connection.MAX_WINDOW_SIZE + 2)); if (opts.contains(PROP_CONGESTION_AVOIDANCE_GROWTH_RATE_FACTOR)) setCongestionAvoidanceGrowthRateFactor(getInt(opts, PROP_CONGESTION_AVOIDANCE_GROWTH_RATE_FACTOR, 2)); diff --git a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java index 53ebb17e1..720c38fdf 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java @@ -12,7 +12,13 @@ import net.i2p.util.SimpleTimer; /** * Receive a packet for a particular connection - placing the data onto the * queue, marking packets as acked, updating various fields, etc. - * + *

+ * I2PSession -> MessageHandler -> PacketHandler -> ConnectionPacketHandler -> MessageInputStream + *

+ * One of these is instantiated per-Destination + * (i.e. per-ConnectionManager, not per-Connection). + * It doesn't store any state. + */ class ConnectionPacketHandler { private final I2PAppContext _context; @@ -94,19 +100,24 @@ class ConnectionPacketHandler { } } - long ready = con.getInputStream().getHighestReadyBockId(); - int available = con.getOptions().getInboundBufferSize() - con.getInputStream().getTotalReadySize(); - int allowedBlocks = available/con.getOptions().getMaxMessageSize(); - if ( (packet.getPayloadSize() > 0) && (packet.getSequenceNum() > ready + allowedBlocks) ) { - if (_log.shouldLog(Log.WARN)) - _log.warn("Inbound buffer exceeded on connection " + con + " (" - + ready + "/"+ (ready+allowedBlocks) + "/" + available - + ": dropping " + packet); - ack(con, packet.getAckThrough(), packet.getNacks(), null, false, choke); - con.getOptions().setChoke(61*1000); - packet.releasePayload(); - con.ackImmediately(); - return; + if (packet.getPayloadSize() > 0) { + // Here, for the purposes of calculating whether the input stream is full, + // we assume all the not-ready blocks are the max message size. + // This prevents us from getting DoSed by accepting unlimited out-of-order small messages + long ready = con.getInputStream().getHighestReadyBockId(); + int available = con.getOptions().getInboundBufferSize() - con.getInputStream().getTotalReadySize(); + int allowedBlocks = available/con.getOptions().getMaxMessageSize(); + if (packet.getSequenceNum() > ready + allowedBlocks) { + if (_log.shouldLog(Log.WARN)) + _log.warn("Inbound buffer exceeded on connection " + con + " (" + + ready + "/"+ (ready+allowedBlocks) + "/" + available + + ": dropping " + packet); + ack(con, packet.getAckThrough(), packet.getNacks(), null, false, choke); + con.getOptions().setChoke(61*1000); + packet.releasePayload(); + con.ackImmediately(); + return; + } } con.getOptions().setChoke(0); @@ -513,12 +524,14 @@ class ConnectionPacketHandler { } private class AckDup implements SimpleTimer.TimedEvent { - private long _created; - private Connection _con; + private final long _created; + private final Connection _con; + public AckDup(Connection con) { _created = _context.clock().now(); _con = con; } + public void timeReached() { if (_con.getLastSendTime() <= _created) { if (_con.getResetReceived() || _con.getResetSent()) { diff --git a/apps/streaming/java/src/net/i2p/client/streaming/MessageHandler.java b/apps/streaming/java/src/net/i2p/client/streaming/MessageHandler.java index d9ca691b4..4aba07d31 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/MessageHandler.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/MessageHandler.java @@ -13,7 +13,8 @@ import net.i2p.util.Log; /** * Receive raw information from the I2PSession and turn it into * Packets, if we can. - * + *

+ * I2PSession -> MessageHandler -> PacketHandler -> ConnectionPacketHandler -> MessageInputStream */ class MessageHandler implements I2PSessionMuxedListener { private final ConnectionManager _manager; diff --git a/apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java b/apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java index 956eb9b51..eb168d110 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java @@ -16,6 +16,11 @@ import net.i2p.util.Log; /** * Stream that can be given messages out of order * yet present them in order. + *

+ * I2PSession -> MessageHandler -> PacketHandler -> ConnectionPacketHandler -> MessageInputStream + *

+ * This buffers unlimited data via messageReceived() - + * limiting / blocking is done in ConnectionPacketHandler.receivePacket(). * */ class MessageInputStream extends InputStream { @@ -113,6 +118,9 @@ class MessageInputStream extends InputStream { } } + /** + * Adds the ack-through and nack fields to a packet we are building for transmission + */ public void updateAcks(PacketLocal packet) { synchronized (_dataLock) { packet.setAckThrough(_highestBlockId); @@ -126,6 +134,7 @@ class MessageInputStream extends InputStream { * * @return block IDs greater than the highest ready block ID, or null if there aren't any. */ +/*** public long[] getOutOfOrderBlocks() { long blocks[] = null; synchronized (_dataLock) { @@ -140,15 +149,18 @@ class MessageInputStream extends InputStream { Arrays.sort(blocks); return blocks; } +***/ /** how many blocks have we received that we still have holes before? * @return Count of blocks received that still have holes */ +/*** public int getOutOfOrderBlockCount() { synchronized (_dataLock) { return _notYetReadyBlocks.size(); } } +***/ /** * how long a read() call should block (if less than 0, block indefinitely, @@ -205,9 +217,9 @@ class MessageInputStream extends InputStream { * @return true if this is a new packet, false if it is a dup */ public boolean messageReceived(long messageId, ByteArray payload) { + if (_log.shouldLog(Log.DEBUG)) + _log.debug("received " + messageId + " with " + (payload != null ? payload.getValid()+"" : "no payload")); synchronized (_dataLock) { - if (_log.shouldLog(Log.DEBUG)) - _log.debug("received " + messageId + " with " + (payload != null ? payload.getValid()+"" : "no payload")); if (messageId <= _highestReadyBlockId) { if (_log.shouldLog(Log.DEBUG)) _log.debug("ignoring dup message " + messageId); @@ -237,7 +249,6 @@ class MessageInputStream extends InputStream { cur++; _highestReadyBlockId++; } - _dataLock.notifyAll(); } else { if (_log.shouldLog(Log.DEBUG)) _log.debug("message is out of order: " + messageId); @@ -245,8 +256,8 @@ class MessageInputStream extends InputStream { _notYetReadyBlocks.put(Long.valueOf(messageId), new ByteArray(null)); else _notYetReadyBlocks.put(Long.valueOf(messageId), payload); - _dataLock.notifyAll(); } + _dataLock.notifyAll(); } return true; } @@ -278,7 +289,7 @@ class MessageInputStream extends InputStream { while (_readyDataBlocks.isEmpty()) { if (_locallyClosed) - throw new IOException("Already closed, you wanker"); + throw new IOException("Already closed"); if ( (_notYetReadyBlocks.isEmpty()) && (_closeReceived) ) { if (_log.shouldLog(Log.INFO)) @@ -360,7 +371,7 @@ class MessageInputStream extends InputStream { @Override public int available() throws IOException { - if (_locallyClosed) throw new IOException("Already closed, you wanker"); + if (_locallyClosed) throw new IOException("Already closed"); throwAnyError(); int numBytes = 0; synchronized (_dataLock) { @@ -384,6 +395,7 @@ class MessageInputStream extends InputStream { * * @return Count of bytes waiting to be read */ +/*** public int getTotalQueuedSize() { synchronized (_dataLock) { if (_locallyClosed) return 0; @@ -401,7 +413,11 @@ class MessageInputStream extends InputStream { return numBytes; } } +***/ + /** + * Same as available() but doesn't throw IOE + */ public int getTotalReadySize() { synchronized (_dataLock) { if (_locallyClosed) return 0; diff --git a/apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java b/apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java index 3a4cd4206..1470bd502 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java @@ -14,6 +14,8 @@ 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. + *

+ * MessageOutputStream -> ConnectionDataReceiver -> Connection -> PacketQueue -> I2PSession */ class MessageOutputStream extends OutputStream { private final I2PAppContext _context; @@ -21,17 +23,17 @@ class MessageOutputStream extends OutputStream { private byte _buf[]; private int _valid; private final Object _dataLock; - private DataReceiver _dataReceiver; + private final DataReceiver _dataReceiver; private IOException _streamError; - private boolean _closed; + private volatile boolean _closed; private long _written; private int _writeTimeout; private ByteCache _dataCache; private final Flusher _flusher; private long _lastFlushed; - private long _lastBuffered; + private volatile long _lastBuffered; /** if we enqueue data but don't flush it in this period, flush it passively */ - private int _passiveFlushDelay; + private final int _passiveFlushDelay; /** * if we are changing the buffer size during operation, set this to the new * buffer size, and next time we are flushing, update the _buf array to the new @@ -39,9 +41,9 @@ class MessageOutputStream extends OutputStream { */ private volatile int _nextBufferSize; // rate calc helpers - private long _sendPeriodBeginTime; - private long _sendPeriodBytes; - private int _sendBps; + //private long _sendPeriodBeginTime; + //private long _sendPeriodBytes; + //private int _sendBps; /** * Since this is less than i2ptunnel's i2p.streaming.connectDelay default of 1000, @@ -73,11 +75,11 @@ class MessageOutputStream extends OutputStream { _writeTimeout = -1; _passiveFlushDelay = passiveFlushDelay; _nextBufferSize = -1; - _sendPeriodBeginTime = ctx.clock().now(); - _context.statManager().createRateStat("stream.sendBps", "How fast we pump data through the stream", "Stream", new long[] { 60*1000, 5*60*1000, 60*60*1000 }); + //_sendPeriodBeginTime = ctx.clock().now(); + //_context.statManager().createRateStat("stream.sendBps", "How fast we pump data through the stream", "Stream", new long[] { 60*1000, 5*60*1000, 60*60*1000 }); _flusher = new Flusher(timer); - if (_log.shouldLog(Log.DEBUG)) - _log.debug("MessageOutputStream created"); + //if (_log.shouldLog(Log.DEBUG)) + // _log.debug("MessageOutputStream created"); } public void setWriteTimeout(int ms) { @@ -131,15 +133,9 @@ class MessageOutputStream extends OutputStream { remaining -= toWrite; cur += toWrite; _valid = _buf.length; - // avoid NPE from race with destroy() - DataReceiver rcvr = _dataReceiver; - if (rcvr == null) { - throwAnyError(); - return; - } if (_log.shouldLog(Log.INFO)) _log.info("write() direct valid = " + _valid); - ws = rcvr.writeData(_buf, 0, _valid); + ws = _dataReceiver.writeData(_buf, 0, _valid); _written += _valid; _valid = 0; throwAnyError(); @@ -167,17 +163,18 @@ class MessageOutputStream extends OutputStream { _log.info("After waitForAccept of " + ws); } } else { - if (_log.shouldLog(Log.INFO)) - _log.info("Queued " + len + " without sending to the receiver"); + if (_log.shouldLog(Log.DEBUG)) + _log.debug("Queued " + len + " without sending to the receiver"); } } long elapsed = _context.clock().now() - begin; if ( (elapsed > 10*1000) && (_log.shouldLog(Log.INFO)) ) _log.info("wtf, took " + elapsed + "ms to write to the stream?", new Exception("foo")); throwAnyError(); - updateBps(len); + //updateBps(len); } +/**** private void updateBps(int len) { long now = _context.clock().now(); int periods = (int)Math.floor((now - _sendPeriodBeginTime) / 1000d); @@ -191,7 +188,9 @@ class MessageOutputStream extends OutputStream { _sendPeriodBytes += len; } } +****/ + /** */ public void write(int b) throws IOException { write(new byte[] { (byte)b }, 0, 1); throwAnyError(); @@ -240,14 +239,15 @@ class MessageOutputStream extends OutputStream { _enqueued = true; } public void timeReached() { + if (_closed) + return; _enqueued = false; - DataReceiver rec = _dataReceiver; long timeLeft = (_lastBuffered + _passiveFlushDelay - _context.clock().now()); if (_log.shouldLog(Log.DEBUG)) _log.debug("flusher time reached: left = " + timeLeft); if (timeLeft > 0) enqueue(); - else if ( (rec != null) && (rec.writeInProcess()) ) + else if (_dataReceiver.writeInProcess()) enqueue(); // don't passive flush if there is a write being done (unacked outbound) else doFlush(); @@ -261,10 +261,8 @@ class MessageOutputStream extends OutputStream { if ( (_valid > 0) && (flushTime <= _context.clock().now()) ) { if (_log.shouldLog(Log.INFO)) _log.info("doFlush() valid = " + _valid); - // avoid NPE from race with destroy() - DataReceiver rcvr = _dataReceiver; - if ( (_buf != null) && (rcvr != null) ) { - ws = rcvr.writeData(_buf, 0, _valid); + if (_buf != null) { + ws = _dataReceiver.writeData(_buf, 0, _valid); _written += _valid; _valid = 0; _lastFlushed = _context.clock().now(); @@ -317,25 +315,18 @@ class MessageOutputStream extends OutputStream { if (_log.shouldLog(Log.INFO) && _valid > 0) _log.info("flush() valid = " + _valid); - // avoid NPE from race with destroy() - DataReceiver rcvr = _dataReceiver; synchronized (_dataLock) { if (_buf == null) { _dataLock.notifyAll(); throw new IOException("closed (buffer went away)"); } - if (rcvr == null) { - _dataLock.notifyAll(); - throwAnyError(); - return; - } // if valid == 0 return ??? - no, this could flush a CLOSE packet too. // Yes, flush here, inside the data lock, and do all the waitForCompletion() stuff below // (disabled) if (!wait_for_accept_only) { - ws = rcvr.writeData(_buf, 0, _valid); + ws = _dataReceiver.writeData(_buf, 0, _valid); _written += _valid; _valid = 0; locked_updateBufferSize(); @@ -347,7 +338,7 @@ class MessageOutputStream extends OutputStream { // Skip all the waitForCompletion() stuff below, which is insanity, as of 0.8.1 // must do this outside the data lock if (wait_for_accept_only) { - flushAvailable(rcvr, true); + flushAvailable(_dataReceiver, true); return; } @@ -387,6 +378,7 @@ class MessageOutputStream extends OutputStream { } // setting _closed before flush() will force flush() to send a CLOSE packet _closed = true; + _flusher.cancel(); // In 0.8.1 we rewrote flush() to only wait for accept into the window, // not "completion" (i.e. ack from the far end). @@ -415,10 +407,11 @@ class MessageOutputStream extends OutputStream { /** * nonblocking close - - * Use outside of this package is deprecated, should be made package local + * Only for use inside package */ public void closeInternal() { _closed = true; + _flusher.cancel(); if (_streamError == null) _streamError = new IOException("Closed internally"); clearData(true); @@ -429,12 +422,10 @@ class MessageOutputStream extends OutputStream { if (_log.shouldLog(Log.INFO) && _valid > 0) _log.info("clearData() valid = " + _valid); - // avoid NPE from race with destroy() - DataReceiver rcvr = _dataReceiver; synchronized (_dataLock) { // flush any data, but don't wait for it - if ( (rcvr != null) && (_valid > 0) && shouldFlush) - rcvr.writeData(_buf, 0, _valid); + if (_valid > 0 && shouldFlush) + _dataReceiver.writeData(_buf, 0, _valid); _written += _valid; _valid = 0; @@ -503,15 +494,15 @@ class MessageOutputStream extends OutputStream { throw new InterruptedIOException("Flush available timed out (" + _writeTimeout + "ms)"); } long afterAccept = System.currentTimeMillis(); - if ( (afterAccept - afterBuild > 1000) && (_log.shouldLog(Log.DEBUG)) ) - _log.debug("Took " + (afterAccept-afterBuild) + "ms to accept a packet? " + ws); + if ( (afterAccept - afterBuild > 1000) && (_log.shouldLog(Log.INFO)) ) + _log.info("Took " + (afterAccept-afterBuild) + "ms to accept a packet? " + ws); return; } void destroy() { - _dataReceiver = null; + _closed = true; + _flusher.cancel(); synchronized (_dataLock) { - _closed = true; _dataLock.notifyAll(); } } diff --git a/apps/streaming/java/src/net/i2p/client/streaming/Packet.java b/apps/streaming/java/src/net/i2p/client/streaming/Packet.java index bb3e723b6..a79f67cb7 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/Packet.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/Packet.java @@ -395,6 +395,7 @@ class Packet { DataHelper.toLong(buffer, cur, 4, _ackThrough > 0 ? _ackThrough : 0); cur += 4; if (_nacks != null) { + // if max win is ever > 255, limit to 255 DataHelper.toLong(buffer, cur, 1, _nacks.length); cur++; for (int i = 0; i < _nacks.length; i++) { @@ -461,7 +462,7 @@ class Packet { * @return How large the current packet would be * @throws IllegalStateException */ - public int writtenSize() throws IllegalStateException { + private int writtenSize() { int size = 0; size += 4; // _sendStreamId.length; size += 4; // _receiveStreamId.length; @@ -469,6 +470,7 @@ class Packet { size += 4; // ackThrough if (_nacks != null) { size++; // nacks length + // if max win is ever > 255, limit to 255 size += 4 * _nacks.length; } else { size++; // nacks length @@ -671,10 +673,11 @@ class Packet { buf.append(toId(_sendStreamId)); //buf.append("<-->"); buf.append(toId(_receiveStreamId)).append(": #").append(_sequenceNum); - if (_sequenceNum < 10) - buf.append(" \t"); // so the tab lines up right - else - buf.append('\t'); + //if (_sequenceNum < 10) + // buf.append(" \t"); // so the tab lines up right + //else + // buf.append('\t'); + buf.append(' '); buf.append(toFlagString()); buf.append(" ACK ").append(getAckThrough()); if (_nacks != null) { diff --git a/apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java b/apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java index ef145179c..34a59475e 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java @@ -2,7 +2,6 @@ package net.i2p.client.streaming; import java.text.SimpleDateFormat; import java.util.Date; -import java.util.Iterator; import java.util.Set; import net.i2p.I2PAppContext; @@ -13,7 +12,8 @@ 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. - * + *

+ * I2PSession -> MessageHandler -> PacketHandler -> ConnectionPacketHandler -> MessageInputStream */ class PacketHandler { private final ConnectionManager _manager; @@ -86,6 +86,7 @@ class PacketHandler { } *****/ + /** */ void receivePacket(Packet packet) { //boolean ok = choke(packet); //if (ok) @@ -202,15 +203,13 @@ class PacketHandler { // someone is sending us a packet on the wrong stream // It isn't a SYN so it isn't likely to have a FROM to send a reset back to if (_log.shouldLog(Log.ERROR)) { - Set cons = _manager.listConnections(); StringBuilder buf = new StringBuilder(512); buf.append("Received a packet on the wrong stream: "); buf.append(packet); buf.append("\nthis connection:\n"); buf.append(con); buf.append("\nall connections:"); - for (Iterator iter = cons.iterator(); iter.hasNext();) { - Connection cur = (Connection)iter.next(); + for (Connection cur : _manager.listConnections()) { buf.append('\n').append(cur); } _log.error(buf.toString(), new Exception("Wrong stream")); @@ -299,9 +298,7 @@ class PacketHandler { } if (_log.shouldLog(Log.DEBUG)) { StringBuilder buf = new StringBuilder(128); - Set cons = _manager.listConnections(); - for (Iterator iter = cons.iterator(); iter.hasNext(); ) { - Connection con = (Connection)iter.next(); + for (Connection con : _manager.listConnections()) { buf.append(con.toString()).append(" "); } _log.debug("connections: " + buf.toString() + " sendId: " diff --git a/apps/streaming/java/src/net/i2p/client/streaming/PacketLocal.java b/apps/streaming/java/src/net/i2p/client/streaming/PacketLocal.java index ca2e25d42..748fe19ca 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/PacketLocal.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/PacketLocal.java @@ -16,11 +16,11 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { private final I2PAppContext _context; private final Log _log; private final Connection _connection; - private Destination _to; + private final Destination _to; private SessionKey _keyUsed; private Set _tagsSent; private final long _createdOn; - private int _numSends; + private volatile int _numSends; private long _lastSend; private long _acceptedOn; private long _ackOn; @@ -45,7 +45,6 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { } public Destination getTo() { return _to; } - public void setTo(Destination to) { _to = to; } /** * @deprecated should always return null @@ -72,6 +71,7 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { public void setTagsSent(Set tags) { if (tags != null && !tags.isEmpty()) _log.error("Who is sending tags thru the streaming lib? " + tags.size()); + /**** if ( (_tagsSent != null) && (!_tagsSent.isEmpty()) && (!tags.isEmpty()) ) { //int old = _tagsSent.size(); //_tagsSent.addAll(tags); @@ -80,6 +80,7 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { } else { _tagsSent = tags; } + ****/ } public boolean shouldSign() { @@ -142,10 +143,15 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { /** @return null if not bound */ public Connection getConnection() { return _connection; } + /** + * Will force a fast restransmit on the 3rd call (FAST_RETRANSMIT_THRESHOLD) + * but only if it's the lowest unacked (see Connection.ResendPacketEvent) + */ public void incrementNACKs() { int cnt = ++_nackCount; SimpleTimer2.TimedEvent evt = _resendEvent; - if ( (cnt >= Connection.FAST_RETRANSMIT_THRESHOLD) && (evt != null) && (!_retransmitted)) { + if (cnt >= Connection.FAST_RETRANSMIT_THRESHOLD && evt != null && (!_retransmitted) && + (_numSends == 1 || _lastSend < _context.clock().now() + 4*1000)) { // Don't fast retx if we recently resent it _retransmitted = true; evt.reschedule(0); } @@ -162,8 +168,11 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { if (con != null) buf.append(" rtt ").append(con.getOptions().getRTT()); - if ( (_tagsSent != null) && (!_tagsSent.isEmpty()) ) - buf.append(" with tags"); + //if ( (_tagsSent != null) && (!_tagsSent.isEmpty()) ) + // buf.append(" with tags"); + + if (_nackCount > 0) + buf.append(" nacked ").append(_nackCount).append(" times"); if (_ackOn > 0) buf.append(" ack after ").append(getAckTime()); @@ -200,8 +209,6 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { * @param maxWaitMs MessageOutputStream is the only caller, generally with -1 */ public void waitForAccept(int maxWaitMs) { - if (_connection == null) - throw new IllegalStateException("Cannot wait for accept with no connection"); long before = _context.clock().now(); int queued = _connection.getUnackedPacketsSent(); int window = _connection.getOptions().getWindowSize(); @@ -216,7 +223,7 @@ class PacketLocal extends Packet implements MessageOutputStream.WriteStatus { int afterQueued = _connection.getUnackedPacketsSent(); if ( (after - before > 1000) && (_log.shouldLog(Log.DEBUG)) ) _log.debug("Took " + (after-before) + "ms to get " - + (accepted ? " accepted" : " rejected") + + (accepted ? "accepted" : "rejected") + (_cancelledOn > 0 ? " and CANCELLED" : "") + ", queued behind " + queued +" with a window size of " + window + ", finally accepted with " + afterQueued + " queued: " diff --git a/apps/streaming/java/src/net/i2p/client/streaming/PacketQueue.java b/apps/streaming/java/src/net/i2p/client/streaming/PacketQueue.java index 724e44202..5e7ae6acc 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/PacketQueue.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/PacketQueue.java @@ -12,7 +12,8 @@ import net.i2p.util.Log; * 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. - * + *

+ * MessageOutputStream -> ConnectionDataReceiver -> Connection -> PacketQueue -> I2PSession */ class PacketQueue { private final I2PAppContext _context; @@ -26,16 +27,17 @@ class PacketQueue { _session = session; _connectionManager = mgr; _log = context.logManager().getLog(PacketQueue.class); - _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.sendDuplicateSize", "Size of a message resent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 }); + // all createRateStats in ConnectionManager } /** * Add a new packet to be sent out ASAP * * keys and tags disabled since dropped in I2PSession + * @return true if sent */ - public void enqueue(PacketLocal packet) { + public boolean enqueue(PacketLocal packet) { + // this updates the ack/nack field packet.prepare(); //SessionKey keyUsed = packet.getKeyUsed(); @@ -52,7 +54,7 @@ class PacketQueue { if (packet.getAckTime() > 0) { if (_log.shouldLog(Log.DEBUG)) _log.debug("Not resending " + packet); - return; + return false; } else { if (_log.shouldLog(Log.DEBUG)) _log.debug("Sending... " + packet); @@ -76,7 +78,7 @@ class PacketQueue { _log.warn("took " + writeTime + "ms to write the packet: " + packet); // last chance to short circuit... - if (packet.getAckTime() > 0) return; + if (packet.getAckTime() > 0) return false; // this should not block! begin = _context.clock().now(); @@ -158,6 +160,7 @@ class PacketQueue { // reset packet.releasePayload(); } + return sent; } } diff --git a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerChooser.java b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerChooser.java index b0167809a..17918a6e1 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerChooser.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerChooser.java @@ -15,7 +15,7 @@ class SchedulerChooser { private final Log _log; private final TaskScheduler _nullScheduler; /** list of TaskScheduler objects */ - private final List _schedulers; + private final List _schedulers; public SchedulerChooser(I2PAppContext context) { _context = context; @@ -26,7 +26,7 @@ class SchedulerChooser { public TaskScheduler getScheduler(Connection con) { for (int i = 0; i < _schedulers.size(); i++) { - TaskScheduler scheduler = (TaskScheduler)_schedulers.get(i); + TaskScheduler scheduler = _schedulers.get(i); if (scheduler.accept(con)) { //if (_log.shouldLog(Log.DEBUG)) // _log.debug("Scheduling for " + con + " with " + scheduler.getClass().getName()); @@ -50,11 +50,7 @@ class SchedulerChooser { } private class NullScheduler implements TaskScheduler { - private final Log _log; - public NullScheduler() { - _log = _context.logManager().getLog(NullScheduler.class); - } - + public void eventOccurred(Connection con) { if (_log.shouldLog(Log.WARN)) _log.warn("Yell at jrandom: Event occurred on " + con, new Exception("source")); diff --git a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerClosing.java b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerClosing.java index 29e69f6fa..b8754f784 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerClosing.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerClosing.java @@ -45,18 +45,25 @@ class SchedulerClosing extends SchedulerImpl { } public void eventOccurred(Connection con) { - if (con.getNextSendTime() <= 0) - con.setNextSendTime(_context.clock().now() + con.getOptions().getSendAckDelay()); - long remaining = con.getNextSendTime() - _context.clock().now(); + long nextSend = con.getNextSendTime(); + long now = _context.clock().now(); + long remaining; + if (nextSend <= 0) { + remaining = con.getOptions().getSendAckDelay(); + nextSend = now + remaining; + con.setNextSendTime(nextSend); + } else { + remaining = nextSend - now; + } if (_log.shouldLog(Log.DEBUG)) _log.debug("Event occurred w/ remaining: " + remaining + " on " + con); if (remaining <= 0) { if (con.getCloseSentOn() <= 0) { con.sendAvailable(); - con.setNextSendTime(_context.clock().now() + con.getOptions().getSendAckDelay()); } else { - con.ackImmediately(); + //con.ackImmediately(); } + con.setNextSendTime(now + con.getOptions().getSendAckDelay()); } else { //if (remaining < 5*1000) // remaining = 5*1000; diff --git a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerImpl.java b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerImpl.java index ace591629..fd36b7723 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/SchedulerImpl.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/SchedulerImpl.java @@ -19,4 +19,9 @@ abstract class SchedulerImpl implements TaskScheduler { protected void reschedule(long msToWait, Connection con) { SimpleScheduler.getInstance().addEvent(con.getConnectionEvent(), msToWait); } + + @Override + public String toString() { + return getClass().getSimpleName(); + } } diff --git a/apps/streaming/java/src/net/i2p/client/streaming/TCBShare.java b/apps/streaming/java/src/net/i2p/client/streaming/TCBShare.java index e4c4ec917..15b8e937e 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/TCBShare.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/TCBShare.java @@ -130,8 +130,8 @@ class TCBShare { super(timer); } public void timeReached() { - for (Iterator iter = _cache.keySet().iterator(); iter.hasNext(); ) { - if (_cache.get(iter.next()).isExpired()) + for (Iterator iter = _cache.values().iterator(); iter.hasNext(); ) { + if (iter.next().isExpired()) iter.remove(); } schedule(CLEAN_TIME); diff --git a/history.txt b/history.txt index 1955fa030..a06d63851 100644 --- a/history.txt +++ b/history.txt @@ -1,3 +1,24 @@ +2012-06-29 zzz + * HTTP Proxy: Change the error code for unknown host from 404 to 500 + * SimpleTimer: Fix logging + * Streaming: + - Allow at least 3 packets and up to half the window to be active resends + instead of just 1, to reduce stall time after a packet drop + - Increase fast retransmit threshold back to 3 to reduce retransmissions + - Don't fast retransmit if we recently retransmitted it already + - Allow double the window as long as gaps are less than the window + - Don't set the MSS in a resent packet (saves 2 bytes) + - Remove redundant calls to updateAcks() + - Update activity timer when resending a packet + - Reset unackedPacketsReceived counter at all places where acks are sent + so it isn't wrong + - Fix some places where the activeResends count could become wrong + - Prevent storm of CLOSE packets + - Never resend the whole packet in ackImmediately(), just send an ack + - Cancel flusher timer in MessageOutputStream when closed + - Move some createRateStats to ConnectionManager to reduce repeated calls + - Cleanups, javadocs, logging, volatile, finals + 2012-06-24 zzz * ElGamalAESEngine: Fix bad size estimate when tags are included, resulting in trailing zeros after the padding diff --git a/router/java/src/net/i2p/router/RouterVersion.java b/router/java/src/net/i2p/router/RouterVersion.java index 5b1ba79e3..f5c926454 100644 --- a/router/java/src/net/i2p/router/RouterVersion.java +++ b/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 = 19; + public final static long BUILD = 20; /** for example "-test" */ public final static String EXTRA = "";