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=98d715ece3962f5eb77669d912a44ffa28124e1c;hb=098e868d40b2130c6735b1547247c61a5d88da95;hp=678e6f679ae809322e81125b79fece1074489a50;hpb=a91feabac0f3f603a73936b4447599402fba71d0;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 678e6f679..98d715ece 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,84 +1,217 @@ +/* + * 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.execution.ExecutionStep; 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. */ +@SuppressWarnings("deprecation") 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; - public ProcessThread( - AbstractExecutionModulesManager executionModulesManager, - SlcExecution slcExecution) { - super(executionModulesManager.getProcessesThreadGroup(), - "SLC Process #" + slcExecution.getUuid()); - this.executionModulesManager = executionModulesManager; - this.slcProcess = slcExecution; - processThreadGroup = new ThreadGroup("SLC Process #" - + slcExecution.getUuid() + " thread group"); - } + private Set executionThreads = Collections + .synchronizedSet(new HashSet()); - public void run() { - log.info("\n##\n## Process SLC Execution " + slcProcess + "\n##\n"); + private Boolean hadAnError = false; + private Boolean killed = false; - slcProcess.setStatus(SlcExecution.STATUS_RUNNING); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_SCHEDULED, - SlcExecution.STATUS_RUNNING); - - flowsToProcess.addAll(slcProcess.getRealizedFlows()); + public ProcessThread(ThreadGroup processesThreadGroup, + ExecutionModulesManager executionModulesManager, + ExecutionProcess process) { + super(processesThreadGroup, "SLC Process #" + process.getUuid()); + this.executionModulesManager = executionModulesManager; + this.process = process; + processThreadGroup = new ProcessThreadGroup(executionModulesManager, + this); + } - while (flowsToProcess.size() > 0) { - RealizedFlow flow = flowsToProcess.remove(0); - ExecutionThread thread = new ExecutionThread(this, flow); - thread.start(); + public final void run() { + // authenticate thread + // Authentication authentication = getProcessThreadGroup() + // .getAuthentication(); + // if (authentication == null) + // throw new SlcException("Can only execute authenticated threads"); + // SecurityContextHolder.getContext().setAuthentication(authentication); + + // log.info("\n##\n## SLC Process #" + process.getUuid() + + // " STARTED by " + // + authentication.getName() + "\n##\n"); + log.info("\n##\n## SLC Process #" + process.getUuid() + + " STARTED\n##\n"); + + // Start logging + new LoggingThread().start(); + + String oldStatus = process.getStatus(); + process.setStatus(ExecutionProcess.RUNNING); + executionModulesManager.dispatchUpdateStatus(process, oldStatus, + ExecutionProcess.RUNNING); + + 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; } } } - slcProcess.setStatus(SlcExecution.STATUS_FINISHED); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING, - SlcExecution.STATUS_FINISHED); + 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 + process.setStatus(ExecutionProcess.COMPLETED); + executionModulesManager.dispatchUpdateStatus(process, oldStatus, + process.getStatus()); + log.info("\n## SLC Process #" + process.getUuid() + " " + + process.getStatus() + "\n"); } - protected void dispatchUpdateStatus(SlcExecution slcExecution, - String oldStatus, String newStatus) { - for (Iterator it = executionModulesManager - .getSlcExecutionNotifiers().iterator(); it.hasNext();) { - it.next().updateStatus(slcExecution, oldStatus, newStatus); + /** 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(); + } + + /** + * Implementation specific execution. To be overridden in order to deal with + * custom process types. Default expects an {@link SlcExecution}. + */ + 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; } + + private class LoggingThread extends Thread { + + public LoggingThread() { + super("SLC Process Logger #" + process.getUuid()); + } + + public void run() { + boolean run = true; + while (run) { + List newSteps = new ArrayList(); + processThreadGroup.getSteps().drainTo(newSteps); + if (newSteps.size() > 0) { + // System.out.println(steps.size() + " steps"); + process.addSteps(newSteps); + } + + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + break; + } + + if (!ProcessThread.this.isAlive() + && processThreadGroup.getSteps().size() == 0) + run = false; + } + } + + } }