add lock for multi-threaded directory creation (#1040)

Author: ionox0

cwltool/command_line_tool.py

@@ -95,7 +95,10 @@ def __init__(self,
             self.script = script
             self.prov_obj = None  # type: Optional[ProvenanceProfile]
 
-        def run(self, runtimeContext):  # type: (RuntimeContext) -> None
+        def run(self,
+                runtimeContext,   # type: RuntimeContext
+                tmpdir_lock=None  # type: threading.Lock
+               ):  # type: (...) -> None
             try:
                 normalizeFilesDirs(self.builder.job)
                 ev = self.builder.do_eval(self.script)

cwltool/executors.py

@@ -4,6 +4,7 @@
 import os
 import tempfile
 import threading
+from threading import Lock
 from abc import ABCMeta, abstractmethod
 from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union
 
@@ -25,6 +26,8 @@
 from .utils import DEFAULT_TMP_PREFIX
 from .workflow import Workflow, WorkflowJob, WorkflowJobStep
 
+TMPDIR_LOCK = Lock()
+
 
 class JobExecutor(with_metaclass(ABCMeta, object)):
     """ Abstract base job executor. """
@@ -229,10 +232,10 @@ def select_resources(self, request, runtime_context):  # pylint: disable=unused-
 
         return result
 
-    def _runner(self, job, runtime_context):
+    def _runner(self, job, runtime_context, TMPDIR_LOCK):
         """ Job running thread. """
         try:
-            job.run(runtime_context)
+            job.run(runtime_context, TMPDIR_LOCK)
         except WorkflowException as err:
             _logger.exception("Got workflow error")
             self.exceptions.append(err)
@@ -293,7 +296,7 @@ def run_job(self,
                         n += 1
                         continue
 
-                thread = threading.Thread(target=self._runner, args=(job, runtime_context))
+                thread = threading.Thread(target=self._runner, args=(job, runtime_context, TMPDIR_LOCK))
                 thread.daemon = True
                 self.threads.add(thread)
                 if isinstance(job, JobBase):

cwltool/job.py

@@ -4,6 +4,7 @@
 import functools
 import itertools
 import logging
+import threading
 import os
 import re
 import shutil
@@ -207,7 +208,8 @@ def __repr__(self):
 
     @abstractmethod
     def run(self,
-            runtimeContext  # type: RuntimeContext
+            runtimeContext,   # type: RuntimeContext
+            tmpdir_lock=None  # type: threading.Lock
             ):  # type: (...) -> None
         pass
 
@@ -421,11 +423,18 @@ def get_tree_mem_usage(memory_usage):
 
 class CommandLineJob(JobBase):
     def run(self,
-            runtimeContext  # type: RuntimeContext
+            runtimeContext,         # type: RuntimeContext
+            tmpdir_lock=None        # type: threading.Lock
             ):  # type: (...) -> None
 
-        if not os.path.exists(self.tmpdir):
-            os.makedirs(self.tmpdir)
+        if tmpdir_lock:
+            with tmpdir_lock:
+                if not os.path.exists(self.tmpdir):
+                    os.makedirs(self.tmpdir)
+        else:
+            if not os.path.exists(self.tmpdir):
+                os.makedirs(self.tmpdir)
+
         self._setup(runtimeContext)
 
         env = self.environment
@@ -587,9 +596,18 @@ def add_volumes(self,
                 pathmapper.update(
                     key, new_path, vol.target, vol.type, vol.staged)
 
-    def run(self, runtimeContext):  # type: (RuntimeContext) -> None
-        if not os.path.exists(self.tmpdir):
-            os.makedirs(self.tmpdir)
+    def run(self,
+            runtimeContext,   # type: RuntimeContext
+            tmpdir_lock=None  # type: threading.Lock
+            ):  # type: (...) -> None
+        if tmpdir_lock:
+            with tmpdir_lock:
+                if not os.path.exists(self.tmpdir):
+                    os.makedirs(self.tmpdir)
+        else:
+            if not os.path.exists(self.tmpdir):
+                os.makedirs(self.tmpdir)
+
         (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
         self.prov_obj = runtimeContext.prov_obj
         img_id = None

cwltool/workflow.py

@@ -412,7 +412,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
             step.completed = True
 
 
-    def run(self, runtimeContext):
+    def run(self, runtimeContext, tmpdir_lock=None):
         '''
         logs the start of each workflow
         '''