|
|
|
@ -33,7 +33,6 @@ import java.util.concurrent.Executors;
|
|
|
|
|
import java.util.concurrent.LinkedBlockingQueue;
|
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import net.yacy.kelondro.logging.Log;
|
|
|
|
|
import net.yacy.kelondro.util.NamePrefixThreadFactory;
|
|
|
|
|
|
|
|
|
@ -71,10 +70,10 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
this.executor.submit(new InstantBlockingThread<J>(env, jobExecMethod, this));
|
|
|
|
|
}
|
|
|
|
|
// init statistics
|
|
|
|
|
blockTime = 0;
|
|
|
|
|
execTime = 0;
|
|
|
|
|
passOnTime = 0;
|
|
|
|
|
execCount = 0;
|
|
|
|
|
this.blockTime = 0;
|
|
|
|
|
this.execTime = 0;
|
|
|
|
|
this.passOnTime = 0;
|
|
|
|
|
this.execCount = 0;
|
|
|
|
|
|
|
|
|
|
// store this object for easy monitoring
|
|
|
|
|
processMonitor.add(this);
|
|
|
|
@ -99,8 +98,8 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
public J take() throws InterruptedException {
|
|
|
|
|
// read from the input queue
|
|
|
|
|
if (this.input == null) return null;
|
|
|
|
|
long t = System.currentTimeMillis();
|
|
|
|
|
J j = this.input.take();
|
|
|
|
|
final long t = System.currentTimeMillis();
|
|
|
|
|
final J j = this.input.take();
|
|
|
|
|
this.blockTime += System.currentTimeMillis() - t;
|
|
|
|
|
return j;
|
|
|
|
|
}
|
|
|
|
@ -109,7 +108,7 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
// don't mix this method up with enQueue()!
|
|
|
|
|
// this method enqueues into the _next_ queue, not this queue!
|
|
|
|
|
if (this.output == null) return;
|
|
|
|
|
long t = System.currentTimeMillis();
|
|
|
|
|
final long t = System.currentTimeMillis();
|
|
|
|
|
this.output.enQueue(next);
|
|
|
|
|
this.passOnTime += System.currentTimeMillis() - t;
|
|
|
|
|
}
|
|
|
|
@ -121,7 +120,7 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
public synchronized void relaxCapacity() {
|
|
|
|
|
if (this.input.isEmpty()) return;
|
|
|
|
|
if (this.input.remainingCapacity() > 1000) return;
|
|
|
|
|
BlockingQueue<J> i = new LinkedBlockingQueue<J>();
|
|
|
|
|
final BlockingQueue<J> i = new LinkedBlockingQueue<J>();
|
|
|
|
|
J e;
|
|
|
|
|
while (!this.input.isEmpty()) {
|
|
|
|
|
e = this.input.poll();
|
|
|
|
@ -134,11 +133,11 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
@SuppressWarnings("unchecked")
|
|
|
|
|
public void enQueue(final J in) throws InterruptedException {
|
|
|
|
|
// ensure that enough job executors are running
|
|
|
|
|
if ((this.input == null) || (executor == null) || (executor.isShutdown()) || (executor.isTerminated())) {
|
|
|
|
|
if ((this.input == null) || (this.executor == null) || (this.executor.isShutdown()) || (this.executor.isTerminated())) {
|
|
|
|
|
// execute serialized without extra thread
|
|
|
|
|
Log.logWarning("PROCESSOR", "executing job " + environment.getClass().getName() + "." + methodName + " serialized");
|
|
|
|
|
//Log.logWarning("PROCESSOR", "executing job " + environment.getClass().getName() + "." + methodName + " serialized");
|
|
|
|
|
try {
|
|
|
|
|
final J out = (J) InstantBlockingThread.execMethod(this.environment, this.methodName).invoke(environment, new Object[]{in});
|
|
|
|
|
final J out = (J) InstantBlockingThread.execMethod(this.environment, this.methodName).invoke(this.environment, new Object[]{in});
|
|
|
|
|
if (out != null && this.output != null) this.output.enQueue(out);
|
|
|
|
|
} catch (final IllegalArgumentException e) {
|
|
|
|
|
Log.logException(e);
|
|
|
|
@ -154,41 +153,41 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
try {
|
|
|
|
|
this.input.put(in);
|
|
|
|
|
break;
|
|
|
|
|
} catch (InterruptedException e) {
|
|
|
|
|
try {Thread.sleep(10);} catch (InterruptedException ee) {}
|
|
|
|
|
} catch (final InterruptedException e) {
|
|
|
|
|
try {Thread.sleep(10);} catch (final InterruptedException ee) {}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@SuppressWarnings("unchecked")
|
|
|
|
|
public void announceShutdown() {
|
|
|
|
|
if (executor == null) return;
|
|
|
|
|
if (executor.isShutdown()) return;
|
|
|
|
|
if (this.executor == null) return;
|
|
|
|
|
if (this.executor.isShutdown()) return;
|
|
|
|
|
// before we put pills into the queue, make sure that they will take them
|
|
|
|
|
relaxCapacity();
|
|
|
|
|
// put poison pills into the queue
|
|
|
|
|
for (int i = 0; i < poolsize; i++) {
|
|
|
|
|
for (int i = 0; i < this.poolsize; i++) {
|
|
|
|
|
try {
|
|
|
|
|
Log.logInfo("serverProcessor", "putting poison pill in queue " + this.processName + ", thread " + i);
|
|
|
|
|
input.put((J) WorkflowJob.poisonPill); // put a poison pill into the queue which will kill the job
|
|
|
|
|
this.input.put((J) WorkflowJob.poisonPill); // put a poison pill into the queue which will kill the job
|
|
|
|
|
Log.logInfo("serverProcessor", ".. poison pill is in queue " + this.processName + ", thread " + i + ". awaiting termination");
|
|
|
|
|
} catch (final InterruptedException e) { }
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void awaitShutdown(final long millisTimeout) {
|
|
|
|
|
if (executor != null & !executor.isShutdown()) {
|
|
|
|
|
if (this.executor != null & !this.executor.isShutdown()) {
|
|
|
|
|
// wait for shutdown
|
|
|
|
|
try {
|
|
|
|
|
executor.shutdown();
|
|
|
|
|
executor.awaitTermination(millisTimeout, TimeUnit.MILLISECONDS);
|
|
|
|
|
this.executor.shutdown();
|
|
|
|
|
this.executor.awaitTermination(millisTimeout, TimeUnit.MILLISECONDS);
|
|
|
|
|
} catch (final InterruptedException e) {}
|
|
|
|
|
}
|
|
|
|
|
Log.logInfo("serverProcessor", "queue " + this.processName + ": shutdown.");
|
|
|
|
|
this.executor = null;
|
|
|
|
|
this.input = null;
|
|
|
|
|
// remove entry from monitor
|
|
|
|
|
Iterator<WorkflowProcessor<?>> i = processes();
|
|
|
|
|
final Iterator<WorkflowProcessor<?>> i = processes();
|
|
|
|
|
WorkflowProcessor<?> p;
|
|
|
|
|
while (i.hasNext()) {
|
|
|
|
|
p = i.next();
|
|
|
|
@ -217,9 +216,9 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public String getChilds() {
|
|
|
|
|
StringBuilder s = new StringBuilder(this.childs.length * 40 + 1);
|
|
|
|
|
for (int i = 0; i < this.childs.length; i++) {
|
|
|
|
|
s.append(this.childs[i]);
|
|
|
|
|
final StringBuilder s = new StringBuilder(this.childs.length * 40 + 1);
|
|
|
|
|
for (final String child : this.childs) {
|
|
|
|
|
s.append(child);
|
|
|
|
|
s.append(' ');
|
|
|
|
|
}
|
|
|
|
|
return s.toString();
|
|
|
|
@ -230,7 +229,7 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
* @return
|
|
|
|
|
*/
|
|
|
|
|
public long getBlockTime() {
|
|
|
|
|
return blockTime;
|
|
|
|
|
return this.blockTime;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
@ -238,10 +237,10 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
* @return
|
|
|
|
|
*/
|
|
|
|
|
public long getExecTime() {
|
|
|
|
|
return execTime;
|
|
|
|
|
return this.execTime;
|
|
|
|
|
}
|
|
|
|
|
public long getExecCount() {
|
|
|
|
|
return execCount;
|
|
|
|
|
return this.execCount;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
@ -250,7 +249,7 @@ public class WorkflowProcessor<J extends WorkflowJob> {
|
|
|
|
|
* @return
|
|
|
|
|
*/
|
|
|
|
|
public long getPassOnTime() {
|
|
|
|
|
return passOnTime;
|
|
|
|
|
return this.passOnTime;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|