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

I2PTunnel:

Browse Source 
- Fix bug that left server acceptor thread running after close
- Add destroy() methods to release all resources when closing a tunnel for good,
  particularly the streaming timer threads
- Use COWAL to prevent concurrency problems
- Javadocs
Streaming:
- Don't return null from accept() any more; actually throw
  ConnectException as the javadocs have always specified
- Throw ConnectException from accept() if interrupted; previously caught and ignored
- Throw exceptions from ConnectionHandler.accept(), not higher up
- Close ServerSocket when ConnectionManager is shut down
- Synchronize setActive(), clear queue when starting to accept,
  better handling of calls that don't change state
- Javadocs
ConfigClientsHelper: Call isPluginRunning() less often
PluginStarter: Simplify detection of active threads

Above changes mostly in support of zzzot plugin implementing ClientApp
and being able to shut down completely so there are no threads
in its thread group, so /configclients will all show status as stopped.
Previously, the I2PTunnelServer acceptor thread and
one or more streaming timer threads would remain.
This commit is contained in:
zzz
2014-11-13 20:12:55 +00:00
parent 0773a30578
commit 2f2aa7f5a8
16 changed files with 283 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnel.java
View File

@ -52,6 +52,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicLong;
@ -107,7 +108,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
private static final String nocli_args[] = { "-nocli", "-die"};
private final List<I2PTunnelTask> tasks = new ArrayList<I2PTunnelTask>();
private final List<I2PTunnelTask> tasks = new CopyOnWriteArrayList<I2PTunnelTask>();
private int next_task_id = 1;
private final Set<ConnectionEventListener> listeners = new CopyOnWriteArraySet<ConnectionEventListener>();
@ -123,6 +124,9 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
new LongOpt("wait", LongOpt.NO_ARGUMENT, null, 'w')
};
/** @since 0.9.17 */
private enum CloseMode { NORMAL, FORCED, DESTROY }
public static void main(String[] args) {
try {
new I2PTunnel(args);
@ -455,7 +459,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
" auth <username> <password>\n" +
" client <port> <pubkey>[,<pubkey,...]|file:<pubkeyfile> [<sharedClient>]\n" +
" clientoptions [-acx] [key=value ]*\n" +
" close [forced] <jobnumber>|all\n" +
" close [forced|destroy] <jobnumber>|all\n" +
" config [-s] <i2phost> <i2pport>\n" +
" connectclient <port> [<sharedClient>] [<proxy>]\n" +
" genkeys <privkeyfile> [<pubkeyfile>]\n" +
@ -1531,21 +1535,24 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
*
* Sets the event "closeResult" = "ok" after the closing is complete
*
* @param args {jobNumber}, {"forced", jobNumber}, {"forced", "all"}
* @param args {jobNumber}, {"forced", jobNumber}, {"forced", "all"}, {"destroy", jobNumber}, {"destroy", "all"}
* @param l logger to receive events and output
*/
public void runClose(String args[], Logging l) {
if (args.length == 0 || args.length > 2) {
l.log("close [forced] <jobnumber>|all\n" +
l.log("close [forced|destroy] <jobnumber>|all\n" +
" stop running tasks. either only one or all.\n"
+ " use 'forced' to also stop tasks with active connections.\n"
+ " use the 'list' command to show the job numbers");
notifyEvent("closeResult", "error");
} else {
int argindex = 0; // parse optional 'forced' keyword
boolean forced = false;
CloseMode mode = CloseMode.NORMAL;
if (args[argindex].equalsIgnoreCase("forced")) {
forced = true;
mode = CloseMode.FORCED;
argindex++;
} else if (args[argindex].equalsIgnoreCase("destroy")) {
mode = CloseMode.DESTROY;
argindex++;
}
if (args[argindex].equalsIgnoreCase("all")) {
@ -1555,7 +1562,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
_log.info(getPrefix() + " runClose(all) no tasks");
}
for (I2PTunnelTask t : tasks) {
if (!closetask(t, forced, l)) {
if (!closetask(t, mode, l)) {
notifyEvent("closeResult", "error");
error = true;
} else if (!error) { // If there's an error, don't hide it
@ -1564,7 +1571,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
}
} else {
try {
if (!closetask(Integer.parseInt(args[argindex]), forced, l)) {
if (!closetask(Integer.parseInt(args[argindex]), mode, l)) {
notifyEvent("closeResult", "error");
} else {
notifyEvent("closeResult", "ok");
@ -1669,7 +1676,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
* closure)
*
*/
private boolean closetask(int num, boolean forced, Logging l) {
private boolean closetask(int num, CloseMode mode, Logging l) {
boolean closed = false;
_log.debug(getPrefix() + "closetask(): looking for task " + num);
@ -1678,7 +1685,7 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
if (_log.shouldLog(Log.DEBUG))
_log.debug(getPrefix() + "closetask(): parsing task " + id + " (" + t.toString() + ")");
if (id == num) {
closed = closetask(t, forced, l);
closed = closetask(t, mode, l);
break;
} else if (id > num) {
break;
@ -1692,17 +1699,23 @@ public class I2PTunnel extends EventDispatcherImpl implements Logging {
* (optionally forcing closure)
*
*/
private boolean closetask(I2PTunnelTask t, boolean forced, Logging l) {
private boolean closetask(I2PTunnelTask t, CloseMode mode, Logging l) {
if (_log.shouldLog(Log.INFO))
_log.info("Closing task " + t.getId() + (forced ? " forced..." : "..."));
_log.info("Closing task " + t.getId() + " mode: " + mode);
//l.log("Closing task " + t.getId() + (forced ? " forced..." : "..."));
if (t.close(forced)) {
boolean success;
if (mode == CloseMode.NORMAL)
success = t.close(false);
else if (mode == CloseMode.FORCED)
success = t.close(true);
else // DESTROY
success = t.destroy();
if (success) {
if (_log.shouldLog(Log.INFO))
_log.info("Task " + t.getId() + " closed.");
//l.log("Task " + t.getId() + " closed.");
return true;
}
return false;
return success;
}
/**

16
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelClientBase.java
View File

@ -796,6 +796,22 @@ public abstract class I2PTunnelClientBase extends I2PTunnelTask implements Runna
return true;
}
/**
* Note that the tunnel cannot be reopened after this by calling startRunning(),
* as it will destroy the underlying socket manager.
* This releases all resources if not a shared client.
* For shared client, the router will kill all the remaining streaming timers at shutdown.
*
* @since 0.9.17
*/
@Override
public synchronized boolean destroy() {
close(true);
if (_ownDest)
sockMgr.destroySocketManager();
return true;
}
public static void closeSocket(Socket s) {
try {
s.close();

53
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelServer.java
View File

@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadFactory;
import net.i2p.I2PException;
import net.i2p.client.I2PSession;
import net.i2p.client.I2PSessionException;
import net.i2p.client.streaming.I2PServerSocket;
import net.i2p.client.streaming.I2PSocket;
@ -45,7 +46,7 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
protected final Log _log;
protected final I2PSocketManager sockMgr;
protected I2PServerSocket i2pss;
protected volatile I2PServerSocket i2pss;
private final Object lock = new Object();
protected final Object slock = new Object();
@ -213,7 +214,7 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
try {
I2PSocketManager rv = I2PSocketManagerFactory.createDisconnectedManager(privData, getTunnel().host,
portNum, props);
rv.setName("Server");
rv.setName("I2PTunnel Server");
getTunnel().addSession(rv.getSession());
return rv;
} catch (I2PSessionException ise) {
@ -317,6 +318,13 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
return readTimeout;
}
/**
* Note that the tunnel can be reopened after this by calling startRunning().
* This does not release all resources. In particular, the I2PSocketManager remains
* and it may have timer threads that continue running.
*
* To release all resources permanently, call destroy().
*/
public synchronized boolean close(boolean forced) {
if (!open) return true;
if (task != null) {
@ -331,16 +339,20 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
return false;
}
l.log("Stopping tunnels for server at " + this.remoteHost + ':' + this.remotePort);
open = false;
try {
if (i2pss != null) i2pss.close();
getTunnel().removeSession(sockMgr.getSession());
sockMgr.getSession().destroySession();
if (i2pss != null) {
i2pss.close();
i2pss = null;
}
I2PSession session = sockMgr.getSession();
getTunnel().removeSession(session);
session.destroySession();
} catch (I2PException ex) {
_log.error("Error destroying the session", ex);
//System.exit(1);
}
//l.log("Server shut down.");
open = false;
if (_usePool && _executor != null) {
_executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
_executor.shutdownNow();
@ -349,6 +361,20 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
}
}
/**
* Note that the tunnel cannot be reopened after this by calling startRunning(),
* as it will destroy the underlying socket manager.
* This releases all resources.
*
* @since 0.9.17
*/
@Override
public synchronized boolean destroy() {
close(true);
sockMgr.destroySocketManager();
return true;
}
/**
* Update the I2PSocketManager.
* And since 0.9.15, the target host and port.
@ -434,7 +460,7 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
* hands each I2P socket to the executor or runs it in-line.
*/
public void run() {
I2PServerSocket i2pS_S = sockMgr.getServerSocket();
i2pss = sockMgr.getServerSocket();
if (_log.shouldLog(Log.WARN)) {
if (_usePool)
_log.warn("Starting executor with " + getHandlerCount() + " threads max");
@ -446,7 +472,10 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
}
while (open) {
try {
final I2PSocket i2ps = i2pS_S.accept();
I2PServerSocket ci2pss = i2pss;
if (ci2pss == null)
throw new I2PException("I2PServerSocket closed");
final I2PSocket i2ps = ci2pss.accept();
if (i2ps == null) throw new I2PException("I2PServerSocket closed");
if (_usePool) {
try {
@ -473,10 +502,8 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
} catch (ConnectException ce) {
if (_log.shouldLog(Log.ERROR))
_log.error("Error accepting", ce);
// not killing the server..
try {
Thread.sleep(500);
} catch (InterruptedException ie) {}
open = false;
break;
} catch(SocketTimeoutException ste) {
// ignored, we never set the timeout
} catch (Exception e) {
@ -489,7 +516,7 @@ public class I2PTunnelServer extends I2PTunnelTask implements Runnable {
} catch (InterruptedException ie) {}
}
}
if (_executor != null)
if (_executor != null && !_executor.isTerminating() && !_executor.isShutdown())
_executor.shutdownNow();
}

41
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/I2PTunnelTask.java
View File

@ -8,9 +8,15 @@ import net.i2p.util.EventDispatcher;
import net.i2p.util.EventDispatcherImpl;
/**
* Either a Server or a Client.
* Either a Server or a Client.
*
* Use caution if extending externally.
* This class should be maintained as a stable API,
* but ask to be sure.
*
* Note that there is no startRunning() method,
* however all extending classes implement one.
*/
public abstract class I2PTunnelTask extends EventDispatcherImpl {
private int id;
@ -56,12 +62,37 @@ public abstract class I2PTunnelTask extends EventDispatcherImpl {
tunnel.routerDisconnected();
}
/**
* Note that the tunnel can be reopened after this by calling startRunning().
* This may not release all resources. In particular, the I2PSocketManager remains
* and it may have timer threads that continue running.
*
* To release all resources permanently, call destroy().
*
* @return success
*/
public abstract boolean close(boolean forced);
/**
* Note that the tunnel cannot be reopened after this by calling startRunning(),
* as it may destroy the underlying socket manager, depending on implementation.
* This should release all resources.
*
* The implementation here simply calls close(true).
* Extending classes should override to release all resources.
*
* @return success
* @since 0.9.17
*/
public boolean destroy() {
return close(true);
}
/**
* Notify the task that I2PTunnel's options have been updated.
* Extending classes should override and call I2PTunnel.getClientOptions(),
* then update the I2PSocketManager.
* Does nothing here.
*
* @since 0.9.1
*/
@ -80,9 +111,15 @@ public abstract class I2PTunnelTask extends EventDispatcherImpl {
getTunnel().removeSession(session);
}
/**
* Does nothing here. Extending classes may override.
*/
public void errorOccurred(I2PSession session, String message, Throwable error) {
}
/**
* Does nothing here. Extending classes may override.
*/
public void reportAbuse(I2PSession session, int severity) {
}

20
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/TunnelController.java
View File

@ -548,6 +548,11 @@ public class TunnelController implements Logging {
}
}
/**
* May be restarted with restartTunnel() or startTunnel() later.
* This may not release all resources. In particular, the I2PSocketManager remains
* and it may have timer threads that continue running.
*/
public void stopTunnel() {
// I2PTunnel removes the session in close(),
// so save the sessions to pass to release() and TCG
@ -556,6 +561,21 @@ public class TunnelController implements Logging {
release(sessions);
_running = false;
}
/**
* May NOT be restarted with restartTunnel() or startTunnel() later.
* This should release all resources.
*
* @since 0.9.17
*/
public void destroyTunnel() {
// I2PTunnel removes the session in close(),
// so save the sessions to pass to release() and TCG
Collection<I2PSession> sessions = getAllSessions();
_tunnel.runClose(new String[] { "destroy", "all" }, this);
release(sessions);
_running = false;
}
public void restartTunnel() {
stopTunnel();

19
apps/i2ptunnel/java/src/net/i2p/i2ptunnel/TunnelControllerGroup.java
View File

@ -268,7 +268,7 @@ public class TunnelControllerGroup implements ClientApp {
*
*/
public synchronized void unloadControllers() {
stopAllControllers();
destroyAllControllers();
_controllers.clear();
if (_log.shouldLog(Log.INFO))
_log.info("All controllers stopped and unloaded");
@ -296,7 +296,7 @@ public class TunnelControllerGroup implements ClientApp {
}
/**
* Stop all tunnels
* Stop all tunnels. May be restarted.
*
* @return list of messages the tunnels generate when stopped
*/
@ -312,6 +312,21 @@ public class TunnelControllerGroup implements ClientApp {
return msgs;
}
/**
* Stop all tunnels. They may not be restarted, you must reload.
* Caller must synch. Caller must clear controller list.
*
* @since 0.9.17
*/
private void destroyAllControllers() {
for (int i = 0; i < _controllers.size(); i++) {
TunnelController controller = _controllers.get(i);
controller.destroyTunnel();
}
if (_log.shouldLog(Log.INFO))
_log.info(_controllers.size() + " controllers stopped");
}
/**
* Start all tunnels
*