* moved the inbound partial messages to the PeerState itself, reducing lock contention in the InboundMessageFragments and transparently dropping failed messages when we drop old peer states

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

@@ -91,7 +91,7 @@ public class ACKSender implements Runnable {
             if (peer != null) {
                 long lastSend = peer.getLastACKSend();
                 long wanted = peer.getWantedACKSendSince();
-                List ackBitfields = peer.retrieveACKBitfields(_transport.getPartialACKSource());
+                List ackBitfields = peer.retrieveACKBitfields();
                 
                 if (wanted < 0)
                     _log.error("wtf, why are we acking something they dont want?  remaining=" + remaining + ", peer=" + peer + ", bitfields=" + ackBitfields);

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

@@ -20,16 +20,10 @@ import net.i2p.util.Log;
  * up in the router we have full blown replay detection, its nice to have a
  * basic line of defense here).
  *
- * TODO: add in some sensible code to drop expired fragments from peers we 
- * don't hear from again (either a periodic culling for expired peers, or
- * a scheduled event)
- *
  */
-public class InboundMessageFragments implements UDPTransport.PartialACKSource {
+public class InboundMessageFragments /*implements UDPTransport.PartialACKSource */{
     private RouterContext _context;
     private Log _log;
-    /** Map of peer (Hash) to a Map of messageId (Long) to InboundMessageState objects */
-    private Map _inboundMessages;
     /** list of message IDs recently received, so we can ignore in flight dups */
     private DecayingBloomFilter _recentlyCompletedMessages;
     private OutboundMessageFragments _outbound;
@@ -44,7 +38,7 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
     public InboundMessageFragments(RouterContext ctx, OutboundMessageFragments outbound, UDPTransport transport) {
         _context = ctx;
         _log = ctx.logManager().getLog(InboundMessageFragments.class);
-        _inboundMessages = new HashMap(64);
+        //_inboundMessages = new HashMap(64);
         _outbound = outbound;
         _transport = transport;
         _ackSender = new ACKSender(_context, _transport);
@@ -73,9 +67,6 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
         _recentlyCompletedMessages = null;
         _ackSender.shutdown();
         _messageReceiver.shutdown();
-        synchronized (_inboundMessages) {
-            _inboundMessages.clear();
-        }
     }
     public boolean isAlive() { return _alive; }
 
@@ -103,12 +94,9 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
     private void receiveMessages(PeerState from, UDPPacketReader.DataReader data) {
         int fragments = data.readFragmentCount();
         if (fragments <= 0) return;
-        synchronized (_inboundMessages) { // XXX: CHOKE POINT (to what extent?)
+        Hash fromPeer = from.getRemotePeer();
-            Map messages = (Map)_inboundMessages.get(from.getRemotePeer());
+            
-            if (messages == null) {
+        Map messages = from.getInboundMessages();
-                messages = new HashMap(fragments);
-                _inboundMessages.put(from.getRemotePeer(), messages);
-            }
             
         for (int i = 0; i < fragments; i++) {
             Long messageId = new Long(data.readMessageId(i));
@@ -129,20 +117,31 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
             boolean messageComplete = false;
             boolean messageExpired = false;
             boolean fragmentOK = false;
+            boolean partialACK = false;
+         
+            // perhaps compact the synchronized block further by synchronizing on the
+            // particular state once its found?
+            synchronized (messages) {
                 state = (InboundMessageState)messages.get(messageId);
                 if (state == null) {
-                    state = new InboundMessageState(_context, messageId.longValue(), from.getRemotePeer());
+                    state = new InboundMessageState(_context, messageId.longValue(), fromPeer);
                     messages.put(messageId, state);
                 }
+                
                 fragmentOK = state.receiveFragment(data, i);
+             
                 if (state.isComplete()) {
                     messageComplete = true;
                     messages.remove(messageId);
-                    if (messages.size() <= 0)
+                } else if (state.isExpired()) {
-                        _inboundMessages.remove(from.getRemotePeer());
+                    messageExpired = true;
+                    messages.remove(messageId);
+                } else {
+                    partialACK = true;
+                }
 
+                if (messageComplete) {
                     _recentlyCompletedMessages.add(messageId.longValue());
-
                     _messageReceiver.receiveMessage(state);
 
                     from.messageFullyReceived(messageId, state.getCompleteSize());
@@ -153,15 +152,11 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
 
                     _context.statManager().addRateData("udp.receivedCompleteTime", state.getLifetime(), state.getLifetime());
                     _context.statManager().addRateData("udp.receivedCompleteFragments", state.getFragmentCount(), state.getLifetime());
-                } else if (state.isExpired()) {
+                } else if (messageExpired) {
-                    messageExpired = true;
+                    state.releaseResources();
-                    messages.remove(messageId);
-                    if (messages.size() <= 0)
-                        _inboundMessages.remove(from.getRemotePeer());
                     if (_log.shouldLog(Log.WARN))
                         _log.warn("Message expired while only being partially read: " + state);
-                    state.releaseResources();
+                } else if (partialACK) {
-                } else {
                     // not expired but not yet complete... lets queue up a partial ACK
                     if (_log.shouldLog(Log.DEBUG))
                         _log.debug("Queueing up a partial ACK for peer: " + from + " for " + state);
@@ -209,23 +204,4 @@ public class InboundMessageFragments implements UDPTransport.PartialACKSource {
         else
             from.dataReceived();
     }
-    
-    public void fetchPartialACKs(Hash fromPeer, List ackBitfields) {
-        synchronized (_inboundMessages) {
-            Map messages = (Map)_inboundMessages.get(fromPeer);
-            if (messages == null)
-                return;
-            for (Iterator iter = messages.values().iterator(); iter.hasNext(); ) {
-                InboundMessageState state = (InboundMessageState)iter.next();
-                if (state.isExpired()) {
-                    iter.remove();
-                } else {
-                    if (!state.isComplete())
-                        ackBitfields.add(state.createACKBitfield());
-                }
-            }
-            if (messages.size() <= 0)
-                _inboundMessages.remove(fromPeer);
-        }
-    }
 }

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

@@ -1,7 +1,10 @@
 package net.i2p.router.transport.udp;
 
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -144,6 +147,9 @@ public class PeerState {
     private long _packetsReceivedDuplicate;
     private long _packetsReceived;
     
+    /** Message (Long) to InboundMessageState for active message */
+    private Map _inboundMessages;
+    
     private static final int DEFAULT_SEND_WINDOW_BYTES = 8*1024;
     private static final int MINIMUM_WINDOW_BYTES = DEFAULT_SEND_WINDOW_BYTES;
     private static final int MAX_SEND_WINDOW_BYTES = 1024*1024;
@@ -202,6 +208,7 @@ public class PeerState {
         _retransmissionPeriodStart = 0;
         _packetsReceived = 0;
         _packetsReceivedDuplicate = 0;
+        _inboundMessages = new HashMap(8);
         _context.statManager().createRateStat("udp.congestionOccurred", "How large the cwin was when congestion occurred (duration == sendBps)", "udp", new long[] { 60*60*1000, 24*60*60*1000 });
         _context.statManager().createRateStat("udp.congestedRTO", "retransmission timeout after congestion (duration == rtt dev)", "udp", new long[] { 60*60*1000, 24*60*60*1000 });
         _context.statManager().createRateStat("udp.sendACKPartial", "Number of partial ACKs sent (duration == number of full ACKs in that ack packet)", "udp", new long[] { 60*60*1000, 24*60*60*1000 });
@@ -456,6 +463,12 @@ public class PeerState {
             _wantACKSendSince = _context.clock().now();
     }
     
+    /** 
+     * Fetch the internal id (Long) to InboundMessageState for incomplete inbound messages.
+     * Access to this map must be synchronized explicitly!
+     */
+    public Map getInboundMessages() { return _inboundMessages; }
+    
     /** 
      * either they told us to back off, or we had to resend to get 
      * the data through.  
@@ -489,7 +502,7 @@ public class PeerState {
      * will be unchanged if there are ACKs remaining.
      *
      */
-    public List retrieveACKBitfields(UDPTransport.PartialACKSource partialACKSource) {
+    public List retrieveACKBitfields() {
         List rv = new ArrayList(16);
         int bytesRemaining = countMaxACKData();
         synchronized (_currentACKs) {
@@ -502,12 +515,12 @@ public class PeerState {
         }
             
         int partialIncluded = 0;
-        if ( (bytesRemaining > 4) && (partialACKSource != null) ) {
+        if (bytesRemaining > 4) {
             // ok, there's room to *try* to fit in some partial ACKs, so
             // we should try to find some packets to partially ACK 
             // (preferably the ones which have the most received fragments)
             List partial = new ArrayList();
-            partialACKSource.fetchPartialACKs(_remotePeer, partial);
+            fetchPartialACKs(partial);
             // we may not be able to use them all, but lets try...
             for (int i = 0; (bytesRemaining > 4) && (i < partial.size()); i++) {
                 ACKBitfield bitfield = (ACKBitfield)partial.get(i);
@@ -528,6 +541,34 @@ public class PeerState {
         return rv;
     }
     
+    private void fetchPartialACKs(List rv) {
+        InboundMessageState states[] = null;
+        int curState = 0;
+        synchronized (_inboundMessages) {
+            int numMessages = _inboundMessages.size();
+            if (numMessages <= 0) 
+                return;
+            for (Iterator iter = _inboundMessages.values().iterator(); iter.hasNext(); ) {
+                InboundMessageState state = (InboundMessageState)iter.next();
+                if (state.isExpired()) {
+                    iter.remove();
+                } else {
+                    if (!state.isComplete()) {
+                        if (states == null)
+                            states = new InboundMessageState[numMessages];
+                        states[curState++] = state;
+                    }
+                }
+            }
+        }
+        if (states != null) {
+            for (int i = curState-1; i >= 0; i--) {
+                if (states[i] != null)
+                    rv.add(states[i].createACKBitfield());
+            }
+        }
+    }
+    
     /** represent a full ACK of a message */
     private class FullACKBitfield implements ACKBitfield {
         private long _msgId;

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

@@ -59,6 +59,7 @@ class UDPFlooder implements Runnable {
     }
     
     public void run() {
+        long nextSend = _context.clock().now();
         while (_alive) {
             try {
                 synchronized (_peers) {
@@ -67,6 +68,8 @@ class UDPFlooder implements Runnable {
                 }
             } catch (InterruptedException ie) {}
             
+            long now = _context.clock().now();
+            if (now >= nextSend) {
                 // peers always grows, so this is fairly safe
                 for (int i = 0; i < _peers.size(); i++) {
                     PeerState peer = (PeerState)_peers.get(i);
@@ -92,8 +95,20 @@ class UDPFlooder implements Runnable {
                         _transport.send(m, peer);
                     }
                 }
-            long floodDelay = calcFloodDelay();
+                nextSend = now + calcFloodDelay();
-            try { Thread.sleep(floodDelay); } catch (InterruptedException ie) {}
+            }
+            
+            long delay = nextSend - now;
+            if (delay > 0) {
+                if (delay > 10*1000) {
+                    long fd = calcFloodDelay();
+                    if (delay > fd) {
+                        nextSend = now + fd;
+                        delay = fd;
+                    }
+                }
+                try { Thread.sleep(delay); } catch (InterruptedException ie) {}
+            }
         }
     }
     

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

@@ -667,17 +667,6 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
         out.write("</table>\n");
     }
     
-    public PartialACKSource getPartialACKSource() { return _inboundFragments; }
-    
-    /** help us grab partial ACKs */
-    public interface PartialACKSource {
-        /** 
-         * build partial ACKs of messages received from the peer and store
-         * them in the ackBitfields
-         */
-        public void fetchPartialACKs(Hash fromPeer, List ackBitfields);
-    }
-    
     private static final DecimalFormat _fmt = new DecimalFormat("#,##0.00");
     private static final String formatKBps(int bps) {
         synchronized (_fmt) {