* i2ptunnel, I2CP, EepGet: Buffer socket input streams (ticket #666)

* I2PSessionImpl: One more volatile (ticket #659)

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

@@ -3,6 +3,7 @@
  */
 package net.i2p.i2ptunnel;
 
+import java.io.BufferedInputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
@@ -219,6 +220,8 @@ public class I2PTunnelHTTPServer extends I2PTunnelServer {
         // shadows _log in super()
         private final Log _log;
 
+        private static final int BUF_SIZE = 16*1024;
+
         public CompressedRequestor(Socket webserver, I2PSocket browser, String headers, I2PAppContext ctx, Log log) {
             _webserver = webserver;
             _browser = browser;
@@ -259,7 +262,7 @@ public class I2PTunnelHTTPServer extends I2PTunnelServer {
                 //     at java.lang.Thread.run(Thread.java:619)
                 //     at net.i2p.util.I2PThread.run(I2PThread.java:71)
                 try {
-                    serverin = _webserver.getInputStream();
+                    serverin = new BufferedInputStream(_webserver.getInputStream(), BUF_SIZE);
                 } catch (NullPointerException npe) {
                     throw new IOException("getInputStream NPE");
                 }

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

@@ -3,6 +3,7 @@
  */
 package net.i2p.i2ptunnel;
 
+import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InterruptedIOException;
@@ -17,6 +18,7 @@ import net.i2p.data.ByteArray;
 import net.i2p.util.ByteCache;
 import net.i2p.util.Clock;
 import net.i2p.util.I2PAppThread;
+import net.i2p.util.InternalSocket;
 import net.i2p.util.Log;
 
 public class I2PTunnelRunner extends I2PAppThread implements I2PSocket.SocketErrorListener {
@@ -167,6 +169,8 @@ public class I2PTunnelRunner extends I2PAppThread implements I2PSocket.SocketErr
                 _log.debug("Initial data " + (initialI2PData != null ? initialI2PData.length : 0) 
                            + " written to I2P, " + (initialSocketData != null ? initialSocketData.length : 0)
                            + " written to the socket, starting forwarders");
+            if (!(s instanceof InternalSocket))
+                in = new BufferedInputStream(in, 2*NETWORK_BUFFER_SIZE);
             Thread t1 = new StreamForwarder(in, i2pout, true);
             Thread t2 = new StreamForwarder(i2pin, out, false);
             synchronized (finishLock) {

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

@@ -9,6 +9,7 @@ package net.i2p.client;
  *
  */
 
+import java.io.BufferedInputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
@@ -116,7 +117,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
     protected volatile boolean _closing;
 
     /** have we received the current date from the router yet? */
-    private boolean _dateReceived;
+    private volatile boolean _dateReceived;
     /** lock that we wait upon, that the SetDateMessageHandler notifies */
     private final Object _dateReceivedLock = new Object();
 
@@ -155,6 +156,8 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
 
     public static final int LISTEN_PORT = 7654;
 
+    private static final int BUF_SIZE = 32*1024;
+    
     /** for extension */
     protected I2PSessionImpl(I2PAppContext context, Properties options) {
         _context = context;
@@ -353,7 +356,7 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
                 _out.write(I2PClient.PROTOCOL_BYTE);
                 _out.flush();
                 _writer = new ClientWriterRunner(_out, this);
-                InputStream in = _socket.getInputStream();
+                InputStream in = new BufferedInputStream(_socket.getInputStream(), BUF_SIZE);
                 _reader = new I2CPMessageReader(in, this);
             }
             Thread notifier = new I2PAppThread(_availabilityNotifier, "ClientNotifier " + getPrefix(), true);

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

@@ -5,6 +5,7 @@ package net.i2p.client;
  * with no warranty of any kind, either expressed or implied.  
  */
 
+import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.Socket;
@@ -28,6 +29,8 @@ import net.i2p.internal.QueuedI2CPMessageReader;
  */
 class I2PSimpleSession extends I2PSessionImpl2 {
 
+    private static final int BUF_SIZE = 1024;
+
     /**
      * Create a new session for doing naming and bandwidth queries only. Do not create a destination.
      *
@@ -68,7 +71,7 @@ class I2PSimpleSession extends I2PSessionImpl2 {
                 _out.write(I2PClient.PROTOCOL_BYTE);
                 _out.flush();
                 _writer = new ClientWriterRunner(_out, this);
-                InputStream in = _socket.getInputStream();
+                InputStream in = new BufferedInputStream(_socket.getInputStream(), BUF_SIZE);
                 _reader = new I2CPMessageReader(in, this);
             }
             // we do not receive payload messages, so we do not need an AvailabilityNotifier

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

@@ -1,5 +1,6 @@
 package net.i2p.util;
 
+import java.io.BufferedInputStream;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
@@ -1057,6 +1058,8 @@ public class EepGet {
             //}
         }
         _proxyIn = _proxy.getInputStream();
+        if (!(_proxy instanceof InternalSocket))
+              _proxyIn = new BufferedInputStream(_proxyIn);
         _proxyOut = _proxy.getOutputStream();
         
         if (timeout != null)

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

@@ -27,7 +27,7 @@ public class InternalServerSocket extends ServerSocket {
     private static final ConcurrentHashMap<Integer, InternalServerSocket> _sockets = new ConcurrentHashMap(4);
     private final BlockingQueue<InternalSocket> _acceptQueue;
     private final Integer _port;
-    private boolean _running;
+    private volatile boolean _running;
     //private static Log _log = I2PAppContext.getGlobalContext().logManager().getLog(InternalServerSocket.class);
 
     /**

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

@@ -37,6 +37,7 @@ package net.i2p.util;
  * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+import java.io.BufferedInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
@@ -676,6 +677,8 @@ public class SSLEepGet extends EepGet {
             throw sslhe;
         }
 
+        _proxyIn = new BufferedInputStream(_proxyIn);
+        
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("Request flushed");
     }

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

@@ -8,6 +8,7 @@ package net.i2p.router.client;
  *
  */
 
+import java.io.BufferedInputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -96,6 +97,8 @@ class ClientConnectionRunner {
     // e.g. on local access
     private static final int MAX_MESSAGE_ID = 0x4000000;
 
+    private static final int BUF_SIZE = 32*1024;
+
     /** @since 0.9.2 */
     private static final String PROP_TAGS = "crypto.tagsToSend";
     private static final String PROP_THRESH = "crypto.lowTagThreshold";
@@ -124,7 +127,8 @@ class ClientConnectionRunner {
      */
     public void startRunning() {
         try {
-            _reader = new I2CPMessageReader(_socket.getInputStream(), new ClientMessageEventListener(_context, this, true));
+            _reader = new I2CPMessageReader(new BufferedInputStream(_socket.getInputStream(), BUF_SIZE),
+                                            new ClientMessageEventListener(_context, this, true));
             _writer = new ClientWriterRunner(_context, this);
             I2PThread t = new I2PThread(_writer);
             t.setName("I2CP Writer " + ++__id);