runtime/org.argeo.slc.core/src/main/java/org/argeo/slc/core/execution/ProcessThread.java

   1 /*
   2  * Copyright (C) 2010 Mathieu Baudier <mbaudier@argeo.org>
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *         http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.argeo.slc.core.execution;
  18
  19 import java.util.ArrayList;
  20 import java.util.Collections;
  21 import java.util.HashSet;
  22 import java.util.List;
  23 import java.util.Set;
  24
  25 import org.apache.commons.logging.Log;
  26 import org.apache.commons.logging.LogFactory;
  27 import org.argeo.slc.SlcException;
  28 import org.argeo.slc.execution.ExecutionModulesManager;
  29 import org.argeo.slc.execution.ExecutionProcess;
  30 import org.argeo.slc.process.RealizedFlow;
  31 import org.argeo.slc.process.SlcExecution;
  32
  33 /** Thread of the SLC Process, starting the sub executions. */
  34 public class ProcessThread extends Thread {
  35         private final static Log log = LogFactory.getLog(ProcessThread.class);
  36
  37         private final ExecutionModulesManager executionModulesManager;
  38         private final ExecutionProcess process;
  39         private final ProcessThreadGroup processThreadGroup;
  40
  41         private Set<ExecutionThread> executionThreads = Collections
  42                         .synchronizedSet(new HashSet<ExecutionThread>());
  43
  44         private Boolean hadAnError = false;
  45         private Boolean killed = false;
  46
  47         public ProcessThread(ThreadGroup processesThreadGroup,
  48                         ExecutionModulesManager executionModulesManager,
  49                         ExecutionProcess process) {
  50                 super(processesThreadGroup, "SLC Process #" + process.getUuid());
  51                 this.executionModulesManager = executionModulesManager;
  52                 this.process = process;
  53                 processThreadGroup = new ProcessThreadGroup(executionModulesManager,
  54                                 this);
  55         }
  56
  57         public final void run() {
  58                 log.info("\n##\n## SLC Process #" + process.getUuid()
  59                                 + " STARTED\n##\n");
  60
  61                 String oldStatus = process.getStatus();
  62                 process.setStatus(ExecutionProcess.RUNNING);
  63                 executionModulesManager.dispatchUpdateStatus(process, oldStatus,
  64                                 ExecutionProcess.RUNNING);
  65
  66                 try {
  67                         process();
  68                 } catch (InterruptedException e) {
  69                         die();
  70                         return;
  71                 }
  72
  73                 // waits for all execution threads to complete (in case they were
  74                 // started asynchronously)
  75                 for (ExecutionThread executionThread : executionThreads) {
  76                         if (executionThread.isAlive()) {
  77                                 try {
  78                                         executionThread.join();
  79                                 } catch (InterruptedException e) {
  80                                         die();
  81                                         return;
  82                                 }
  83                         }
  84                 }
  85
  86                 computeFinalStatus();
  87         }
  88
  89         /** Make sure this is called BEFORE all the threads are interrupted. */
  90         private void computeFinalStatus() {
  91                 String oldStatus = process.getStatus();
  92                 // TODO: error management at flow level?
  93                 if (killed)
  94                         process.setStatus(ExecutionProcess.KILLED);
  95                 else if (hadAnError)
  96                         process.setStatus(ExecutionProcess.ERROR);
  97                 else
  98                         process.setStatus(ExecutionProcess.COMPLETED);
  99                 executionModulesManager.dispatchUpdateStatus(process, oldStatus,
 100                                 process.getStatus());
 101                 log.info("\n## SLC Process #" + process.getUuid() + " "
 102                                 + process.getStatus() + "\n");
 103         }
 104
 105         /** Called when being killed */
 106         private synchronized void die() {
 107                 killed = true;
 108                 computeFinalStatus();
 109                 for (ExecutionThread executionThread : executionThreads) {
 110                         try {
 111                                 executionThread.interrupt();
 112                         } catch (Exception e) {
 113                                 log.error("Cannot interrupt " + executionThread);
 114                         }
 115                 }
 116                 processThreadGroup.interrupt();
 117         }
 118
 119         /**
 120          * Implementation specific execution. To be overridden in order to deal with
 121          * custom process types. Default expects an {@link SlcExecution}.
 122          */
 123         @SuppressWarnings("deprecation")
 124         protected void process() throws InterruptedException {
 125                 if (!(process instanceof SlcExecution))
 126                         throw new SlcException("Unsupported process type "
 127                                         + process.getClass());
 128                 SlcExecution slcExecution = (SlcExecution) process;
 129                 List<RealizedFlow> flowsToProcess = new ArrayList<RealizedFlow>();
 130                 flowsToProcess.addAll(slcExecution.getRealizedFlows());
 131
 132                 while (flowsToProcess.size() > 0) {
 133                         RealizedFlow realizedFlow = flowsToProcess.remove(0);
 134                         execute(realizedFlow, true);
 135                 }
 136         }
 137
 138         /** @return the (distinct) thread used for this execution */
 139         protected final void execute(RealizedFlow realizedFlow, Boolean synchronous)
 140                         throws InterruptedException {
 141                 if (killed)
 142                         return;
 143
 144                 ExecutionThread thread = new ExecutionThread(this, realizedFlow);
 145                 executionThreads.add(thread);
 146                 thread.start();
 147
 148                 if (synchronous)
 149                         thread.join();
 150
 151                 return;
 152         }
 153
 154         public void notifyError() {
 155                 hadAnError = true;
 156         }
 157
 158         public synchronized void flowCompleted() {
 159                 // notifyAll();
 160         }
 161
 162         public ExecutionProcess getProcess() {
 163                 return process;
 164         }
 165
 166         public ProcessThreadGroup getProcessThreadGroup() {
 167                 return processThreadGroup;
 168         }
 169
 170         public ExecutionModulesManager getExecutionModulesManager() {
 171                 return executionModulesManager;
 172         }
 173 }