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=b64639a8cd55be014788e9e1e419277248c0f857;hb=b301144d5f9dc598da7d75bd1eb515f02f32f5ca;hp=7c0e7b10db0a4d3a7111916041b1f134d3676c77;hpb=30f4c6af6c20077e5e36b61faf5edb22c1aae6c6;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 7c0e7b10d..b64639a8c 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,5 +1,5 @@ /* - * Copyright (C) 2010 Mathieu Baudier + * Copyright (C) 2007-2012 Argeo GmbH * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,89 +13,165 @@ * 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.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.process.RealizedFlow; -import org.argeo.slc.process.SlcExecution; - -/** Thread of the SLC Process, starting the sub executions. */ +import org.argeo.slc.execution.ExecutionProcess; +import org.argeo.slc.execution.ExecutionStep; +import org.argeo.slc.execution.RealizedFlow; +import org.springframework.security.Authentication; +import org.springframework.security.context.SecurityContextHolder; + +/** + * Main thread coordinating an {@link ExecutionProcess}, launching parallel or + * sequential {@link ExecutionThread}s. + */ public class ProcessThread extends Thread { private final static Log log = LogFactory.getLog(ProcessThread.class); private final ExecutionModulesManager executionModulesManager; - private final SlcExecution slcProcess; + private final ExecutionProcess process; private final ProcessThreadGroup processThreadGroup; - private final List flowsToProcess = new ArrayList(); - private Boolean hadAnError = false; + private Set executionThreads = Collections + .synchronizedSet(new HashSet()); + + // private Boolean hadAnError = false; + private Boolean killed = false; - public ProcessThread(ExecutionModulesManager 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 ProcessThreadGroup(executionModulesManager, - this); + this.process = process; + processThreadGroup = new ProcessThreadGroup(process); } - public void run() { - log.info("\n##\n## SLC Process " + slcProcess + " STARTED\n##"); + 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\n##\n"); + + // Start logging + new LoggingThread().start(); + + process.setStatus(ExecutionProcess.RUNNING); + try { + process(); + } catch (InterruptedException e) { + die(); + return; + } catch (Exception e) { + String msg = "Process " + getProcess().getUuid() + + " failed unexpectedly."; + log.error(msg, e); + getProcessThreadGroup().dispatchAddStep( + new ExecutionStep("Process", ExecutionStep.ERROR, msg + " " + + e.getMessage())); + } - slcProcess.setStatus(SlcExecution.STATUS_RUNNING); - executionModulesManager.dispatchUpdateStatus(slcProcess, - SlcExecution.STATUS_SCHEDULED, SlcExecution.STATUS_RUNNING); + // waits for all execution threads to complete (in case they were + // started asynchronously) + for (ExecutionThread executionThread : executionThreads) { + if (executionThread.isAlive()) { + try { + executionThread.join(); + } catch (InterruptedException e) { + die(); + return; + } + } + } - flowsToProcess.addAll(slcProcess.getRealizedFlows()); + computeFinalStatus(); + } - while (flowsToProcess.size() > 0) { - RealizedFlow flow = flowsToProcess.remove(0); - ExecutionThread thread = new ExecutionThread(this, flow); - thread.start(); + /** 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 (processThreadGroup.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"); + } + /** Called when being killed */ + private synchronized void die() { + killed = true; + computeFinalStatus(); + for (ExecutionThread executionThread : executionThreads) { try { - thread.join(); - } catch (InterruptedException e) { - log.error("Flow " + flow + " was interrupted", e); + executionThread.interrupt(); + } catch (Exception e) { + log.error("Cannot interrupt " + executionThread); } + } + processThreadGroup.interrupt(); + } - // synchronized (this) { - // try { - // wait(); - // } catch (InterruptedException e) { - // // silent - // } - // } + /** + * Implementation specific execution. To be overridden in order to deal with + * custom process types. Default expects an {@link SlcExecution}. + */ + protected void process() throws InterruptedException { + List flowsToProcess = new ArrayList(); + flowsToProcess.addAll(process.getRealizedFlows()); + while (flowsToProcess.size() > 0) { + RealizedFlow realizedFlow = flowsToProcess.remove(0); + execute(realizedFlow, true); } + } - // TODO: error management at flow level? - if (hadAnError) - slcProcess.setStatus(SlcExecution.STATUS_ERROR); - else - slcProcess.setStatus(SlcExecution.STATUS_FINISHED); - executionModulesManager.dispatchUpdateStatus(slcProcess, - SlcExecution.STATUS_RUNNING, slcProcess.getStatus()); + /** @return the (distinct) thread used for this execution */ + protected final void execute(RealizedFlow realizedFlow, Boolean synchronous) + throws InterruptedException { + if (killed) + return; - log.info("## SLC Process " + slcProcess + " COMPLETED"); - } + ExecutionThread thread = new ExecutionThread(processThreadGroup, + executionModulesManager, realizedFlow); + executionThreads.add(thread); + thread.start(); - public void notifyError() { - hadAnError = true; - } + if (synchronous) + thread.join(); - public synchronized void flowCompleted() { - // notifyAll(); + return; } - public SlcExecution getSlcProcess() { - return slcProcess; + // public void notifyError() { + // hadAnError = true; + // } + // + // public synchronized void flowCompleted() { + // // notifyAll(); + // } + + public ExecutionProcess getProcess() { + return process; } public ProcessThreadGroup getProcessThreadGroup() { @@ -105,4 +181,34 @@ public class ProcessThread extends Thread { 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; + } + } + + } }