diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java index 355d7fd1f..8bdb09dfb 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java @@ -505,26 +505,25 @@ class Connection { * @return List of packets acked for the first time, or null if none */ public List ackPackets(long ackThrough, long nacks[]) { - // FIXME synch this part too? - if (ackThrough < _highestAckedThrough) { - // dupack which won't tell us anything - } else { - if (nacks == null) { - _highestAckedThrough = ackThrough; - } else { - long lowest = -1; - for (int i = 0; i < nacks.length; i++) { - if ( (lowest < 0) || (nacks[i] < lowest) ) - lowest = nacks[i]; - } - if (lowest - 1 > _highestAckedThrough) - _highestAckedThrough = lowest - 1; - } - } - List acked = null; - boolean anyLeft = false; synchronized (_outboundPackets) { + if (ackThrough < _highestAckedThrough) { + // dupack which won't tell us anything + } else { + if (nacks == null) { + _highestAckedThrough = ackThrough; + } else { + long lowest = -1; + for (int i = 0; i < nacks.length; i++) { + if ( (lowest < 0) || (nacks[i] < lowest) ) + lowest = nacks[i]; + } + if (lowest - 1 > _highestAckedThrough) + _highestAckedThrough = lowest - 1; + } + } + + boolean anyLeft = false; if (!_outboundPackets.isEmpty()) { // short circuit iterator for (Iterator> iter = _outboundPackets.entrySet().iterator(); iter.hasNext(); ) { Map.Entry e = iter.next(); @@ -1178,6 +1177,15 @@ class Connection { return _outboundPackets.size(); } } + + /** + * For ConnectionPacketHandler.adjustWindow() + * + * @since 0.9.71 + */ + public Object getWindowLock() { + return _outboundPackets; + } public long getCongestionWindowEnd() { return _congestionWindowEnd; } public void setCongestionWindowEnd(long endMsg) { _congestionWindowEnd = endMsg; } diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionPacketHandler.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionPacketHandler.java index 72ade38f0..f64a8c554 100644 --- a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionPacketHandler.java +++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionPacketHandler.java @@ -432,88 +432,86 @@ class ConnectionPacketHandler { congested = false; } - long lowest = con.getHighestAckedThrough(); - // RFC 2581 - // Why wait until we get a whole cwin to start updating the window? - // That means we don't start increasing the window until after 1 RTT. - // And whether we increase the window or not (probably not since 1/N), - // we reset the CongestionWindowEnd and have to wait another RTT. - // So we add the acked > 1 and UnackedPacketsSent > 0 cases, - // so we almost always go through the window adjustment code, - // unless we're just sending a single packet now and then. - // This keeps the window size from going sky-high from ping traffic alone. - // Since we don't adjust the window down after idle? (RFC 2581 sec. 4.1) - if (lowest >= con.getCongestionWindowEnd() || - acked > 1 || - con.getUnackedPacketsSent() > 0) { - // new packet that ack'ed uncongested data, or an empty ack - int oldWindow = con.getOptions().getWindowSize(); - int newWindowSize = oldWindow; + synchronized(con.getWindowLock()) { + long lowest = con.getHighestAckedThrough(); + // RFC 2581 + // Why wait until we get a whole cwin to start updating the window? + // That means we don't start increasing the window until after 1 RTT. + // And whether we increase the window or not (probably not since 1/N), + // we reset the CongestionWindowEnd and have to wait another RTT. + // So we add the acked > 1 and UnackedPacketsSent > 0 cases, + // so we almost always go through the window adjustment code, + // unless we're just sending a single packet now and then. + // This keeps the window size from going sky-high from ping traffic alone. + // Since we don't adjust the window down after idle? (RFC 2581 sec. 4.1) + if (lowest >= con.getCongestionWindowEnd() || + acked > 1 || + con.getUnackedPacketsSent() > 0) { + // new packet that ack'ed uncongested data, or an empty ack + int oldWindow = con.getOptions().getWindowSize(); + int newWindowSize = oldWindow; - //int trend = con.getOptions().getRTTTrend(); - - //_context.statManager().addRateData("stream.trend", trend, newWindowSize); - - if ( (!congested) && (acked > 0) ) { - int ssthresh = con.getSSThresh(); - if (newWindowSize < ssthresh) { - // slow start - exponential growth - // grow acked/N times (where N = the slow start factor) - // always grow at least 1 - int factor = con.getOptions().getSlowStartGrowthRateFactor(); - if (factor <= 1) { - // above a certain point, don't grow exponentially - // as it often leads to a big packet loss (30-50) all at once that - // takes quite a while (a minute or more) to recover from, - // especially if crypto tags are lost - newWindowSize = Math.min(ssthresh, newWindowSize + acked); - } else if (acked < factor) - newWindowSize++; - else - newWindowSize += acked / factor; - if (_log.shouldLog(Log.DEBUG)) - _log.debug("slow start acks = " + acked + " for " + con); - // this is too fast since we mostly disabled the CongestionWindowEnd test above - //} else if (trend < 0) { - // // rtt is shrinking, so lets increment the cwin - // newWindowSize++; - // if (_log.shouldLog(Log.DEBUG)) - // _log.debug("trend < 0 for " + con); + if ( (!congested) && (acked > 0) ) { + int ssthresh = con.getSSThresh(); + if (newWindowSize < ssthresh) { + // slow start - exponential growth + // grow acked/N times (where N = the slow start factor) + // always grow at least 1 + int factor = con.getOptions().getSlowStartGrowthRateFactor(); + if (factor <= 1) { + // above a certain point, don't grow exponentially + // as it often leads to a big packet loss (30-50) all at once that + // takes quite a while (a minute or more) to recover from, + // especially if crypto tags are lost + newWindowSize = Math.min(ssthresh, newWindowSize + acked); + } else if (acked < factor) + newWindowSize++; + else + newWindowSize += acked / factor; + if (_log.shouldDebug()) + _log.debug("slow start acks = " + acked + " for " + con); + // this is too fast since we mostly disabled the CongestionWindowEnd test above + //} else if (trend < 0) { + // // rtt is shrinking, so lets increment the cwin + // newWindowSize++; + // if (_log.shouldLog(Log.DEBUG)) + // _log.debug("trend < 0 for " + con); + } else { + // congestion avoidance + // linear growth - increase window 1/N per RTT + // we can't use newWindowSize += acked/(oldWindow*N) (where N = the cong. avoid. factor), since we're + // integers, so lets use a random distribution instead + int shouldIncrement = _context.random().nextInt(con.getOptions().getCongestionAvoidanceGrowthRateFactor()*newWindowSize); + if (shouldIncrement < acked) + newWindowSize++; + if (_log.shouldDebug()) + _log.debug("cong. avoid acks = " + acked + " for " + con); + } } else { - // congestion avoidance - // linear growth - increase window 1/N per RTT - // we can't use newWindowSize += acked/(oldWindow*N) (where N = the cong. avoid. factor), since we're - // integers, so lets use a random distribution instead - int shouldIncrement = _context.random().nextInt(con.getOptions().getCongestionAvoidanceGrowthRateFactor()*newWindowSize); - if (shouldIncrement < acked) - newWindowSize++; - if (_log.shouldLog(Log.DEBUG)) - _log.debug("cong. avoid acks = " + acked + " for " + con); + if (_log.shouldDebug()) + _log.debug("No change to window: " + con.getOptions().getWindowSize() + + " congested? " + congested + " acked: " + acked + " resends: " + numResends); } - } else { - if (_log.shouldLog(Log.DEBUG)) - _log.debug("No change to window: " + con.getOptions().getWindowSize() + - " congested? " + congested + " acked: " + acked + " resends: " + numResends); - } - if (newWindowSize <= 0) - newWindowSize = 1; + if (newWindowSize <= 0) + newWindowSize = 1; - con.getOptions().setWindowSize(newWindowSize); - con.setCongestionWindowEnd(newWindowSize + lowest); + con.getOptions().setWindowSize(newWindowSize); + con.setCongestionWindowEnd(newWindowSize + lowest); - if (_log.shouldLog(Log.INFO)) - _log.info("New window size " + newWindowSize + "/" + oldWindow + "/" + con.getOptions().getWindowSize() - + " (#resends: " + numResends - + ") for " + con); - } else { - if (_log.shouldLog(Log.DEBUG)) - _log.debug("No change to window: " + con.getOptions().getWindowSize() + - " highestAckedThrough: " + lowest + " congestionWindowEnd: " + con.getCongestionWindowEnd() + - " acked: " + acked + " unacked: " + con.getUnackedPacketsSent()); - } + if (_log.shouldInfo()) + _log.info("New window size " + newWindowSize + "/" + oldWindow + "/" + con.getOptions().getWindowSize() + + " (#resends: " + numResends + + ") for " + con); + } else { + if (_log.shouldDebug()) + _log.debug("No change to window: " + con.getOptions().getWindowSize() + + " highestAckedThrough: " + lowest + " congestionWindowEnd: " + con.getCongestionWindowEnd() + + " acked: " + acked + " unacked: " + con.getUnackedPacketsSent()); + } - con.windowAdjusted(); + con.windowAdjusted(); + } return congested; }