echelon/i2p.i2p
1
0
Fork 0
forked from I2P_Developers/i2p.i2p
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Allow multiple simultaneous ACCEPT sockets.

Browse Source 
Add support for parallel accepts in sink client
This commit is contained in:
zzz
2015-11-27 19:39:32 +00:00
parent bafec18093
commit cb979fb685
2 changed files with 59 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
apps/sam/java/src/net/i2p/sam/SAMv3StreamSession.java
View File

@ -14,8 +14,14 @@ import java.net.ConnectException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.NoRouteToHostException; import java.net.NoRouteToHostException;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.security.GeneralSecurityException; import java.security.GeneralSecurityException;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
import javax.net.ssl.SSLException; import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSocket; import javax.net.ssl.SSLSocket;
@ -30,11 +36,6 @@ import net.i2p.data.Destination;
import net.i2p.util.I2PAppThread; import net.i2p.util.I2PAppThread;
import net.i2p.util.I2PSSLSocketFactory; import net.i2p.util.I2PSSLSocketFactory;
import net.i2p.util.Log; import net.i2p.util.Log;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
/** /**
* SAMv3 STREAM session class. * SAMv3 STREAM session class.
@ -48,7 +49,11 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
private static final int BUFFER_SIZE = 1024 ; private static final int BUFFER_SIZE = 1024 ;
private final Object socketServerLock = new Object(); private final Object socketServerLock = new Object();
/** this is ONLY set for FORWARD, not for ACCEPT */
private I2PServerSocket socketServer; private I2PServerSocket socketServer;
/** this is the count of active ACCEPT sockets */
private final AtomicInteger _acceptors = new AtomicInteger();
private static I2PSSLSocketFactory _sslSocketFactory; private static I2PSSLSocketFactory _sslSocketFactory;
private final String nick ; private final String nick ;
@ -154,6 +159,8 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
/** /**
* Accept a single incoming STREAM on the socket stolen from the handler. * Accept a single incoming STREAM on the socket stolen from the handler.
* As of version 3.2 (0.9.24), multiple simultaneous accepts are allowed.
* Accepts and forwarding may not be done at the same time.
* *
* @param handler The handler that communicates with the requesting client * @param handler The handler that communicates with the requesting client
* @param verbose If true, SAM will send the Base64-encoded peer Destination of an * @param verbose If true, SAM will send the Base64-encoded peer Destination of an
@ -170,23 +177,22 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
public void accept(SAMv3Handler handler, boolean verbose) public void accept(SAMv3Handler handler, boolean verbose)
throws I2PException, InterruptedIOException, IOException, SAMException { throws I2PException, InterruptedIOException, IOException, SAMException {
synchronized( this.socketServerLock ) synchronized(this.socketServerLock) {
{ if (this.socketServer != null) {
if (this.socketServer!=null) { if (_log.shouldWarn())
if (_log.shouldLog(Log.DEBUG)) _log.warn("a forwarding server is already defined for this destination");
_log.debug("a socket server is already defined for this destination"); throw new SAMException("a forwarding server is already defined for this destination");
throw new SAMException("a socket server is already defined for this destination"); }
} }
this.socketServer = this.socketMgr.getServerSocket();
} I2PSocket i2ps;
_acceptors.incrementAndGet();
I2PSocket i2ps = this.socketServer.accept(); try {
i2ps = socketMgr.getServerSocket().accept();
} finally {
_acceptors.decrementAndGet();
}
synchronized( this.socketServerLock )
{
this.socketServer = null ;
}
SAMv3Handler.SessionRecord rec = SAMv3Handler.sSessionsHash.get(nick); SAMv3Handler.SessionRecord rec = SAMv3Handler.sSessionsHash.get(nick);
if ( rec==null || i2ps==null ) throw new InterruptedIOException() ; if ( rec==null || i2ps==null ) throw new InterruptedIOException() ;
@ -212,7 +218,8 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
/** /**
* Forward sockets from I2P to the host/port provided * Forward sockets from I2P to the host/port provided.
* Accepts and forwarding may not be done at the same time.
*/ */
public void startForwardingIncoming(Properties props, boolean sendPorts) throws SAMException, InterruptedIOException public void startForwardingIncoming(Properties props, boolean sendPorts) throws SAMException, InterruptedIOException
{ {
@ -236,14 +243,17 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
_log.debug("no host specified. Taken from the client socket : " + host +':'+port); _log.debug("no host specified. Taken from the client socket : " + host +':'+port);
} }
boolean isSSL = Boolean.parseBoolean(props.getProperty("SSL")); boolean isSSL = Boolean.parseBoolean(props.getProperty("SSL"));
if (_acceptors.get() > 0) {
synchronized( this.socketServerLock ) if (_log.shouldWarn())
{ _log.warn("an accepting server is already defined for this destination");
if (this.socketServer!=null) { throw new SAMException("an accepting server is already defined for this destination");
if (_log.shouldLog(Log.DEBUG)) }
_log.debug("a socket server is already defined for this destination"); synchronized(this.socketServerLock) {
throw new SAMException("a socket server is already defined for this destination"); if (this.socketServer!=null) {
} if (_log.shouldWarn())
_log.warn("a forwarding server is already defined for this destination");
throw new SAMException("a forwarding server is already defined for this destination");
}
this.socketServer = this.socketMgr.getServerSocket(); this.socketServer = this.socketMgr.getServerSocket();
} }
@ -427,7 +437,7 @@ class SAMv3StreamSession extends SAMStreamSession implements SAMv3Handler.Sessi
} }
} }
public I2PServerSocket getSocketServer() private I2PServerSocket getSocketServer()
{ {
synchronized ( this.socketServerLock ) { synchronized ( this.socketServerLock ) {
return this.socketServer ; return this.socketServer ;

30
apps/sam/java/src/net/i2p/sam/client/SAMStreamSink.java
View File

@ -168,18 +168,22 @@ public class SAMStreamSink {
t.start(); t.start();
} }
if (_isV3 && mode == STREAM) { if (_isV3 && mode == STREAM) {
Socket sock2 = connect(isSSL); // test multiple acceptors, only works in 3.2
out = sock2.getOutputStream(); int acceptors = isV32 ? 4 : 1;
eventHandler = new SinkEventHandler2(_context, sock2.getInputStream(), out); for (int i = 0; i < acceptors; i++) {
_reader2 = new SAMReader(_context, sock2.getInputStream(), eventHandler); Socket sock2 = connect(isSSL);
_reader2.startReading(); out = sock2.getOutputStream();
if (_log.shouldLog(Log.DEBUG)) eventHandler = new SinkEventHandler2(_context, sock2.getInputStream(), out);
_log.debug("Reader2 created"); _reader2 = new SAMReader(_context, sock2.getInputStream(), eventHandler);
String ok = handshake(out, version, false, eventHandler, mode, user, password, ""); _reader2.startReading();
if (ok == null) if (_log.shouldLog(Log.DEBUG))
throw new IOException("2nd handshake failed"); _log.debug("Reader " + (2 + i) + " created");
if (_log.shouldLog(Log.DEBUG)) String ok = handshake(out, version, false, eventHandler, mode, user, password, "");
_log.debug("Handshake2 complete."); if (ok == null)
throw new IOException("handshake " + (2 + i) + " failed");
if (_log.shouldLog(Log.DEBUG))
_log.debug("Handshake " + (2 + i) + " complete.");
}
} else if (_isV3 && (mode == DG || mode == RAW || mode == RAWHDR)) { } else if (_isV3 && (mode == DG || mode == RAW || mode == RAWHDR)) {
// set up a listening DatagramSocket // set up a listening DatagramSocket
(new DGRcvr(mode)).start(); (new DGRcvr(mode)).start();
@ -622,6 +626,8 @@ public class SAMStreamSink {
sinkDir.mkdirs(); sinkDir.mkdirs();
File out = File.createTempFile("sink", ".dat", sinkDir); File out = File.createTempFile("sink", ".dat", sinkDir);
if (_log.shouldWarn())
_log.warn("outputting to " + out);
_out = new FileOutputStream(out); _out = new FileOutputStream(out);
_started = _context.clock().now(); _started = _context.clock().now();
} }