2005-03-01 jrandom

* Really disable the streaming lib packet caching
    * Synchronized a message handling point in the SDK (even though its use is
      already essentially single threaded, its better to play it safe)
    * Don't add new RepublishLeaseSetJobs on failure, just requeue up the
      existing one (duh)
    * Throttle the number of concurrent pending tunnel builds across all
      pools, in addition to simply throttling the number of new requests per
      minute for each pool individually.  This should avoid the cascading
      failure when tunnel builds take too long, as no new builds will be
      created until the previous ones are handled.
    * Factored out and extended the DataHelper's unit tests for dealing with
      long and date formatting.
    * Explicitly specify the HTTP auth realm as "i2prouter", though this
      alone doesn't address the bug where jetty asks for authentication too
      much.  (thanks orion!)
    * Updated the StreamSinkServer to ignore all read bytes, rather than write
      them to the filesystem.

router/java/src/net/i2p/router/RouterVersion.java

@@ -15,9 +15,9 @@ import net.i2p.CoreVersion;
  *
  */
 public class RouterVersion {
-    public final static String ID = "$Revision: 1.154 $ $Date: 2005/02/26 19:03:42 $";
+    public final static String ID = "$Revision: 1.155 $ $Date: 2005/02/27 17:09:37 $";
     public final static String VERSION = "0.5.0.1";
-    public final static long BUILD = 6;
+    public final static long BUILD = 7;
     public static void main(String args[]) {
         System.out.println("I2P Router version: " + VERSION);
         System.out.println("Router ID: " + RouterVersion.ID);

router/java/src/net/i2p/router/networkdb/kademlia/RepublishLeaseSetJob.java

@@ -76,9 +76,7 @@ public class RepublishLeaseSetJob extends JobImpl {
         public void runJob() { 
             if (_log.shouldLog(Log.WARN))
                 _log.warn("FAILED publishing of the leaseSet for " + _dest.toBase64());
-            LeaseSet ls = _facade.lookupLeaseSetLocally(_dest);
-            if ( (ls != null) && (ls.isCurrent(0)) )
-                getContext().jobQueue().addJob(new RepublishLeaseSetJob(getContext(), _facade, _dest));
+            RepublishLeaseSetJob.this.requeue(30*1000);
         }
     }
 }

router/java/src/net/i2p/router/tunnel/pool/OnCreatedJob.java

@@ -29,6 +29,8 @@ class OnCreatedJob extends JobImpl {
         } else {
             getContext().tunnelDispatcher().joinOutbound(_cfg);
         }
+        
+        _pool.getManager().buildComplete();
         _pool.addTunnel(_cfg);
         TestJob testJob = (_cfg.getLength() > 1 ? new TestJob(getContext(), _cfg, _pool) : null);
         RebuildJob rebuildJob = new RebuildJob(getContext(), _cfg, _pool);

router/java/src/net/i2p/router/tunnel/pool/TunnelBuilder.java

@@ -31,12 +31,16 @@ public class TunnelBuilder {
         buildTunnel(ctx, pool, false);
     }
     public void buildTunnel(RouterContext ctx, TunnelPool pool, boolean zeroHop) {
-        if (!pool.isAlive()) return;
+        if (!pool.isAlive()) {
+            pool.getManager().buildComplete();
+            return;
+        }
         // this is probably overkill (ya think?)
         pool.refreshSettings();
         
         PooledTunnelCreatorConfig cfg = configTunnel(ctx, pool, zeroHop);
         if (cfg == null) {
+            pool.getManager().buildComplete();
             return;
         }
         OnCreatedJob onCreated = new OnCreatedJob(ctx, pool, cfg);
@@ -111,6 +115,7 @@ public class TunnelBuilder {
         public String getName() { return "Tunnel create failed"; }
         public void runJob() {
             // yikes, nothing left, lets get some backup (if we're allowed)
+            _pool.getManager().buildComplete();
             _pool.refreshBuilders();
         }
     }

router/java/src/net/i2p/router/tunnel/pool/TunnelPool.java

@@ -120,6 +120,13 @@ public class TunnelPool {
             if (build > (MAX_BUILDS_PER_MINUTE - _buildsThisMinute))
                 build = (MAX_BUILDS_PER_MINUTE - _buildsThisMinute);
             
+            int wanted = build;
+            build = _manager.allocateBuilds(build);
+            
+            if ( (wanted != build) && (_log.shouldLog(Log.ERROR)) )
+                _log.error("Wanted to build " + wanted + " tunnels, but throttled down to " 
+                           + build + ", due to concurrent requests (cpu overload?)");
+            
             for (int i = 0; i < build; i++)
                 _builder.buildTunnel(_context, this);
             _buildsThisMinute += build;
@@ -130,6 +137,8 @@ public class TunnelPool {
         }
     }
     
+    TunnelPoolManager getManager() { return _manager; }
+    
     void refreshSettings() {
         if (_settings.getDestination() != null) {
             return; // don't override client specified settings

router/java/src/net/i2p/router/tunnel/pool/TunnelPoolManager.java

@@ -40,6 +40,10 @@ public class TunnelPoolManager implements TunnelManagerFacade {
     private Map _clientOutboundPools;
     private TunnelPool _inboundExploratory;
     private TunnelPool _outboundExploratory;
+    /** how many build requests are in process */
+    private int _outstandingBuilds;
+    /** max # of concurrent build requests */
+    private int _maxOutstandingBuilds;
     
     public TunnelPoolManager(RouterContext ctx) {
         _context = ctx;
@@ -53,6 +57,16 @@ public class TunnelPoolManager implements TunnelManagerFacade {
 
         _clientInboundPools = new HashMap(4);
         _clientOutboundPools = new HashMap(4);
+        _outstandingBuilds = 0;
+        _maxOutstandingBuilds = 10;
+        String max = ctx.getProperty("router.tunnel.maxConcurrentBuilds", "10");
+        if (max != null) {
+            try {
+                _maxOutstandingBuilds = Integer.parseInt(max);
+            } catch (NumberFormatException nfe) {
+                _maxOutstandingBuilds = 10;
+            }
+        }
         
         ctx.statManager().createRateStat("tunnel.testSuccessTime", 
                                          "How long do successful tunnel tests take?", "Tunnels", 
@@ -248,6 +262,35 @@ public class TunnelPoolManager implements TunnelManagerFacade {
             outbound.shutdown();
     }
     
+    /** 
+     * Check to make sure we can build this many new tunnels (throttled so
+     * we don't build too many at a time across all pools).
+     *
+     * @param wanted how many tunnels the pool wants to build
+     * @return how many are allowed to be built
+     */
+    int allocateBuilds(int wanted) {
+        synchronized (this) {
+            if (_outstandingBuilds >= _maxOutstandingBuilds)
+                return 0;
+            if (_outstandingBuilds + wanted < _maxOutstandingBuilds) {
+                _outstandingBuilds += wanted;
+                return wanted;
+            } else {
+                int allowed = _maxOutstandingBuilds - _outstandingBuilds;
+                _outstandingBuilds = _maxOutstandingBuilds;
+                return allowed;
+            }
+        }
+    }
+
+    void buildComplete() {
+        synchronized (this) {
+            if (_outstandingBuilds > 0)
+                _outstandingBuilds--;
+        }
+    }
+    
     public void startup() { 
         TunnelBuilder builder = new TunnelBuilder();
         ExploratoryPeerSelector selector = new ExploratoryPeerSelector();