*) further improvements in shutdown behaviour

git-svn-id: https://svn.berlios.de/svnroot/repos/yacy/trunk@392 6c8d7289-2bf4-0310-a012-ef5d649a1542
pull/1/head
theli 20 years ago
parent 96386d6fc6
commit 55d10b864c

@ -46,8 +46,10 @@ import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
import de.anomic.server.serverCore;
import de.anomic.server.serverSemaphore; import de.anomic.server.serverSemaphore;
import de.anomic.server.logging.serverLog; import de.anomic.server.logging.serverLog;
import de.anomic.server.serverCore.Session;
import org.apache.commons.pool.impl.GenericObjectPool; import org.apache.commons.pool.impl.GenericObjectPool;
@ -310,32 +312,54 @@ final class CrawlerPool extends GenericObjectPool {
} }
public synchronized void close() throws Exception { public synchronized void close() throws Exception {
/*
* shutdown all still running session threads ...
*/
// interrupting all still running or pooled threads ...
this.theThreadGroup.interrupt();
/* waiting for all threads to finish */
int threadCount = this.theThreadGroup.activeCount();
Thread[] threadList = new Thread[threadCount];
threadCount = this.theThreadGroup.enumerate(threadList);
try { try {
/*
* shutdown all still running session threads ...
*/
this.isClosed = true;
/* waiting for all threads to finish */
int threadCount = this.theThreadGroup.activeCount();
Thread[] threadList = new Thread[threadCount];
threadCount = this.theThreadGroup.enumerate(threadList);
// signaling shutdown to all still running or pooled threads ...
serverLog.logInfo("CRAWLER","Signaling shutdown to " + threadCount + " remaining crawler threads ...");
for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) { for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) {
((plasmaCrawlWorker)threadList[currentThreadIdx]).setStopped(true); ((plasmaCrawlWorker)threadList[currentThreadIdx]).setStopped(true);
}
// giving the crawlers some time to finish shutdown
try { Thread.sleep(500); } catch(Exception e) {}
// sending interrupted signal to all remaining threads
serverLog.logInfo("CRAWLER","Sending interruption signal to " + this.theThreadGroup.activeCount() + " remaining crawler threads ...");
this.theThreadGroup.interrupt();
// aborting all crawlers by closing all still open httpc sockets
serverLog.logInfo("CRAWLER","Trying to abort " + this.theThreadGroup.activeCount() + " remaining crawler threads ...");
for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) {
Thread currentThread = threadList[currentThreadIdx];
if (currentThread.isAlive()) {
serverLog.logInfo("CRAWLER","Trying to shutdown crawler thread '" + currentThread.getName() + "' [" + currentThreadIdx + "].");
((plasmaCrawlWorker)currentThread).close();
}
} }
serverLog.logInfo("CRAWLER","Waiting for " + this.theThreadGroup.activeCount() + " remaining crawler threads to finish shutdown ...");
for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) { for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) {
// we need to use a timeout here because of missing interruptable session threads ... Thread currentThread = threadList[currentThreadIdx];
if (threadList[currentThreadIdx].isAlive()) threadList[currentThreadIdx].join(500); if (currentThread.isAlive()) {
serverLog.logInfo("CRAWLER","Waiting for crawler thread '" + currentThread.getName() + "' [" + currentThreadIdx + "] to finish shutdown.");
try { currentThread.join(500); } catch (InterruptedException ex) {}
}
} }
serverLog.logWarning("CRAWLER","Shutdown of remaining crawler threads finish.");
} }
catch (InterruptedException e) { catch (Exception e) {
System.err.println("Interruption while trying to shutdown all crawler threads."); serverLog.logWarning("CRAWLER","Unexpected error while trying to shutdown all remaining crawler threads.",e);
} }
this.isClosed = true;
super.close(); super.close();
} }

@ -54,6 +54,7 @@ import java.util.logging.Logger;
import de.anomic.http.httpHeader; import de.anomic.http.httpHeader;
import de.anomic.http.httpc; import de.anomic.http.httpc;
import de.anomic.http.httpdProxyHandler; import de.anomic.http.httpdProxyHandler;
import de.anomic.server.serverCore;
import de.anomic.server.logging.serverLog; import de.anomic.server.logging.serverLog;
import de.anomic.server.logging.serverMiniLogFormatter; import de.anomic.server.logging.serverMiniLogFormatter;
@ -219,6 +220,18 @@ public final class plasmaCrawlWorker extends Thread {
public boolean isRunning() { public boolean isRunning() {
return this.running; return this.running;
} }
public void close() {
if (this.isAlive()) {
try {
// trying to close all still open httpc-Sockets first
int closedSockets = httpc.closeOpenSockets(this);
if (closedSockets > 0) {
this.log.logInfo(closedSockets + " http-client sockets of thread '" + this.getName() + "' closed.");
}
} catch (Exception e) {}
}
}
public static void load( public static void load(
URL url, URL url,

@ -521,13 +521,13 @@ public final class serverCore extends serverAbstractThread implements serverThre
((Session)threadList[currentThreadIdx]).setStopped(true); ((Session)threadList[currentThreadIdx]).setStopped(true);
} }
// waiting a frew ms for the session objects to continue processing
try { Thread.sleep(500); } catch (InterruptedException ex) {}
// interrupting all still running or pooled threads ... // interrupting all still running or pooled threads ...
serverCore.this.log.logInfo("Sending interruption signal to " + serverCore.this.theSessionThreadGroup.activeCount() + " remaining session threads ..."); serverCore.this.log.logInfo("Sending interruption signal to " + serverCore.this.theSessionThreadGroup.activeCount() + " remaining session threads ...");
serverCore.this.theSessionThreadGroup.interrupt(); serverCore.this.theSessionThreadGroup.interrupt();
// waiting a frew ms for the session objects to continue processing
try { Thread.sleep(500); } catch (InterruptedException ex) {}
// if there are some sessions that are blocking in IO, we simply close the socket // if there are some sessions that are blocking in IO, we simply close the socket
serverCore.this.log.logDebug("Trying to abort " + serverCore.this.theSessionThreadGroup.activeCount() + " remaining session threads ..."); serverCore.this.log.logDebug("Trying to abort " + serverCore.this.theSessionThreadGroup.activeCount() + " remaining session threads ...");
for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) { for ( int currentThreadIdx = 0; currentThreadIdx < threadCount; currentThreadIdx++ ) {
@ -675,10 +675,7 @@ public final class serverCore extends serverAbstractThread implements serverThre
int closedSockets = httpc.closeOpenSockets(this); int closedSockets = httpc.closeOpenSockets(this);
if (closedSockets > 0) { if (closedSockets > 0) {
serverCore.this.log.logInfo(closedSockets + " http-client sockets of thread '" + this.getName() + "' closed."); serverCore.this.log.logInfo(closedSockets + " http-client sockets of thread '" + this.getName() + "' closed.");
} }
// waiting some time
this.join(300);
// closing the socket to the client // closing the socket to the client
if ((this.controlSocket != null)&&(this.controlSocket.isConnected())) { if ((this.controlSocket != null)&&(this.controlSocket.isConnected())) {

Loading…
Cancel
Save