2004-12-03 jrandom

* Toss in a small pool of threads (3) to execute the events queued up with
      the SimpleTimer, as we do currently see the occational event
      notification spiking up to a second or so.
    * Implement a SAM client API in java, useful for event based streaming (or
      for testing the SAM bridge)
    * Added support to shut down the SAM bridge on OOM (useful if the SAM
      bridge is being run outside of the router).
    * Include the SAM test code in the sam.jar
    * Remove an irrelevent warning message from SAM, which was caused by
      perfectly normal operation due to a session being closed.
    * Removed some unnecessary synchronization in the streaming lib's
      PacketQueue
    * More quickly clean up the memory used by the streaming lib by
      immediately killing each packet's resend job as soon as it is ACKed (or
      cancelled), so that there are no longer any valid pointers to the
      (potentially 32KB) packet.
    * Fixed the timestamps dumped to stdout when debugging the PacketHandler.
    * Drop packets that would expand our inbound window beyond our maximum
      buffer size (default 32 messages)
    * Always read the ACK/NACK data from the verified packets received, even
      if we are going to drop them
    * Always adjust the window when there are messages ACKed, though do not
      change its size except as before.
    * Streamlined some synchronization in the router's I2CP handling
    * Streamlined some memory allocation in the SAM bridge
    * Default the streaming lib to disconnect on inactivity, rather than send
      an empty message.
this still doesnt get the BT to where it needs to be, or fix the timeout problem,
but i dont like having so many commits outstanding and these updates are sound

apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelRunner.java

@@ -242,8 +242,8 @@ public class I2PTunnelRunner extends I2PThread implements I2PSocket.SocketErrorL
                               + ex.getMessage() + "\")");
             } catch (IOException ex) {
                 if (!finished) {
-                    if (_log.shouldLog(Log.ERROR))
-                        _log.error(direction + ": Error forwarding", ex);
+                    if (_log.shouldLog(Log.WARN))
+                        _log.warn(direction + ": Error forwarding", ex);
                 }
                 //else
                 //    _log.warn("You may ignore this", ex);

apps/ministreaming/java/src/net/i2p/client/streaming/I2PSocketManagerFactory.java

@@ -100,7 +100,8 @@ public class I2PSocketManagerFactory {
             //p.setProperty("tunnels.depthInbound", "0");
         }
 
-        opts.setProperty(I2PClient.PROP_TCP_HOST, i2cpHost);
+        if (i2cpHost != null)
+            opts.setProperty(I2PClient.PROP_TCP_HOST, i2cpHost);
         opts.setProperty(I2PClient.PROP_TCP_PORT, "" + i2cpPort);
         
         try {

apps/ministreaming/java/src/net/i2p/client/streaming/TestSwarm.java

@@ -0,0 +1,218 @@
+package net.i2p.client.streaming;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileInputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+
+import java.util.Map;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Properties;
+import java.util.StringTokenizer;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PClientFactory;
+import net.i2p.data.DataHelper;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+import net.i2p.util.I2PThread;
+
+/**
+ * Sit around on a destination, receiving lots of data and sending lots of
+ * data to whomever talks to us.
+ *
+ * Usage: TestSwarm myKeyFile [peerDestFile ]*
+ *
+ */
+public class TestSwarm {
+    private I2PAppContext _context;
+    private Log _log;
+    private String _destFile;
+    private String _peerDestFiles[];
+    private String _conOptions;
+    private I2PSocketManager _manager;
+    private boolean _dead;
+    
+    public static void main(String args[]) {
+        if (args.length < 1) {
+            System.err.println("Usage: TestSwarm myDestFile [peerDestFile ]*");
+            return;
+        }
+        I2PAppContext ctx = new I2PAppContext();
+        String files[] = new String[args.length - 1];
+        System.arraycopy(args, 1, files, 0, files.length);
+        TestSwarm swarm = new TestSwarm(ctx, args[0], files);
+        swarm.startup();
+    }
+    
+    public TestSwarm(I2PAppContext ctx, String destFile, String peerDestFiles[]) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(TestSwarm.class);
+        _dead = false;
+        _destFile = destFile;
+        _peerDestFiles = peerDestFiles;
+        _conOptions = "";
+    }
+    
+    public void startup() {
+        _log.debug("Starting up");
+        File keys = new File(_destFile);
+        if (!keys.exists()) {
+            try {
+                I2PClientFactory.createClient().createDestination(new FileOutputStream(keys));
+            } catch (Exception e) {
+                _log.error("Error creating a new destination on " + keys, e);
+                return;
+            }
+        }
+        try {
+            _manager = I2PSocketManagerFactory.createManager(new FileInputStream(_destFile), "localhost", 10001, null);
+        } catch (Exception e) {
+            _log.error("Error creatign the manager", e);
+            return;
+        }
+
+        I2PThread listener = new I2PThread(new Listener(), "Listener");
+        listener.start();
+        
+        connectWithPeers();
+    }
+    
+    
+    private void connectWithPeers() {
+        if (_peerDestFiles != null) {
+            for (int i = 0; i < _peerDestFiles.length; i++) {
+                try {
+                    FileInputStream fin = new FileInputStream(_peerDestFiles[i]);
+                    Destination dest = new Destination();
+                    dest.readBytes(fin);
+                    
+                    I2PThread flooder = new I2PThread(new Flooder(dest), "Flooder+" + dest.calculateHash().toBase64().substring(0,4));
+                    flooder.start();
+                } catch (Exception e) {
+                    _log.error("Unable to read the peer from " + _peerDestFiles[i], e);
+                }
+            }
+        }
+    }
+    
+    private class Listener implements Runnable {
+        public void run() {
+            try {
+                I2PServerSocket ss = _manager.getServerSocket();
+                I2PSocket s = null;
+                while ( (s = ss.accept()) != null) {
+                    I2PThread flooder = new I2PThread(new Flooder(s), "Flooder-" + s.getPeerDestination().calculateHash().toBase64().substring(0,4));
+                    flooder.start();
+                }
+            } catch (Exception e) {
+                _log.error("Error listening", e);
+            }
+        }
+    }
+    
+    private static volatile long __conId = 0;
+    private class Flooder implements Runnable {
+        private Destination _remoteDestination;
+        private I2PSocket _socket;
+        private boolean _closed;
+        private long _started;
+        private long _totalSent;
+        private long _totalReceived;
+        private long _lastReceived;
+        private long _lastReceivedOn;
+        private long _connectionId;
+        
+        public Flooder(Destination dest) {
+            _socket = null;
+            _remoteDestination = dest;
+            _connectionId = ++__conId;
+            _closed = false;
+            _lastReceived = -1;
+            _lastReceivedOn = _context.clock().now();
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalReceived", "Data size received", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalSent", "Data size sent", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".started", "When we start", "swarm", new long[] { 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".lifetime", "How long we talk to a peer", "swarm", new long[] { 5*60*1000 });
+        }
+        
+        public Flooder(I2PSocket socket) {
+            _socket = socket;
+            _remoteDestination = socket.getPeerDestination();
+            _connectionId = ++__conId;
+            _closed = false;
+            _lastReceived = -1;
+            _lastReceivedOn = _context.clock().now();
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalReceived", "Data size received", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalSent", "Data size sent", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".started", "When we start", "swarm", new long[] { 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".lifetime", "How long we talk to a peer", "swarm", new long[] { 5*60*1000 });
+        }
+        
+        public long getConnectionId() { return _connectionId; }
+        public Destination getDestination() { return _remoteDestination; }
+        
+        public void run() {
+            _started = _context.clock().now();
+            _context.statManager().addRateData("swarm." + _connectionId + ".started", 1, 0);
+            byte data[] = new byte[32*1024];
+            long value = 0;
+            long lastSend = _context.clock().now();
+            if (_socket == null) {
+                try {
+                    _socket = _manager.connect(_remoteDestination);
+                } catch (Exception e) {
+                    _log.error("Error connecting to " + _remoteDestination.calculateHash().toBase64().substring(0,4));
+                    return;
+                }
+            }
+            
+            I2PThread floodListener = new I2PThread(new FloodListener(), "FloodListener" + _connectionId);
+            floodListener.start();
+            
+            try {
+                OutputStream out = _socket.getOutputStream();
+                while (!_closed) {
+                    out.write(data);
+                    // out.flush();
+                    _totalSent += data.length;
+                    _context.statManager().addRateData("swarm." + _connectionId + ".totalSent", _totalSent, 0);
+                    //try { Thread.sleep(100); } catch (InterruptedException ie) {}
+                    long now = _context.clock().now();
+                    _log.debug("Sending " + _connectionId + " after " + (now-lastSend));
+                    lastSend = now;
+                    try { Thread.sleep(20); } catch (InterruptedException ie) {}
+                }
+            } catch (Exception e) {
+                _log.error("Error sending", e);
+            }
+        }
+        
+        private class FloodListener implements Runnable {
+            public void run() {
+                long lastRead = System.currentTimeMillis();
+                long now = lastRead;
+                try {
+                    InputStream in = _socket.getInputStream();
+                    byte buf[] = new byte[32*1024];
+                    int read = 0;
+                    while ( (read = in.read(buf)) != -1) {
+                        now = System.currentTimeMillis();
+                        _totalReceived += read;
+                        _context.statManager().addRateData("swarm." + getConnectionId() + ".totalReceived", _totalReceived, 0);
+                        _log.debug("Receiving " + _connectionId + " with " + read + " after " + (now-lastRead));
+                        lastRead = now;
+                    }
+                } catch (Exception e) {
+                    _log.error("Error listening to the flood", e);
+                }
+            }
+        }
+    }
+}

apps/sam/java/build.xml

@@ -10,7 +10,7 @@
         <mkdir dir="./build" />
         <mkdir dir="./build/obj" />
         <javac 
-            srcdir="./src" 
+            srcdir="./src:./test" 
             debug="true" deprecation="on" source="1.3" target="1.3" 
             destdir="./build/obj" 
             classpath="../../../core/java/build/i2p.jar:../../ministreaming/java/build/mstreaming.jar" />

apps/sam/java/src/net/i2p/sam/SAMStreamSession.java

@@ -28,8 +28,10 @@ import net.i2p.client.streaming.I2PSocketManager;
 import net.i2p.client.streaming.I2PSocketManagerFactory;
 import net.i2p.client.streaming.I2PSocketOptions;
 import net.i2p.data.Base64;
+import net.i2p.data.ByteArray;
 import net.i2p.data.DataFormatException;
 import net.i2p.data.Destination;
+import net.i2p.util.ByteCache;
 import net.i2p.util.I2PThread;
 import net.i2p.util.Log;
 
@@ -199,7 +201,7 @@ public class SAMStreamSession {
      *
      * @return True if the data was sent, false otherwise
      */
-    public boolean sendBytes(int id, byte[] data) {
+    public boolean sendBytes(int id, InputStream in, int size) { 
         Destination d = new Destination();
         SAMStreamSessionSocketHandler handler = getSocketHandler(id);
         
@@ -208,7 +210,7 @@ public class SAMStreamSession {
             return false;
         }
 
-        return handler.sendBytes(data);
+        return handler.sendBytes(in, size);
     }
 
     /**
@@ -313,8 +315,7 @@ public class SAMStreamSession {
         }
 
         if (removed == null) {
-            _log.error("BUG! Trying to remove inexistent SAM STREAM session socket handler " + id);
-            recv.stopStreamReceiving();
+            // ignore - likely the socket handler was already removed by removeAllSocketHandlers
         } else {
             removed.stopRunning();
             _log.debug("Removed SAM STREAM session socket handler " + id);
@@ -462,17 +463,29 @@ public class SAMStreamSession {
          *
          * @return True if data has been sent without errors, false otherwise
          */
-        public boolean sendBytes(byte[] data) {
+        public boolean sendBytes(InputStream in, int size) { // byte[] data) {
             if (_log.shouldLog(Log.DEBUG)) {
-                _log.debug("Handler " + id + ": sending " + data.length
+                _log.debug("Handler " + id + ": sending " + size
                            + " bytes");
             }
+            ByteCache cache = ByteCache.getInstance(1024, 4);
+            ByteArray ba = cache.acquire();
             try {
-                i2pSocketOS.write(data);
+                int sent = 0;
+                byte buf[] = ba.getData(); 
+                while (sent < size) {
+                    int read = in.read(buf);
+                    if (read == -1)
+                        throw new IOException("Insufficient data from the SAM client (" + sent + "/" + size + ")");
+                    i2pSocketOS.write(buf, 0, read);
+                    sent += read;
+                }
                 //i2pSocketOS.flush();
             } catch (IOException e) {
                 _log.error("Error sending data through I2P socket", e);
                 return false;
+            } finally {
+                cache.release(ba);
             }
 
             return true;

apps/sam/java/src/net/i2p/sam/SAMv1Handler.java

@@ -44,6 +44,9 @@ public class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatag
     private SAMDatagramSession datagramSession = null;
     private SAMStreamSession streamSession = null;
 
+    private long _id;
+    private static volatile long __id = 0;
+    
     /**
      * Create a new SAM version 1 handler.  This constructor expects
      * that the SAM HELLO message has been still answered (and
@@ -68,6 +71,7 @@ public class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatag
      */
     public SAMv1Handler(Socket s, int verMajor, int verMinor, Properties i2cpProps) throws SAMException, IOException {
         super(s, verMajor, verMinor, i2cpProps);
+        _id = ++__id;
         _log.debug("SAM version 1 handler instantiated");
 
         if ((this.verMajor != 1) || (this.verMinor != 0)) {
@@ -82,7 +86,7 @@ public class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatag
         StringTokenizer tok;
         Properties props;
 
-        this.thread.setName("SAMv1Handler");
+        this.thread.setName("SAMv1Handler " + _id);
         _log.debug("SAM handling started");
 
         try {
@@ -550,12 +554,7 @@ public class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatag
         }
 
         try {
-            DataInputStream in = new DataInputStream(getClientSocketInputStream());
-            byte[] data = new byte[size];
-
-            in.readFully(data);
-
-            if (!streamSession.sendBytes(id, data)) {
+            if (!streamSession.sendBytes(id, getClientSocketInputStream(), size)) { // data)) {
                 _log.error("STREAM SEND failed");
                 boolean rv = writeString("STREAM CLOSED RESULT=CANT_REACH_PEER ID=" + id + " MESSAGE=\"Send of " + size + " bytes failed\"\n");
                 streamSession.closeConnection(id);
@@ -801,7 +800,7 @@ public class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatag
     }
 
     public void stopStreamReceiving() {
-        _log.debug("stopStreamReceiving() invoked");
+        _log.debug("stopStreamReceiving() invoked", new Exception("stopped"));
 
         if (streamSession == null) {
             _log.error("BUG! Got stream receiving stop, but session is null!");

apps/sam/java/src/net/i2p/sam/client/SAMClientEventListenerImpl.java

@@ -0,0 +1,18 @@
+package net.i2p.sam.client;
+
+import java.util.Properties;
+
+/**
+ * Basic noop client event listener
+ */
+public class SAMClientEventListenerImpl implements SAMReader.SAMClientEventListener {
+    public void destReplyReceived(String publicKey, String privateKey) {}
+    public void helloReplyReceived(boolean ok) {}
+    public void namingReplyReceived(String name, String result, String value, String message) {}
+    public void sessionStatusReceived(String result, String destination, String message) {}
+    public void streamClosedReceived(String result, int id, String message) {}
+    public void streamConnectedReceived(String remoteDestination, int id) {}
+    public void streamDataReceived(int id, byte[] data, int offset, int length) {}
+    public void streamStatusReceived(String result, int id, String message) {}
+    public void unknownMessageReceived(String major, String minor, Properties params) {}
+}

apps/sam/java/src/net/i2p/sam/client/SAMEventHandler.java

@@ -0,0 +1,127 @@
+package net.i2p.sam.client;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * Simple helper implementation of a the SAMClientEventListener
+ *
+ */
+public class SAMEventHandler extends SAMClientEventListenerImpl {
+    private I2PAppContext _context;
+    private Log _log;
+    private Boolean _helloOk;
+    private Object _helloLock = new Object();
+    private Boolean _sessionCreateOk;
+    private Object _sessionCreateLock = new Object();
+    private Object _namingReplyLock = new Object();
+    private Map _namingReplies = new HashMap();
+
+    public SAMEventHandler(I2PAppContext ctx) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(getClass());
+    }
+    
+    public void helloReplyReceived(boolean ok) {
+        synchronized (_helloLock) {
+            if (ok)
+                _helloOk = Boolean.TRUE;
+            else
+                _helloOk = Boolean.FALSE;
+            _helloLock.notifyAll();
+        }
+    }
+
+    public void sessionStatusReceived(String result, String destination, String msg) {
+        synchronized (_sessionCreateLock) {
+            if (SAMReader.SAMClientEventListener.SESSION_STATUS_OK.equals(result))
+                _sessionCreateOk = Boolean.TRUE;
+            else 
+                _sessionCreateOk = Boolean.FALSE;
+            _sessionCreateLock.notifyAll();
+        }
+    }
+
+    public void namingReplyReceived(String name, String result, String value, String msg) {
+        synchronized (_namingReplyLock) {
+            if (SAMReader.SAMClientEventListener.NAMING_REPLY_OK.equals(result)) 
+                _namingReplies.put(name, value);
+            else
+                _namingReplies.put(name, result);
+            _namingReplyLock.notifyAll();
+        }
+    }
+
+    public void unknownMessageReceived(String major, String minor, Properties params) {
+        _log.error("wrt, [" + major + "] [" + minor + "] [" + params + "]");
+    }
+
+    
+    //
+    // blocking lookup calls below
+    //
+
+    /**
+     * Wait for the connection to be established, returning true if everything 
+     * went ok
+     */
+    public boolean waitForHelloReply() {
+        while (true) {
+            try {
+                synchronized (_helloLock) {
+                    if (_helloOk == null)
+                        _helloLock.wait();
+                    else 
+                        return _helloOk.booleanValue();
+                }
+            } catch (InterruptedException ie) {}
+        }
+    }
+
+    /**
+     * Wait for the session to be created, returning true if everything went ok
+     *
+     */
+    public boolean waitForSessionCreateReply() {
+        while (true) {
+            try {
+                synchronized (_sessionCreateLock) {
+                    if (_sessionCreateOk == null)
+                        _sessionCreateLock.wait();
+                    else
+                        return _sessionCreateOk.booleanValue();
+                }
+            } catch (InterruptedException ie) {}
+        }
+    }
+    
+    /**
+     * Return the destination found matching the name, or null if the key was
+     * not able to be retrieved.
+     *
+     * @param name name to be looked for, or "ME"
+     */
+    public String waitForNamingReply(String name) {
+        while (true) {
+            try {
+                synchronized (_namingReplyLock) {
+                    String val = (String)_namingReplies.remove(name);
+                    if (val == null) {
+                        _namingReplyLock.wait();
+                    } else {
+                        if (SAMReader.SAMClientEventListener.NAMING_REPLY_INVALID_KEY.equals(val))
+                            return null;
+                        else if (SAMReader.SAMClientEventListener.NAMING_REPLY_KEY_NOT_FOUND.equals(val))
+                            return null;
+                        else
+                            return val;
+                    }
+                }
+            } catch (InterruptedException ie) {}
+        }
+    }
+}

apps/sam/java/src/net/i2p/sam/client/SAMReader.java

@@ -0,0 +1,253 @@
+package net.i2p.sam.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.IOException;
+
+import java.util.Properties;
+import java.util.StringTokenizer;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.DataHelper;
+import net.i2p.util.Log;
+import net.i2p.util.I2PThread;
+
+/**
+ * Read from a socket, producing events for any SAM message read
+ *
+ */
+public class SAMReader {
+    private Log _log;
+    private InputStream _inRaw;
+    private SAMClientEventListener _listener;
+    private boolean _live;
+    
+    public SAMReader(I2PAppContext context, InputStream samIn, SAMClientEventListener listener) {
+        _log = context.logManager().getLog(SAMReader.class);
+        _inRaw = samIn;
+        _listener = listener;
+    }
+    
+    public void startReading() {
+        _live = true;
+        I2PThread t = new I2PThread(new Runner(), "SAM reader");
+        t.start();
+    }
+    public void stopReading() { _live = false; }
+    
+    /**
+     * Async event notification interface for SAM clients
+     *
+     */
+    public interface SAMClientEventListener {
+        public static final String SESSION_STATUS_OK = "OK";
+        public static final String SESSION_STATUS_DUPLICATE_DEST = "DUPLICATE_DEST";
+        public static final String SESSION_STATUS_I2P_ERROR = "I2P_ERROR";
+        public static final String SESSION_STATUS_INVALID_KEY = "INVALID_KEY";
+        
+        public static final String STREAM_STATUS_OK = "OK";
+        public static final String STREAM_STATUS_CANT_REACH_PEER = "CANT_REACH_PEER";
+        public static final String STREAM_STATUS_I2P_ERROR = "I2P_ERROR";
+        public static final String STREAM_STATUS_INVALID_KEY = "INVALID_KEY";
+        public static final String STREAM_STATUS_TIMEOUT = "TIMEOUT";
+        
+        public static final String STREAM_CLOSED_OK = "OK";
+        public static final String STREAM_CLOSED_CANT_REACH_PEER = "CANT_REACH_PEER";
+        public static final String STREAM_CLOSED_I2P_ERROR = "I2P_ERROR";
+        public static final String STREAM_CLOSED_PEER_NOT_FOUND = "PEER_NOT_FOUND";
+        public static final String STREAM_CLOSED_TIMEOUT = "CLOSED";
+        
+        public static final String NAMING_REPLY_OK = "OK";
+        public static final String NAMING_REPLY_INVALID_KEY = "INVALID_KEY";
+        public static final String NAMING_REPLY_KEY_NOT_FOUND = "KEY_NOT_FOUND";
+        
+        public void helloReplyReceived(boolean ok);
+        public void sessionStatusReceived(String result, String destination, String message);
+        public void streamStatusReceived(String result, int id, String message);
+        public void streamConnectedReceived(String remoteDestination, int id);
+        public void streamClosedReceived(String result, int id, String message);
+        public void streamDataReceived(int id, byte data[], int offset, int length);
+        public void namingReplyReceived(String name, String result, String value, String message);
+        public void destReplyReceived(String publicKey, String privateKey);
+        
+        public void unknownMessageReceived(String major, String minor, Properties params);
+    }
+    
+    private class Runner implements Runnable {
+        public void run() {
+            Properties params = new Properties();
+            ByteArrayOutputStream baos = new ByteArrayOutputStream(80);
+            while (_live) {
+                
+                try {
+                    int c = -1;
+                    while ((c = _inRaw.read()) != -1) {
+                        if (c == '\n') {
+                            break;
+                        }
+                        baos.write(c);
+                    }
+                    if (c == -1) {
+                        _log.error("Error reading from the SAM bridge");
+                        return;
+                    }
+                } catch (IOException ioe) {
+                    _log.error("Error reading from SAM", ioe);
+                }
+                
+                String line = new String(baos.toByteArray());
+                baos.reset();
+                
+                if (line == null) {
+                    _log.info("No more data from the SAM bridge");
+                    break;
+                }
+                
+                _log.debug("Line read from the bridge: " + line);
+                
+                StringTokenizer tok = new StringTokenizer(line);
+                
+                if (tok.countTokens() < 2) {
+                    _log.error("Invalid SAM line: [" + line + "]");
+                    _live = false;
+                    return;
+                }
+                
+                String major = tok.nextToken();
+                String minor = tok.nextToken();
+                
+                params.clear();
+                while (tok.hasMoreTokens()) {
+                    String pair = tok.nextToken();
+                    int eq = pair.indexOf('=');
+                    if ( (eq > 0) && (eq < pair.length() - 1) ) {
+                        String name = pair.substring(0, eq);
+                        String val = pair.substring(eq+1);
+                        while ( (val.charAt(0) == '\"') && (val.length() > 0) )
+                            val = val.substring(1);
+                        while ( (val.length() > 0) && (val.charAt(val.length()-1) == '\"') )
+                            val = val.substring(0, val.length()-1);
+                        params.setProperty(name, val);
+                    }
+                }
+                
+                processEvent(major, minor, params);
+            }
+        }
+    }
+    
+    /**
+     * Big ugly method parsing everything.  If I cared, I'd factor this out into
+     * a dozen tiny methods.
+     *
+     */
+    private void processEvent(String major, String minor, Properties params) {
+        if ("HELLO".equals(major)) {
+            if ("REPLY".equals(minor)) {
+                String result = params.getProperty("RESULT");
+                if ("OK".equals(result))
+                    _listener.helloReplyReceived(true);
+                else
+                    _listener.helloReplyReceived(false);
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else if ("SESSION".equals(major)) {
+            if ("STATUS".equals(minor)) {
+                String result = params.getProperty("RESULT");
+                String dest = params.getProperty("DESTINATION");
+                String msg = params.getProperty("MESSAGE");
+                _listener.sessionStatusReceived(result, dest, msg);
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else if ("STREAM".equals(major)) {
+            if ("STATUS".equals(minor)) {
+                String result = params.getProperty("RESULT");
+                String id = params.getProperty("ID");
+                String msg = params.getProperty("MESSAGE");
+                if (id != null) {
+                    try {
+                        _listener.streamStatusReceived(result, Integer.parseInt(id), msg);
+                    } catch (NumberFormatException nfe) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    }
+                } else {
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else if ("CONNECTED".equals(minor)) {
+                String dest = params.getProperty("DESTINATION");
+                String id = params.getProperty("ID");
+                if (id != null) {
+                    try {
+                        _listener.streamConnectedReceived(dest, Integer.parseInt(id));
+                    } catch (NumberFormatException nfe) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    }
+                } else {
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else if ("CLOSED".equals(minor)) {
+                String result = params.getProperty("RESULT");
+                String id = params.getProperty("ID");
+                String msg = params.getProperty("MESSAGE");
+                if (id != null) {
+                    try {
+                        _listener.streamClosedReceived(result, Integer.parseInt(id), msg);
+                    } catch (NumberFormatException nfe) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    }
+                } else {
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else if ("RECEIVED".equals(minor)) {
+                String id = params.getProperty("ID");
+                String size = params.getProperty("SIZE");
+                if (id != null) {
+                    try {
+                        int idVal = Integer.parseInt(id);
+                        int sizeVal = Integer.parseInt(size);
+                        
+                        byte data[] = new byte[sizeVal];
+                        int read = DataHelper.read(_inRaw, data);
+                        if (read != sizeVal) {
+                            _listener.unknownMessageReceived(major, minor, params);
+                        } else {
+                            _listener.streamDataReceived(idVal, data, 0, sizeVal);
+                        }
+                    } catch (NumberFormatException nfe) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    } catch (IOException ioe) {
+                        _live = false;
+                        _listener.unknownMessageReceived(major, minor, params);
+                    }
+                } else {
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else if ("NAMING".equals(major)) {
+            if ("REPLY".equals(minor)) {
+                String name = params.getProperty("NAME");
+                String result = params.getProperty("RESULT");
+                String value = params.getProperty("VALUE");
+                String msg = params.getProperty("MESSAGE");
+                _listener.namingReplyReceived(name, result, value, msg);
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else if ("DEST".equals(major)) {
+            if ("REPLY".equals(minor)) {
+                String pub = params.getProperty("PUB");
+                String priv = params.getProperty("PRIV");
+                _listener.destReplyReceived(pub, priv);
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else {
+            _listener.unknownMessageReceived(major, minor, params);
+        }
+    }
+}

apps/sam/java/test/net/i2p/sam/TestSwarm.java

@@ -0,0 +1,312 @@
+package net.i2p.sam;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.FileOutputStream;
+import java.io.FileInputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+
+import java.util.Map;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Properties;
+import java.util.StringTokenizer;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.DataHelper;
+import net.i2p.util.Log;
+import net.i2p.util.I2PThread;
+
+import net.i2p.sam.client.SAMEventHandler;
+import net.i2p.sam.client.SAMClientEventListenerImpl;
+import net.i2p.sam.client.SAMReader;
+
+/**
+ * Sit around on a SAM destination, receiving lots of data and sending lots of
+ * data to whomever talks to us.
+ *
+ * Usage: TestSwarm samHost samPort myKeyFile [peerDestFile ]*
+ *
+ */
+public class TestSwarm {
+    private I2PAppContext _context;
+    private Log _log;
+    private String _samHost;
+    private String _samPort;
+    private String _destFile;
+    private String _peerDestFiles[];
+    private String _conOptions;
+    private Socket _samSocket;
+    private OutputStream _samOut;
+    private InputStream _samIn;
+    private SAMReader _reader;
+    private boolean _dead;
+    private SAMEventHandler _eventHandler;
+    /** Connection id (Integer) to peer (Flooder) */
+    private Map _remotePeers;
+    
+    public static void main(String args[]) {
+        if (args.length < 3) {
+            System.err.println("Usage: TestSwarm samHost samPort myDestFile [peerDestFile ]*");
+            return;
+        }
+        I2PAppContext ctx = new I2PAppContext();
+        String files[] = new String[args.length - 3];
+        System.arraycopy(args, 3, files, 0, files.length);
+        TestSwarm swarm = new TestSwarm(ctx, args[0], args[1], args[2], files);
+        swarm.startup();
+    }
+    
+    public TestSwarm(I2PAppContext ctx, String samHost, String samPort, String destFile, String peerDestFiles[]) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(TestSwarm.class);
+        _dead = false;
+        _samHost = samHost;
+        _samPort = samPort;
+        _destFile = destFile;
+        _peerDestFiles = peerDestFiles;
+        _conOptions = "";
+        _eventHandler = new SwarmEventHandler(_context);
+        _remotePeers = new HashMap();
+    }
+    
+    public void startup() {
+        _log.debug("Starting up");
+        boolean ok = connect();
+        _log.debug("Connected: " + ok);
+        if (ok) {
+            _reader = new SAMReader(_context, _samIn, _eventHandler);
+            _reader.startReading();
+            _log.debug("Reader created");
+            String ourDest = handshake();
+            _log.debug("Handshake complete.  we are " + ourDest);
+            if (ourDest != null) {
+                boolean written = writeDest(ourDest);
+                _log.debug("Dest written");
+                if (written) {
+                    connectWithPeers();
+                    _log.debug("connected with peers");
+                }
+            }
+        }
+    }
+    
+    private class SwarmEventHandler extends SAMEventHandler {
+        public SwarmEventHandler(I2PAppContext ctx) { super(ctx); }
+        public void streamClosedReceived(String result, int id, String message) {
+            Flooder flooder = null;
+            synchronized (_remotePeers) {
+                flooder = (Flooder)_remotePeers.remove(new Integer(id));
+            }
+            if (flooder != null) {
+                flooder.closed();
+                _log.debug("Connection " + flooder.getConnectionId() + " closed to " + flooder.getDestination());
+            } else {
+                _log.error("wtf, not connected to " + id + " but we were just closed?");
+            }
+        }
+        public void streamDataReceived(int id, byte data[], int offset, int length) {
+            Flooder flooder = null;
+            synchronized (_remotePeers) {
+                flooder = (Flooder)_remotePeers.get(new Integer(id));
+            }
+            long value = DataHelper.fromLong(data, 0, 4);
+            if (flooder != null) {
+                flooder.received(length, value);
+            } else {
+                _log.error("wtf, not connected to " + id + " but we received " + value + "?");
+            }
+        }
+        public void streamConnectedReceived(String dest, int id) {  
+            _log.debug("Connection " + id + " received from " + dest);
+
+            Flooder flooder = new Flooder(id, dest);
+            synchronized (_remotePeers) {
+                _remotePeers.put(new Integer(id), flooder);
+            }
+            I2PThread t = new I2PThread(flooder, "Flood " + id);
+            t.start();
+        }
+
+    }
+    
+    private boolean connect() {
+        try {
+            _samSocket = new Socket(_samHost, Integer.parseInt(_samPort));
+            _samOut = _samSocket.getOutputStream();
+            _samIn = _samSocket.getInputStream();
+            return true;
+        } catch (Exception e) {
+            _log.error("Unable to connect to SAM at " + _samHost + ":" + _samPort, e);
+            return false;
+        }
+    }
+    
+    private String handshake() {
+        synchronized (_samOut) {
+            try {
+                _samOut.write("HELLO VERSION MIN=1.0 MAX=1.0\n".getBytes());
+                _samOut.flush();
+                _log.debug("Hello sent");
+                boolean ok = _eventHandler.waitForHelloReply();
+                _log.debug("Hello reply found: " + ok);
+                if (!ok) 
+                    throw new IOException("wtf, hello failed?");
+                String req = "SESSION CREATE STYLE=STREAM DESTINATION=" + _destFile + " " + _conOptions + "\n";
+                _samOut.write(req.getBytes());
+                _samOut.flush();
+                _log.debug("Session create sent");
+                ok = _eventHandler.waitForSessionCreateReply();
+                _log.debug("Session create reply found: " + ok);
+
+                req = "NAMING LOOKUP NAME=ME\n";
+                _samOut.write(req.getBytes());
+                _samOut.flush();
+                _log.debug("Naming lookup sent");
+                String destination = _eventHandler.waitForNamingReply("ME");
+                _log.debug("Naming lookup reply found: " + destination);
+                if (destination == null) {
+                    _log.error("No naming lookup reply found!");
+                    return null;
+                } else {
+                    _log.info(_destFile + " is located at " + destination);
+                }
+                return destination;
+            } catch (Exception e) {
+                _log.error("Error handshaking", e);
+                return null;
+            }
+        }
+    }
+    
+    private boolean writeDest(String dest) {
+        try {
+            FileOutputStream fos = new FileOutputStream(_destFile);
+            fos.write(dest.getBytes());
+            fos.close();
+            return true;
+        } catch (Exception e) {
+            _log.error("Error writing to " + _destFile, e);
+            return false;
+        }
+    }
+    
+    private void connectWithPeers() {
+        if (_peerDestFiles != null) {
+            for (int i = 0; i < _peerDestFiles.length; i++) {
+                try {
+                    FileInputStream fin = new FileInputStream(_peerDestFiles[i]);
+                    byte dest[] = new byte[1024];
+                    int read = DataHelper.read(fin, dest);
+                    
+                    String remDest = new String(dest, 0, read);
+                    int con = 0;
+                    Flooder flooder = null;
+                    synchronized (_remotePeers) {
+                        con = _remotePeers.size() + 1;
+                        flooder = new Flooder(con, remDest);
+                        _remotePeers.put(new Integer(con), flooder);
+                    }
+                    
+                    byte msg[] = ("STREAM CONNECT ID=" + con + " DESTINATION=" + remDest + "\n").getBytes();
+                    synchronized (_samOut) {
+                        _samOut.write(msg);
+                        _samOut.flush();
+                    }
+                    I2PThread flood = new I2PThread(flooder, "Flood " + con);
+                    flood.start();
+                    _log.debug("Starting flooder with peer from " + _peerDestFiles[i] + ": " + con);
+                } catch (IOException ioe) {
+                    _log.error("Unable to read the peer from " + _peerDestFiles[i]);
+                }
+            }
+        }
+    }
+    
+    private class Flooder implements Runnable {
+        private int _connectionId; 
+        private String _remoteDestination;
+        private boolean _closed;
+        private long _started;
+        private long _totalSent;
+        private long _totalReceived;
+        private long _lastReceived;
+        private long _lastReceivedOn;
+        private boolean _outOfSync;
+        
+        public Flooder(int conId, String remDest) {
+            _connectionId = conId;
+            _remoteDestination = remDest;
+            _closed = false;
+            _outOfSync = false;
+            _lastReceived = -1;
+            _lastReceivedOn = _context.clock().now();
+            _context.statManager().createRateStat("swarm." + conId + ".totalReceived", "Data size received", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + conId + ".totalSent", "Data size sent", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + conId + ".started", "When we start", "swarm", new long[] { 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + conId + ".lifetime", "How long we talk to a peer", "swarm", new long[] { 5*60*1000 });
+        }
+        
+        public int getConnectionId() { return _connectionId; }
+        public String getDestination() { return _remoteDestination; }
+        
+        public void closed() {
+            _closed = true;
+            long lifetime = _context.clock().now() - _started;
+            _context.statManager().addRateData("swarm." + _connectionId + ".lifetime", lifetime, lifetime);
+        }
+        public void run() {
+            _started = _context.clock().now();
+            _context.statManager().addRateData("swarm." + _connectionId + ".started", 1, 0);
+            byte data[] = new byte[32*1024];
+            long value = 0;
+            long lastSend = _context.clock().now();
+            while (!_closed) {
+                byte msg[] = ("STREAM SEND ID=" + _connectionId + " SIZE=" + data.length + "\n").getBytes();
+                DataHelper.toLong(data, 0, 4, value);
+                try {
+                    synchronized (_samOut) {
+                        _samOut.write(msg);
+                        _samOut.write(data);
+                        _samOut.flush();
+                    }
+                } catch (IOException ioe) {
+                    _log.error("Error talking to SAM", ioe);
+                    return;
+                }
+                _totalSent += data.length;
+                _context.statManager().addRateData("swarm." + _connectionId + ".totalSent", _totalSent, 0);
+                value++;
+                try { Thread.sleep(20); } catch (InterruptedException ie) {}
+                long now = _context.clock().now();
+                _log.debug("Sending " + value + " on " + _connectionId + " after " + (now-lastSend));
+                lastSend = now;
+            }
+        }
+        public void received(int len, long value) {
+            _totalReceived += len;
+            if ( (!_outOfSync) && (len % 32*1024 != 0) ) {
+                _outOfSync = true;
+                if (_log.shouldLog(Log.ERROR))
+                    _log.error("Out of sync (len=" + len + " after " + (_totalReceived-len) + ")");
+            }
+            _context.statManager().addRateData("swarm." + getConnectionId() + ".totalReceived", _totalReceived, 0);
+            if (value != _lastReceived + 1) {
+                if (!_outOfSync)
+                    _log.error("Received " + value + " when expecting " + (_lastReceived+1) + " on " 
+                               + _connectionId + " with " + _remoteDestination.substring(0,6));
+                else
+                    _log.debug("(out of sync) Received " + value + " when expecting " + (_lastReceived+1) + " on " 
+                               + _connectionId + " with " + _remoteDestination.substring(0,6));
+            } else {
+                _log.debug("Received " + value + " on " + _connectionId + " after " + (_context.clock().now()-_lastReceivedOn)
+                           + "ms with " + _remoteDestination.substring(0,6));
+            }
+            _lastReceived = value;
+            _lastReceivedOn = _context.clock().now();
+        }
+    }
+}

apps/streaming/java/src/net/i2p/client/streaming/Connection.java

@@ -112,6 +112,8 @@ public class Connection {
         _connectLock = new Object();
         _activeResends = 0;
         _context.statManager().createRateStat("stream.con.windowSizeAtCongestion", "How large was our send window when we send a dup?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.chokeSizeBegin", "How many messages were outstanding when we started to choke?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
+        _context.statManager().createRateStat("stream.chokeSizeEnd", "How many messages were outstanding when we stopped being choked?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
     }
     
     public long getNextOutboundPacketNum() { 
@@ -135,9 +137,14 @@ public class Connection {
     boolean packetSendChoke(long timeoutMs) {
         if (false) return true;
         long writeExpire = timeoutMs;
+        long start = _context.clock().now();
+        boolean started = false;
         while (true) {
             long timeLeft = writeExpire - _context.clock().now();
             synchronized (_outboundPackets) {
+                if (!started)
+                    _context.statManager().addRateData("stream.chokeSizeBegin", _outboundPackets.size(), timeoutMs);
+                started = true;
                 if (_outboundPackets.size() >= _options.getWindowSize()) {
                     if (writeExpire > 0) {
                         if (timeLeft <= 0) {
@@ -154,6 +161,7 @@ public class Connection {
                         try { _outboundPackets.wait(); } catch (InterruptedException ie) {}
                     }
                 } else {
+                    _context.statManager().addRateData("stream.chokeSizeEnd", _outboundPackets.size(), _context.clock().now() - start);
                     return true;
                 }
             }
@@ -325,14 +333,19 @@ public class Connection {
             _occurredEventCount++;
         } else {
             _occurredTime = now;
-            if (_occurredEventCount > 100) {
-                _log.log(Log.CRIT, "More than 100 events (" + _occurredEventCount + ") in a second on " 
-                                   + toString() + ": scheduler = " + sched);
+            if ( (_occurredEventCount > 1000) && (_log.shouldLog(Log.WARN)) ) {
+                _log.warn("More than 1000 events (" + _occurredEventCount + ") in a second on " 
+                          + toString() + ": scheduler = " + sched);
             }
             _occurredEventCount = 0;
         }
+        
+        long before = System.currentTimeMillis();
             
         sched.eventOccurred(this);
+        long elapsed = System.currentTimeMillis() - before;
+        if ( (elapsed > 1000) && (_log.shouldLog(Log.WARN)) )
+            _log.warn("Took " + elapsed + "ms to pump through " + sched);
     }
     
     void resetReceived() {
@@ -714,6 +727,7 @@ public class Connection {
         public ResendPacketEvent(PacketLocal packet) {
             _packet = packet;
             _currentIsActiveResend = false;
+            packet.setResendPacketEvent(ResendPacketEvent.this);
         }
         
         public void timeReached() {

apps/streaming/java/src/net/i2p/client/streaming/ConnectionDataReceiver.java

@@ -64,7 +64,7 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
         if (doSend) {
             PacketLocal packet = send(buf, off, size);
             //dont wait for non-acks
-            if ( (packet.getPayloadSize() > 0) || (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) )
+            if ( (packet.getSequenceNum() > 0) || (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) )
                 return packet;
             else
                 return _dummyStatus;
@@ -95,8 +95,16 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
      * @return the packet sent
      */
     public PacketLocal send(byte buf[], int off, int size, boolean forceIncrement) {
+        long before = System.currentTimeMillis();
         PacketLocal packet = buildPacket(buf, off, size, forceIncrement);
+        long built = System.currentTimeMillis();
         _connection.sendPacket(packet);
+        long sent = System.currentTimeMillis();
+        
+        if ( (built-before > 1000) && (_log.shouldLog(Log.WARN)) )
+            _log.warn("wtf, took " + (built-before) + "ms to build a packet: " + packet);
+        if ( (sent-built> 1000) && (_log.shouldLog(Log.WARN)) )
+            _log.warn("wtf, took " + (sent-built) + "ms to send a packet: " + packet);
         return packet;
     }
     

apps/streaming/java/src/net/i2p/client/streaming/ConnectionOptions.java

@@ -87,7 +87,7 @@ public class ConnectionOptions extends I2PSocketOptionsImpl {
         setMaxResends(getInt(opts, PROP_MAX_RESENDS, 5));
         setWriteTimeout(getInt(opts, PROP_WRITE_TIMEOUT, -1));
         setInactivityTimeout(getInt(opts, PROP_INACTIVITY_TIMEOUT, 5*60*1000));
-        setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, INACTIVITY_ACTION_SEND));
+        setInactivityAction(getInt(opts, PROP_INACTIVITY_ACTION, INACTIVITY_ACTION_DISCONNECT));
         setInboundBufferSize((getMaxMessageSize() + 2) * Connection.MAX_WINDOW_SIZE);
     }
     

apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java

@@ -30,12 +30,22 @@ public class ConnectionPacketHandler {
     /** distribute a packet to the connection specified */
     void receivePacket(Packet packet, Connection con) throws I2PException {
         boolean ok = verifyPacket(packet, con);
-        if (!ok) return;
+        if (!ok) {
+            if ( (!packet.isFlagSet(Packet.FLAG_RESET)) && (_log.shouldLog(Log.ERROR)) )
+                _log.error("Packet does NOT verify: " + packet);
+            return;
+        }
         con.packetReceived();
         
-        if (con.getInputStream().getTotalQueuedSize() > con.getOptions().getInboundBufferSize()) {
+        long ready = con.getInputStream().getHighestReadyBockId();
+        int available = con.getOptions().getInboundBufferSize() - con.getInputStream().getTotalReadySize();
+        int allowedBlocks = available/con.getOptions().getMaxMessageSize();
+        if (packet.getSequenceNum() > ready + allowedBlocks) {
             if (_log.shouldLog(Log.WARN))
-                _log.warn("Inbound buffer exceeded on connection " + con + ": dropping " + packet);
+                _log.warn("Inbound buffer exceeded on connection " + con + " (" 
+                          + ready + "/"+ (ready+allowedBlocks) + "/" + available
+                          + ": dropping " + packet);
+            ack(con, packet.getAckThrough(), packet.getNacks(), null, false);
             con.getOptions().setChoke(5*1000);
             return;
         }
@@ -95,8 +105,20 @@ public class ConnectionPacketHandler {
             }
         }
         
+        boolean fastAck = ack(con, packet.getAckThrough(), packet.getNacks(), packet, isNew);
+        con.eventOccurred();
+        if (fastAck) {
+            if (con.getLastSendTime() + con.getOptions().getRTT() < _context.clock().now()) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Fast ack for dup " + packet);
+                con.ackImmediately();
+            }
+        }
+    }
+    
+    private boolean ack(Connection con, long ackThrough, long nacks[], Packet packet, boolean isNew) {
         int numResends = 0;
-        List acked = con.ackPackets(packet.getAckThrough(), packet.getNacks());
+        List acked = con.ackPackets(ackThrough, nacks);
         if ( (acked != null) && (acked.size() > 0) ) {
             if (_log.shouldLog(Log.DEBUG))
                 _log.debug(acked.size() + " of our packets acked with " + packet);
@@ -130,18 +152,15 @@ public class ConnectionPacketHandler {
             _context.statManager().addRateData("stream.con.packetsAckedPerMessageReceived", acked.size(), highestRTT);
         }
 
-        boolean fastAck = adjustWindow(con, isNew, packet.getSequenceNum(), numResends, (acked != null ? acked.size() : 0));
-        con.eventOccurred();
-        if (fastAck) {
-            if (con.getLastSendTime() + con.getOptions().getRTT() < _context.clock().now()) {
-                if (_log.shouldLog(Log.DEBUG))
-                    _log.debug("Fast ack for dup " + packet);
-                con.ackImmediately();
-            }
-        }
+        if (packet != null)
+            return adjustWindow(con, isNew, packet.getSequenceNum(), numResends, (acked != null ? acked.size() : 0));
+        else
+            return adjustWindow(con, false, -1, numResends, (acked != null ? acked.size() : 0));
     }
     
+    
     private boolean adjustWindow(Connection con, boolean isNew, long sequenceNum, int numResends, int acked) {
+        boolean congested = false;
         if ( (!isNew) && (sequenceNum > 0) ) {
             // dup real packet
             int oldSize = con.getOptions().getWindowSize();
@@ -156,64 +175,38 @@ public class ConnectionPacketHandler {
                            + con.getLastCongestionSeenAt() + " (#resends: " + numResends 
                            + ") for " + con);
 
-            return true;
-        //} else if (numResends > 0) {
-            // window sizes are shrunk on resend, not on ack
-        } else {
-            if (acked > 0) { 
-                long lowest = con.getHighestAckedThrough();
-                if (lowest >= con.getCongestionWindowEnd()) {
-                    // new packet that ack'ed uncongested data, or an empty ack
-                    int newWindowSize = con.getOptions().getWindowSize();
-                    
-                    if (numResends <= 0) {
-                        if (newWindowSize > con.getLastCongestionSeenAt() / 2) {
-                            // congestion avoidance
+            congested = true;
+        } 
+        
+        long lowest = con.getHighestAckedThrough();
+        if (lowest >= con.getCongestionWindowEnd()) {
+            // new packet that ack'ed uncongested data, or an empty ack
+            int newWindowSize = con.getOptions().getWindowSize();
 
-                            // we can't use newWindowSize += 1/newWindowSize, since we're
-                            // integers, so lets use a random distribution instead
-                            int shouldIncrement = _context.random().nextInt(newWindowSize);
-                            if (shouldIncrement <= 0)
-                                newWindowSize += 1;
-                        } else {
-                            // slow start
-                            newWindowSize += 1;
-                        }
-                    }
-                    
-                    if (_log.shouldLog(Log.DEBUG))
-                        _log.debug("New window size " + newWindowSize + " congestionSeenAt: "
-                                   + con.getLastCongestionSeenAt() + " (#resends: " + numResends 
-                                   + ") for " + con);
-                    con.getOptions().setWindowSize(newWindowSize);
-                    con.setCongestionWindowEnd(newWindowSize + lowest);
+            if ( (!congested) && (acked > 0) && (numResends <= 0) ) {
+                if (newWindowSize > con.getLastCongestionSeenAt() / 2) {
+                    // congestion avoidance
+
+                    // we can't use newWindowSize += 1/newWindowSize, since we're
+                    // integers, so lets use a random distribution instead
+                    int shouldIncrement = _context.random().nextInt(newWindowSize);
+                    if (shouldIncrement <= 0)
+                        newWindowSize += 1;
+                } else {
+                    // slow start
+                    newWindowSize += 1;
                 }
-            } else {
-                // received a message that doesn't contain a new ack
-                
-                // ehh. cant do this, as we SACK and the acks may be 
-                // received out of order: 
-                // Alice: RECEIVE 2
-                // Alice: SEND    ack 2 nack 1
-                // Alice: RECEIVE 1
-                // Alice: SEND    ack 2
-                // Bob:   RECEIVE ack 2
-                // Bob:   RECEIVE ack 2 nack 1 <-- NOT bad
-                
-                /*
-                if (con.getUnackedPacketsSent() > 0) {
-                    // peer got a dup
-                    int oldSize = con.getOptions().getWindowSize();
-                    oldSize >>>= 1;
-                    if (oldSize <= 0)
-                        oldSize = 1;
-                    con.getOptions().setWindowSize(oldSize);
-                    return false;
-                }
-                */
             }
+                    
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("New window size " + newWindowSize + " congestionSeenAt: "
+                           + con.getLastCongestionSeenAt() + " (#resends: " + numResends 
+                           + ") for " + con);
+            con.getOptions().setWindowSize(newWindowSize);
+            con.setCongestionWindowEnd(newWindowSize + lowest);
         }
-        return false;
+        
+        return congested;
     }
     
     /**

apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java

@@ -33,9 +33,9 @@ public class MessageInputStream extends InputStream {
     private List _readyDataBlocks;
     private int _readyDataBlockIndex;
     /** highest message ID used in the readyDataBlocks */
-    private long _highestReadyBlockId;
+    private volatile long _highestReadyBlockId;
     /** highest overall message ID */
-    private long _highestBlockId;
+    private volatile long _highestBlockId;
     /** 
      * Message ID (Long) to ByteArray for blocks received
      * out of order when there are lower IDs not yet 
@@ -74,15 +74,13 @@ public class MessageInputStream extends InputStream {
     
     /** What is the highest block ID we've completely received through? */
     public long getHighestReadyBockId() { 
-        synchronized (_dataLock) {
-            return _highestReadyBlockId; 
-        }
+        // not synchronized as it doesnt hurt to read a too-low value
+        return _highestReadyBlockId; 
     }
     
     public long getHighestBlockId() { 
-        synchronized (_dataLock) {
-            return _highestBlockId;
-        }
+        // not synchronized as it doesnt hurt to read a too-low value
+        return _highestBlockId;
     }
     
     /**
@@ -394,6 +392,21 @@ public class MessageInputStream extends InputStream {
         }
     }
     
+    public int getTotalReadySize() {
+        synchronized (_dataLock) {
+            if (_locallyClosed) return 0;
+            int numBytes = 0;
+            for (int i = 0; i < _readyDataBlocks.size(); i++) {
+                ByteArray cur = (ByteArray)_readyDataBlocks.get(i);
+                if (i == 0)
+                    numBytes += cur.getData().length - _readyDataBlockIndex;
+                else
+                    numBytes += cur.getData().length;
+            }
+            return numBytes;
+        }
+    }
+    
     public void close() {
         synchronized (_dataLock) {
             _readyDataBlocks.clear();

apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java

@@ -47,7 +47,7 @@ public class MessageOutputStream extends OutputStream {
         _written = 0;
         _closed = false;
         _writeTimeout = -1;
-        _passiveFlushDelay = 5*1000;
+        _passiveFlushDelay = 2*1000;
         _flusher = new Flusher();
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("MessageOutputStream created");
@@ -83,8 +83,7 @@ public class MessageOutputStream extends OutputStream {
                     remaining = 0;
                     _lastBuffered = _context.clock().now();
                     if (_passiveFlushDelay > 0) {
-                        // if it is already enqueued, this just pushes it further out
-                        SimpleTimer.getInstance().addEvent(_flusher, _passiveFlushDelay);
+                        _flusher.enqueue();
                     }
                 } else {
                     // buffer whatever we can fit then flush,
@@ -115,9 +114,9 @@ public class MessageOutputStream extends OutputStream {
                 ws.waitForAccept(_writeTimeout);
                 if (!ws.writeAccepted()) {
                     if (_writeTimeout > 0)
-                        throw new InterruptedIOException("Write not accepted within timeout");
+                        throw new InterruptedIOException("Write not accepted within timeout: " + ws);
                     else
-                        throw new IOException("Write not accepted into the queue");
+                        throw new IOException("Write not accepted into the queue: " + ws);
                 }
             } else {
                 if (_log.shouldLog(Log.DEBUG))
@@ -140,7 +139,24 @@ public class MessageOutputStream extends OutputStream {
      * period of inactivity
      */
     private class Flusher implements SimpleTimer.TimedEvent {
+        private boolean _enqueued;
+        public void enqueue() {
+            // no need to be overly worried about duplicates - it would just 
+            // push it further out
+            if (!_enqueued)
+                SimpleTimer.getInstance().addEvent(_flusher, _passiveFlushDelay);
+            _enqueued = true;
+        }
         public void timeReached() {
+            _enqueued = false;
+            long timeLeft = (_lastBuffered + _passiveFlushDelay - _context.clock().now());
+            if (timeLeft > 0)
+                enqueue();
+            else
+                doFlush();
+        }
+        
+        private void doFlush() {
             boolean sent = false;
             WriteStatus ws = null;
             synchronized (_dataLock) {
@@ -159,7 +175,6 @@ public class MessageOutputStream extends OutputStream {
             if (sent && _log.shouldLog(Log.DEBUG)) 
                 _log.debug("Passive flush of " + ws);
         }
-        
     }
     
     /** 
@@ -275,6 +290,7 @@ public class MessageOutputStream extends OutputStream {
     }
     void flushAvailable(DataReceiver target, boolean blocking) throws IOException {
         WriteStatus ws = null;
+        long before = System.currentTimeMillis();
         synchronized (_dataLock) {
             // _buf may be null, but the data receiver can handle that just fine,
             // deciding whether or not to send a packet
@@ -284,6 +300,10 @@ public class MessageOutputStream extends OutputStream {
             _dataLock.notifyAll();
             _lastFlushed = _context.clock().now();
         }
+        long afterBuild = System.currentTimeMillis();
+        if ( (afterBuild - before > 1000) && (_log.shouldLog(Log.DEBUG)) )
+            _log.debug("Took " + (afterBuild-before) + "ms to build a packet?  " + ws);
+        
         if (blocking && ws != null) {
             ws.waitForAccept(_writeTimeout);
             if (ws.writeFailed())
@@ -291,6 +311,9 @@ public class MessageOutputStream extends OutputStream {
             else if (!ws.writeAccepted())
                 throw new InterruptedIOException("Flush available timed out");
         }
+        long afterAccept = System.currentTimeMillis();
+        if ( (afterAccept - afterBuild > 1000) && (_log.shouldLog(Log.DEBUG)) )
+            _log.debug("Took " + (afterAccept-afterBuild) + "ms to accept a packet? " + ws);
         return;
     }
     

apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java

@@ -104,7 +104,7 @@ public class PacketHandler {
         }
     }
     
-    private static final SimpleDateFormat _fmt = new SimpleDateFormat("hh:mm:ss.SSS");
+    private static final SimpleDateFormat _fmt = new SimpleDateFormat("HH:mm:ss.SSS");
     void displayPacket(Packet packet, String prefix) {
         String msg = null;
         synchronized (_fmt) {

apps/streaming/java/src/net/i2p/client/streaming/PacketLocal.java

@@ -5,6 +5,8 @@ import java.util.Set;
 import net.i2p.I2PAppContext;
 import net.i2p.data.Destination;
 import net.i2p.data.SessionKey;
+import net.i2p.util.Log;
+import net.i2p.util.SimpleTimer;
 
 /**
  * coordinate local attributes about a packet - send time, ack time, number of
@@ -12,6 +14,7 @@ import net.i2p.data.SessionKey;
  */
 public class PacketLocal extends Packet implements MessageOutputStream.WriteStatus {
     private I2PAppContext _context;
+    private Log _log;
     private Connection _connection;
     private Destination _to;
     private SessionKey _keyUsed;
@@ -22,6 +25,7 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
     private long _acceptedOn;
     private long _ackOn;
     private long _cancelledOn;
+    private SimpleTimer.TimedEvent _resendEvent;
     
     public PacketLocal(I2PAppContext ctx, Destination to) {
         this(ctx, to, null);
@@ -29,6 +33,7 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
     public PacketLocal(I2PAppContext ctx, Destination to, Connection con) {
         _context = ctx;
         _createdOn = ctx.clock().now();
+        _log = ctx.logManager().getLog(PacketLocal.class);
         _to = to;
         _connection = con;
         _lastSend = -1;
@@ -78,12 +83,16 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
                 _ackOn = _context.clock().now(); 
             notifyAll();
         }
+        SimpleTimer.getInstance().removeEvent(_resendEvent);
     }
     public void cancelled() { 
         synchronized (this) {
             _cancelledOn = _context.clock().now();
             notifyAll();
         }
+        SimpleTimer.getInstance().removeEvent(_resendEvent);
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Cancelled! " + toString(), new Exception("cancelled"));
     }
     
     /** how long after packet creation was it acked? */
@@ -97,6 +106,8 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
     public long getLastSend() { return _lastSend; }
     public Connection getConnection() { return _connection; }
     
+    public void setResendPacketEvent(SimpleTimer.TimedEvent evt) { _resendEvent = evt; }
+    
     public String toString() {
         String str = super.toString();
         if (_ackOn > 0)
@@ -108,12 +119,24 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
     public void waitForAccept(int maxWaitMs) {
         if (_connection == null) 
             throw new IllegalStateException("Cannot wait for accept with no connection");
-        long expiration = _context.clock().now()+maxWaitMs;
+        long before = _context.clock().now();
+        long expiration = before+maxWaitMs;
+        int queued = _connection.getUnackedPacketsSent();
+        int window = _connection.getOptions().getWindowSize();
         boolean accepted = _connection.packetSendChoke(maxWaitMs);
+        long after = _context.clock().now();
         if (accepted)
-            _acceptedOn = _context.clock().now();
+            _acceptedOn = after;
         else
             _acceptedOn = -1;
+        int afterQueued = _connection.getUnackedPacketsSent();
+        if ( (after - before > 1000) && (_log.shouldLog(Log.DEBUG)) )
+            _log.debug("Took " + (after-before) + "ms to get " 
+                       + (accepted ? " accepted" : " rejected")
+                       + (_cancelledOn > 0 ? " and CANCELLED" : "")
+                       + ", queued behind " + queued +" with a window size of " + window 
+                       + ", finally accepted with " + afterQueued + " queued: " 
+                       + toString());
     }
     
     public void waitForCompletion(int maxWaitMs) {

apps/streaming/java/src/net/i2p/client/streaming/PacketQueue.java

@@ -24,24 +24,17 @@ class PacketQueue {
     private Log _log;
     private I2PSession _session;
     private ConnectionManager _connectionManager;
-    private byte _buf[];
     private ByteCache _cache = ByteCache.getInstance(64, 36*1024);
     
     public PacketQueue(I2PAppContext context, I2PSession session, ConnectionManager mgr) {
         _context = context;
         _session = session;
         _connectionManager = mgr;
-        _buf = _cache.acquire().getData(); // new byte[36*1024];
         _log = context.logManager().getLog(PacketQueue.class);
         _context.statManager().createRateStat("stream.con.sendMessageSize", "Size of a message sent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
         _context.statManager().createRateStat("stream.con.sendDuplicateSize", "Size of a message resent on a connection", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
     }
     
-    protected void finalize() throws Throwable {
-        _cache.release(new ByteArray(_buf));
-        super.finalize();
-    }
-    
     /**
      * Add a new packet to be sent out ASAP
      */
@@ -53,7 +46,7 @@ class PacketQueue {
             keyUsed = new SessionKey();
         Set tagsSent = packet.getTagsSent();
         if (tagsSent == null)
-            tagsSent = new HashSet();
+            tagsSent = new HashSet(0);
 
         // cache this from before sendMessage
         String conStr = (packet.getConnection() != null ? packet.getConnection().toString() : "");
@@ -63,29 +56,36 @@ class PacketQueue {
         } else {
             _log.debug("Sending... " + packet);
         }
+    
+        ByteArray ba = _cache.acquire();
+        byte buf[] = ba.getData();
 
         long begin = 0;
         long end = 0;
         boolean sent = false;
         try {
             int size = 0;
-            synchronized (this) {
-                Arrays.fill(_buf, (byte)0x0);
-                if (packet.shouldSign())
-                    size = packet.writeSignedPacket(_buf, 0, _context, _session.getPrivateKey());
-                else
-                    size = packet.writePacket(_buf, 0);
+            long beforeWrite = System.currentTimeMillis();
+            if (packet.shouldSign())
+                size = packet.writeSignedPacket(buf, 0, _context, _session.getPrivateKey());
+            else
+                size = packet.writePacket(buf, 0);
+            long writeTime = System.currentTimeMillis() - beforeWrite;
+            if ( (writeTime > 1000) && (_log.shouldLog(Log.WARN)) )
+                _log.warn("took " + writeTime + "ms to write the packet: " + packet);
 
-                // this should not block!
-                begin = _context.clock().now();
-                sent = _session.sendMessage(packet.getTo(), _buf, 0, size, keyUsed, tagsSent);
-                end = _context.clock().now();
-            }
+            // this should not block!
+            begin = _context.clock().now();
+            sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent);
+            end = _context.clock().now();
+            
+            if ( (end-begin > 1000) && (_log.shouldLog(Log.WARN)) ) 
+                _log.warn("Took " + (end-begin) + "ms to sendMessage(...) " + packet);
+            
             _context.statManager().addRateData("stream.con.sendMessageSize", size, packet.getLifetime());
             if (packet.getNumSends() > 1)
                 _context.statManager().addRateData("stream.con.sendDuplicateSize", size, packet.getLifetime());
             
-            
             Connection con = packet.getConnection();
             if (con != null) {
                 con.incrementBytesSent(size);
@@ -97,6 +97,8 @@ class PacketQueue {
                 _log.warn("Unable to send the packet " + packet, ise);
         }
         
+        _cache.release(ba);
+        
         if (!sent) {
             if (_log.shouldLog(Log.WARN))
                 _log.warn("Send failed for " + packet);

apps/streaming/java/src/net/i2p/client/streaming/SchedulerImpl.java

@@ -32,5 +32,6 @@ abstract class SchedulerImpl implements TaskScheduler {
             //    _log.debug("firing event on " + _connection, _addedBy);
             _connection.eventOccurred(); 
         }
+        public String toString() { return "event on " + _connection; }
     }
 }

build.xml

@@ -239,8 +239,6 @@
         <copy file="build/routerconsole.jar" todir="pkg-temp/lib/" />
         <copy file="build/i2ptunnel.war" todir="pkg-temp/webapps/" />
         <copy file="build/routerconsole.war" todir="pkg-temp/webapps/" />
-        <copy file="installer/resources/wrapper.config" todir="pkg-temp/" />
-        <copy file="installer/resources/wrapper.config" tofile="pkg-temp/wrapper.config.updated" />
         <copy file="history.txt" todir="pkg-temp/" />
         <copy file="hosts.txt" todir="pkg-temp/" />
         <mkdir dir="pkg-temp/eepsite" />

core/java/src/net/i2p/client/I2PSessionImpl2.java

@@ -108,6 +108,8 @@ class I2PSessionImpl2 extends I2PSessionImpl {
 
     private boolean sendBestEffort(Destination dest, byte payload[], SessionKey keyUsed, Set tagsSent)
                     throws I2PSessionException {
+        long begin = _context.clock().now();
+        
         SessionKey key = _context.sessionKeyManager().getCurrentKey(dest.getPublicKey());
         if (key == null) key = _context.sessionKeyManager().createSession(dest.getPublicKey());
         SessionTag tag = _context.sessionKeyManager().consumeNextAvailableTag(dest.getPublicKey(), key);
@@ -180,9 +182,17 @@ class I2PSessionImpl2 extends I2PSessionImpl {
                        + " sync took " + (inSendingSync-beforeSendingSync) 
                        + " add took " + (afterSendingSync-inSendingSync));
         _producer.sendMessage(this, dest, nonce, payload, tag, key, sentTags, newKey);
+        
+        // since this is 'best effort', all we're waiting for is a status update 
+        // saying that the router received it - in theory, that should come back
+        // immediately, but in practice can take up to a second (though usually
+        // much quicker).  setting this to false will short-circuit that delay
+        boolean actuallyWait = true;
+        
         long beforeWaitFor = _context.clock().now();
-        state.waitFor(MessageStatusMessage.STATUS_SEND_ACCEPTED, 
-                      _context.clock().now() + getTimeout());
+        if (actuallyWait)
+            state.waitFor(MessageStatusMessage.STATUS_SEND_ACCEPTED, 
+                          _context.clock().now() + getTimeout());
         long afterWaitFor = _context.clock().now();
         long inRemovingSync = 0;
         synchronized (_sendingStates) {
@@ -190,7 +200,7 @@ class I2PSessionImpl2 extends I2PSessionImpl {
             _sendingStates.remove(state);
         }
         long afterRemovingSync = _context.clock().now();
-        boolean found = state.received(MessageStatusMessage.STATUS_SEND_ACCEPTED);
+        boolean found = !actuallyWait || state.received(MessageStatusMessage.STATUS_SEND_ACCEPTED);
         if (_log.shouldLog(Log.DEBUG))
             _log.debug(getPrefix() + "After waitFor sending state " + state.getMessageId()
                        + " / " + state.getNonce() + " found = " + found);
@@ -200,6 +210,13 @@ class I2PSessionImpl2 extends I2PSessionImpl {
             _log.warn("wtf, took " + timeToSend + "ms to send the message?!", new Exception("baz"));
         }
         
+        if ( (afterRemovingSync - begin > 500) && (_log.shouldLog(Log.WARN) ) ) {
+            _log.warn("Took " + (afterRemovingSync-begin) + "ms to sendBestEffort, "
+                      + (afterSendingSync-begin) + "ms to prepare, "
+                      + (beforeWaitFor-afterSendingSync) + "ms to send, "
+                      + (afterRemovingSync-beforeWaitFor) + "ms waiting for reply");
+        }
+        
         if (found) {
             if (_log.shouldLog(Log.INFO))
                 _log.info(getPrefix() + "Message sent after " + state.getElapsed() + "ms with "

core/java/src/net/i2p/client/MessagePayloadMessageHandler.java

@@ -16,6 +16,7 @@ import net.i2p.data.i2cp.I2CPMessage;
 import net.i2p.data.i2cp.MessageId;
 import net.i2p.data.i2cp.MessagePayloadMessage;
 import net.i2p.data.i2cp.ReceiveMessageEndMessage;
+import net.i2p.util.Log;
 
 /**
  * Handle I2CP MessagePayloadMessages from the router delivering the contents
@@ -30,7 +31,8 @@ class MessagePayloadMessageHandler extends HandlerImpl {
     }
 
     public void handleMessage(I2CPMessage message, I2PSessionImpl session) {
-        _log.debug("Handle message " + message);
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Handle message " + message);
         try {
             MessagePayloadMessage msg = (MessagePayloadMessage) message;
             MessageId id = msg.getMessageId();
@@ -55,9 +57,8 @@ class MessagePayloadMessageHandler extends HandlerImpl {
         Payload payload = msg.getPayload();
         byte[] data = _context.elGamalAESEngine().decrypt(payload.getEncryptedData(), session.getDecryptionKey());
         if (data == null) {
-            _log
-                .error("Error decrypting the payload to public key "
-                       + session.getMyDestination().getPublicKey().toBase64() + "\nPayload: " + payload.calculateHash());
+            if (_log.shouldLog(Log.ERROR))
+                _log.error("Error decrypting the payload");
             throw new DataFormatException("Unable to decrypt the payload");
         }
         payload.setUnencryptedData(data);

core/java/src/net/i2p/data/DataHelper.java

@@ -735,7 +735,7 @@ public class DataHelper {
 
     /** decompress the GZIP compressed data (returning null on error) */
     public static byte[] decompress(byte orig[]) throws IOException {
-        return decompress(orig, 0, orig.length);
+        return (orig != null ? decompress(orig, 0, orig.length) : null);
     }
     public static byte[] decompress(byte orig[], int offset, int length) throws IOException {
         if ((orig == null) || (orig.length <= 0)) return orig;

core/java/src/net/i2p/util/SimpleTimer.java

@@ -18,20 +18,30 @@ import net.i2p.I2PAppContext;
 public class SimpleTimer {
     private static final SimpleTimer _instance = new SimpleTimer();
     public static SimpleTimer getInstance() { return _instance; }
+    private I2PAppContext _context;
     private Log _log;
     /** event time (Long) to event (TimedEvent) mapping */
     private TreeMap _events;
     /** event (TimedEvent) to event time (Long) mapping */
     private Map _eventTimes;
+    private List _readyEvents;
     
     private SimpleTimer() {
-        _log = I2PAppContext.getGlobalContext().logManager().getLog(SimpleTimer.class);
+        _context = I2PAppContext.getGlobalContext();
+        _log = _context.logManager().getLog(SimpleTimer.class);
         _events = new TreeMap();
         _eventTimes = new HashMap();
+        _readyEvents = new ArrayList(4);
         I2PThread runner = new I2PThread(new SimpleTimerRunner());
         runner.setName("SimpleTimer");
         runner.setDaemon(true);
         runner.start();
+        for (int i = 0; i < 3; i++) {
+            I2PThread executor = new I2PThread(new Executor());
+            executor.setName("SimpleTimerExecutor " + i);
+            executor.setDaemon(true);
+            executor.start();
+        }
     }
     
     /**
@@ -40,18 +50,42 @@ public class SimpleTimer {
      */
     public void addEvent(TimedEvent event, long timeoutMs) {
         long eventTime = System.currentTimeMillis() + timeoutMs;
+        Long time = new Long(eventTime);
         synchronized (_events) {
             // remove the old scheduled position, then reinsert it
             if (_eventTimes.containsKey(event))
                 _events.remove(_eventTimes.get(event));
-            while (_events.containsKey(new Long(eventTime)))
-                eventTime++;
-            _events.put(new Long(eventTime), event);
-            _eventTimes.put(event, new Long(eventTime));
+            while (_events.containsKey(time))
+                time = new Long(time.longValue() + 1);
+            _events.put(time, event);
+            _eventTimes.put(event, time);
+            
+            if ( (_events.size() != _eventTimes.size()) ) {
+                _log.error("Skewed events: " + _events.size() + " for " + _eventTimes.size());
+                for (Iterator iter = _eventTimes.keySet().iterator(); iter.hasNext(); ) {
+                    TimedEvent evt = (TimedEvent)iter.next();
+                    Long when = (Long)_eventTimes.get(evt);
+                    TimedEvent cur = (TimedEvent)_events.get(when);
+                    if (cur != evt) {
+                        _log.error("event " + evt + " @ " + when + ": " + cur);
+                    }
+                }
+            }
+            
             _events.notifyAll();
         }
     }
     
+    public boolean removeEvent(TimedEvent evt) {
+        if (evt == null) return false;
+        synchronized (_events) {
+            Long when = (Long)_eventTimes.remove(evt);
+            if (when != null)
+                _events.remove(when);
+            return null != when;
+        }
+    }
+    
     /**
      * Simple interface for events to be queued up and notified on expiration
      */
@@ -82,8 +116,8 @@ public class SimpleTimer {
             while (true) {
                 try {
                     synchronized (_events) {
-                        if (_events.size() <= 0)
-                            _events.wait();
+                        //if (_events.size() <= 0)
+                        //    _events.wait();
                         //if (_events.size() > 100)
                         //    _log.warn("> 100 events!  " + _events.values());
                         long now = System.currentTimeMillis();
@@ -97,7 +131,7 @@ public class SimpleTimer {
                                 if (evt != null) {                            
                                     _eventTimes.remove(evt);
                                     eventsToFire.add(evt);
-                                }
+                                } 
                             } else {
                                 nextEventDelay = when.longValue() - now;
                                 nextEvent = _events.get(when);
@@ -128,32 +162,20 @@ public class SimpleTimer {
                 long now = System.currentTimeMillis();
                 now = now - (now % 1000);
 
-                for (int i = 0; i < eventsToFire.size(); i++) {
-                    TimedEvent evt = (TimedEvent)eventsToFire.get(i);
-                    try {
-                        evt.timeReached();
-                    } catch (Throwable t) {
-                        log("wtf, event borked: " + evt, t);
-                    }
-                    _recentEvents[4] = _recentEvents[3];
-                    _recentEvents[3] = _recentEvents[2];
-                    _recentEvents[2] = _recentEvents[1];
-                    _recentEvents[1] = _recentEvents[0];
-                    _recentEvents[0] = evt;
+                synchronized (_readyEvents) {
+                    for (int i = 0; i < eventsToFire.size(); i++) 
+                        _readyEvents.add(eventsToFire.get(i));
+                    _readyEvents.notifyAll();
                 }
 
                 if (_occurredTime == now) {
                     _occurredEventCount += eventsToFire.size();
                 } else {
                     _occurredTime = now;
-                    if (_occurredEventCount > 100) {
-                        StringBuffer buf = new StringBuffer(256);
+                    if (_occurredEventCount > 1000) {
+                        StringBuffer buf = new StringBuffer(128);
                         buf.append("Too many simpleTimerJobs (").append(_occurredEventCount);
-                        buf.append(") in a second!  Last 5: \n");
-                        for (int i = 0; i < _recentEvents.length; i++) {
-                            if (_recentEvents[i] != null)
-                                buf.append(_recentEvents[i]).append('\n');
-                        }
+                        buf.append(") in a second!");
                         _log.log(Log.CRIT, buf.toString());
                     }
                     _occurredEventCount = 0;
@@ -163,4 +185,30 @@ public class SimpleTimer {
             }
         }
     }
+    
+    private class Executor implements Runnable {
+        public void run() {
+            while (true) {
+                TimedEvent evt = null;
+                synchronized (_readyEvents) {
+                    if (_readyEvents.size() <= 0) 
+                        try { _readyEvents.wait(); } catch (InterruptedException ie) {}
+                    if (_readyEvents.size() > 0) 
+                        evt = (TimedEvent)_readyEvents.remove(0);
+                }
+                
+                if (evt != null) {
+                    long before = _context.clock().now();
+                    try {
+                        evt.timeReached();
+                    } catch (Throwable t) {
+                        log("wtf, event borked: " + evt, t);
+                    }
+                    long time = _context.clock().now() - before;
+                    if ( (time > 1000) && (_log != null) && (_log.shouldLog(Log.WARN)) )
+                        _log.warn("wtf, event execution took " + time + ": " + evt);
+                }
+            }
+        }
+    }
 }

history.txt

@@ -1,4 +1,33 @@
-$Id: history.txt,v 1.92 2004/12/01 19:27:27 jrandom Exp $
+$Id: history.txt,v 1.93 2004/12/01 22:20:04 jrandom Exp $
+
+2004-12-03  jrandom
+    * Toss in a small pool of threads (3) to execute the events queued up with
+      the SimpleTimer, as we do currently see the occational event 
+      notification spiking up to a second or so.
+    * Implement a SAM client API in java, useful for event based streaming (or
+      for testing the SAM bridge)
+    * Added support to shut down the SAM bridge on OOM (useful if the SAM 
+      bridge is being run outside of the router).
+    * Include the SAM test code in the sam.jar
+    * Remove an irrelevent warning message from SAM, which was caused by 
+      perfectly normal operation due to a session being closed.
+    * Removed some unnecessary synchronization in the streaming lib's 
+      PacketQueue
+    * More quickly clean up the memory used by the streaming lib by 
+      immediately killing each packet's resend job as soon as it is ACKed (or
+      cancelled), so that there are no longer any valid pointers to the 
+      (potentially 32KB) packet.
+    * Fixed the timestamps dumped to stdout when debugging the PacketHandler.
+    * Drop packets that would expand our inbound window beyond our maximum 
+      buffer size (default 32 messages)
+    * Always read the ACK/NACK data from the verified packets received, even
+      if we are going to drop them
+    * Always adjust the window when there are messages ACKed, though do not
+      change its size except as before.
+    * Streamlined some synchronization in the router's I2CP handling
+    * Streamlined some memory allocation in the SAM bridge
+    * Default the streaming lib to disconnect on inactivity, rather than send
+      an empty message.
 
 2004-12-01  jrandom
     * Fix for a race in the streaming lib as caused by some odd SAM activity

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

@@ -15,9 +15,9 @@ import net.i2p.CoreVersion;
  *
  */
 public class RouterVersion {
-    public final static String ID = "$Revision: 1.97 $ $Date: 2004/12/01 19:27:27 $";
+    public final static String ID = "$Revision: 1.98 $ $Date: 2004/12/01 22:20:03 $";
     public final static String VERSION = "0.4.2.2";
-    public final static long BUILD = 1;
+    public final static long BUILD = 2;
     public static void main(String args[]) {
         System.out.println("I2P Router version: " + VERSION);
         System.out.println("Router ID: " + RouterVersion.ID);

router/java/src/net/i2p/router/client/ClientWriterRunner.java

@@ -25,8 +25,6 @@ class ClientWriterRunner implements Runnable {
     
     private static final long MAX_WAIT = 5*1000;
     
-    /** notify this lock when there are messages to write */
-    private Object _activityLock = new Object();
     /** lock on this when updating the class level data structs */
     private Object _dataLock = new Object();
     
@@ -47,9 +45,7 @@ class ClientWriterRunner implements Runnable {
         synchronized (_dataLock) {
             _messagesToWrite.add(msg);
             _messagesToWriteTimes.add(new Long(_context.clock().now()));
-        }
-        synchronized (_activityLock) {
-            _activityLock.notifyAll();
+            _dataLock.notifyAll();
         }
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("["+_id+"] addMessage completed for " + msg.getClass().getName());
@@ -60,8 +56,8 @@ class ClientWriterRunner implements Runnable {
      *
      */
     public void stopWriting() {
-        synchronized (_activityLock) {
-            _activityLock.notifyAll();
+        synchronized (_dataLock) {
+            _dataLock.notifyAll();
         }
     }
     public void run() {
@@ -70,6 +66,9 @@ class ClientWriterRunner implements Runnable {
             List messageTimes = null;
             
             synchronized (_dataLock) {
+                if (_messagesToWrite.size() <= 0) 
+                    try { _dataLock.wait(); } catch (InterruptedException ie) {}
+                
                 if (_messagesToWrite.size() > 0) {
                     messages = new ArrayList(_messagesToWrite.size());
                     messageTimes = new ArrayList(_messagesToWriteTimes.size());
@@ -80,16 +79,7 @@ class ClientWriterRunner implements Runnable {
                 } 
             }
             
-            if (messages == null) {
-                try {
-                    synchronized (_activityLock) {
-                        _activityLock.wait();
-                    }
-                } catch (InterruptedException ie) {
-                    if (_log.shouldLog(Log.WARN))
-                        _log.warn("Interrupted while waiting for activity", ie);
-                }
-            } else {
+            if (messages != null) {
                 for (int i = 0; i < messages.size(); i++) {
                     I2CPMessage msg = (I2CPMessage)messages.get(i);
                     Long when = (Long)messageTimes.get(i);

router/java/src/net/i2p/router/transport/tcp/TCPTransport.java

@@ -769,8 +769,8 @@ public class TCPTransport extends TransportImpl {
     /** Make this stuff pretty (only used in the old console) */
     public String renderStatusHTML() { 
         StringBuffer buf = new StringBuffer(1024);
-        buf.append("<b>Connections:</b><ul>\n");
         synchronized (_connectionLock) {
+            buf.append("<b>Connections (").append(_connectionsByIdent.size()).append("):</b><ul>\n");
             for (Iterator iter = _connectionsByIdent.values().iterator(); iter.hasNext(); ) {
                 TCPConnection con = (TCPConnection)iter.next();
                 buf.append("<li>");