X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;ds=sidebyside;f=runtime%2Forg.argeo.slc.core%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fcore%2Fexecution%2FProcessThread.java;h=dc9e7efdd8a1233d01c1ea7f05e0d69d41c34d92;hb=86c47402780f41526382267ff1597d2f3a0d0dd7;hp=c119127ad20b71f1ab269321db7df1f08deb33fc;hpb=1fdb1b4e7b1d2b0cabb6483238301b857a6392fa;p=gpl%2Fargeo-slc.git diff --git a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java index c119127ad..dc9e7efdd 100644 --- a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java +++ b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java @@ -17,73 +17,114 @@ package org.argeo.slc.core.execution; import java.util.ArrayList; -import java.util.Iterator; +import java.util.HashSet; import java.util.List; +import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.argeo.slc.SlcException; +import org.argeo.slc.execution.ExecutionModulesManager; +import org.argeo.slc.execution.ExecutionProcess; import org.argeo.slc.process.RealizedFlow; import org.argeo.slc.process.SlcExecution; -import org.argeo.slc.process.SlcExecutionNotifier; /** Thread of the SLC Process, starting the sub executions. */ public class ProcessThread extends Thread { private final static Log log = LogFactory.getLog(ProcessThread.class); - private final AbstractExecutionModulesManager executionModulesManager; - private final SlcExecution slcProcess; + private final ExecutionModulesManager executionModulesManager; + private final ExecutionProcess process; private final ProcessThreadGroup processThreadGroup; - private final List flowsToProcess = new ArrayList(); + + private Set executionThreads = new HashSet(); private Boolean hadAnError = false; - public ProcessThread( - AbstractExecutionModulesManager executionModulesManager, - SlcExecution slcExecution) { + public ProcessThread(ExecutionModulesManager executionModulesManager, + ExecutionProcess process) { super(executionModulesManager.getProcessesThreadGroup(), - "SLC Process #" + slcExecution.getUuid()); + "SLC Process #" + process.getUuid()); this.executionModulesManager = executionModulesManager; - this.slcProcess = slcExecution; - processThreadGroup = new ProcessThreadGroup(this); + this.process = process; + processThreadGroup = new ProcessThreadGroup(executionModulesManager, + this); } public void run() { - log.info("\n##\n## Process SLC Execution " + slcProcess + "\n##\n"); - - slcProcess.setStatus(SlcExecution.STATUS_RUNNING); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_SCHEDULED, - SlcExecution.STATUS_RUNNING); + log.info("\n##\n## SLC Process #" + process.getUuid() + + " STARTED\n##\n"); - flowsToProcess.addAll(slcProcess.getRealizedFlows()); + process.setStatus(SlcExecution.RUNNING); + executionModulesManager.dispatchUpdateStatus(process, + SlcExecution.SCHEDULED, SlcExecution.RUNNING); - while (flowsToProcess.size() > 0) { - RealizedFlow flow = flowsToProcess.remove(0); - ExecutionThread thread = new ExecutionThread(this, flow); - thread.start(); + process(); - synchronized (this) { + // waits for all execution threads to complete (in case they were + // started asynchronously) + for (ExecutionThread executionThread : executionThreads) { + if (executionThread.isAlive()) { try { - wait(); + executionThread.join(); } catch (InterruptedException e) { - // silent + log.error("Execution thread " + executionThread + + " was interrupted"); } } } + // TODO: error management at flow level? if (hadAnError) - slcProcess.setStatus(SlcExecution.STATUS_ERROR); + process.setStatus(SlcExecution.ERROR); else - slcProcess.setStatus(SlcExecution.STATUS_FINISHED); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING, - slcProcess.getStatus()); + process.setStatus(SlcExecution.COMPLETED); + executionModulesManager.dispatchUpdateStatus(process, + SlcExecution.RUNNING, process.getStatus()); + + log.info("\n## SLC Process #" + process.getUuid() + " COMPLETED\n"); + } + + /** + * Implementation specific execution. To be overridden in order to deal with + * custom process types. Default expects an {@link SlcExecution}. + */ + protected void process() { + if (!(process instanceof SlcExecution)) + throw new SlcException("Unsupported process type " + + process.getClass()); + SlcExecution slcExecution = (SlcExecution) process; + List flowsToProcess = new ArrayList(); + flowsToProcess.addAll(slcExecution.getRealizedFlows()); + + while (flowsToProcess.size() > 0) { + RealizedFlow realizedFlow = flowsToProcess.remove(0); + execute(realizedFlow, true); + } } - protected void dispatchUpdateStatus(SlcExecution slcExecution, - String oldStatus, String newStatus) { - for (Iterator it = executionModulesManager - .getSlcExecutionNotifiers().iterator(); it.hasNext();) { - it.next().updateStatus(slcExecution, oldStatus, newStatus); + /** @return the (distinct) thread used for this execution */ + protected Thread execute(RealizedFlow realizedFlow, Boolean synchronous) { + ExecutionThread thread = new ExecutionThread(this, realizedFlow); + executionThreads.add(thread); + thread.start(); + + if (synchronous) { + try { + thread.join(); + } catch (InterruptedException e) { + log.error("Flow " + realizedFlow + " was interrupted", e); + } } + return thread; + + // synchronized (this) { + // try { + // wait(); + // } catch (InterruptedException e) { + // // silent + // } + // } } public void notifyError() { @@ -91,18 +132,18 @@ public class ProcessThread extends Thread { } public synchronized void flowCompleted() { - notifyAll(); + // notifyAll(); } - public SlcExecution getSlcProcess() { - return slcProcess; + public ExecutionProcess getProcess() { + return process; } public ProcessThreadGroup getProcessThreadGroup() { return processThreadGroup; } - public AbstractExecutionModulesManager getExecutionModulesManager() { + public ExecutionModulesManager getExecutionModulesManager() { return executionModulesManager; } }