X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;f=runtime%2Forg.argeo.slc.core%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fcore%2Fexecution%2FProcessThread.java;h=c11e875b8083df0d627c51dba32587360bdfed22;hb=32f9566efa0d0ce29b31ee0779faf2b42f46321a;hp=ec600cb5776deeaa6a9020cba225f2c284d59ea0;hpb=35f2455f24c3c8570a9c8ae7137be046e40fa84d;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 ec600cb57..c11e875b8 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 @@ -1,93 +1,173 @@ +/* + * Copyright (C) 2010 Mathieu Baudier + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package org.argeo.slc.core.execution; import java.util.ArrayList; -import java.util.Iterator; +import java.util.Collections; +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 ThreadGroup processThreadGroup; - private final List flowsToProcess = new ArrayList(); + private final ExecutionModulesManager executionModulesManager; + private final ExecutionProcess process; + private final ProcessThreadGroup processThreadGroup; + + private Set executionThreads = Collections + .synchronizedSet(new HashSet()); private Boolean hadAnError = false; + private Boolean killed = false; - public ProcessThread( - AbstractExecutionModulesManager executionModulesManager, - SlcExecution slcExecution) { - super(executionModulesManager.getProcessesThreadGroup(), - "SLC Process #" + slcExecution.getUuid()); + public ProcessThread(ThreadGroup processesThreadGroup, + ExecutionModulesManager executionModulesManager, + ExecutionProcess process) { + super(processesThreadGroup, "SLC Process #" + process.getUuid()); this.executionModulesManager = executionModulesManager; - this.slcProcess = slcExecution; - processThreadGroup = new ThreadGroup("SLC Process #" - + slcExecution.getUuid() + " thread group"); + 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); + public final void run() { + log.info("\n##\n## SLC Process #" + process.getUuid() + + " STARTED\n##\n"); - flowsToProcess.addAll(slcProcess.getRealizedFlows()); + String oldStatus = process.getStatus(); + process.setStatus(ExecutionProcess.RUNNING); + executionModulesManager.dispatchUpdateStatus(process, oldStatus, + ExecutionProcess.RUNNING); - while (flowsToProcess.size() > 0) { - RealizedFlow flow = flowsToProcess.remove(0); - ExecutionThread thread = new ExecutionThread(this, flow); - thread.start(); + try { + process(); + } catch (InterruptedException e) { + die(); + return; + } - 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 + die(); + return; } } } - if (hadAnError) - slcProcess.setStatus(SlcExecution.STATUS_ERROR); + computeFinalStatus(); + } + + /** Make sure this is called BEFORE all the threads are interrupted. */ + private void computeFinalStatus() { + String oldStatus = process.getStatus(); + // TODO: error management at flow level? + if (killed) + process.setStatus(ExecutionProcess.KILLED); + else if (hadAnError) + process.setStatus(ExecutionProcess.ERROR); else - slcProcess.setStatus(SlcExecution.STATUS_FINISHED); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING, - slcProcess.getStatus()); + process.setStatus(ExecutionProcess.COMPLETED); + executionModulesManager.dispatchUpdateStatus(process, oldStatus, + process.getStatus()); + log.info("\n## SLC Process #" + process.getUuid() + " " + + process.getStatus() + "\n"); + } + + /** Called when being killed */ + private synchronized void die() { + killed = true; + computeFinalStatus(); + for (ExecutionThread executionThread : executionThreads) { + try { + executionThread.interrupt(); + } catch (Exception e) { + log.error("Cannot interrupt " + executionThread); + } + } + processThreadGroup.interrupt(); } - protected void dispatchUpdateStatus(SlcExecution slcExecution, - String oldStatus, String newStatus) { - for (Iterator it = executionModulesManager - .getSlcExecutionNotifiers().iterator(); it.hasNext();) { - it.next().updateStatus(slcExecution, oldStatus, newStatus); + /** + * Implementation specific execution. To be overridden in order to deal with + * custom process types. Default expects an {@link SlcExecution}. + */ + @SuppressWarnings("deprecation") + protected void process() throws InterruptedException { + 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); } } + /** @return the (distinct) thread used for this execution */ + protected final void execute(RealizedFlow realizedFlow, Boolean synchronous) + throws InterruptedException { + if (killed) + return; + + ExecutionThread thread = new ExecutionThread(this, realizedFlow); + executionThreads.add(thread); + thread.start(); + + if (synchronous) + thread.join(); + + return; + } + public void notifyError() { hadAnError = true; } public synchronized void flowCompleted() { - notifyAll(); + // notifyAll(); } - public SlcExecution getSlcProcess() { - return slcProcess; + public ExecutionProcess getProcess() { + return process; } - public ThreadGroup getProcessThreadGroup() { + public ProcessThreadGroup getProcessThreadGroup() { return processThreadGroup; } - public AbstractExecutionModulesManager getExecutionModulesManager() { + public ExecutionModulesManager getExecutionModulesManager() { return executionModulesManager; } }