messageIds, String msg) {
if (_config != null)
_routerContext.messageHistory().fragmentMessage(messageId, numFragments, totalLength, messageIds, _config, msg);
else
diff --git a/router/java/src/net/i2p/router/tunnel/FragmentHandler.java b/router/java/src/net/i2p/router/tunnel/FragmentHandler.java
index 43e1d1314..f294d5c78 100644
--- a/router/java/src/net/i2p/router/tunnel/FragmentHandler.java
+++ b/router/java/src/net/i2p/router/tunnel/FragmentHandler.java
@@ -21,6 +21,65 @@ import net.i2p.util.SimpleTimer;
* Handle fragments at the endpoint of a tunnel, peeling off fully completed
* I2NPMessages when they arrive, and dropping fragments if they take too long
* to arrive.
+ *
+ * From tunnel-alt.html:
+
+When the gateway wants to deliver data through the tunnel, it first
+gathers zero or more I2NP messages, selects how much padding will be used,
+fragments it across the necessary number of 1KB tunnel messages, and decides how
+each I2NP message should be handled by the tunnel endpoint, encoding that
+data into the raw tunnel payload:
+
+- the first 4 bytes of the SHA256 of (the remaining preprocessed data concatenated
+ with the IV), using the IV as will be seen on the tunnel endpoint (for
+ outbound tunnels), or the IV as was seen on the tunnel gateway (for inbound
+ tunnels) (see below for IV processing).
+- 0 or more bytes containing random nonzero integers
+- 1 byte containing 0x00
+- a series of zero or more { instructions, message } pairs
+
+
+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:
+
+- 1 byte control byte:
+ bit 0: is follow on fragment? (1 = true, 0 = false, must be 0)
+ bits 1-2: delivery type
+ (0x0 = LOCAL, 0x01 = TUNNEL, 0x02 = ROUTER)
+ bit 3: delay included? (1 = true, 0 = false)
+ bit 4: fragmented? (1 = true, 0 = false)
+ bit 5: extended options? (1 = true, 0 = false)
+ bits 6-7: reserved
+- if the delivery type was TUNNEL, a 4 byte tunnel ID
+- if the delivery type was TUNNEL or ROUTER, a 32 byte router hash
+- if the delay included flag is true, a 1 byte value:
+ bit 0: type (0 = strict, 1 = randomized)
+ bits 1-7: delay exponent (2^value minutes)
+- if the fragmented flag is true, a 4 byte message ID
+- if the extended options flag is true:
+ = a 1 byte option size (in bytes)
+ = that many bytes
+- 2 byte size of the I2NP message or this fragment
+
+
+If the first bit being 1, the instructions are:
+
+- 1 byte control byte:
+ bit 0: is follow on fragment? (1 = true, 0 = false, must be 1)
+ bits 1-6: fragment number
+ bit 7: is last? (1 = true, 0 = false)
+- 4 byte message ID (same one defined in the first fragment)
+- 2 byte size of this fragment
+
+
+The I2NP message is encoded in its standard form, and the
+preprocessed payload must be padded to a multiple of 16 bytes.
+
*
*/
public class FragmentHandler {
@@ -149,7 +208,7 @@ public class FragmentHandler {
if (_log.shouldLog(Log.WARN))
_log.warn("cannot verify, going past the end [off="
+ offset + " len=" + length + " paddingEnd="
- + paddingEnd + " data:\n"
+ + paddingEnd + " data: "
+ Base64.encode(preprocessed, offset, length));
return false;
}
@@ -165,21 +224,19 @@ public class FragmentHandler {
_log.debug("endpoint IV: " + Base64.encode(preV, validLength - HopProcessor.IV_LENGTH, HopProcessor.IV_LENGTH));
Hash v = _context.sha().calculateHash(preV, 0, validLength);
+ _validateCache.release(ba);
- //Hash v = _context.sha().calculateHash(preV, 0, validLength);
boolean eq = DataHelper.eq(v.getData(), 0, preprocessed, offset + HopProcessor.IV_LENGTH, 4);
if (!eq) {
- if (_log.shouldLog(Log.WARN))
- _log.warn("Corrupt tunnel message - verification fails: \n" + Base64.encode(preprocessed, offset+HopProcessor.IV_LENGTH, 4)
- + "\n" + Base64.encode(v.getData(), 0, 4));
- if (_log.shouldLog(Log.WARN))
- _log.warn("nomatching endpoint: # pad bytes: " + (paddingEnd-(HopProcessor.IV_LENGTH+4)-1) + "\n"
- + " offset=" + offset + " length=" + length + " paddingEnd=" + paddingEnd
+ if (_log.shouldLog(Log.WARN)) {
+ _log.warn("Corrupt tunnel message - verification fails: " + Base64.encode(preprocessed, offset+HopProcessor.IV_LENGTH, 4)
+ + " != " + Base64.encode(v.getData(), 0, 4));
+ _log.warn("No matching endpoint: # pad bytes: " + (paddingEnd-(HopProcessor.IV_LENGTH+4)-1)
+ + " offset=" + offset + " length=" + length + " paddingEnd=" + paddingEnd + ' '
+ Base64.encode(preprocessed, offset, length));
+ }
}
- _validateCache.release(ba);
-
if (eq) {
int excessPadding = paddingEnd - (HopProcessor.IV_LENGTH + 4 + 1);
if (excessPadding > 0) // suboptimal fragmentation
diff --git a/router/java/src/net/i2p/router/tunnel/InboundEndpointProcessor.java b/router/java/src/net/i2p/router/tunnel/InboundEndpointProcessor.java
index 6c572cb42..7376a5046 100644
--- a/router/java/src/net/i2p/router/tunnel/InboundEndpointProcessor.java
+++ b/router/java/src/net/i2p/router/tunnel/InboundEndpointProcessor.java
@@ -1,6 +1,5 @@
package net.i2p.router.tunnel;
-import net.i2p.I2PAppContext;
import net.i2p.data.ByteArray;
import net.i2p.data.Hash;
import net.i2p.router.RouterContext;
@@ -16,7 +15,7 @@ import net.i2p.util.Log;
*
*/
public class InboundEndpointProcessor {
- private I2PAppContext _context;
+ private RouterContext _context;
private Log _log;
private TunnelCreatorConfig _config;
private IVValidator _validator;
@@ -24,10 +23,10 @@ public class InboundEndpointProcessor {
static final boolean USE_ENCRYPTION = HopProcessor.USE_ENCRYPTION;
private static final ByteCache _cache = ByteCache.getInstance(128, HopProcessor.IV_LENGTH);
- public InboundEndpointProcessor(I2PAppContext ctx, TunnelCreatorConfig cfg) {
+ public InboundEndpointProcessor(RouterContext ctx, TunnelCreatorConfig cfg) {
this(ctx, cfg, DummyValidator.getInstance());
}
- public InboundEndpointProcessor(I2PAppContext ctx, TunnelCreatorConfig cfg, IVValidator validator) {
+ public InboundEndpointProcessor(RouterContext ctx, TunnelCreatorConfig cfg, IVValidator validator) {
_context = ctx;
_log = ctx.logManager().getLog(InboundEndpointProcessor.class);
_config = cfg;
@@ -73,23 +72,19 @@ public class InboundEndpointProcessor {
_cache.release(ba);
- // now for a little bookkeeping
- RouterContext ctx = null;
- if (_context instanceof RouterContext)
- ctx = (RouterContext)_context;
- if ( (ctx != null) && (_config.getLength() > 0) ) {
+ if (_config.getLength() > 0) {
int rtt = 0; // dunno... may not be related to an rtt
if (_log.shouldLog(Log.DEBUG))
_log.debug("Received a " + length + "byte message through tunnel " + _config);
for (int i = 0; i < _config.getLength(); i++)
- ctx.profileManager().tunnelDataPushed(_config.getPeer(i), rtt, length);
+ _context.profileManager().tunnelDataPushed(_config.getPeer(i), rtt, length);
_config.incrementVerifiedBytesTransferred(length);
}
return true;
}
- private void decrypt(I2PAppContext ctx, TunnelCreatorConfig cfg, byte iv[], byte orig[], int offset, int length) {
+ private void decrypt(RouterContext ctx, TunnelCreatorConfig cfg, byte iv[], byte orig[], int offset, int length) {
Log log = ctx.logManager().getLog(OutboundGatewayProcessor.class);
ByteArray ba = _cache.acquire();
byte cur[] = ba.getData(); // new byte[HopProcessor.IV_LENGTH]; // so we dont malloc
diff --git a/router/java/src/net/i2p/router/tunnel/TrivialPreprocessor.java b/router/java/src/net/i2p/router/tunnel/TrivialPreprocessor.java
index 4792b14c2..bd7bc7362 100644
--- a/router/java/src/net/i2p/router/tunnel/TrivialPreprocessor.java
+++ b/router/java/src/net/i2p/router/tunnel/TrivialPreprocessor.java
@@ -3,11 +3,11 @@ package net.i2p.router.tunnel;
import java.util.ArrayList;
import java.util.List;
-import net.i2p.I2PAppContext;
import net.i2p.data.Base64;
import net.i2p.data.ByteArray;
import net.i2p.data.DataHelper;
import net.i2p.data.Hash;
+import net.i2p.router.RouterContext;
import net.i2p.util.ByteCache;
import net.i2p.util.Log;
@@ -19,8 +19,8 @@ import net.i2p.util.Log;
*
*/
public class TrivialPreprocessor implements TunnelGateway.QueuePreprocessor {
- protected I2PAppContext _context;
- private Log _log;
+ protected RouterContext _context;
+ protected Log _log;
public static final int PREPROCESSED_SIZE = 1024;
protected static final int IV_SIZE = HopProcessor.IV_LENGTH;
@@ -28,7 +28,7 @@ public class TrivialPreprocessor implements TunnelGateway.QueuePreprocessor {
protected static final ByteCache _ivCache = ByteCache.getInstance(128, IV_SIZE);
protected static final ByteCache _hashCache = ByteCache.getInstance(128, Hash.HASH_LENGTH);
- public TrivialPreprocessor(I2PAppContext ctx) {
+ public TrivialPreprocessor(RouterContext ctx) {
_context = ctx;
_log = ctx.logManager().getLog(TrivialPreprocessor.class);
}
@@ -41,7 +41,7 @@ public class TrivialPreprocessor implements TunnelGateway.QueuePreprocessor {
* a delayed flush to clear them
*
*/
- public boolean preprocessQueue(List pending, TunnelGateway.Sender sender, TunnelGateway.Receiver rec) {
+ public boolean preprocessQueue(List pending, TunnelGateway.Sender sender, TunnelGateway.Receiver rec) {
long begin = System.currentTimeMillis();
StringBuilder buf = null;
if (_log.shouldLog(Log.DEBUG)) {
@@ -49,7 +49,7 @@ public class TrivialPreprocessor implements TunnelGateway.QueuePreprocessor {
buf.append("Trivial preprocessing of ").append(pending.size()).append(" ");
}
while (pending.size() > 0) {
- TunnelGateway.Pending msg = (TunnelGateway.Pending)pending.remove(0);
+ TunnelGateway.Pending msg = pending.remove(0);
long beforePreproc = System.currentTimeMillis();
byte preprocessed[][] = preprocess(msg);
long afterPreproc = System.currentTimeMillis();
@@ -84,7 +84,7 @@ public class TrivialPreprocessor implements TunnelGateway.QueuePreprocessor {
return false;
}
- protected void notePreprocessing(long messageId, int numFragments, int totalLength, List messageIds, String msg) {}
+ protected void notePreprocessing(long messageId, int numFragments, int totalLength, List messageIds, String msg) {}
private byte[][] preprocess(TunnelGateway.Pending msg) {
List fragments = new ArrayList(1);
diff --git a/router/java/src/net/i2p/router/tunnel/TrivialRouterPreprocessor.java b/router/java/src/net/i2p/router/tunnel/TrivialRouterPreprocessor.java
index ad7322cb6..6ddf636e4 100644
--- a/router/java/src/net/i2p/router/tunnel/TrivialRouterPreprocessor.java
+++ b/router/java/src/net/i2p/router/tunnel/TrivialRouterPreprocessor.java
@@ -7,6 +7,7 @@ import net.i2p.router.RouterContext;
/**
* Minor extension to track fragmentation
*
+ * @deprecated unused
*/
public class TrivialRouterPreprocessor extends TrivialPreprocessor {
private RouterContext _routerContext;
@@ -16,7 +17,7 @@ public class TrivialRouterPreprocessor extends TrivialPreprocessor {
_routerContext = ctx;
}
- protected void notePreprocessing(long messageId, int numFragments, int totalLength, List messageIds) {
+ protected void notePreprocessing(long messageId, int numFragments, int totalLength, List messageIds) {
_routerContext.messageHistory().fragmentMessage(messageId, numFragments, totalLength, messageIds, null);
}
}
diff --git a/router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java b/router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java
index 681dadec1..527426416 100644
--- a/router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java
+++ b/router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java
@@ -131,17 +131,20 @@ public class TunnelDispatcher implements Service {
new long[] { 60*1000l, 10*60*1000l, 60*60*1000l });
}
+ /** for IBGW */
private TunnelGateway.QueuePreprocessor createPreprocessor(HopConfig cfg) {
- if (true)
- return new BatchedRouterPreprocessor(_context, cfg);
- else
- return new TrivialRouterPreprocessor(_context);
+ //if (true)
+ return new DroppingBatchedRouterPreprocessor(_context, cfg);
+ //else
+ // return new TrivialRouterPreprocessor(_context);
}
+
+ /** for OBGW */
private TunnelGateway.QueuePreprocessor createPreprocessor(TunnelCreatorConfig cfg) {
- if (true)
+ //if (true)
return new BatchedRouterPreprocessor(_context, cfg);
- else
- return new TrivialRouterPreprocessor(_context);
+ //else
+ // return new TrivialRouterPreprocessor(_context);
}
/**
@@ -605,12 +608,17 @@ public class TunnelDispatcher implements Service {
if (pctDrop <= 0)
return false;
// increase the drop probability for OBEP,
+ // (except lower it for tunnel build messages (type 21)),
// and lower it for IBGW, for network efficiency
double len = length;
- if (type.startsWith("OBEP"))
- len *= 1.5;
- else if (type.startsWith("IBGW"))
+ if (type.startsWith("OBEP")) {
+ if (type.equals("OBEP 21"))
+ len /= 1.5;
+ else
+ len *= 1.5;
+ } else if (type.startsWith("IBGW")) {
len /= 1.5;
+ }
// drop in proportion to size w.r.t. a standard 1024-byte message
// this is a little expensive but we want to adjust the curve between 0 and 1
// Most messages are 1024, only at the OBEP do we see other sizes
diff --git a/router/java/src/net/i2p/router/tunnel/TunnelGateway.java b/router/java/src/net/i2p/router/tunnel/TunnelGateway.java
index 40c935e79..2b394e6da 100644
--- a/router/java/src/net/i2p/router/tunnel/TunnelGateway.java
+++ b/router/java/src/net/i2p/router/tunnel/TunnelGateway.java
@@ -154,12 +154,18 @@ public class TunnelGateway {
public interface QueuePreprocessor {
/**
+ * Caller must synchronize on the list!
+ *
* @param pending list of Pending objects for messages either unsent
* or partly sent. This list should be update with any
* values removed (the preprocessor owns the lock)
+ * Messages are not removed from the list until actually sent.
+ * The status of unsent and partially-sent messages is stored in
+ * the Pending structure.
+ *
* @return true if we should delay before preprocessing again
*/
- public boolean preprocessQueue(List pending, Sender sender, Receiver receiver);
+ public boolean preprocessQueue(List pending, Sender sender, Receiver receiver);
/** how long do we want to wait before flushing */
public long getDelayAmount();
@@ -173,6 +179,9 @@ public class TunnelGateway {
public long receiveEncrypted(byte encrypted[]);
}
+ /**
+ * Stores all the state for an unsent or partially-sent message
+ */
public static class Pending {
protected Hash _toRouter;
protected TunnelId _toTunnel;
@@ -182,7 +191,7 @@ public class TunnelGateway {
protected int _offset;
protected int _fragmentNumber;
protected long _created;
- private List _messageIds;
+ private List _messageIds;
public Pending(I2NPMessage message, Hash toRouter, TunnelId toTunnel) {
this(message, toRouter, toTunnel, System.currentTimeMillis());
@@ -222,7 +231,7 @@ public class TunnelGateway {
_messageIds.add(new Long(id));
}
}
- public List getMessageIds() {
+ public List getMessageIds() {
synchronized (Pending.this) {
if (_messageIds != null)
return new ArrayList(_messageIds);
@@ -231,6 +240,8 @@ public class TunnelGateway {
}
}
}
+
+ /** Extend for debugging */
class PendingImpl extends Pending {
public PendingImpl(I2NPMessage message, Hash toRouter, TunnelId toTunnel) {
super(message, toRouter, toTunnel, _context.clock().now());
diff --git a/router/java/src/net/i2p/router/tunnel/TunnelParticipant.java b/router/java/src/net/i2p/router/tunnel/TunnelParticipant.java
index a73208e3b..6263f8ec7 100644
--- a/router/java/src/net/i2p/router/tunnel/TunnelParticipant.java
+++ b/router/java/src/net/i2p/router/tunnel/TunnelParticipant.java
@@ -101,8 +101,11 @@ public class TunnelParticipant {
}
}
+/****
private int _periodMessagesTransferred;
private long _lastCoallesced = System.currentTimeMillis();
+****/
+
/**
* take note that the peers specified were able to push us data. hmm, is this safe?
* this could be easily gamed to get us to rank some peer of their choosing as quite
diff --git a/router/java/src/net/i2p/router/tunnel/pool/BuildExecutor.java b/router/java/src/net/i2p/router/tunnel/pool/BuildExecutor.java
index e1cec3607..5e4456490 100644
--- a/router/java/src/net/i2p/router/tunnel/pool/BuildExecutor.java
+++ b/router/java/src/net/i2p/router/tunnel/pool/BuildExecutor.java
@@ -54,9 +54,7 @@ class BuildExecutor implements Runnable {
// Get stat manager, get recognized bandwidth tiers
StatManager statMgr = _context.statManager();
- @SuppressWarnings("static-access")
- /* FIXME Accessing static field "BW_CAPABILITY_CHARS" FIXME */
- String bwTiers = _context.router().getRouterInfo().BW_CAPABILITY_CHARS;
+ String bwTiers = RouterInfo.BW_CAPABILITY_CHARS;
// For each bandwidth tier, create tunnel build agree/reject/expire stats
for (int i = 0; i < bwTiers.length(); i++) {
String bwTier = String.valueOf(bwTiers.charAt(i));
diff --git a/router/java/src/net/i2p/router/tunnel/pool/PooledTunnelCreatorConfig.java b/router/java/src/net/i2p/router/tunnel/pool/PooledTunnelCreatorConfig.java
index b2cb7c704..dfbdd3eca 100644
--- a/router/java/src/net/i2p/router/tunnel/pool/PooledTunnelCreatorConfig.java
+++ b/router/java/src/net/i2p/router/tunnel/pool/PooledTunnelCreatorConfig.java
@@ -79,8 +79,9 @@ public class PooledTunnelCreatorConfig extends TunnelCreatorConfig {
public TunnelPool getTunnelPool() { return _pool; }
- /* FIXME Exporting non-public type through public API FIXME */
- public void setTestJob(TestJob job) { _testJob = job; }
+ /** @deprecated unused, which makes _testJob unused - why is it here */
+ void setTestJob(TestJob job) { _testJob = job; }
+ /** does nothing, to be deprecated */
public void setExpireJob(Job job) { /* _expireJob = job; */ }
/**
diff --git a/router/java/src/net/i2p/router/tunnel/pool/TunnelPoolManager.java b/router/java/src/net/i2p/router/tunnel/pool/TunnelPoolManager.java
index 1eae1de45..bbee14a48 100644
--- a/router/java/src/net/i2p/router/tunnel/pool/TunnelPoolManager.java
+++ b/router/java/src/net/i2p/router/tunnel/pool/TunnelPoolManager.java
@@ -29,6 +29,8 @@ import net.i2p.stat.RateStat;
import net.i2p.util.I2PThread;
import net.i2p.util.Log;
import net.i2p.util.ObjectCounter;
+import net.i2p.util.SimpleScheduler;
+import net.i2p.util.SimpleTimer;
/**
*
@@ -42,7 +44,7 @@ public class TunnelPoolManager implements TunnelManagerFacade {
private final Map _clientOutboundPools;
private TunnelPool _inboundExploratory;
private TunnelPool _outboundExploratory;
- private BuildExecutor _executor;
+ private final BuildExecutor _executor;
private boolean _isShutdown;
public TunnelPoolManager(RouterContext ctx) {
@@ -263,6 +265,7 @@ public class TunnelPoolManager implements TunnelManagerFacade {
TunnelPool inbound = null;
TunnelPool outbound = null;
// should we share the clientPeerSelector across both inbound and outbound?
+ // or just one for all clients? why separate?
synchronized (_clientInboundPools) {
inbound = _clientInboundPools.get(dest);
if (inbound == null) {
@@ -284,11 +287,22 @@ public class TunnelPoolManager implements TunnelManagerFacade {
}
}
inbound.startup();
- try { Thread.sleep(3*1000); } catch (InterruptedException ie) {}
- outbound.startup();
+ SimpleScheduler.getInstance().addEvent(new DelayedStartup(outbound), 3*1000);
}
+ private static class DelayedStartup implements SimpleTimer.TimedEvent {
+ private TunnelPool pool;
+
+ public DelayedStartup(TunnelPool p) {
+ this.pool = p;
+ }
+
+ public void timeReached() {
+ this.pool.startup();
+ }
+ }
+
public void removeTunnels(Hash destination) {
if (destination == null) return;
if (_context.clientManager().isLocal(destination)) {
@@ -361,12 +375,11 @@ public class TunnelPoolManager implements TunnelManagerFacade {
_inboundExploratory = new TunnelPool(_context, this, inboundSettings, selector);
_inboundExploratory.startup();
- try { Thread.sleep(3*1000); } catch (InterruptedException ie) {}
TunnelPoolSettings outboundSettings = new TunnelPoolSettings();
outboundSettings.setIsExploratory(true);
outboundSettings.setIsInbound(false);
_outboundExploratory = new TunnelPool(_context, this, outboundSettings, selector);
- _outboundExploratory.startup();
+ SimpleScheduler.getInstance().addEvent(new DelayedStartup(_outboundExploratory), 3*1000);
// try to build up longer tunnels
_context.jobQueue().addJob(new BootstrapPool(_context, _inboundExploratory));