- added blockingThreads which are threads that are not driven by pause times but by BlockingQueue lookup git-svn-id: https://svn.berlios.de/svnroot/repos/yacy/trunk@4606 6c8d7289-2bf4-0310-a012-ef5d649a1542pull/1/head
orbiter
17 years ago
parent
968c775025
commit
bca87f1e38
@ -0,0 +1,91 @@
|
||||
// serverAbstractBlockingThread.java
|
||||
// (C) 2008 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
|
||||
// first published 27.03.2008 on http://yacy.net
|
||||
//
|
||||
// $LastChangedDate: 2006-04-02 22:40:07 +0200 (So, 02 Apr 2006) $
|
||||
// $LastChangedRevision: 1986 $
|
||||
// $LastChangedBy: orbiter $
|
||||
//
|
||||
// LICENSE
|
||||
//
|
||||
// This program is free software; you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation; either version 2 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with this program; if not, write to the Free Software
|
||||
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
||||
|
||||
package de.anomic.server;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import de.anomic.server.logging.serverLog;
|
||||
|
||||
public abstract class serverAbstractBlockingThread<I, O> extends serverAbstractThread implements serverBlockingThread<I, O> {
|
||||
|
||||
private BlockingQueue<I> input = null;
|
||||
private BlockingQueue<O> output = null;
|
||||
|
||||
public void setInputQueue(BlockingQueue<I> queue) {
|
||||
this.input = queue;
|
||||
}
|
||||
public void setOutputQueue(BlockingQueue<O> queue) {
|
||||
this.output = queue;
|
||||
}
|
||||
public BlockingQueue<I> getInputQueue() {
|
||||
return this.input;
|
||||
}
|
||||
public BlockingQueue<O> getOutputQueue() {
|
||||
return this.output;
|
||||
}
|
||||
|
||||
public void run() {
|
||||
this.open();
|
||||
if (log != null) {
|
||||
logSystem("thread '" + this.getName() + "' deployed, starting loop.");
|
||||
}
|
||||
long timestamp;
|
||||
long memstamp0, memstamp1;
|
||||
long busyCycles = 0;
|
||||
|
||||
while (running) {
|
||||
try {
|
||||
// do job
|
||||
timestamp = System.currentTimeMillis();
|
||||
memstamp0 = serverMemory.used();
|
||||
O out = this.job(this.input.take());
|
||||
if (out != null) this.output.add(out);
|
||||
// do memory and busy/idle-count/time monitoring
|
||||
memstamp1 = serverMemory.used();
|
||||
if (memstamp1 >= memstamp0) {
|
||||
// no GC in between. this is not shure but most probable
|
||||
memuse += memstamp1 - memstamp0;
|
||||
} else {
|
||||
// GC was obviously in between. Add an average as simple heuristic
|
||||
if (busyCycles > 0) memuse += memuse / busyCycles;
|
||||
}
|
||||
busytime += System.currentTimeMillis() - timestamp;
|
||||
} catch (Exception e) {
|
||||
// handle exceptions: thread must not die on any unexpected exceptions
|
||||
// if the exception is too bad it should call terminate()
|
||||
this.jobExceptionHandler(e);
|
||||
} finally {
|
||||
busyCycles++;
|
||||
}
|
||||
}
|
||||
this.close();
|
||||
logSystem("thread '" + this.getName() + "' terminated.");
|
||||
}
|
||||
|
||||
private void logSystem(String text) {
|
||||
if (log == null) serverLog.logConfig("THREAD-CONTROL", text);
|
||||
else log.logConfig(text);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,197 @@
|
||||
// serverAbstractBusyThread.java
|
||||
// (C) 2005 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
|
||||
// first published 14.03.2005 on http://yacy.net
|
||||
//
|
||||
// $LastChangedDate: 2006-04-02 22:40:07 +0200 (So, 02 Apr 2006) $
|
||||
// $LastChangedRevision: 1986 $
|
||||
// $LastChangedBy: orbiter $
|
||||
//
|
||||
// LICENSE
|
||||
//
|
||||
// This program is free software; you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation; either version 2 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with this program; if not, write to the Free Software
|
||||
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
||||
|
||||
package de.anomic.server;
|
||||
|
||||
import de.anomic.server.logging.serverLog;
|
||||
|
||||
public abstract class serverAbstractBusyThread extends serverAbstractThread implements serverBusyThread {
|
||||
|
||||
private long startup = 0, intermission = 0, idlePause = 0, busyPause = 0;
|
||||
private long idletime = 0, memprereq = 0;
|
||||
private long idleCycles = 0, busyCycles = 0, outofmemoryCycles = 0;
|
||||
private boolean intermissionObedient = true;
|
||||
|
||||
protected final void announceMoreSleepTime(long millis) {
|
||||
this.idletime += millis;
|
||||
}
|
||||
|
||||
public final void setStartupSleep(long milliseconds) {
|
||||
// sets a sleep time before execution of the job-loop
|
||||
startup = milliseconds;
|
||||
}
|
||||
|
||||
public final long setIdleSleep(long milliseconds) {
|
||||
// sets a sleep time for pauses between two jobs
|
||||
idlePause = milliseconds;
|
||||
return milliseconds;
|
||||
}
|
||||
|
||||
public final long setBusySleep(long milliseconds) {
|
||||
// sets a sleep time for pauses between two jobs
|
||||
busyPause = milliseconds;
|
||||
return milliseconds;
|
||||
}
|
||||
|
||||
public void setMemPreReqisite(long freeBytes) {
|
||||
// sets minimum required amount of memory for the job execution
|
||||
memprereq = freeBytes;
|
||||
}
|
||||
|
||||
public void setObeyIntermission(boolean obey) {
|
||||
// defines if the thread should obey the intermission command
|
||||
intermissionObedient = obey;
|
||||
}
|
||||
|
||||
public final long getIdleCycles() {
|
||||
// returns the total number of cycles of job execution with idle-result
|
||||
return this.idleCycles;
|
||||
}
|
||||
|
||||
public final long getBusyCycles() {
|
||||
// returns the total number of cycles of job execution with busy-result
|
||||
return this.busyCycles;
|
||||
}
|
||||
|
||||
public long getOutOfMemoryCycles() {
|
||||
// returns the total number of cycles where
|
||||
// a job execution was omitted because of memory shortage
|
||||
return this.outofmemoryCycles;
|
||||
}
|
||||
|
||||
public final long getSleepTime() {
|
||||
// returns the total time that this thread has slept so far
|
||||
return this.idletime;
|
||||
}
|
||||
|
||||
public void intermission(long pause) {
|
||||
if (pause == Long.MAX_VALUE)
|
||||
this.intermission = Long.MAX_VALUE;
|
||||
else
|
||||
this.intermission = System.currentTimeMillis() + pause;
|
||||
}
|
||||
|
||||
|
||||
public void run() {
|
||||
if (startup > 0) {
|
||||
// do a startup-delay
|
||||
logSystem("thread '" + this.getName() + "' deployed, delaying start-up.");
|
||||
ratz(startup);
|
||||
if (!(running)) return;
|
||||
}
|
||||
this.open();
|
||||
if (log != null) {
|
||||
if (startup > 0)
|
||||
logSystem("thread '" + this.getName() + "' delayed, " + ((this.busyPause < 0) ? "starting now job." : "starting now loop."));
|
||||
else
|
||||
logSystem("thread '" + this.getName() + "' deployed, " + ((this.busyPause < 0) ? "starting job." : "starting loop."));
|
||||
}
|
||||
long timestamp;
|
||||
long memstamp0, memstamp1;
|
||||
boolean isBusy;
|
||||
//Runtime rt = Runtime.getRuntime();
|
||||
|
||||
while (running) {
|
||||
if ((this.intermissionObedient) && (this.intermission > 0) && (this.intermission != Long.MAX_VALUE)) {
|
||||
long itime = this.intermission - System.currentTimeMillis();
|
||||
if (itime > 0) {
|
||||
if (itime > this.idlePause) itime = this.idlePause;
|
||||
logSystem("thread '" + this.getName()
|
||||
+ "' breaks for intermission: " + (itime / 1000)
|
||||
+ " seconds");
|
||||
ratz(itime);
|
||||
}
|
||||
this.intermission = 0;
|
||||
}
|
||||
|
||||
if (this.intermission == Long.MAX_VALUE) {
|
||||
// omit Job, paused
|
||||
logSystem("thread '" + this.getName() + "' paused");
|
||||
timestamp = System.currentTimeMillis();
|
||||
ratz(this.idlePause);
|
||||
idletime += System.currentTimeMillis() - timestamp;
|
||||
//} else if ((memnow = serverMemory.available()) > memprereq) try {
|
||||
} else if (serverMemory.request(memprereq, false)) try {
|
||||
// do job
|
||||
timestamp = System.currentTimeMillis();
|
||||
memstamp0 = serverMemory.used();
|
||||
isBusy = this.job();
|
||||
// do memory and busy/idle-count/time monitoring
|
||||
if (isBusy) {
|
||||
memstamp1 = serverMemory.used();
|
||||
if (memstamp1 >= memstamp0) {
|
||||
// no GC in between. this is not shure but most probable
|
||||
memuse += memstamp1 - memstamp0;
|
||||
} else {
|
||||
// GC was obviously in between. Add an average as simple heuristic
|
||||
if (busyCycles > 0) memuse += memuse / busyCycles;
|
||||
}
|
||||
busytime += System.currentTimeMillis() - timestamp;
|
||||
busyCycles++;
|
||||
} else {
|
||||
idleCycles++;
|
||||
}
|
||||
// interrupt loop if this is interrupted or supposed to be a one-time job
|
||||
if ((!running) || (this.isInterrupted())) break;
|
||||
if ((this.idlePause < 0) || (this.busyPause < 0)) break; // for one-time jobs
|
||||
// process scheduled pause
|
||||
timestamp = System.currentTimeMillis();
|
||||
ratz((isBusy) ? this.busyPause : this.idlePause);
|
||||
idletime += System.currentTimeMillis() - timestamp;
|
||||
} catch (Exception e) {
|
||||
// handle exceptions: thread must not die on any unexpected exceptions
|
||||
// if the exception is too bad it should call terminate()
|
||||
this.jobExceptionHandler(e);
|
||||
busyCycles++;
|
||||
} else {
|
||||
log.logWarning("Thread '" + this.getName() + "' runs short memory cycle. Free mem: " +
|
||||
(serverMemory.available() / 1024) + " KB, needed: " + (memprereq / 1024) + " KB");
|
||||
// omit job, not enough memory
|
||||
// process scheduled pause
|
||||
timestamp = System.currentTimeMillis();
|
||||
// do a clean-up
|
||||
this.freemem();
|
||||
// sleep a while
|
||||
ratz(this.idlePause);
|
||||
idletime += System.currentTimeMillis() - timestamp;
|
||||
outofmemoryCycles++;
|
||||
}
|
||||
}
|
||||
this.close();
|
||||
logSystem("thread '" + this.getName() + "' terminated.");
|
||||
}
|
||||
|
||||
private void ratz(long millis) {
|
||||
try {
|
||||
Thread.sleep(millis);
|
||||
} catch (InterruptedException e) {
|
||||
if (this.log != null) this.log.logConfig("thread '" + this.getName() + "' interrupted because of shutdown.");
|
||||
}
|
||||
}
|
||||
|
||||
private void logSystem(String text) {
|
||||
if (log == null) serverLog.logConfig("THREAD-CONTROL", text);
|
||||
else log.logConfig(text);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,70 @@
|
||||
// serverBusyThread.java
|
||||
// (C) 2008 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
|
||||
// first published 27.03.2008 on http://yacy.net
|
||||
//
|
||||
// $LastChangedDate: 2006-04-02 22:40:07 +0200 (So, 02 Apr 2006) $
|
||||
// $LastChangedRevision: 1986 $
|
||||
// $LastChangedBy: orbiter $
|
||||
//
|
||||
// LICENSE
|
||||
//
|
||||
// This program is free software; you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation; either version 2 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with this program; if not, write to the Free Software
|
||||
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
||||
|
||||
package de.anomic.server;
|
||||
|
||||
public interface serverBusyThread extends serverThread {
|
||||
|
||||
public void setStartupSleep(long milliseconds);
|
||||
// sets a sleep time before execution of the job-loop
|
||||
|
||||
public long setIdleSleep(long milliseconds);
|
||||
// sets a sleep time for pauses between two jobs if the job returns false (idle)
|
||||
|
||||
public long setBusySleep(long milliseconds);
|
||||
// sets a sleep time for pauses between two jobs if the job returns true (busy)
|
||||
|
||||
public void setMemPreReqisite(long freeBytes);
|
||||
// sets minimum required amount of memory for the job execution
|
||||
|
||||
public void setObeyIntermission(boolean obey);
|
||||
// defines if the thread should obey the intermission command
|
||||
|
||||
public long getIdleCycles();
|
||||
// returns the total number of cycles of job execution with idle-result
|
||||
|
||||
public long getBusyCycles();
|
||||
// returns the total number of cycles of job execution with busy-result
|
||||
|
||||
public long getOutOfMemoryCycles();
|
||||
// returns the total number of cycles where
|
||||
// a job execution was omitted because of memory shortage
|
||||
|
||||
public long getSleepTime();
|
||||
// returns the total time that this thread has slept so far
|
||||
|
||||
public void intermission(long pause);
|
||||
// the thread is forced to pause for a specific time
|
||||
// if the thread is busy meanwhile, the pause is ommitted
|
||||
|
||||
public boolean job() throws Exception;
|
||||
// performes one job procedure; this loopes until terminate() is called
|
||||
// job returns true if it has done something
|
||||
// it returns false if it is idle and does not expect to work on more for a longer time
|
||||
|
||||
public void freemem();
|
||||
// is called when an outOfMemoryCycle is performed
|
||||
// this method should try to free some memory, so that the job can be executed
|
||||
|
||||
}
|
||||
@ -0,0 +1,122 @@
|
||||
// serverInstantBlockingThread.java
|
||||
// (C) 2008 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
|
||||
// first published 27.03.2008 on http://yacy.net
|
||||
//
|
||||
// $LastChangedDate: 2006-04-02 22:40:07 +0200 (So, 02 Apr 2006) $
|
||||
// $LastChangedRevision: 1986 $
|
||||
// $LastChangedBy: orbiter $
|
||||
//
|
||||
// LICENSE
|
||||
//
|
||||
// This program is free software; you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation; either version 2 of the License, or
|
||||
// (at your option) any later version.
|
||||
//
|
||||
// This program is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with this program; if not, write to the Free Software
|
||||
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
||||
|
||||
package de.anomic.server;
|
||||
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.TreeMap;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import de.anomic.server.logging.serverLog;
|
||||
|
||||
public class serverInstantBlockingThread<I, O> extends serverAbstractBlockingThread<I, O> implements serverBlockingThread<I, O> {
|
||||
|
||||
private Method jobExecMethod, jobCountMethod;
|
||||
private Object environment;
|
||||
private Long handle;
|
||||
|
||||
public static int instantThreadCounter = 0;
|
||||
public static TreeMap<Long, String> jobs = new TreeMap<Long, String>();
|
||||
|
||||
public serverInstantBlockingThread(Object env, String jobExec, String jobCount, BlockingQueue<I> input, BlockingQueue<O> output) {
|
||||
// jobExec is the name of a method of the object 'env' that executes the one-step-run
|
||||
// jobCount is the name of a method that returns the size of the job
|
||||
|
||||
// set the blocking queues for input and output
|
||||
this.setInputQueue(input);
|
||||
this.setOutputQueue(output);
|
||||
|
||||
// define execution class
|
||||
Class<?> theClass = (env instanceof Class) ? (Class<?>) env : env.getClass();
|
||||
try {
|
||||
this.jobExecMethod = theClass.getMethod(jobExec, new Class[0]);
|
||||
} catch (NoSuchMethodException e) {
|
||||
throw new RuntimeException("serverInstantThread, wrong declaration of jobExec: " + e.getMessage());
|
||||
}
|
||||
try {
|
||||
if (jobCount == null)
|
||||
this.jobCountMethod = null;
|
||||
else
|
||||
this.jobCountMethod = theClass.getMethod(jobCount, new Class[0]);
|
||||
|
||||
} catch (NoSuchMethodException e) {
|
||||
throw new RuntimeException("serverInstantThread, wrong declaration of jobCount: " + e.getMessage());
|
||||
}
|
||||
this.environment = (env instanceof Class) ? null : env;
|
||||
this.setName(theClass.getName() + "." + jobExec);
|
||||
this.handle = new Long(System.currentTimeMillis() + this.getName().hashCode());
|
||||
}
|
||||
|
||||
public int getJobCount() {
|
||||
if (this.jobCountMethod == null) return Integer.MAX_VALUE;
|
||||
try {
|
||||
Object result = jobCountMethod.invoke(environment, new Object[0]);
|
||||
if (result instanceof Integer)
|
||||
return ((Integer) result).intValue();
|
||||
else
|
||||
return -1;
|
||||
} catch (IllegalAccessException e) {
|
||||
return -1;
|
||||
} catch (IllegalArgumentException e) {
|
||||
return -1;
|
||||
} catch (InvocationTargetException e) {
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "invocation serverInstantThread of thread '" + this.getName() + "': " + e.getMessage(), e);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public O job(I next) throws Exception {
|
||||
instantThreadCounter++;
|
||||
//System.out.println("started job " + this.handle + ": " + this.getName());
|
||||
synchronized(jobs) {jobs.put(this.handle, this.getName());}
|
||||
O out = null;
|
||||
try {
|
||||
out = (O) jobExecMethod.invoke(environment, new Object[0]);
|
||||
} catch (IllegalAccessException e) {
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "Internal Error in serverInstantThread.job: " + e.getMessage());
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "shutting down thread '" + this.getName() + "'");
|
||||
this.terminate(false);
|
||||
} catch (IllegalArgumentException e) {
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "Internal Error in serverInstantThread.job: " + e.getMessage());
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "shutting down thread '" + this.getName() + "'");
|
||||
this.terminate(false);
|
||||
} catch (InvocationTargetException e) {
|
||||
String targetException = e.getTargetException().getMessage();
|
||||
e.getTargetException().printStackTrace();
|
||||
e.printStackTrace();
|
||||
if ((targetException != null) && ((targetException.indexOf("heap space") > 0) || (targetException.indexOf("NullPointerException") > 0))) e.getTargetException().printStackTrace();
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "Runtime Error in serverInstantThread.job, thread '" + this.getName() + "': " + e.getMessage() + "; target exception: " + targetException, e.getTargetException());
|
||||
e.getTargetException().printStackTrace();
|
||||
} catch (OutOfMemoryError e) {
|
||||
serverLog.logSevere("BLOCKINGTHREAD", "OutOfMemory Error in serverInstantThread.job, thread '" + this.getName() + "': " + e.getMessage());
|
||||
e.printStackTrace();
|
||||
}
|
||||
instantThreadCounter--;
|
||||
synchronized(jobs) {jobs.remove(this.handle);}
|
||||
return out;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in new issue