/me waves to our new friend, the MessageStateMonitor, which keeps track of how many messages we're dealing with in memory (and whether they've been processed & discarded yet)

router/java/src/net/i2p/router/MessageStateMonitor.java

@@ -0,0 +1,72 @@
+package net.i2p.router;
+
+import net.i2p.util.Log;
+
+/**
+ * Keep track of the inbound and outbound messages in memory.
+ *
+ */
+public class MessageStateMonitor {
+    private Log _log;
+    private RouterContext _context;
+    private volatile int _inboundLiveCount;
+    private volatile int _inboundReadCount;
+    private volatile int _inboundFinalizedCount;
+    private volatile int _outboundLiveCount;
+    private volatile int _outboundDiscardedCount;
+    
+    public MessageStateMonitor(RouterContext context) {
+        _context = context;
+        _log = context.logManager().getLog(MessageStateMonitor.class);
+        _inboundLiveCount = 0;
+        _inboundReadCount = 0;
+        _inboundFinalizedCount = 0;
+        _outboundLiveCount = 0;
+        _outboundDiscardedCount = 0;
+    }
+    
+    public void inboundMessageAdded() {
+        _inboundLiveCount++;
+        logStatus("inboundAdded     ");
+    }
+    public void inboundMessageRead() {
+        _inboundReadCount++;
+        _inboundLiveCount--;
+        logStatus("inboundRead      ");
+    }
+    public void inboundMessageFinalized() {
+        _inboundReadCount--;
+        _inboundFinalizedCount++;
+        logStatus("inboundFinalized ");
+    }
+    
+    public void outboundMessageAdded() {
+        _outboundLiveCount++;
+        logStatus("outboundAdded    ");
+    }
+    public void outboundMessageDiscarded() {
+        _outboundDiscardedCount++;
+        _outboundLiveCount--;
+        logStatus("outboundDiscarded");
+    }
+    public void outboundMessageFinalized() {
+        _outboundDiscardedCount--;
+        logStatus("outboundFinalized");
+    }
+    
+    private void logStatus(String event) {
+        if (false || (_log.shouldLog(Log.DEBUG)))
+            _log.debug(event + ": outbound (live: " + _outboundLiveCount 
+                       + " discarded:" + _outboundDiscardedCount + ")"
+                       + " inbound (live: " + (_inboundLiveCount) 
+                       //+ " inbound (live: " + (_inboundLiveCount-_inboundFinalizedCount) 
+                       + " read: " + (_inboundReadCount)
+                       //+ " completed: " + _inboundFinalizedCount
+                       + ")");
+    }
+    
+    public int getInboundLiveCount() { return _inboundLiveCount; }
+    public int getInboundReadCount() { return _inboundReadCount; }
+    public int getOutboundLiveCount() { return _outboundLiveCount; }
+    public int getOutboundDiscardedCount() { return _outboundDiscardedCount; }
+}

router/java/src/net/i2p/router/RouterContext.java

@@ -49,6 +49,7 @@ public class RouterContext extends I2PAppContext {
     private StatisticsManager _statPublisher;
     private Shitlist _shitlist;
     private MessageValidator _messageValidator;
+    private MessageStateMonitor _messageStateMonitor;
     private Calculator _isFailingCalc;
     private Calculator _integrationCalc;
     private Calculator _speedCalc;
@@ -68,6 +69,7 @@ public class RouterContext extends I2PAppContext {
         _outNetMessagePool = new OutNetMessagePool(this);
         _messageHistory = new MessageHistory(this);
         _messageRegistry = new OutboundMessageRegistry(this);
+        _messageStateMonitor = new MessageStateMonitor(this);
         _netDb = new KademliaNetworkDatabaseFacade(this);
         _keyManager = new KeyManager(this);
         if ("false".equals(getProperty("i2p.vmCommSystem", "false")))
@@ -123,6 +125,13 @@ public class RouterContext extends I2PAppContext {
      * The registry is used by outbound messages to wait for replies.
      */
     public OutboundMessageRegistry messageRegistry() { return _messageRegistry; }
+    /**
+     * The monitor keeps track of inbound and outbound messages currently held in
+     * memory / queued for processing.  We'll use this to throttle the router so
+     * we don't overflow.
+     *
+     */
+    public MessageStateMonitor messageStateMonitor() { return _messageStateMonitor; }
     /**
      * Our db cache
      */