Changeset 855293d6

Timestamp:

Oct 10, 2008 5:34:25 PM (12 years ago)

Author:

zzz <zzz@…>

Branches:

master

Children:

f3f7537

Parents:

4c2d414

Message:

• Tunnels: Implement random discard to enforce share limit

Files:

• core/java/src/net/i2p/stat/StatManager.java (modified) (1 diff)
• history.txt (modified) (1 diff)
• router/java/src/net/i2p/router/RouterVersion.java (modified) (1 diff)
• router/java/src/net/i2p/router/tunnel/HopConfig.java (modified) (4 diffs)
• router/java/src/net/i2p/router/tunnel/InboundGatewayReceiver.java (modified) (2 diffs)
• router/java/src/net/i2p/router/tunnel/OutboundTunnelEndpoint.java (modified) (1 diff)
• router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java (modified) (6 diffs)
• router/java/src/net/i2p/router/tunnel/TunnelParticipant.java (modified) (5 diffs)

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

@@ -54 +54 @@
         "tunnel.decryptRequestTime,tunnel.fragmentedDropped,tunnel.participatingMessageCount,"+
         "tunnel.participatingTunnels,tunnel.testFailedTime,tunnel.testSuccessTime," +
-        "udp.sendPacketSize,udp.packetsRetransmitted" ;
+        "tunnel.participatingBandwidth,udp.sendPacketSize,udp.packetsRetransmitted" ;
     
     /**

history.txt

@@ +1 @@
+2008-10-10 zzz
+    * Profiles: Reduce reject penalty in
+      capacity calculation to avoid a congestion collapse
+    * Throttle: Change reject to BANDWIDTH from CRIT on shutdown
+      for improved anonymity
+    * Tunnels: Implement random discard to enforce share limit
+    * Tunnel Tests: Add time for outbound delay, to avoid
+      congestion collapse
+    * UDPPacketReader: Adjust logging
+    * build files: Change to source=1.5, target=1.5
+    * configpeer.jsp: Table cleanup
+    * i2psnark: Change default tunnel length from 1+1 to 2+0
+    * peers.jsp: Change <,> to in,out for UDP
+
 2008-10-09 sponge
     * Update version to -3

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

@@ -18 +18 @@
     public final static String ID = "$Revision: 1.548 $ $Date: 2008-06-07 23:00:00 $";
     public final static String VERSION = "0.6.4";
-    public final static long BUILD = 3;
+    public final static long BUILD = 4;
     public static void main(String args[]) {
         System.out.println("I2P Router version: " + VERSION + "-" + BUILD);

router/java/src/net/i2p/router/tunnel/HopConfig.java

@@ -29 +29 @@
     private long _messagesProcessed;
     private long _oldMessagesProcessed;
+    private long _messagesSent;
+    private long _oldMessagesSent;
     
     /** IV length for {@link #getReplyIV} */
@@ -45 +47 @@
         _messagesProcessed = 0;
         _oldMessagesProcessed = 0;
+        _messagesSent = 0;
+        _oldMessagesSent = 0;
     }
     
@@ -116 +120 @@
     
     /** take note of a message being pumped through this tunnel */
+    /** "processed" is for incoming and "sent" is for outgoing (could be dropped in between) */
     public void incrementProcessedMessages() { _messagesProcessed++; }
     public long getProcessedMessagesCount() { return _messagesProcessed; }
@@ -121 +126 @@
         long rv = _messagesProcessed - _oldMessagesProcessed;
         _oldMessagesProcessed = _messagesProcessed;
+        return rv;
+    }
+    public void incrementSentMessages() { _messagesSent++; }
+    public long getSentMessagesCount() { return _messagesSent; }
+    public long getRecentSentMessagesCount() {
+        long rv = _messagesSent - _oldMessagesSent;
+        _oldMessagesSent = _messagesSent;
         return rv;
     }

router/java/src/net/i2p/router/tunnel/InboundGatewayReceiver.java

@@ -23 +23 @@
     }
     public long receiveEncrypted(byte[] encrypted, boolean alreadySearched) {
+        if (!alreadySearched)
+            _config.incrementProcessedMessages();
         if (_target == null) {
             _target = _context.netDb().lookupRouterInfoLocally(_config.getSendTo());
@@ -34 +36 @@
         }
         
-        _config.incrementProcessedMessages();
+        if (_context.tunnelDispatcher().shouldDropParticipatingMessage())
+            return -1;
+        _config.incrementSentMessages();
         TunnelDataMessage msg = new TunnelDataMessage(_context);
         msg.setData(encrypted);

router/java/src/net/i2p/router/tunnel/OutboundTunnelEndpoint.java

@@ -42 +42 @@
                            + (toRouter != null ? toRouter.toBase64().substring(0,4) : "")
                            + (toTunnel != null ? toTunnel.getTunnelId() + "" : ""));
+            // don't drop it if we are the target
+            if ((!_context.routerHash().equals(toRouter)) &&
+                _context.tunnelDispatcher().shouldDropParticipatingMessage())
+                return;
+            _config.incrementSentMessages();
             _outDistributor.distribute(msg, toRouter, toTunnel);
         }

router/java/src/net/i2p/router/tunnel/TunnelDispatcher.java

@@ -19 +19 @@
 import net.i2p.router.Service;
 import net.i2p.router.peermanager.PeerProfile;
+import net.i2p.stat.Rate;
+import net.i2p.stat.RateStat;
 import net.i2p.util.Log;
 
@@ -111 +113 @@
                                          "Participating traffic", "Tunnels", 
                                          new long[] { 60*1000l, 60*10*1000l });
+        ctx.statManager().createRateStat("tunnel.participatingBandwidthOut", 
+                                         "Participating traffic", "Tunnels", 
+                                         new long[] { 60*1000l, 60*10*1000l });
+        ctx.statManager().createRateStat("tunnel.participatingMessageDropped", 
+                                         "Dropped for exceeding share limit", "Tunnels", 
+                                         new long[] { 60*1000l, 60*10*1000l });
         ctx.statManager().createRateStat("tunnel.participatingMessageCount", 
                                          "How many messages are sent through a participating tunnel?", "Tunnels", 
@@ -551 +559 @@
         long count = 0;
         long bw = 0;
+        long bwOut = 0;
         long tcount = 0;
         long tooYoung = _context.clock().now() - 60*1000;
@@ -558 +567 @@
             long c = cfg.getRecentMessagesCount();
             bw += c;
+            bwOut += cfg.getRecentSentMessagesCount();
             long created = cfg.getCreation();
             if (created > tooYoung || created < tooOld)
@@ -568 +578 @@
         _context.statManager().addRateData("tunnel.participatingMessageCount", count, 20*1000);
         _context.statManager().addRateData("tunnel.participatingBandwidth", bw*1024/20, 20*1000);
+        _context.statManager().addRateData("tunnel.participatingBandwidthOut", bwOut*1024/20, 20*1000);
         _context.statManager().addRateData("tunnel.participatingTunnels", size, 0);
+    }
+
+    /**
+     * Implement random early discard (RED) to enforce the share bandwidth limit.
+     * For now, this does not enforce the available bandwidth,
+     * we leave that to Throttle.
+     * This is similar to the code in ../RouterThrottleImpl.java
+     * We drop in proportion to how far over the limit we are.
+     * Perhaps an exponential function would be better?
+     */
+    public boolean shouldDropParticipatingMessage() {
+        RateStat rs = _context.statManager().getRate("tunnel.participatingBandwidth");
+        if (rs == null)
+            return false;
+        Rate r = rs.getRate(60*1000);
+        if (r == null)
+            return false;
+        // weight current period higher
+        long count = r.getLastEventCount() + (3 * r.getCurrentEventCount());
+        int bw = 0;
+        if (count > 0)
+            bw = (int) ((r.getLastTotalValue() + (3 * r.getCurrentTotalValue())) / count);
+        else
+            bw = (int) r.getLifetimeAverageValue();
+
+        int usedIn = Math.min(_context.router().get1sRateIn(), _context.router().get15sRateIn());
+        usedIn = Math.min(usedIn, bw);
+        if (usedIn <= 0)
+            return false;
+        int usedOut = Math.min(_context.router().get1sRate(true), _context.router().get15sRate(true));
+        usedOut = Math.min(usedOut, bw);
+        if (usedOut <= 0)
+            return false;
+        int used = Math.min(usedIn, usedOut);
+        int maxKBps = Math.min(_context.bandwidthLimiter().getInboundKBytesPerSecond(),
+                               _context.bandwidthLimiter().getOutboundKBytesPerSecond());
+        float share = (float) _context.router().getSharePercentage();
+
+        // start dropping at 95% of the limit
+        float maxBps = maxKBps * share * 1024f * 0.95f;
+        float pctDrop = (used - maxBps) / used;
+        if (pctDrop <= 0)
+            return false;
+        float rand = _context.random().nextFloat();
+        boolean reject = rand <= pctDrop;
+        if (reject) {
+            if (_log.shouldLog(Log.WARN)) {
+                int availBps = (int) (((maxKBps*1024)*share) - used);
+                _log.warn("Drop part. msg. avail/max/used " + availBps + "/" + (int) maxBps + "/" 
+                          + used + " %Drop = " + pctDrop);
+            }
+            _context.statManager().addRateData("tunnel.participatingMessageDropped", 1, 0);
+        }
+        return reject;
     }
 
@@ -686 +751 @@
                 }
             }
-            if (_log.shouldLog(Log.INFO)) {
+            if (_log.shouldLog(Log.DEBUG)) {
                 long now = getContext().clock().now();
-                _log.info("Scheduling leave in " + DataHelper.formatDuration(dropTime.longValue()-now) +": " + cfg);
+                _log.debug("Scheduling leave in " + DataHelper.formatDuration(dropTime.longValue()-now) +": " + cfg);
             }
         }

router/java/src/net/i2p/router/tunnel/TunnelParticipant.java

@@ -76 +76 @@
         
         if ( (_config != null) && (_config.getSendTo() != null) ) {
+            _config.incrementProcessedMessages();
             RouterInfo ri = _nextHopCache;
             if (ri == null)
@@ -83 +84 @@
                     _log.debug("Send off to nextHop directly (" + _config.getSendTo().toBase64().substring(0,4) 
                               + " for " + msg);
-                _config.incrementProcessedMessages();
                 send(_config, msg, ri);
-                if (_config != null)
-                    incrementThroughput(_config.getReceiveFrom());
+                // see comments below
+                //if (_config != null)
+                //    incrementThroughput(_config.getReceiveFrom());
             } else {
                 if (_log.shouldLog(Log.WARN))
@@ -110 +111 @@
      * now.
      */
+/****
     private void incrementThroughput(Hash prev) {
         if (true) return;
@@ -124 +126 @@
         }
     }
+****/
     
     public int getCompleteCount() { 
@@ -148 +151 @@
 
     private void send(HopConfig config, TunnelDataMessage msg, RouterInfo ri) {
+        if (_context.tunnelDispatcher().shouldDropParticipatingMessage())
+            return;
+        _config.incrementSentMessages();
         long oldId = msg.getUniqueId();
         long newId = _context.random().nextLong(I2NPMessage.MAX_ID_VALUE);