2006-03-15 jrandom

* Further stat cleanup
    * Keep track of how many peers we are actively trying to communicate with,
      beyond those who are just trying to communicate with us.
    * Further router tunnel participation throttle revisions to avoid spurious
      rejections
    * Rate stat display cleanup (thanks ripple!)
    * Don't even try to send messages that have been queued too long

core/java/src/net/i2p/stat/Rate.java

@@ -189,7 +189,7 @@ public class Rate {
             // how much were we off by?  (so that we can sample down the measured values)
             double periodFactor = measuredPeriod / (double)_period;
             _lastTotalValue = _currentTotalValue / periodFactor;
-            _lastEventCount = (long) (_currentEventCount / periodFactor);
+            _lastEventCount = (long) ( (_currentEventCount + periodFactor - 1) / periodFactor);
             _lastTotalEventTime = (long) (_currentTotalEventTime / periodFactor);
             _lastCoalesceDate = now;
 

history.txt

@@ -1,4 +1,13 @@
-$Id: history.txt,v 1.428 2006/03/05 12:07:15 jrandom Exp $
+$Id: history.txt,v 1.429 2006/03/05 20:57:50 zzz Exp $
+
+2006-03-15  jrandom
+    * Further stat cleanup
+    * Keep track of how many peers we are actively trying to communicate with,
+      beyond those who are just trying to communicate with us.
+    * Further router tunnel participation throttle revisions to avoid spurious
+      rejections
+    * Rate stat display cleanup (thanks ripple!)
+    * Don't even try to send messages that have been queued too long
 
 2006-03-05  zzz
     * Remove the +++--- from the logs on i2psnark startup

router/java/src/net/i2p/router/CommSystemFacade.java

@@ -29,6 +29,7 @@ public abstract class CommSystemFacade implements Service {
     public Set createAddresses() { return new HashSet(); }
     
     public int countActivePeers() { return 0; }
+    public int countActiveSendPeers() { return 0; }
     public List getMostRecentErrorMessages() { return Collections.EMPTY_LIST; }
     
     /**

router/java/src/net/i2p/router/Router.java

@@ -1019,6 +1019,7 @@ class CoalesceStatsJob extends JobImpl {
         ctx.statManager().createRateStat("bw.sendRate", "Low level bandwidth send rate, averaged every minute", "Bandwidth", new long[] { 60*1000l, 5*60*1000l, 10*60*1000l, 60*60*1000l });
         ctx.statManager().createRateStat("bw.recvRate", "Low level bandwidth receive rate, averaged every minute", "Bandwidth", new long[] { 60*1000l, 5*60*1000l, 10*60*1000l, 60*60*1000l });
         ctx.statManager().createRateStat("router.activePeers", "How many peers we are actively talking with", "Throttle", new long[] { 5*60*1000, 60*60*1000 });
+        ctx.statManager().createRateStat("router.activeSendPeers", "How many peers have sent messages to this minute", "Throttle", new long[] { 5*60*1000, 60*60*1000 });
         ctx.statManager().createRateStat("router.highCapacityPeers", "How many high capacity peers we know", "Throttle", new long[] { 5*60*1000, 60*60*1000 });
         ctx.statManager().createRateStat("router.fastPeers", "How many fast peers we know", "Throttle", new long[] { 5*60*1000, 60*60*1000 });
     }
@@ -1027,6 +1028,9 @@ class CoalesceStatsJob extends JobImpl {
         int active = getContext().commSystem().countActivePeers();
         getContext().statManager().addRateData("router.activePeers", active, 60*1000);
 
+        int activeSend = getContext().commSystem().countActiveSendPeers();
+        getContext().statManager().addRateData("router.activeSendPeers", activeSend, 60*1000);
+
         int fast = getContext().profileOrganizer().countFastPeers();
         getContext().statManager().addRateData("router.fastPeers", fast, 60*1000);
 

router/java/src/net/i2p/router/RouterThrottleImpl.java

@@ -249,35 +249,50 @@ class RouterThrottleImpl implements RouterThrottle {
         int maxKBps = Math.min(_context.bandwidthLimiter().getOutboundKBytesPerSecond(), _context.bandwidthLimiter().getInboundKBytesPerSecond());
         int used1s = 0; //get1sRate(_context); // dont throttle on the 1s rate, its too volatile
         int used1m = get1mRate(_context);
-        int used5m = get5mRate(_context);
+        int used5m = 0; //get5mRate(_context); // don't throttle on the 5m rate, as that'd hide available bandwidth
         int used = Math.max(Math.max(used1s, used1m), used5m);
         int availBps = (int)(((maxKBps*1024) - used) * getSharePercentage());
 
         _context.statManager().addRateData("router.throttleTunnelBytesUsed", used, maxKBps);
         _context.statManager().addRateData("router.throttleTunnelBytesAllowed", availBps, (long)bytesAllocated);
 
+        /*
         if (availBps <= 8*1024) {
             // lets be more conservative for people near their limit and assume 1KBps per tunnel
-            return ( (numTunnels + 1)*1024 < availBps);
+            boolean rv = ( (numTunnels + 1)*1024 < availBps);
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Nearly full router (" + availBps + ") with " + numTunnels + " tunnels, allow a new request? " + rv);
+            return rv;
         }
+        */
 
         double growthFactor = ((double)(numTunnels+1))/(double)numTunnels;
         double toAllocate = (numTunnels > 0 ? bytesAllocated * growthFactor : 0);
         
-        double allocatedKBps = toAllocate / (10 * 60 * 1024);
-        double pctFull = allocatedKBps / availBps;
+        double allocatedBps = toAllocate / (10 * 60);
+        double pctFull = allocatedBps / availBps;
         
         if ( (pctFull < 1.0) && (pctFull >= 0.0) ) { // (_context.random().nextInt(100) > 100 * pctFull) {
             if (_log.shouldLog(Log.DEBUG))
-                _log.debug("Probabalistically allowing the tunnel w/ " + pctFull + " of our " + availBps
-                           + "Bps/" + allocatedKBps + "KBps allocated through " + numTunnels + " tunnels");
+                _log.debug("Allowing the tunnel w/ " + pctFull + " of our " + availBps
+                           + "Bps/" + allocatedBps + "KBps allocated through " + numTunnels + " tunnels");
             return true;
         } else {
-            if (_log.shouldLog(Log.WARN))
-                _log.warn("Rejecting the tunnel w/ " + pctFull + " of our " + availBps 
-                           + "Bps allowed (" + toAllocate + "bytes / " + allocatedKBps 
-                           + "KBps) through " + numTunnels + " tunnels");
-            return false;
+            double probAllow = availBps / (allocatedBps + availBps);
+            boolean allow = _context.random().nextDouble() <= probAllow;
+            if (allow) {
+                if (_log.shouldLog(Log.INFO))
+                    _log.info("Probabalistically allowing the tunnel w/ " + (pctFull*100d) + "% of our " + availBps 
+                               + "Bps allowed (" + toAllocate + "bytes / " + allocatedBps 
+                               + "Bps) through " + numTunnels + " tunnels");
+                return true;
+            } else {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Rejecting the tunnel w/ " + (pctFull*100d) + "% of our " + availBps 
+                               + "Bps allowed (" + toAllocate + "bytes / " + allocatedBps 
+                               + "Bps) through " + numTunnels + " tunnels");
+                return false;
+            }
         }
     }
     
@@ -287,7 +302,7 @@ class RouterThrottleImpl implements RouterThrottle {
      *
      */
     private double getSharePercentage() {
-        String pct = _context.getProperty(PROP_BANDWIDTH_SHARE_PERCENTAGE, "0.8");
+        String pct = _context.getProperty(PROP_BANDWIDTH_SHARE_PERCENTAGE);
         if (pct != null) {
             try {
                 double d = Double.parseDouble(pct);

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.369 $ $Date: 2006/03/05 12:07:13 $";
+    public final static String ID = "$Revision: 1.370 $ $Date: 2006/03/05 20:57:47 $";
     public final static String VERSION = "0.6.1.12";
-    public final static long BUILD = 7;
+    public final static long BUILD = 8;
     public static void main(String args[]) {
         System.out.println("I2P Router version: " + VERSION + "-" + BUILD);
         System.out.println("Router ID: " + RouterVersion.ID);

router/java/src/net/i2p/router/StatisticsManager.java

@@ -287,8 +287,14 @@ public class StatisticsManager implements Service {
     
     private String getPeriod(Rate rate) { return DataHelper.formatDuration(rate.getPeriod()); }
 
-    private final String num(double num) { synchronized (_fmt) { return _fmt.format(num); } }
-    private final String pct(double num) { synchronized (_pct) { return _pct.format(num); } }
+    private final String num(double num) { 
+        if (num < 0) num = 0;
+        synchronized (_fmt) { return _fmt.format(num); } 
+    }
+    private final String pct(double num) { 
+        if (num < 0) num = 0;
+        synchronized (_pct) { return _pct.format(num); } 
+    }
    
     public void renderStatusHTML(Writer out) { }
 }

router/java/src/net/i2p/router/message/GarlicMessageBuilder.java

@@ -122,8 +122,10 @@ public class GarlicMessageBuilder {
         msg.setMessageExpiration(config.getExpiration());
         
         long timeFromNow = config.getExpiration() - ctx.clock().now();
-        if (timeFromNow < 1*1000)
+        if (timeFromNow < 1*1000) {
             log.error("Building a message expiring in " + timeFromNow + "ms: " + config, new Exception("created by"));
+            return null;
+        }
         
         if (log.shouldLog(Log.WARN))
             log.warn("CloveSet size for message " + msg.getUniqueId() + " is " + cloveSet.length

router/java/src/net/i2p/router/message/OutboundClientMessageOneShotJob.java

@@ -313,6 +313,10 @@ public class OutboundClientMessageOneShotJob extends JobImpl {
      */
     private void send() {
         if (_finished) return;
+        if (getContext().clock().now() >= _overallExpiration) {
+            dieFatal();
+            return;
+        }
         boolean wantACK = true;
         int existingTags = GarlicMessageBuilder.estimateAvailableTags(getContext(), _leaseSet.getEncryptionKey());
         if (existingTags > 30)
@@ -333,8 +337,9 @@ public class OutboundClientMessageOneShotJob extends JobImpl {
         buildClove();
         if (_log.shouldLog(Log.DEBUG))
             _log.debug(getJobId() + ": Clove built to " + _toString);
+        long msgExpiration = _overallExpiration; // getContext().clock().now() + OVERALL_TIMEOUT_MS_DEFAULT;
         GarlicMessage msg = OutboundClientMessageJobHelper.createGarlicMessage(getContext(), token, 
-                                                                               _overallExpiration, key, 
+                                                                               msgExpiration, key, 
                                                                                _clove, _from.calculateHash(), 
                                                                                _to, _inTunnel,
                                                                                sessKey, tags, 
@@ -344,7 +349,7 @@ public class OutboundClientMessageOneShotJob extends JobImpl {
             // (should we always fail for this? or should we send it anyway, even if
             // we dont receive the reply? hmm...)
             if (_log.shouldLog(Log.WARN))
-                _log.warn(getJobId() + ": Unable to create the garlic message (no tunnels left) to " + _toString);
+                _log.warn(getJobId() + ": Unable to create the garlic message (no tunnels left or too lagged) to " + _toString);
             getContext().statManager().addRateData("client.dispatchNoTunnels", getContext().clock().now() - _start, 0);            
             dieFatal();
             return;

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

@@ -694,10 +694,10 @@ public class KademliaNetworkDatabaseFacade extends NetworkDatabaseFacade {
         if (err != null)
             throw new IllegalArgumentException("Invalid store attempt - " + err);
         
-        if (_log.shouldLog(Log.INFO))
-            _log.info("RouterInfo " + key.toBase64() + " is stored with "
-            + routerInfo.getOptions().size() + " options on "
-            + new Date(routerInfo.getPublished()));
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("RouterInfo " + key.toBase64() + " is stored with "
+                       + routerInfo.getOptions().size() + " options on "
+                       + new Date(routerInfo.getPublished()));
     
         _context.peerManager().setCapabilities(key, routerInfo.getCapabilities());
         _ds.put(key, routerInfo);

router/java/src/net/i2p/router/transport/CommSystemFacadeImpl.java

@@ -54,6 +54,7 @@ public class CommSystemFacadeImpl extends CommSystemFacade {
     }
     
     public int countActivePeers() { return (_manager == null ? 0 : _manager.countActivePeers()); }
+    public int countActiveSendPeers() { return (_manager == null ? 0 : _manager.countActiveSendPeers()); } 
     
     public List getBids(OutNetMessage msg) {
         return _manager.getBids(msg);

router/java/src/net/i2p/router/transport/TransportImpl.java

@@ -64,6 +64,10 @@ public abstract class TransportImpl implements Transport {
      *
      */
     public int countActivePeers() { return 0; }
+    /**
+     * How many peers are we actively sending messages to (this minute)
+     */
+    public int countActiveSendPeers() { return 0; }
     
     public List getMostRecentErrorMessages() { return Collections.EMPTY_LIST; }
     /**

router/java/src/net/i2p/router/transport/TransportManager.java

@@ -120,6 +120,14 @@ public class TransportManager implements TransportEventListener {
         return peers;
     }
     
+    public int countActiveSendPeers() { 
+        int peers = 0;
+        for (int i = 0; i < _transports.size(); i++) {
+            peers += ((Transport)_transports.get(i)).countActiveSendPeers();
+        }
+        return peers;
+    }
+    
     public short getReachabilityStatus() { 
         if (_transports.size() <= 0) return CommSystemFacade.STATUS_UNKNOWN;
         short status[] = new short[_transports.size()];

router/java/src/net/i2p/router/transport/udp/UDPTransport.java

@@ -1164,6 +1164,22 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
         return active;
     }
     
+    public int countActiveSendPeers() {
+        long now = _context.clock().now();
+        int active = 0;
+        int inactive = 0;
+        synchronized (_peersByIdent) {
+            for (Iterator iter = _peersByIdent.values().iterator(); iter.hasNext(); ) {
+                PeerState peer = (PeerState)iter.next();
+                if (now-peer.getLastSendFullyTime() > 1*60*1000)
+                    inactive++;
+                else
+                    active++;
+            }
+        }
+        return active;
+    }
+    
     private static class AlphaComparator implements Comparator {
         private static final AlphaComparator _instance = new AlphaComparator();
         public static final AlphaComparator instance() { return _instance; }