Implement outbound bandwidth limiting for i2psnark

apps/i2psnark/java/build.xml

@@ -14,7 +14,7 @@
             srcdir="./src" 
             debug="true" deprecation="on" source="1.3" target="1.3" 
             destdir="./build/obj" 
-            classpath="../../../core/java/build/i2p.jar:../../jetty/jettylib/org.mortbay.jetty.jar:../../jetty/jettylib/javax.servlet.jar:../../ministreaming/java/build/mstreaming.jar" />
+            classpath="../../../core/java/build/i2p.jar:../../../router/java/build/router.jar:../../jetty/jettylib/org.mortbay.jetty.jar:../../jetty/jettylib/javax.servlet.jar:../../ministreaming/java/build/mstreaming.jar" />
     </target>
     <target name="jar" depends="builddep, compile">
         <jar destfile="./build/i2psnark.jar" basedir="./build/obj" includes="**/*.class" excludes="**/*Servlet.class">

apps/i2psnark/java/src/org/klomp/snark/I2PSnarkUtil.java

@@ -34,6 +34,7 @@ public class I2PSnarkUtil {
     private boolean _configured;
     private Set _shitlist;
     private int _maxUploaders;
+    private int _maxUpBW;
     
     private I2PSnarkUtil() {
         _context = I2PAppContext.getGlobalContext();
@@ -79,6 +80,11 @@ public class I2PSnarkUtil {
         _configured = true;
     }
     
+    public void setMaxUpBW(int limit) {
+        _maxUpBW = limit;
+        _configured = true;
+    }
+    
     public String getI2CPHost() { return _i2cpHost; }
     public int getI2CPPort() { return _i2cpPort; }
     public Map getI2CPOptions() { return _opts; }
@@ -86,6 +92,7 @@ public class I2PSnarkUtil {
     public int getEepProxyPort() { return _proxyPort; }
     public boolean getEepProxySet() { return _shouldProxy; }
     public int getMaxUploaders() { return _maxUploaders; }
+    public int getMaxUpBW() { return _maxUpBW; }
     
     /**
      * Connect to the router, if we aren't already
@@ -197,6 +204,8 @@ public class I2PSnarkUtil {
     }
     
     String getOurIPString() {
+        if (_manager == null)
+            return "unknown";
         I2PSession sess = _manager.getSession();
         if (sess != null) {
             Destination dest = sess.getMyDestination();

apps/i2psnark/java/src/org/klomp/snark/PeerCheckerTask.java

@@ -68,6 +68,7 @@ class PeerCheckerTask extends TimerTask
         // we will add them back to the end of the list.
         List removed = new ArrayList();
         int uploadLimit = coordinator.allowedUploaders();
+        boolean overBWLimit = coordinator.overUpBWLimit();
         while (it.hasNext())
           {
             Peer peer = (Peer)it.next();
@@ -109,7 +110,8 @@ class PeerCheckerTask extends TimerTask
             // (Note use of coordinator.uploaders)
             if (((coordinator.uploaders == uploadLimit
                 && coordinator.interestedAndChoking > 0)
-                || coordinator.uploaders > uploadLimit)
+                || coordinator.uploaders > uploadLimit
+                || overBWLimit)
                 && !peer.isChoking())
               {
                 // Check if it still wants pieces from us.
@@ -125,6 +127,15 @@ class PeerCheckerTask extends TimerTask
                     it.remove();
                     removed.add(peer);
                   }
+                else if (overBWLimit)
+                  {
+                    Snark.debug("BW limit, choke peer: " + peer,
+                                Snark.INFO);
+                    peer.setChoking(true);
+                    uploaders--;
+                    coordinator.uploaders--;
+                    removedCount++;
+                  }
                 else if (peer.isInteresting() && peer.isChoked())
                   {
                     // If they are choking us make someone else a downloader
@@ -209,7 +220,8 @@ class PeerCheckerTask extends TimerTask
           }
         
         // Optimistically unchoke a peer
-        coordinator.unchokePeer();
+        if (!overBWLimit)
+            coordinator.unchokePeer();
 
         // Put peers back at the end of the list that we removed earlier.
         coordinator.peers.addAll(removed);

apps/i2psnark/java/src/org/klomp/snark/PeerCoordinator.java

@@ -52,8 +52,8 @@ public class PeerCoordinator implements PeerListener
   private long uploaded;
   private long downloaded;
   final static int RATE_DEPTH = 6; // make following arrays RATE_DEPTH long
-  private long uploaded_old[] = {0,0,0,0,0,0};
+  private long uploaded_old[] = {-1,-1,-1,-1,-1,-1};
-  private long downloaded_old[] = {0,0,0,0,0,0};
+  private long downloaded_old[] = {-1,-1,-1,-1,-1,-1};
 
   // synchronize on this when changing peers or downloaders
   final List peers = new ArrayList();
@@ -195,11 +195,17 @@ public class PeerCoordinator implements PeerListener
   private long getRate(long array[])
   {
     long rate = 0;
+    int i = 0;
     synchronized(array) {
-      for (int i = 0; i < RATE_DEPTH; i++)
+      for ( ; i < RATE_DEPTH; i++) {
+        if (array[i] < 0)
+            break;
         rate += array[i];
+      }
     }
-    return rate / (RATE_DEPTH * CHECK_PERIOD / 1000);
+    if (i == 0)
+        return 0;
+    return rate / (i * CHECK_PERIOD / 1000);
   }
 
   public MetaInfo getMetaInfo()
@@ -819,5 +825,10 @@ public class PeerCoordinator implements PeerListener
     else
         return MAX_UPLOADERS;
   }
+
+  public boolean overUpBWLimit()
+  {
+    return Snark.overUpBWLimit();
+  }
 }
 

apps/i2psnark/java/src/org/klomp/snark/Snark.java

@@ -776,4 +776,18 @@ public class Snark
     // Snark.debug("Total uploaders: " + totalUploaders + " Limit: " + limit, Snark.DEBUG);
     return totalUploaders > limit;
   }
+
+  public static boolean overUpBWLimit() {
+    PeerCoordinatorSet coordinators = PeerCoordinatorSet.instance();
+    if (coordinators == null)
+      return false;
+    long total = 0;
+    for (Iterator iter = coordinators.iterator(); iter.hasNext(); ) {
+      PeerCoordinator c = (PeerCoordinator)iter.next();
+      if (!c.halted())
+        total += c.getUploadRate();
+    }
+    long limit = 1024l * I2PSnarkUtil.instance().getMaxUpBW();
+    return total > limit;
+  }
 }

apps/i2psnark/java/src/org/klomp/snark/SnarkManager.java

@@ -5,6 +5,7 @@ import java.util.*;
 import net.i2p.I2PAppContext;
 import net.i2p.data.Base64;
 import net.i2p.data.DataHelper;
+import net.i2p.router.RouterContext;
 import net.i2p.util.I2PThread;
 import net.i2p.util.Log;
 
@@ -30,6 +31,7 @@ public class SnarkManager implements Snark.CompleteListener {
     public static final String PROP_EEP_HOST = "i2psnark.eepHost";
     public static final String PROP_EEP_PORT = "i2psnark.eepPort";
     public static final String PROP_UPLOADERS_TOTAL = "i2psnark.uploaders.total";
+    public static final String PROP_UPBW_MAX = "i2psnark.upbw.max";
     public static final String PROP_DIR = "i2psnark.dir";
     public static final String PROP_META_PREFIX = "i2psnark.zmeta.";
     public static final String PROP_META_BITFIELD_SUFFIX = ".bitfield";
@@ -41,6 +43,9 @@ public class SnarkManager implements Snark.CompleteListener {
     public static final String PROP_OPENTRACKERS = "i2psnark.opentrackers";
     public static final String DEFAULT_OPENTRACKERS = "http://tracker.welterde.i2p/a";
     
+    public static final int MIN_UP_BW = 2;
+    public static final int DEFAULT_MAX_UP_BW = 10;
+
     private SnarkManager() {
         _snarks = new HashMap();
         _addSnarkLock = new Object();
@@ -112,6 +117,12 @@ public class SnarkManager implements Snark.CompleteListener {
             _config.setProperty(PROP_EEP_PORT, "4444");
         if (!_config.containsKey(PROP_UPLOADERS_TOTAL))
             _config.setProperty(PROP_UPLOADERS_TOTAL, "" + Snark.MAX_TOTAL_UPLOADERS);
+        if (!_config.containsKey(PROP_UPBW_MAX)) {
+            if (_context instanceof RouterContext)
+                _config.setProperty(PROP_UPBW_MAX, "" + (((RouterContext)_context).bandwidthLimiter().getOutboundKBytesPerSecond() / 3));
+            else
+                _config.setProperty(PROP_UPBW_MAX, "" + DEFAULT_MAX_UP_BW);
+        }
         if (!_config.containsKey(PROP_DIR))
             _config.setProperty(PROP_DIR, "i2psnark");
         if (!_config.containsKey(PROP_AUTO_START))
@@ -143,6 +154,7 @@ public class SnarkManager implements Snark.CompleteListener {
         if (eepHost != null)
             I2PSnarkUtil.instance().setProxy(eepHost, eepPort);
         I2PSnarkUtil.instance().setMaxUploaders(getInt(PROP_UPLOADERS_TOTAL, Snark.MAX_TOTAL_UPLOADERS));
+        I2PSnarkUtil.instance().setMaxUpBW(getInt(PROP_UPBW_MAX, DEFAULT_MAX_UP_BW));
         getDataDir().mkdirs();
     }
     
@@ -159,7 +171,7 @@ public class SnarkManager implements Snark.CompleteListener {
     
     public void updateConfig(String dataDir, boolean autoStart, String seedPct, String eepHost, 
                              String eepPort, String i2cpHost, String i2cpPort, String i2cpOpts,
-                             String upLimit, boolean useOpenTrackers, String openTrackers) {
+                             String upLimit, String upBW, boolean useOpenTrackers, String openTrackers) {
         boolean changed = false;
         if (eepHost != null) {
             int port = I2PSnarkUtil.instance().getEepProxyPort();
@@ -188,6 +200,20 @@ public class SnarkManager implements Snark.CompleteListener {
                 }
             }
         }
+        if (upBW != null) {
+            int limit = I2PSnarkUtil.instance().getMaxUpBW();
+            try { limit = Integer.parseInt(upBW); } catch (NumberFormatException nfe) {}
+            if ( limit != I2PSnarkUtil.instance().getMaxUpBW()) {
+                if ( limit >= MIN_UP_BW ) {
+                    I2PSnarkUtil.instance().setMaxUpBW(limit);
+                    changed = true;
+                    _config.setProperty(PROP_UPBW_MAX, "" + limit);
+                    addMessage("Up BW limit changed to " + limit + "KBps");
+                } else {
+                    addMessage("Minimum Up BW limit is " + MIN_UP_BW + "KBps");
+                }
+            }
+        }
         if (i2cpHost != null) {
             int oldI2CPPort = I2PSnarkUtil.instance().getI2CPPort();
             String oldI2CPHost = I2PSnarkUtil.instance().getI2CPHost();

apps/i2psnark/java/src/org/klomp/snark/web/I2PSnarkServlet.java

@@ -293,9 +293,10 @@ public class I2PSnarkServlet extends HttpServlet {
             String i2cpPort = req.getParameter("i2cpPort");
             String i2cpOpts = req.getParameter("i2cpOpts");
             String upLimit = req.getParameter("upLimit");
+            String upBW = req.getParameter("upBW");
             boolean useOpenTrackers = req.getParameter("useOpenTrackers") != null;
             String openTrackers = req.getParameter("openTrackers");
-            _manager.updateConfig(dataDir, autoStart, seedPct, eepHost, eepPort, i2cpHost, i2cpPort, i2cpOpts, upLimit, useOpenTrackers, openTrackers);
+            _manager.updateConfig(dataDir, autoStart, seedPct, eepHost, eepPort, i2cpHost, i2cpPort, i2cpOpts, upLimit, upBW, useOpenTrackers, openTrackers);
         } else if ("Create torrent".equals(action)) {
             String baseData = req.getParameter("baseFile");
             if (baseData != null) {
@@ -700,7 +701,9 @@ public class I2PSnarkServlet extends HttpServlet {
         out.write("</select><br />\n");
 */
         out.write("Total uploader limit: <input type=\"text\" name=\"upLimit\" value=\""
-                  + I2PSnarkUtil.instance().getMaxUploaders() + "\" size=\"3\" /> peers<br />\n");
+                  + I2PSnarkUtil.instance().getMaxUploaders() + "\" size=\"3\" maxlength=\"3\" /> peers<br />\n");
+        out.write("Up bandwidth limit: <input type=\"text\" name=\"upBW\" value=\""
+                  + I2PSnarkUtil.instance().getMaxUpBW() + "\" size=\"3\" maxlength=\"3\" /> KBps <i>(Router Up BW / 3 recommended)</i><br />\n");
         
         out.write("Use open trackers also: <input type=\"checkbox\" name=\"useOpenTrackers\" value=\"true\" " 
                   + (useOpenTrackers ? "checked " : "") 
@@ -712,11 +715,11 @@ public class I2PSnarkServlet extends HttpServlet {
         out.write("EepProxy host: <input type=\"text\" name=\"eepHost\" value=\""
                   + I2PSnarkUtil.instance().getEepProxyHost() + "\" size=\"15\" /> ");
         out.write("port: <input type=\"text\" name=\"eepPort\" value=\""
-                  + I2PSnarkUtil.instance().getEepProxyPort() + "\" size=\"5\" /><br />\n");
+                  + I2PSnarkUtil.instance().getEepProxyPort() + "\" size=\"5\" maxlength=\"5\" /><br />\n");
         out.write("I2CP host: <input type=\"text\" name=\"i2cpHost\" value=\"" 
                   + I2PSnarkUtil.instance().getI2CPHost() + "\" size=\"15\" /> ");
         out.write("port: <input type=\"text\" name=\"i2cpPort\" value=\"" +
-                  + I2PSnarkUtil.instance().getI2CPPort() + "\" size=\"5\" /> <br />\n");
+                  + I2PSnarkUtil.instance().getI2CPPort() + "\" size=\"5\" maxlength=\"5\" /> <br />\n");
         StringBuffer opts = new StringBuffer(64);
         Map options = new TreeMap(I2PSnarkUtil.instance().getI2CPOptions());
         for (Iterator iter = options.keySet().iterator(); iter.hasNext(); ) {
@@ -850,7 +853,8 @@ class FetchAndAdd implements Runnable {
     }
     public void run() {
         _url = _url.trim();
-        File file = I2PSnarkUtil.instance().get(_url, false);
+        // 3 retries
+        File file = I2PSnarkUtil.instance().get(_url, false, 3);
         try {
             if ( (file != null) && (file.exists()) && (file.length() > 0) ) {
                 _manager.addMessage("Torrent fetched from " + _url);