diff --git a/history.txt b/history.txt index 4974fa3da..e23cb9cf0 100644 --- a/history.txt +++ b/history.txt @@ -1,3 +1,13 @@ +2009-12-22 zzz + * Tunnels: + - Do RED dropping before the IBGW fragmenter, not after + - Change batch time to 250ms for IBGWs (was 100ms) + - Change batch time to 150ms for exploratory OBGWs (was 100ms) + - Start a new message in the fragmenter if almost full + - Fix a major, longstanding synchronization bug in the FragmentHandler + which led to corrupt messages at the endpoints + - More cleanups and comments + 2009-12-20 zzz * Console: - Fix status to show a disconnected network error rather than diff --git a/router/java/src/net/i2p/router/RouterVersion.java b/router/java/src/net/i2p/router/RouterVersion.java index 1440f81c0..217d41d54 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 = 10; + public final static long BUILD = 11; /** for example "-test" */ public final static String EXTRA = ""; public final static String FULL_VERSION = VERSION + "-" + BUILD + EXTRA; diff --git a/router/java/src/net/i2p/router/tunnel/BatchedPreprocessor.java b/router/java/src/net/i2p/router/tunnel/BatchedPreprocessor.java index 30fbbdd27..cfaabcfcb 100644 --- a/router/java/src/net/i2p/router/tunnel/BatchedPreprocessor.java +++ b/router/java/src/net/i2p/router/tunnel/BatchedPreprocessor.java @@ -67,6 +67,7 @@ public class BatchedPreprocessor extends TrivialPreprocessor { ctx.statManager().createRateStat("tunnel.batchFragmentation", "Avg. number of fragments per msg", "Tunnels", new long[] { 10*60*1000, 60*60*1000 }); } + /** 1003 */ private static final int FULL_SIZE = PREPROCESSED_SIZE - IV_SIZE - 1 // 0x00 ending the padding @@ -76,11 +77,28 @@ public class BatchedPreprocessor extends TrivialPreprocessor { /* not final or private so the test code can adjust */ static long DEFAULT_DELAY = 100; - /** Wait up to this long before sending (flushing) a small tunnel message */ + /** + * Wait up to this long before sending (flushing) a small tunnel message + * Warning - overridden in BatchedRouterPreprocessor + */ protected long getSendDelay() { return DEFAULT_DELAY; } - /** if we have 50 messages queued that are too small, flush them anyway */ - private static final int FORCE_BATCH_FLUSH = 50; + /** + * if we have this many messages queued that are too small, flush them anyway + * Even small messages take up about 200 bytes or so. + */ + private static final int FORCE_BATCH_FLUSH = 5; + + /** If we have this much allocated, flush anyway. + * Tune this to trade off padding vs. fragmentation. + * The lower the value, the more we are willing to send off + * a tunnel msg that isn't full so the next message can start + * in a new tunnel msg to minimize fragmentation. + * + * This should be at most FULL_SIZE - (39 + a few), since + * you want to at least fit in the instructions and a few bytes. + */ + private static final int FULL_ENOUGH_SIZE = (FULL_SIZE * 80) / 100; /** how long do we want to wait before flushing */ @Override @@ -100,6 +118,8 @@ public class BatchedPreprocessor extends TrivialPreprocessor { /* See TunnelGateway.QueuePreprocessor for Javadoc */ @Override public boolean preprocessQueue(List pending, TunnelGateway.Sender sender, TunnelGateway.Receiver rec) { + if (_log.shouldLog(Log.INFO)) + display(0, pending, "Starting"); StringBuilder timingBuf = null; if (_log.shouldLog(Log.DEBUG)) { _log.debug("Preprocess queue with " + pending.size() + " to send"); @@ -155,9 +175,10 @@ public class BatchedPreprocessor extends TrivialPreprocessor { _context.statManager().addRateData("tunnel.batchFullFragments", 1, 0); long afterSend = System.currentTimeMillis(); if (_log.shouldLog(Log.INFO)) - _log.info("Allocated=" + allocated + " so we sent " + (i+1) - + " (last complete? " + (msg.getOffset() >= msg.getData().length) - + ", off=" + msg.getOffset() + ", count=" + pending.size() + ")"); + display(allocated, pending, "Sent the message with " + (i+1)); + //_log.info(_name + ": Allocated=" + allocated + "B, Sent " + (i+1) + // + " msgs (last complete? " + (msg.getOffset() >= msg.getData().length) + // + ", off=" + msg.getOffset() + ", pending=" + pending.size() + ")"); // Remove what we sent from the pending queue for (int j = 0; j < i; j++) { @@ -197,7 +218,6 @@ public class BatchedPreprocessor extends TrivialPreprocessor { timingBuf.append(" After pending loop " + (System.currentTimeMillis()-beforePendingLoop)).append("."); } // for - long afterCleared = System.currentTimeMillis(); if (_log.shouldLog(Log.INFO)) display(allocated, pending, "after looping to clear " + (beforeLooping - pending.size())); long afterDisplayed = System.currentTimeMillis(); @@ -205,7 +225,12 @@ public class BatchedPreprocessor extends TrivialPreprocessor { // After going through the entire pending list, we have only a partial message. // We might flush it or might not, but we are returning either way. - if ( (pending.size() > FORCE_BATCH_FLUSH) || ( (_pendingSince > 0) && (getDelayAmount() <= 0) ) ) { + if ( (pending.size() > FORCE_BATCH_FLUSH) || // enough msgs - or + ( (_pendingSince > 0) && (getDelayAmount() <= 0) ) || // time to flush - or + (allocated >= FULL_ENOUGH_SIZE)) { // full enough + //(pending.get(0).getFragmentNumber() > 0)) { // don't delay anybody's last fragment, + // // which would be the first fragment in the message + // not even a full message, but we want to flush it anyway if (pending.size() > 1) @@ -215,22 +240,24 @@ public class BatchedPreprocessor extends TrivialPreprocessor { send(pending, 0, pending.size()-1, sender, rec); _context.statManager().addRateData("tunnel.batchSmallFragments", FULL_SIZE - allocated, 0); - // Remove everything in the message from the pending queue + // Remove everything in the outgoing message from the pending queue int beforeSize = pending.size(); - for (int i = 0; i < pending.size(); i++) { - TunnelGateway.Pending cur = pending.get(i); - if (cur.getOffset() >= cur.getData().length) { - pending.remove(i); - notePreprocessing(cur.getMessageId(), cur.getFragmentNumber(), cur.getData().length, cur.getMessageIds(), "flushed remaining"); - _context.statManager().addRateData("tunnel.batchFragmentation", cur.getFragmentNumber() + 1, 0); - _context.statManager().addRateData("tunnel.writeDelay", cur.getLifetime(), cur.getData().length); - i--; - } + for (int i = 0; i < beforeSize; i++) { + TunnelGateway.Pending cur = pending.get(0); + if (cur.getOffset() < cur.getData().length) + break; + pending.remove(0); + notePreprocessing(cur.getMessageId(), cur.getFragmentNumber(), cur.getData().length, cur.getMessageIds(), "flushed remaining"); + _context.statManager().addRateData("tunnel.batchFragmentation", cur.getFragmentNumber() + 1, 0); + _context.statManager().addRateData("tunnel.writeDelay", cur.getLifetime(), cur.getData().length); } + if (pending.size() > 0) { + // rare _pendingSince = _context.clock().now(); _context.statManager().addRateData("tunnel.batchFlushRemaining", pending.size(), beforeSize); - display(allocated, pending, "flushed, some remain"); + if (_log.shouldLog(Log.INFO)) + display(allocated, pending, "flushed, some remain"); if (timingBuf != null) { timingBuf.append(" flushed, some remain (displayed to now: " + (System.currentTimeMillis()-afterDisplayed) + ")"); @@ -239,12 +266,15 @@ public class BatchedPreprocessor extends TrivialPreprocessor { } return true; } else { - long delayAmount = _context.clock().now() - _pendingSince; - _pendingSince = 0; + long delayAmount = 0; + if (_pendingSince > 0) { + delayAmount = _context.clock().now() - _pendingSince; + _pendingSince = 0; + } if (batchCount > 1) _context.statManager().addRateData("tunnel.batchCount", batchCount, 0); if (_log.shouldLog(Log.INFO)) - display(allocated, pending, "flushed " + (beforeSize) + ", no remaining after " + delayAmount); + display(allocated, pending, "flushed " + (beforeSize) + ", no remaining after " + delayAmount + "ms"); if (timingBuf != null) { timingBuf.append(" flushed, none remain (displayed to now: " + (System.currentTimeMillis()-afterDisplayed) + ")"); @@ -262,7 +292,8 @@ public class BatchedPreprocessor extends TrivialPreprocessor { if (batchCount > 1) _context.statManager().addRateData("tunnel.batchCount", batchCount, 0); // not yet time to send the delayed flush - display(allocated, pending, "dont flush"); + if (_log.shouldLog(Log.INFO)) + display(allocated, pending, "dont flush"); if (timingBuf != null) { timingBuf.append(" dont flush (displayed to now: " + (System.currentTimeMillis()-afterDisplayed) + ")"); @@ -293,20 +324,25 @@ public class BatchedPreprocessor extends TrivialPreprocessor { return false; } + /* + * Only if Log.INFO + * + * title: allocated: X pending: X (delay: X) [0]:offset/length/lifetime [1]:etc. + */ private void display(long allocated, List pending, String title) { if (_log.shouldLog(Log.INFO)) { long highestDelay = 0; - StringBuilder buf = new StringBuilder(); + StringBuilder buf = new StringBuilder(128); buf.append(_name).append(": "); buf.append(title); - buf.append(" allocated: ").append(allocated); + buf.append(" - allocated: ").append(allocated); buf.append(" pending: ").append(pending.size()); if (_pendingSince > 0) buf.append(" delay: ").append(getDelayAmount(false)); for (int i = 0; i < pending.size(); i++) { TunnelGateway.Pending curPending = pending.get(i); - buf.append(" pending[").append(i).append("]: "); - buf.append(curPending.getOffset()).append("/").append(curPending.getData().length).append('/'); + buf.append(" [").append(i).append("]:"); + buf.append(curPending.getOffset()).append('/').append(curPending.getData().length).append('/'); buf.append(curPending.getLifetime()); if (curPending.getLifetime() > highestDelay) highestDelay = curPending.getLifetime(); diff --git a/router/java/src/net/i2p/router/tunnel/BatchedRouterPreprocessor.java b/router/java/src/net/i2p/router/tunnel/BatchedRouterPreprocessor.java index 6af65e88e..71d80b011 100644 --- a/router/java/src/net/i2p/router/tunnel/BatchedRouterPreprocessor.java +++ b/router/java/src/net/i2p/router/tunnel/BatchedRouterPreprocessor.java @@ -14,68 +14,96 @@ public class BatchedRouterPreprocessor extends BatchedPreprocessor { protected RouterContext _routerContext; private TunnelCreatorConfig _config; protected HopConfig _hopConfig; + private final long _sendDelay; /** * How frequently should we flush non-full messages, in milliseconds + * This goes in I2CP custom options for the pool. + * Only applies to OBGWs. */ public static final String PROP_BATCH_FREQUENCY = "batchFrequency"; + /** This goes in router advanced config */ public static final String PROP_ROUTER_BATCH_FREQUENCY = "router.batchFrequency"; - public static final int DEFAULT_BATCH_FREQUENCY = 100; + /** for client OBGWs only (our data) */ + public static final int OB_CLIENT_BATCH_FREQ = 100; + /** for exploratory OBGWs only (our tunnel tests and build messages) */ + public static final int OB_EXPL_BATCH_FREQ = 150; + /** for IBGWs for efficiency (not our data) */ + public static final int DEFAULT_BATCH_FREQUENCY = 250; - public BatchedRouterPreprocessor(RouterContext ctx) { - this(ctx, (HopConfig)null); - } + /** for OBGWs */ public BatchedRouterPreprocessor(RouterContext ctx, TunnelCreatorConfig cfg) { super(ctx, getName(cfg)); _routerContext = ctx; _config = cfg; + _sendDelay = initialSendDelay(); } + + /** for IBGWs */ public BatchedRouterPreprocessor(RouterContext ctx, HopConfig cfg) { super(ctx, getName(cfg)); _routerContext = ctx; _hopConfig = cfg; + _sendDelay = initialSendDelay(); } private static String getName(HopConfig cfg) { - if (cfg == null) return "[unknown]"; + if (cfg == null) return "IB??"; if (cfg.getReceiveTunnel() != null) - return cfg.getReceiveTunnel().getTunnelId() + ""; + return "IB " + cfg.getReceiveTunnel().getTunnelId(); else if (cfg.getSendTunnel() != null) - return cfg.getSendTunnel().getTunnelId() + ""; + return "IB " + cfg.getSendTunnel().getTunnelId(); else - return "[n/a]"; + return "IB??"; } private static String getName(TunnelCreatorConfig cfg) { - if (cfg == null) return "[unknown]"; + if (cfg == null) return "OB??"; if (cfg.getReceiveTunnelId(0) != null) - return cfg.getReceiveTunnelId(0).getTunnelId() + ""; + return "OB " + cfg.getReceiveTunnelId(0).getTunnelId(); else if (cfg.getSendTunnelId(0) != null) - return cfg.getSendTunnelId(0).getTunnelId() + ""; + return "OB " + cfg.getSendTunnelId(0).getTunnelId(); else - return "[n/a]"; + return "OB??"; } - /** how long should we wait before flushing */ + /** + * how long should we wait before flushing + */ @Override - protected long getSendDelay() { - String freq = null; + protected long getSendDelay() { return _sendDelay; } + + /* + * Extend the batching time for exploratory OBGWs, they have a lot of small + * tunnel test messages, and build messages that don't fit perfectly. + * And these are not as delay-sensitive. + * + * We won't pick up config changes after the preprocessor is created, + * but a preprocessor lifetime is only 10 minutes, so just wait... + */ + private long initialSendDelay() { if (_config != null) { Properties opts = _config.getOptions(); - if (opts != null) - freq = opts.getProperty(PROP_BATCH_FREQUENCY); - } - if (freq == null) - freq = _routerContext.getProperty(PROP_ROUTER_BATCH_FREQUENCY); - - if (freq != null) { - try { - return Integer.parseInt(freq); - } catch (NumberFormatException nfe) { - return DEFAULT_BATCH_FREQUENCY; + if (opts != null) { + String freq = opts.getProperty(PROP_BATCH_FREQUENCY); + if (freq != null) { + try { + return Integer.parseInt(freq); + } catch (NumberFormatException nfe) {} + } } } - return DEFAULT_BATCH_FREQUENCY; + + int def; + if (_config != null) { + if (_config.getDestination() != null) + def = OB_CLIENT_BATCH_FREQ; + else + def = OB_EXPL_BATCH_FREQ; + } else { + def = DEFAULT_BATCH_FREQUENCY; + } + return _routerContext.getProperty(PROP_ROUTER_BATCH_FREQUENCY, def); } @Override diff --git a/router/java/src/net/i2p/router/tunnel/FragmentHandler.java b/router/java/src/net/i2p/router/tunnel/FragmentHandler.java index f294d5c78..6e4c00e1c 100644 --- a/router/java/src/net/i2p/router/tunnel/FragmentHandler.java +++ b/router/java/src/net/i2p/router/tunnel/FragmentHandler.java @@ -39,29 +39,32 @@ data into the raw tunnel payload:

  • a series of zero or more { instructions, message } pairs
    • +

    Note that the padding, if any, must be before the instruction/message pairs. +there is no provision for padding at the end.

    +

    The instructions are encoded with a single control byte, followed by any necessary additional information. The first bit in that control byte determines how the remainder of the header is interpreted - if it is not set, the message is either not fragmented or this is the first fragment in the message. If it is set, this is a follow on fragment.

    -

    With the first bit being 0, the instructions are:

    +

    With the first (leftmost or MSB) bit being 0, the instructions are: