Merge branch 'master' into workflowstep_valuefrom_dir_listing

Author: mr-c

cwltool/command_line_tool.py

@@ -9,6 +9,7 @@
 import re
 import shutil
 import tempfile
+import threading
 from functools import cmp_to_key, partial
 from typing import (Any, Callable, Dict,  # pylint: disable=unused-import
                     Generator, List, Optional, Set, Text, Type, TYPE_CHECKING,
@@ -49,8 +50,8 @@
 ACCEPTLIST_EN_STRICT_RE = re.compile(r"^[a-zA-Z0-9._+-]+$")
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")  # Accept anything
 ACCEPTLIST_RE = ACCEPTLIST_EN_STRICT_RE
-DEFAULT_CONTAINER_MSG =\
-"""We are on Microsoft Windows and not all components of this CWL description have a
+DEFAULT_CONTAINER_MSG = """
+We are on Microsoft Windows and not all components of this CWL description have a
 container specified. This means that these steps will be executed in the default container,
 which is %s.
 
@@ -295,10 +296,10 @@ def job(self,
                                                  separateDirs=False)
             _check_adjust = partial(check_adjust, cachebuilder)
             visit_class([cachebuilder.files, cachebuilder.bindings],
-                       ("File", "Directory"), _check_adjust)
+                        ("File", "Directory"), _check_adjust)
 
             cmdline = flatten(list(map(cachebuilder.generate_arg, cachebuilder.bindings)))
-            (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
+            (docker_req, _) = self.get_requirement("DockerRequirement")
             if docker_req and runtimeContext.use_container:
                 dockerimg = docker_req.get("dockerImageId") or docker_req.get("dockerPull")
             elif runtimeContext.default_container is not None and runtimeContext.use_container:
@@ -311,19 +312,23 @@ def job(self,
                 # not really run using docker, just for hashing purposes
             keydict = {u"cmdline": cmdline}
 
-            if "stdout" in self.tool:
-                keydict["stdout"] = self.tool["stdout"]
-            for location, f in cachebuilder.pathmapper.items():
-                if f.type == "File":
-                    checksum = next((e['checksum'] for e in cachebuilder.files
-                            if 'location' in e and e['location'] == location
-                            and 'checksum' in e
-                            and e['checksum'] != 'sha1$hash'), None)
-                    st = os.stat(f.resolved)
+            for shortcut in ["stdout", "stderr"]:  # later, add "stdin"
+                if shortcut in self.tool:
+                    keydict[shortcut] = self.tool[shortcut]
+
+            for location, fobj in cachebuilder.pathmapper.items():
+                if fobj.type == "File":
+                    checksum = next(
+                        (e['checksum'] for e in cachebuilder.files
+                         if 'location' in e and e['location'] == location
+                         and 'checksum' in e
+                         and e['checksum'] != 'sha1$hash'), None)
+                    fobj_stat = os.stat(fobj.resolved)
                     if checksum:
-                        keydict[f.resolved] = [st.st_size, checksum]
+                        keydict[fobj.resolved] = [fobj_stat.st_size, checksum]
                     else:
-                        keydict[f.resolved] = [st.st_size, int(st.st_mtime * 1000)]
+                        keydict[fobj.resolved] = [fobj_stat.st_size,
+                                                  int(fobj_stat.st_mtime * 1000)]
 
             interesting = {"DockerRequirement",
                            "EnvVarRequirement",
@@ -342,7 +347,8 @@ def job(self,
                           keydictstr, cachekey)
 
             jobcache = os.path.join(runtimeContext.cachedir, cachekey)
-            jobcachepending = jobcache + ".pending"
+            jobcachepending = "{}.{}.pending".format(
+                jobcache, threading.current_thread().ident)
 
             if os.path.isdir(jobcache) and not os.path.isfile(jobcachepending):
                 if docker_req and runtimeContext.use_container:

cwltool/job.py

@@ -10,7 +10,7 @@
 import tempfile
 import uuid
 import datetime
-from threading import Lock, Timer
+from threading import Timer
 from abc import ABCMeta, abstractmethod
 from io import IOBase, open  # pylint: disable=redefined-builtin
 from typing import (IO, Any, AnyStr, Callable,  # pylint: disable=unused-import
@@ -196,10 +196,11 @@ def _setup(self, runtimeContext):  # type: (RuntimeContext) -> None
             self.generatemapper = self.make_path_mapper(
                 cast(List[Any], self.generatefiles["listing"]),
                 self.builder.outdir, runtimeContext, False)
-            _logger.debug(u"[job %s] initial work dir %s", self.name,
-                          json_dumps({p: self.generatemapper.mapper(p)
-                                      for p in self.generatemapper.files()},
-                                     indent=4))
+            if _logger.isEnabledFor(logging.DEBUG):
+                _logger.debug(
+                    u"[job %s] initial work dir %s", self.name,
+                    json_dumps({p: self.generatemapper.mapper(p)
+                                for p in self.generatemapper.files()}, indent=4))
 
     def _execute(self,
                  runtime,                # type: List[Text]
@@ -250,12 +251,12 @@ def _execute(self,
             stdout_path = None
             if self.stdout:
                 absout = os.path.join(self.outdir, self.stdout)
-                dn = os.path.dirname(absout)
-                if dn and not os.path.exists(dn):
-                    os.makedirs(dn)
+                dnout = os.path.dirname(absout)
+                if dnout and not os.path.exists(dnout):
+                    os.makedirs(dnout)
                 stdout_path = absout
 
-            commands = [Text(x) for x in (runtime + self.command_line)]
+            commands = [Text(x) for x in runtime + self.command_line]
             if runtimeContext.secret_store:
                 commands = runtimeContext.secret_store.retrieve(commands)
                 env = runtimeContext.secret_store.retrieve(env)
@@ -308,8 +309,8 @@ def _execute(self,
             else:
                 _logger.exception("Exception while running job")
             processStatus = "permanentFail"
-        except WorkflowException as e:
-            _logger.error(u"[job %s] Job error:\n%s" % (self.name, e))
+        except WorkflowException as err:
+            _logger.error(u"[job %s] Job error:\n%s", self.name, err)
             processStatus = "permanentFail"
         except Exception as e:
             _logger.exception("Exception while running job")

cwltool/workflow.py

@@ -254,7 +254,8 @@ def do_output_callback(self, final_output_callback):
                 datetime.datetime.now())
             self.prov_obj.finalize_prov_profile(str(self.name))
         _logger.info(u"[%s] completed %s", self.name, self.processStatus)
-        _logger.debug(u"[%s] %s", self.name, json_dumps(wo, indent=4))
+        if _logger.isEnabledFor(logging.DEBUG):
+            _logger.debug(u"[%s] %s", self.name, json_dumps(wo, indent=4))
 
         self.did_callback = True
 
@@ -417,8 +418,8 @@ def job(self,
         self.state = {}
         self.processStatus = "success"
 
-        _logger.debug(u"[%s] %s", self.name, json_dumps(joborder,
-                                                        indent=4))
+        if _logger.isEnabledFor(logging.DEBUG):
+            _logger.debug(u"[%s] %s", self.name, json_dumps(joborder, indent=4))
 
         runtimeContext = runtimeContext.copy()
         runtimeContext.outdir = None