From: Mathieu Baudier Date: Sat, 16 Apr 2011 16:34:53 +0000 (+0000) Subject: Improve execution X-Git-Tag: argeo-slc-2.1.7~984 X-Git-Url: http://git.argeo.org/?a=commitdiff_plain;h=30f4c6af6c20077e5e36b61faf5edb22c1aae6c6;hp=633f99ce62e7c18f91d0798682b45528fd822d62;p=gpl%2Fargeo-slc.git Improve execution NEW - bug 17: Generalize agent management and registration beyond JMS https://bugzilla.argeo.org/show_bug.cgi?id=17 git-svn-id: https://svn.argeo.org/slc/trunk@4440 4cfe0d0a-d680-48aa-b62c-e0a02a3f76cc --- diff --git a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/AbstractExecutionModulesManager.java b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/AbstractExecutionModulesManager.java index f471d65ca..4e9856fe6 100644 --- a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/AbstractExecutionModulesManager.java +++ b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/AbstractExecutionModulesManager.java @@ -17,6 +17,7 @@ package org.argeo.slc.core.execution; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import java.util.Map; @@ -30,7 +31,9 @@ import org.argeo.slc.execution.ExecutionModulesManager; import org.argeo.slc.process.RealizedFlow; import org.argeo.slc.process.SlcExecution; import org.argeo.slc.process.SlcExecutionNotifier; +import org.argeo.slc.process.SlcExecutionStep; +/** Provides the base feature of an execution module manager. */ public abstract class AbstractExecutionModulesManager implements ExecutionModulesManager { private final static Log log = LogFactory @@ -81,6 +84,23 @@ public abstract class AbstractExecutionModulesManager implements // } + public void dispatchUpdateStatus(SlcExecution slcExecution, + String oldStatus, String newStatus) { + for (Iterator it = getSlcExecutionNotifiers() + .iterator(); it.hasNext();) { + it.next().updateStatus(slcExecution, oldStatus, newStatus); + } + } + + public void dispatchAddStep(SlcExecution slcExecution, SlcExecutionStep step) { + List steps = new ArrayList(); + steps.add(step); + for (Iterator it = getSlcExecutionNotifiers() + .iterator(); it.hasNext();) { + it.next().addSteps(slcExecution, steps); + } + } + public void setSlcExecutionNotifiers( List slcExecutionNotifiers) { this.slcExecutionNotifiers = slcExecutionNotifiers; @@ -94,7 +114,7 @@ public abstract class AbstractExecutionModulesManager implements return processesThreadGroup; } - public List getExecutionModulesListeners() { + protected List getExecutionModulesListeners() { return executionModulesListeners; } diff --git a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionAspect.java b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionAspect.java index 09e2afe5b..e9e5b4cdd 100644 --- a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionAspect.java +++ b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ExecutionAspect.java @@ -48,14 +48,14 @@ public class ExecutionAspect { if (log.isDebugEnabled()) logStackEvent("=> ", executionFlow); - + try { // Actually execute the flow pjp.proceed(); } finally { if (log.isDebugEnabled()) logStackEvent("<= ", executionFlow); - + executionStack.leaveFlow(executionFlow); } } @@ -104,16 +104,21 @@ public class ExecutionAspect { protected void logStackEvent(String symbol, ExecutionFlow executionFlow) { Integer stackSize = executionStack.getStackSize(); - log.debug(depthSpaces(stackSize) + symbol + executionFlow + " #" - + executionStack.getCurrentStackLevelUuid() + ", depth=" - + stackSize); + if (log.isTraceEnabled()) + log.debug(depthSpaces(stackSize) + symbol + executionFlow + " #" + + executionStack.getCurrentStackLevelUuid() + ", depth=" + + stackSize); + else if (log.isDebugEnabled()) + log.debug(depthSpaces(stackSize) + symbol + executionFlow); } protected void logRunnableExecution(ExecutionFlow executionFlow, Runnable runnable) { Integer stackSize = executionStack.getStackSize(); - log.debug(depthSpaces(stackSize + 1) - + runnable.getClass().getSimpleName() + " in " + executionFlow); + if (log.isDebugEnabled()) + log.debug(depthSpaces(stackSize + 1) + + runnable.getClass().getSimpleName() + " in " + + executionFlow); } private String depthSpaces(int depth) { 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..7c0e7b10d 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,42 +17,41 @@ package org.argeo.slc.core.execution; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.argeo.slc.execution.ExecutionModulesManager; 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 ExecutionModulesManager executionModulesManager; private final SlcExecution slcProcess; private final ProcessThreadGroup processThreadGroup; private final List flowsToProcess = new ArrayList(); private Boolean hadAnError = false; - public ProcessThread( - AbstractExecutionModulesManager executionModulesManager, + public ProcessThread(ExecutionModulesManager executionModulesManager, SlcExecution slcExecution) { super(executionModulesManager.getProcessesThreadGroup(), "SLC Process #" + slcExecution.getUuid()); this.executionModulesManager = executionModulesManager; this.slcProcess = slcExecution; - processThreadGroup = new ProcessThreadGroup(this); + processThreadGroup = new ProcessThreadGroup(executionModulesManager, + this); } public void run() { - log.info("\n##\n## Process SLC Execution " + slcProcess + "\n##\n"); + log.info("\n##\n## SLC Process " + slcProcess + " STARTED\n##"); slcProcess.setStatus(SlcExecution.STATUS_RUNNING); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_SCHEDULED, - SlcExecution.STATUS_RUNNING); + executionModulesManager.dispatchUpdateStatus(slcProcess, + SlcExecution.STATUS_SCHEDULED, SlcExecution.STATUS_RUNNING); flowsToProcess.addAll(slcProcess.getRealizedFlows()); @@ -61,29 +60,30 @@ public class ProcessThread extends Thread { ExecutionThread thread = new ExecutionThread(this, flow); thread.start(); - synchronized (this) { - try { - wait(); - } catch (InterruptedException e) { - // silent - } + try { + thread.join(); + } catch (InterruptedException e) { + log.error("Flow " + flow + " was interrupted", e); } + + // synchronized (this) { + // try { + // wait(); + // } catch (InterruptedException e) { + // // silent + // } + // } } + // TODO: error management at flow level? if (hadAnError) slcProcess.setStatus(SlcExecution.STATUS_ERROR); else slcProcess.setStatus(SlcExecution.STATUS_FINISHED); - dispatchUpdateStatus(slcProcess, SlcExecution.STATUS_RUNNING, - slcProcess.getStatus()); - } + executionModulesManager.dispatchUpdateStatus(slcProcess, + SlcExecution.STATUS_RUNNING, slcProcess.getStatus()); - protected void dispatchUpdateStatus(SlcExecution slcExecution, - String oldStatus, String newStatus) { - for (Iterator it = executionModulesManager - .getSlcExecutionNotifiers().iterator(); it.hasNext();) { - it.next().updateStatus(slcExecution, oldStatus, newStatus); - } + log.info("## SLC Process " + slcProcess + " COMPLETED"); } public void notifyError() { @@ -91,7 +91,7 @@ public class ProcessThread extends Thread { } public synchronized void flowCompleted() { - notifyAll(); + // notifyAll(); } public SlcExecution getSlcProcess() { @@ -102,7 +102,7 @@ public class ProcessThread extends Thread { return processThreadGroup; } - public AbstractExecutionModulesManager getExecutionModulesManager() { + public ExecutionModulesManager getExecutionModulesManager() { return executionModulesManager; } } diff --git a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThreadGroup.java b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThreadGroup.java index f7e2bea1c..1574b97d6 100644 --- a/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThreadGroup.java +++ b/runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThreadGroup.java @@ -16,20 +16,20 @@ package org.argeo.slc.core.execution; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; - +import org.argeo.slc.execution.ExecutionModulesManager; import org.argeo.slc.process.SlcExecution; -import org.argeo.slc.process.SlcExecutionNotifier; import org.argeo.slc.process.SlcExecutionStep; +/** The thread group attached to a given {@link SlcExecution}. */ public class ProcessThreadGroup extends ThreadGroup { + private final ExecutionModulesManager executionModulesManager; private final ProcessThread processThread; - public ProcessThreadGroup(ProcessThread processThread) { + public ProcessThreadGroup(ExecutionModulesManager executionModulesManager, + ProcessThread processThread) { super("SLC Process #" + processThread.getSlcProcess().getUuid() + " thread group"); + this.executionModulesManager = executionModulesManager; this.processThread = processThread; } @@ -38,14 +38,9 @@ public class ProcessThreadGroup extends ThreadGroup { } public void dispatchAddStep(SlcExecutionStep step) { - processThread.getSlcProcess().getSteps().add(step); - List steps = new ArrayList(); - steps.add(step); - for (Iterator it = processThread - .getExecutionModulesManager().getSlcExecutionNotifiers() - .iterator(); it.hasNext();) { - it.next().addSteps(processThread.getSlcProcess(), steps); - } + SlcExecution slcProcess = processThread.getSlcProcess(); + slcProcess.getSteps().add(step); + executionModulesManager.dispatchAddStep(slcProcess, step); } } diff --git a/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesListener.java b/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesListener.java index c515e1073..54f5cb3e6 100644 --- a/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesListener.java +++ b/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesListener.java @@ -18,6 +18,7 @@ package org.argeo.slc.execution; import org.argeo.slc.deploy.Module; +/** Listen to events on execution modules. */ public interface ExecutionModulesListener { public void executionModuleAdded(Module module, ExecutionContext executionContext); diff --git a/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesManager.java b/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesManager.java index c1b9e47a3..26890ad05 100644 --- a/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesManager.java +++ b/runtime/org.argeo.slc.specs/src/main/java/org/argeo/slc/execution/ExecutionModulesManager.java @@ -21,6 +21,7 @@ import java.util.List; import org.argeo.slc.deploy.ModulesManager; import org.argeo.slc.process.RealizedFlow; import org.argeo.slc.process.SlcExecution; +import org.argeo.slc.process.SlcExecutionStep; /** Provides access to the execution modules */ public interface ExecutionModulesManager extends ModulesManager { @@ -34,9 +35,19 @@ public interface ExecutionModulesManager extends ModulesManager { */ public List listExecutionModules(); - /** Asynchronously prepare and executes an {@link SlcExecution} */ + /** Asynchronously prepares and executes an {@link SlcExecution} */ public void process(SlcExecution slcExecution); + /** The thread group to which all process threads will belong. */ + public ThreadGroup getProcessesThreadGroup(); + /** Synchronously finds and executes an {@link ExecutionFlow}. */ public void execute(RealizedFlow realizedFlow); + + /** Notify of a status update status of the {@link SlcExecution} */ + public void dispatchUpdateStatus(SlcExecution slcExecution, + String oldStatus, String newStatus); + + /** Notify that a step was added in an {@link SlcExecution} */ + public void dispatchAddStep(SlcExecution slcExecution, SlcExecutionStep step); }