Correct locking for multiprocess executor.

Author: Peter Amstutz

cwltool/context.py

@@ -1,4 +1,6 @@
 import copy
+import threading
+
 from .utils import DEFAULT_TMP_PREFIX
 from .stdfsaccess import StdFsAccess
 from typing import (Any, Callable, Dict,  # pylint: disable=unused-import
@@ -108,6 +110,8 @@ def __init__(self, kwargs=None):
         self.cidfile_dir = None
         self.cidfile_prefix = None
 
+        self.workflow_eval_lock = None  # type: threading.Condition
+
         super(RuntimeContext, self).__init__(kwargs)
 
 

cwltool/docker.py

@@ -339,4 +339,7 @@ def create_runtime(self, env, runtimeContext):
         for t, v in self.environment.items():
             runtime.append(u"--env=%s=%s" % (t, v))
 
+        runtime.append("--cpu-count=%d" % self.builder.resources.cores)
+        runtime.append("--memory=%dm" % self.builder.resources.ram)
+
         return runtime

cwltool/executors.py

@@ -69,6 +69,7 @@ def execute(self,
         self.output_dirs.add(runtimeContext.outdir)
         runtimeContext.mutation_manager = MutationManager()
         runtimeContext.toplevel = True
+        runtimeContext.workflow_eval_lock = threading.Condition(threading.RLock())
 
         job_reqs = None
         if "cwl:requirements" in job_order_object:
@@ -148,15 +149,19 @@ def runner():
                 self.exceptions.append(err)
             except Exception as err:
                 self.exceptions.append(WorkflowException(Text(err)))
-            self.threads.remove(thread)
+            finally:
+                with runtimeContext.workflow_eval_lock:
+                    self.threads.remove(thread)
+                    runtimeContext.notifyAll()
 
         thread = threading.Thread(target=runner)
         thread.daemon = True
         self.threads.add(thread)
         thread.start()
 
-    def wait_for_next_completion(self):  # type: () -> None
-        """ Check for exceptions while waiting for the jobs to finish. """
+    def wait_for_next_completion(self, runtimeContext):  # type: () -> None
+        """ Wait for jobs to finish. """
+        runtimeContext.workflow_eval_lock.wait()
         if self.exceptions:
             raise self.exceptions[0]
 
@@ -169,19 +174,20 @@ def run_jobs(self,
 
         jobiter = process.job(job_order_object, self.output_callback, runtimeContext)
 
+        runtimeContext.workflow_eval_lock.acquire()
         for job in jobiter:
-            if job:
+            if job is not None:
                 if runtimeContext.builder is not None:
                     job.builder = runtimeContext.builder
                 if job.outdir:
                     self.output_dirs.add(job.outdir)
                 self.run_job(job, runtimeContext)
             else:
                 if self.threads:
-                    self.wait_for_next_completion()
+                    self.wait_for_next_completion(runtimeContext)
                 else:
                     logger.error("Workflow cannot make any more progress.")
                     break
 
         while self.threads:
-            self.wait_for_next_completion()
+            self.wait_for_next_completion(runtimeContext)

cwltool/job.py

@@ -33,8 +33,6 @@
 
 needs_shell_quoting_re = re.compile(r"""(^$|[\s|&;()<>\'"$@])""")
 
-job_output_lock = Lock()
-
 FORCE_SHELLED_POPEN = os.getenv("CWLTOOL_FORCE_SHELL_POPEN", "0") == "1"
 
 SHELL_COMMAND_TEMPLATE = """#!/bin/bash
@@ -326,7 +324,7 @@ def _execute(self,
                                 host_outdir, p.target[len(container_outdir)+1:])
                         os.remove(host_outdir_tgt)
 
-        with job_output_lock:
+        with runtimeContext.workflow_eval_lock:
             self.output_callback(outputs, processStatus)
 
         if self.stagedir and os.path.exists(self.stagedir):