Changeset 1119612

Timestamp:

Nov 21, 2011 6:22:13 PM (9 years ago)

Author:

zzz <zzz@…>

Branches:

master

Children:

9d0bafb

Parents:

dc6c568

Message:

• NTCP Pumper:
  • Ensure failsafe pumper code gets run on schedule
  • Don't copy the read buffers
  • Adjust minimum read buffers based on memory
  • New i2np.ntcp.useDirectBuffer option (default false)
  • Mark peer unreachable when read failure is during establishment
  • Change some Reader Lists to Sets to avoid linear search
  • Log tweaks, debugging, new loop stats

Files:

• history.txt (modified) (1 diff)
• router/java/src/net/i2p/router/RouterVersion.java (modified) (1 diff)
• router/java/src/net/i2p/router/transport/ntcp/EstablishState.java (modified) (4 diffs)
• router/java/src/net/i2p/router/transport/ntcp/EventPumper.java (modified) (21 diffs)
• router/java/src/net/i2p/router/transport/ntcp/NTCPConnection.java (modified) (7 diffs)
• router/java/src/net/i2p/router/transport/ntcp/NTCPTransport.java (modified) (3 diffs)
• router/java/src/net/i2p/router/transport/ntcp/Reader.java (modified) (11 diffs)

history.txt

@@ +1 @@
+2011-11-21 zzz
+  * NTCP Pumper:
+    - Ensure failsafe pumper code gets run on schedule
+    - Don't copy the read buffers
+    - Adjust minimum read buffers based on memory
+    - New i2np.ntcp.useDirectBuffer option (default false)
+    - Mark peer unreachable when read failure is during establishment
+    - Change some Reader Lists to Sets to avoid linear search
+    - Log tweaks, debugging, new loop stats
+
 2011-11-18 zzz
   * NTCP:

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

@@ -19 +19 @@
     public final static String ID = "Monotone";
     public final static String VERSION = CoreVersion.VERSION;
-    public final static long BUILD = 5;
+    public final static long BUILD = 6;
 
     /** for example "-test" */

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

@@ -148 +148 @@
     /**
      * parse the contents of the buffer as part of the handshake.  if the
-     * handshake is completed and there is more data remaining, the buffer is
-     * updated so that the next read will be the (still encrypted) remaining
+     * handshake is completed and there is more data remaining, the data are
+     * copieed out so that the next read will be the (still encrypted) remaining
      * data (available from getExtraBytes)
+     *
+     * All data must be copied out of the buffer as Reader.processRead()
+     * will return it to the pool.
      */
     public void receive(ByteBuffer src) {
@@ -177 +180 @@
      *  we are Bob, so receive these bytes as part of an inbound connection
      *  This method receives messages 1 and 3, and sends messages 2 and 4.
+     *
+     *  All data must be copied out of the buffer as Reader.processRead()
+     *  will return it to the pool.
      */
     private void receiveInbound(ByteBuffer src) {
@@ -341 +347 @@
      *  We are Alice, so receive these bytes as part of an outbound connection.
      *  This method receives messages 2 and 4, and sends message 3.
+     *
+     *  All data must be copied out of the buffer as Reader.processRead()
+     *  will return it to the pool.
      */
     private void receiveOutbound(ByteBuffer src) {
@@ -685 +694 @@
     }
 
-    /** anything left over in the byte buffer after verification is extra */
+    /** Anything left over in the byte buffer after verification is extra
+     *
+     *  All data must be copied out of the buffer as Reader.processRead()
+     *  will return it to the pool.
+     */
     private void prepareExtra(ByteBuffer buf) {
         int remaining = buf.remaining();

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

@@ -20 +20 @@
 import java.util.concurrent.LinkedBlockingQueue;
 
+import net.i2p.I2PAppContext;
 import net.i2p.data.RouterIdentity;
 import net.i2p.data.RouterInfo;
@@ -45 +46 @@
     private final NTCPTransport _transport;
     private long _expireIdleWriteTime;
-    
+    private boolean _useDirect;
+    
+    /**
+     *  This probably doesn't need to be bigger than the largest typical
+     *  message, which is a 5-slot VTBM (~2700 bytes).
+     *  The occasional larger message can use multiple buffers.
+     */
     private static final int BUF_SIZE = 8*1024;
     private static final int MAX_CACHE_SIZE = 64;
 
     /**
+     *  Read buffers. (write buffers use wrap())
      *  Shared if there are multiple routers in the JVM
+     *  Note that if the routers have different PROP_DIRECT settings this will have a mix,
+     *  so don't do that.
      */
     private static final LinkedBlockingQueue<ByteBuffer> _bufCache = new LinkedBlockingQueue<ByteBuffer>(MAX_CACHE_SIZE);
@@ -68 +78 @@
     private static final long MAX_EXPIRE_IDLE_TIME = 15*60*1000l;
 
+    /**
+     *  Do we use direct buffers for reading? Default false.
+     *  @see java.nio.ByteBuffer
+     */
+    private static final String PROP_DIRECT = "i2np.ntcp.useDirectBuffers";
+
+    private static final int MIN_MINB = 4;
+    private static final int MAX_MINB = 12;
+    private static final int MIN_BUFS;
+    static {
+        long maxMemory = Runtime.getRuntime().maxMemory();
+        if (maxMemory == Long.MAX_VALUE)
+            maxMemory = 96*1024*1024l;
+        MIN_BUFS = (int) Math.max(MIN_MINB, Math.min(MAX_MINB, 1 + (maxMemory / (16*1024*1024))));
+    }
+
     public EventPumper(RouterContext ctx, NTCPTransport transport) {
         _context = ctx;
@@ -73 +99 @@
         _transport = transport;
         _expireIdleWriteTime = MAX_EXPIRE_IDLE_TIME;
+        _context.statManager().createRateStat("ntcp.pumperKeySetSize", "", "ntcp", new long[] {10*60*1000} );
+        _context.statManager().createRateStat("ntcp.pumperKeysPerLoop", "", "ntcp", new long[] {10*60*1000} );
+        _context.statManager().createRateStat("ntcp.pumperLoopsPerSecond", "", "ntcp", new long[] {10*60*1000} );
     }
     
@@ -131 +160 @@
      */
     public void run() {
+        int loopCount = 0;
         long lastFailsafeIteration = System.currentTimeMillis();
         while (_alive && _selector.isOpen()) {
             try {
+                loopCount++;
                 runDelayedEvents();
-                int count = 0;
+
                 try {
                     //if (_log.shouldLog(Log.DEBUG))
                     //    _log.debug("before select...");
-                    count = _selector.select(SELECTOR_LOOP_DELAY);
+                    int count = _selector.select(SELECTOR_LOOP_DELAY);
+                    if (count > 0) {
+                        //if (_log.shouldLog(Log.DEBUG))
+                        //    _log.debug("select returned " + count);
+                        Set<SelectionKey> selected = _selector.selectedKeys();
+                        _context.statManager().addRateData("ntcp.pumperKeysPerLoop", selected.size());
+                        processKeys(selected);
+                        // does clear() do anything useful?
+                        selected.clear();
+                    }
+                } catch (ClosedSelectorException cse) {
+                    continue;
                 } catch (IOException ioe) {
                     if (_log.shouldLog(Log.WARN))
                         _log.warn("Error selecting", ioe);
                 }
-                if (count <= 0)
-                    continue;
-                //if (_log.shouldLog(Log.DEBUG))
-                //    _log.debug("select returned " + count);
-
-                Set<SelectionKey> selected;
-                try {
-                    selected = _selector.selectedKeys();
-                } catch (ClosedSelectorException cse) {
-                    continue;
-                }
-
-                processKeys(selected);
-                selected.clear();
                 
                 if (lastFailsafeIteration + FAILSAFE_ITERATION_FREQ < System.currentTimeMillis()) {
@@ -167 +195 @@
                     try {
                         Set<SelectionKey> all = _selector.keys();
+                        _context.statManager().addRateData("ntcp.pumperKeySetSize", all.size());
+                        _context.statManager().addRateData("ntcp.pumperLoopsPerSecond", loopCount / (FAILSAFE_ITERATION_FREQ / 1000));
+                        loopCount = 0;
                         
                         int failsafeWrites = 0;
@@ -204 +235 @@
                                     con.getTimeSinceCreated() > 2 * NTCPTransport.ESTABLISH_TIMEOUT) {
                                     if (_log.shouldLog(Log.INFO))
-                                        _log.info("Invalid key " + con);
+                                        _log.info("Removing invalid key for " + con);
                                     // this will cancel the key, and it will then be removed from the keyset
                                     con.close();
@@ -240 +271 @@
                     }
                 }
+                // Clear the cache if the user changes the setting,
+                // so we can test the effect.
+                boolean newUseDirect = _context.getBooleanProperty(PROP_DIRECT);
+                if (_useDirect != newUseDirect) {
+                    _useDirect = newUseDirect;
+                    _bufCache.clear();
+                }
             } catch (RuntimeException re) {
                 _log.error("Error in the event pumper", re);
@@ -311 +349 @@
                 if (read) {
                     //_context.statManager().addRateData("ntcp.read", 1, 0);
-                    key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
                     processRead(key);
                 }
                 if (write) {
                     //_context.statManager().addRateData("ntcp.write", 1, 0);
-                    key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                     processWrite(key);
                 }
+                //if (!(accept || connect || read || write)) {
+                //    if (_log.shouldLog(Log.INFO))
+                //        _log.info("key wanted nothing? con: " + key.attachment());
+                //}
             } catch (CancelledKeyException cke) {
                 if (_log.shouldLog(Log.DEBUG))
@@ -366 +406 @@
         _selector.wakeup();
     }
-    
-    private static final int MIN_BUFS = 5;
 
     /**
@@ -374 +412 @@
      */
     private static int _numBufs = MIN_BUFS;
-    private static int __liveBufs = 0;
     private static int __consecutiveExtra;
 
@@ -380 +417 @@
      *  High-frequency path in thread.
      */
-    private static ByteBuffer acquireBuf() {
+    private ByteBuffer acquireBuf() {
         ByteBuffer rv = _bufCache.poll();
-        if (rv == null) {
-            rv = ByteBuffer.allocate(BUF_SIZE);
-            _numBufs = ++__liveBufs;
+        // discard buffer if _useDirect setting changes
+        if (rv == null || rv.isDirect() != _useDirect) {
+            if (_useDirect)
+                rv = ByteBuffer.allocateDirect(BUF_SIZE);
+            else
+                rv = ByteBuffer.allocate(BUF_SIZE);
+            _numBufs++;
             //if (_log.shouldLog(Log.DEBUG))
             //    _log.debug("creating a new read buffer " + System.identityHashCode(rv) + " with " + __liveBufs + " live: " + rv);            
@@ -396 +437 @@
     
     /**
+     *  Return a read buffer to the pool.
+     *  These buffers must be from acquireBuf(), i.e. capacity() == BUF_SIZE.
      *  High-frequency path in thread.
      */
-    private static void releaseBuf(ByteBuffer buf) {
+    public static void releaseBuf(ByteBuffer buf) {
         //if (false) return;
         //if (_log.shouldLog(Log.DEBUG))
         //    _log.debug("releasing read buffer " + System.identityHashCode(buf) + " with " + __liveBufs + " live: " + buf);
+
+        // double check
+        if (buf.capacity() < BUF_SIZE) {
+            I2PAppContext.getGlobalContext().logManager().getLog(EventPumper.class).error("Bad size " + buf.capacity(), new Exception());
+            return;
+        }
         buf.clear();
         int extra = _bufCache.size();
         boolean cached = extra < _numBufs;
 
+        // TODO always offer if direct?
         if (cached) {
             _bufCache.offer(buf);
-            if (extra > 5) {
+            if (extra > MIN_BUFS) {
                 __consecutiveExtra++;
                 if (__consecutiveExtra >= 20) {
-                    _numBufs = Math.max(_numBufs - 1, MIN_BUFS);
+                    if (_numBufs > MIN_BUFS)
+                        _numBufs--;
                     __consecutiveExtra = 0;
                 }
             }
-        } else {
-            __liveBufs--;
         }
         //if (cached && _log.shouldLog(Log.DEBUG))
@@ -466 +515 @@
             boolean connected = chan.finishConnect();
             if (_log.shouldLog(Log.DEBUG))
-                _log.debug("processing connect for " + key + " / " + con + ": connected? " + connected);
+                _log.debug("processing connect for " + con + ": connected? " + connected);
             if (connected) {
                 // BUGFIX for firewalls. --Sponge
@@ -479 +528 @@
             }
         } catch (IOException ioe) {   // this is the usual failure path for a timeout or connect refused
-            if (_log.shouldLog(Log.WARN))
-                _log.warn("Failed outbound connection to " + con.getRemotePeer().calculateHash(), ioe);
+            if (_log.shouldLog(Log.INFO))
+                _log.info("Failed outbound " + con, ioe);
             con.close();
             //_context.shitlist().shitlistRouter(con.getRemotePeer().calculateHash(), "Error connecting", NTCPTransport.STYLE);
@@ -487 +536 @@
         } catch (NoConnectionPendingException ncpe) {
             // ignore
-        }
-    }
-    
-    /**
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("error connecting on " + con, ncpe);
+        }
+    }
+    
+    /**
+     *  OP_READ will always be set before this is called.
+     *  This method will disable the interest if no more reads remain because of inbound bandwidth throttling.
      *  High-frequency path in thread.
      */
@@ -500 +553 @@
             if (read == -1) {
                 //if (_log.shouldLog(Log.DEBUG)) _log.debug("EOF on " + con);
-                _context.statManager().addRateData("ntcp.readEOF", 1);
+                //_context.statManager().addRateData("ntcp.readEOF", 1);
                 con.close();
                 releaseBuf(buf);
@@ -506 +559 @@
                 //if (_log.shouldLog(Log.DEBUG))
                 //    _log.debug("nothing to read for " + con + ", but stay interested");
-                key.interestOps(key.interestOps() | SelectionKey.OP_READ);
+                // stay interested
+                //key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                 releaseBuf(buf);
             } else if (read > 0) {
-                byte data[] = new byte[read];
+                // ZERO COPY. The buffer will be returned in Reader.processRead()
                 buf.flip();
-                buf.get(data);
-                releaseBuf(buf);
-                buf = null;
-                ByteBuffer rbuf = ByteBuffer.wrap(data);
                 FIFOBandwidthLimiter.Request req = _context.bandwidthLimiter().requestInbound(read, "NTCP read"); //con, buf);
                 if (req.getPendingInboundRequested() > 0) {
@@ -522 +572 @@
                     //    _log.debug("bw throttled reading for " + con + ", so we don't want to read anymore");
                     _context.statManager().addRateData("ntcp.queuedRecv", read);
-                    con.queuedRecv(rbuf, req);
+                    con.queuedRecv(buf, req);
                 } else {
                     // fully allocated
                     //if (_log.shouldLog(Log.DEBUG))
                     //    _log.debug("not bw throttled reading for " + con);
-                    key.interestOps(key.interestOps() | SelectionKey.OP_READ);
-                    con.recv(rbuf);
+                    // stay interested
+                    //key.interestOps(key.interestOps() | SelectionKey.OP_READ);
+                    con.recv(buf);
+                    _context.statManager().addRateData("ntcp.read", read);
                 }
             }
         } catch (CancelledKeyException cke) {
-            if (_log.shouldLog(Log.WARN)) _log.warn("error reading", cke);
+            releaseBuf(buf);
+            if (_log.shouldLog(Log.WARN)) _log.warn("error reading on " + con, cke);
             con.close();
             _context.statManager().addRateData("ntcp.readError", 1);
-            if (buf != null) releaseBuf(buf);
         } catch (IOException ioe) {
-            if (_log.shouldLog(Log.WARN)) _log.warn("error reading", ioe);
+            // common, esp. at outbound connect time
+            releaseBuf(buf);
+            if (_log.shouldLog(Log.INFO))
+                _log.info("error reading on " + con, ioe);
+            if (con.isEstablished()) {
+                _context.statManager().addRateData("ntcp.readError", 1);
+            } else {
+                // Usually "connection reset by peer", probably a conn limit rejection?
+                // although it could be a read failure during the DH handshake
+                // Same stat as in processConnect()
+                _context.statManager().addRateData("ntcp.connectFailedTimeoutIOE", 1);
+                _transport.markUnreachable(con.getRemotePeer().calculateHash());
+            }
             con.close();
-            _context.statManager().addRateData("ntcp.readError", 1);
-            if (buf != null) releaseBuf(buf);
         } catch (NotYetConnectedException nyce) {
+            releaseBuf(buf);
             // ???
-        }
-    }
-    
-    /**
+            key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("error reading on " + con, nyce);
+        }
+    }
+    
+    /**
+     *  OP_WRITE will always be set before this is called.
+     *  This method will disable the interest if no more writes remain.
      *  High-frequency path in thread.
      */
@@ -574 +642 @@
                         if ( (buf.remaining() > 0) || (!con.isWriteBufEmpty()) ) {
                             //if (_log.shouldLog(Log.DEBUG)) _log.debug("done writing, but data remains...");
-                            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
+                            // stay interested
+                            //key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                         } else {
                             //if (_log.shouldLog(Log.DEBUG)) _log.debug("done writing, no data remains...");
+                            key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                         }
                         break;
                     } else if (buf.remaining() > 0) {
                         //if (_log.shouldLog(Log.DEBUG)) _log.debug("buffer data remaining...");
-                        key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
+                        // stay interested
+                        //key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                         break;
                     } else {
@@ -593 +664 @@
                         //buffers++;
                         //if (buffer time is too much, add OP_WRITe to the interest ops and break?)
+                        // LOOP
                     }
                 } else {
+                    // Nothing more to write
+                    key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                     break;
                 }
             }
         } catch (CancelledKeyException cke) {
-            if (_log.shouldLog(Log.WARN)) _log.warn("error writing", cke);
+            if (_log.shouldLog(Log.WARN)) _log.warn("error writing on " + con, cke);
             _context.statManager().addRateData("ntcp.writeError", 1);
             con.close();
         } catch (IOException ioe) {
-            if (_log.shouldLog(Log.WARN)) _log.warn("error writing", ioe);
+            if (_log.shouldLog(Log.WARN)) _log.warn("error writing on " + con, ioe);
             _context.statManager().addRateData("ntcp.writeError", 1);
             con.close();

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

@@ -166 +166 @@
         _readBufs = new ConcurrentLinkedQueue();
         _writeBufs = new ConcurrentLinkedQueue();
-        _bwRequests = new ConcurrentHashSet(2);
+        _bwRequests = new ConcurrentHashSet(8);
         // TODO possible switch to CLQ but beware non-constant size() - see below
         _outbound = new LinkedBlockingQueue();
@@ -258 +258 @@
         for (Iterator<FIFOBandwidthLimiter.Request> iter = _bwRequests.iterator(); iter.hasNext(); ) {
             iter.next().abort();
+            // we would like to return read ByteBuffers via EventPumper.releaseBuf(),
+            // but we can't risk releasing it twice
         }
         _bwRequests.clear();
+
+        _writeBufs.clear();
+        ByteBuffer bb;
+        while ((bb = _readBufs.poll()) != null) {
+            EventPumper.releaseBuf(bb);
+        }
 
         OutNetMessage msg;
@@ -790 +798 @@
     }
 
+    /**
+     *  The FifoBandwidthLimiter.CompleteListener callback.
+     *  Does the delayed read or write.
+     */
     public void complete(FIFOBandwidthLimiter.Request req) {
         removeRequest(req);
         ByteBuffer buf = (ByteBuffer)req.attachment();
         if (req.getTotalInboundRequested() > 0) {
+            if (_closed) {
+                EventPumper.releaseBuf(buf);
+                return;
+            }
             _context.statManager().addRateData("ntcp.throttledReadComplete", (System.currentTimeMillis()-req.getRequestTime()));
             recv(buf);
@@ -801 +817 @@
             _transport.getPumper().wantsRead(this);
             //_transport.getReader().wantsRead(this);
-        } else if (req.getTotalOutboundRequested() > 0) {
+        } else if (req.getTotalOutboundRequested() > 0 && !_closed) {
             _context.statManager().addRateData("ntcp.throttledWriteComplete", (System.currentTimeMillis()-req.getRequestTime()));
             write(buf);
@@ -837 +853 @@
      * The contents of the buffer have been read and can be processed asap.
      * This should not block, and the NTCP connection now owns the buffer
-     * to do with as it pleases.
+     * to do with as it pleases BUT it should eventually copy out the data
+     * and call EventPumper.releaseBuf().
      */
     public void recv(ByteBuffer buf) {
@@ -978 +995 @@
      * encoded as "sizeof(data)+data+pad+crc", and those are encrypted
      * with the session key and the last 16 bytes of the previous encrypted
-     * i2np message.  the contents of the buffer is owned by the EventPumper,
-     * so data should be copied out.
+     * i2np message.
+     *
+     * The NTCP connection now owns the buffer
+     * BUT it must copy out the data
+     * as reader will call EventPumper.releaseBuf().
      */
     synchronized void recvEncryptedI2NP(ByteBuffer buf) {
@@ -1011 +1031 @@
     }
     
-    /** _decryptBlockBuf contains another cleartext block of I2NP to parse */
+    /**
+     *  Append the next 16 bytes of cleartext to the read state.
+     *  _decryptBlockBuf contains another cleartext block of I2NP to parse.
+     *  Caller must synchronize!
+     *  @return success
+     */
     private boolean recvUnencryptedI2NP() {
         _curReadState.receiveBlock(_decryptBlockBuf);

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

@@ -90 +90 @@
         _context.statManager().createRateStat("ntcp.connectFailedInvalidPort", "", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.bidRejectedLocalAddress", "", "ntcp", RATES);
-        _context.statManager().createRateStat("ntcp.bidRejectedNoNTCPAddress", "", "ntcp", RATES);
+        //_context.statManager().createRateStat("ntcp.bidRejectedNoNTCPAddress", "", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.connectFailedTimeout", "", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.connectFailedTimeoutIOE", "", "ntcp", RATES);
@@ -125 +125 @@
         _context.statManager().createRateStat("ntcp.noBidTooLargeI2NP", "send size", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.queuedRecv", "", "ntcp", RATES);
-        //_context.statManager().createRateStat("ntcp.read", "", "ntcp", RATES);
-        _context.statManager().createRateStat("ntcp.readEOF", "", "ntcp", RATES);
+        _context.statManager().createRateStat("ntcp.read", "", "ntcp", RATES);
+        //_context.statManager().createRateStat("ntcp.readEOF", "", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.readError", "", "ntcp", RATES);
         _context.statManager().createRateStat("ntcp.receiveCorruptEstablishment", "", "ntcp", RATES);
@@ -290 +290 @@
         if (addr == null) {
             markUnreachable(peer);
-            _context.statManager().addRateData("ntcp.bidRejectedNoNTCPAddress", 1);
+            //_context.statManager().addRateData("ntcp.bidRejectedNoNTCPAddress", 1);
             //_context.shitlist().shitlistRouter(toAddress.getIdentity().calculateHash(), "No NTCP address", STYLE);
             if (_log.shouldLog(Log.DEBUG))

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

@@ -3 +3 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 import net.i2p.router.RouterContext;
@@ -20 +22 @@
     // TODO change to LBQ ??
     private final List<NTCPConnection> _pendingConnections;
-    private final List<NTCPConnection> _liveReads;
-    private final List<NTCPConnection> _readAfterLive;
+    private final Set<NTCPConnection> _liveReads;
+    private final Set<NTCPConnection> _readAfterLive;
     private final List<Runner> _runners;
     
@@ -28 +30 @@
         _log = ctx.logManager().getLog(getClass());
         _pendingConnections = new ArrayList(16);
-        _runners = new ArrayList(5);
-        _liveReads = new ArrayList(5);
-        _readAfterLive = new ArrayList();
+        _runners = new ArrayList(8);
+        _liveReads = new HashSet(8);
+        _readAfterLive = new HashSet(8);
     }
     
@@ -41 +43 @@
         }
     }
+
     public void stopReading() {
         while (!_runners.isEmpty()) {
@@ -56 +59 @@
         synchronized (_pendingConnections) {
             if (_liveReads.contains(con)) {
-                if (!_readAfterLive.contains(con)) {
-                    _readAfterLive.add(con);
-                }
+                _readAfterLive.add(con);
                 already = true;
             } else if (!_pendingConnections.contains(con)) {
@@ -79 +80 @@
     private class Runner implements Runnable {
         private boolean _stop;
-        public Runner() { _stop = false; }
+
+        public Runner() {}
+
         public void stop() { _stop = true; }
+
         public void run() {
             if (_log.shouldLog(Log.INFO)) _log.info("Starting reader");
@@ -119 +123 @@
     
     /**
-     * process everything read
+     * Process everything read.
+     * Return read buffers back to the pool as we process them.
      */
     private void processRead(NTCPConnection con) {
@@ -130 +135 @@
                 _log.debug("Processing read buffer as an establishment for " + con + " with [" + est + "]");
             if (est == null) {
+                EventPumper.releaseBuf(buf);
                 if (!con.isEstablished()) {
                     // establish state is only removed when the connection is fully established,
@@ -145 +151 @@
                     _log.error("establishment state [" + est + "] is complete, yet the connection isn't established? " 
                                + con.isEstablished() + " (inbound? " + con.isInbound() + " " + con + ")");
+                EventPumper.releaseBuf(buf);
                 break;
             }
             est.receive(buf);
+            EventPumper.releaseBuf(buf);
             if (est.isCorrupt()) {
                 if (_log.shouldLog(Log.WARN))
@@ -155 +163 @@
                 con.close();
                 return;
-            } else if (buf.remaining() <= 0) {
-                // not necessary, getNextReadBuf() removes
-                //con.removeReadBuf(buf);
             }
             if (est.isComplete() && est.getExtraBytes() != null)
@@ -170 +175 @@
                 _log.debug("Processing read buffer as part of an i2np message (" + buf.remaining() + " bytes)");
             con.recvEncryptedI2NP(buf);
-            // not necessary, getNextReadBuf() removes
-            //con.removeReadBuf(buf);
+            EventPumper.releaseBuf(buf);
         }
     }