* Floodfill Peer Selector: Prefer already-connected floodfill

peer for direct RouterInfo stores, to mimimize floodfill
      connections
    * Peer Profiles: Classify connected peers as "active",
      which will help improve the fast pool
    * Transport Manager: Add isEstablished(Hash)

history.txt

@@ -1,3 +1,13 @@
+2008-08-27 zzz
+    * Floodfill Peer Selector: Prefer already-connected floodfill
+      peer for direct RouterInfo stores, to mimimize floodfill
+      connections
+    * Peer Profiles: Classify connected peers as "active",
+      which will help improve the fast pool
+    * Transport Manager: Add isEstablished(Hash)
+    * NTCP: Reduce max idle time from 20m to 15m
+    * NetDb stats: Post-0.6.3 clean up
+
 * 2008-08-24  0.6.3 released
 
 2008-08-24 Complication

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

@@ -56,6 +56,7 @@ public abstract class CommSystemFacade implements Service {
     public void recheckReachability() {}
     public boolean isBacklogged(Hash dest) { return false; }
     public boolean wasUnreachable(Hash dest) { return false; }
+    public boolean isEstablished(Hash dest) { return false; }
     
     /** 
      * Tell other transports our address changed

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

@@ -17,7 +17,7 @@ import net.i2p.CoreVersion;
 public class RouterVersion {
     public final static String ID = "$Revision: 1.548 $ $Date: 2008-06-07 23:00:00 $";
     public final static String VERSION = "0.6.3";
-    public final static long BUILD = 0;
+    public final static long BUILD = 1;
     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/networkdb/kademlia/FloodfillPeerSelector.java

@@ -31,14 +31,22 @@ class FloodfillPeerSelector extends PeerSelector {
      *
      * @return List of Hash for the peers selected
      */
+    public List selectMostReliablePeers(Hash key, int maxNumRouters, Set peersToIgnore, KBucketSet kbuckets) { 
+        return selectNearestExplicitThin(key, maxNumRouters, peersToIgnore, kbuckets, true);
+    }
+
     public List selectNearestExplicitThin(Hash key, int maxNumRouters, Set peersToIgnore, KBucketSet kbuckets) { 
+        return selectNearestExplicitThin(key, maxNumRouters, peersToIgnore, kbuckets, false);
+    }
+
+    public List selectNearestExplicitThin(Hash key, int maxNumRouters, Set peersToIgnore, KBucketSet kbuckets, boolean preferConnected) { 
         if (peersToIgnore == null)
             peersToIgnore = new HashSet(1);
         peersToIgnore.add(_context.router().getRouterInfo().getIdentity().getHash());
         FloodfillSelectionCollector matches = new FloodfillSelectionCollector(key, peersToIgnore, maxNumRouters);
         if (kbuckets == null) return new ArrayList();
         kbuckets.getAll(matches);
-        List rv = matches.get(maxNumRouters);
+        List rv = matches.get(maxNumRouters, preferConnected);
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("Searching for " + maxNumRouters + " peers close to " + key + ": " 
                        + rv + " (not including " + peersToIgnore + ") [allHashes.size = " 
@@ -100,9 +108,14 @@ class FloodfillPeerSelector extends PeerSelector {
         }
         /** get the first $howMany entries matching */
         public List get(int howMany) {
+            return get(howMany, false);
+        }
+
+        public List get(int howMany, boolean preferConnected) {
             Collections.shuffle(_floodfillMatches, _context.random());
             List rv = new ArrayList(howMany);
             List badff = new ArrayList(howMany);
+            List unconnectedff = new ArrayList(howMany);
             int found = 0;
             long now = _context.clock().now();
             // Only add in "good" floodfills here...
@@ -121,12 +134,21 @@ class FloodfillPeerSelector extends PeerSelector {
                         badff.add(entry);
                         if (_log.shouldLog(Log.DEBUG))
                             _log.debug("Skipping, recent failed send: " + entry);
+                    } else if (preferConnected && !_context.commSystem().isEstablished(entry)) {
+                        unconnectedff.add(entry);
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("Skipping, unconnected: " + entry);
                     } else {
                         rv.add(entry);
                         found++;
                     }
                 }
             }
+            // Put the unconnected floodfills after the connected floodfills
+            for (int i = 0; found < howMany && i < unconnectedff.size(); i++) {
+                rv.add(unconnectedff.get(i));
+                found++;
+            }
             // Put the "bad" floodfills at the end of the floodfills but before the kademlias
             for (int i = 0; found < howMany && i < badff.size(); i++) {
                 rv.add(badff.get(i));

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

@@ -141,7 +141,16 @@ class StoreJob extends JobImpl {
         if (toCheck > getParallelization())
             toCheck = getParallelization();
 
-        List closestHashes = getClosestRouters(_state.getTarget(), toCheck, _state.getAttempted());
+        // We are going to send the RouterInfo directly, rather than through a lease,
+        // so select a floodfill peer we are already connected to.
+        // This will help minimize active connections for floodfill peers and allow
+        // the network to scale.
+        // Perhaps the ultimate solution is to send RouterInfos through a lease also.
+        List closestHashes;
+        if (_state.getData() instanceof RouterInfo) 
+            closestHashes = getMostReliableRouters(_state.getTarget(), toCheck, _state.getAttempted());
+        else
+            closestHashes = getClosestRouters(_state.getTarget(), toCheck, _state.getAttempted());
         if ( (closestHashes == null) || (closestHashes.size() <= 0) ) {
             if (_state.getPending().size() <= 0) {
                 if (_log.shouldLog(Log.INFO))
@@ -216,6 +225,13 @@ class StoreJob extends JobImpl {
         return _peerSelector.selectNearestExplicit(rkey, numClosest, alreadyChecked, ks);
     }
 
+    private List getMostReliableRouters(Hash key, int numClosest, Set alreadyChecked) {
+        Hash rkey = getContext().routingKeyGenerator().getRoutingKey(key);
+        KBucketSet ks = _facade.getKBuckets();
+        if (ks == null) return new ArrayList();
+        return _peerSelector.selectMostReliablePeers(rkey, numClosest, alreadyChecked, ks);
+    }
+
     /**
      * Send a store to the given peer through a garlic route, including a reply 
      * DeliveryStatusMessage so we know it got there

router/java/src/net/i2p/router/peermanager/PeerProfile.java

@@ -94,12 +94,15 @@ public class PeerProfile {
     /**
      * Is this peer active at the moment (sending/receiving messages within the 
      * given period?)
+     * Also mark active if it is connected, as this will tend to encourage use
+     * of already-connected peers.
      */
     public boolean getIsActive(long period) {
         if ( (getSendSuccessSize().getRate(period).getCurrentEventCount() > 0) ||
              (getSendSuccessSize().getRate(period).getLastEventCount() > 0) ||
              (getReceiveSize().getRate(period).getCurrentEventCount() > 0) ||
-             (getReceiveSize().getRate(period).getLastEventCount() > 0) )
+             (getReceiveSize().getRate(period).getLastEventCount() > 0) ||
+             _context.commSystem().isEstablished(_peer) )
             return true;
         else
             return false;

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

@@ -125,6 +125,10 @@ public class CommSystemFacadeImpl extends CommSystemFacade {
         return _manager.isBacklogged(dest); 
     }
     
+    public boolean isEstablished(Hash dest) { 
+        return _manager.isEstablished(dest); 
+    }
+    
     public boolean wasUnreachable(Hash dest) { 
         return _manager.wasUnreachable(dest); 
     }

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

@@ -50,4 +50,5 @@ public interface Transport {
     public boolean wasUnreachable(Hash dest);
     
     public boolean isUnreachable(Hash peer);
+    public boolean isEstablished(Hash peer);
 }

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

@@ -395,6 +395,7 @@ public abstract class TransportImpl implements Transport {
     public short getReachabilityStatus() { return CommSystemFacade.STATUS_UNKNOWN; }
     public void recheckReachability() {}
     public boolean isBacklogged(Hash dest) { return false; }
+    public boolean isEstablished(Hash dest) { return false; }
     
     private static final long UNREACHABLE_PERIOD = 5*60*1000;
     public boolean isUnreachable(Hash peer) {

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

@@ -203,6 +203,15 @@ public class TransportManager implements TransportEventListener {
         return false;
     }    
     
+    public boolean isEstablished(Hash dest) {
+        for (int i = 0; i < _transports.size(); i++) {
+            Transport t = (Transport)_transports.get(i);
+            if (t.isEstablished(dest))
+                return true;
+        }
+        return false;
+    }    
+    
     /**
      * Was the peer UNreachable (outbound only) on any transport,
      * based on the last time we tried it for each transport?

router/java/src/net/i2p/router/transport/ntcp/NTCPTransport.java

@@ -336,8 +336,12 @@ public class NTCPTransport extends TransportImpl {
     }
 
     private boolean isEstablished(RouterIdentity peer) {
+        return isEstablished(peer.calculateHash());
+    }
+    
+    public boolean isEstablished(Hash dest) {
         synchronized (_conLock) {
-            NTCPConnection con = (NTCPConnection)_conByIdent.get(peer.calculateHash());
+            NTCPConnection con = (NTCPConnection)_conByIdent.get(dest);
             return (con != null) && con.isEstablished() && !con.isClosed();
         }
     }

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

@@ -1266,6 +1266,10 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
         return active;
     }
     
+    public boolean isEstablished(Hash dest) {
+        return getPeerState(dest) != null;
+    }
+
     /**
      * Return our peer clock skews on this transport.
      * Vector composed of Long, each element representing a peer skew in seconds.