Merge branch 'master' into fix-648

Author: ThomasHickman

appveyor.yml

@@ -31,6 +31,7 @@ install:
   - "python -c \"import struct; print(struct.calcsize('P') * 8)\""
 
 build_script:
+  - "%CMD_IN_ENV% pip install -U setuptools pip"
   - "%CMD_IN_ENV% pip install ."
 
 

cwltool/argparser.py

@@ -28,7 +28,10 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     parser.add_argument("--no-container", action="store_false", default=True,
                         help="Do not execute jobs in a Docker container, even when specified by the CommandLineTool",
                         dest="use_container")
-
+    parser.add_argument("--parallel", action="store_true", default=False,
+                        help="[experimental] Run jobs in parallel. "
+                             "Does not currently keep track of ResourceRequirements like the number of cores"
+                             "or memory and can overload this system")
     parser.add_argument("--preserve-environment", type=Text, action="append",
                         help="Preserve specific environment variable when running CommandLineTools.  May be provided multiple times.",
                         metavar="ENVVAR",
@@ -49,6 +52,27 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                          default=True, help="Do not delete Docker container used by jobs after they exit",
                          dest="rm_container")
 
+    cidgroup = parser.add_argument_group("Options for recording the Docker "
+        "container identifier into a file")
+    cidgroup.add_argument("--record-container-id", action="store_true",
+                          default=False,
+                          help="If enabled, store the Docker container ID into a file. "
+                          "See --cidfile-dir to specify the directory.",
+                          dest="record_container_id")
+
+    cidgroup.add_argument("--cidfile-dir", type=Text,
+                        help="Directory for storing the Docker container ID file. "
+                             "The default is the current directory",
+                        default="",
+                        dest="cidfile_dir")
+
+    cidgroup.add_argument("--cidfile-prefix", type=Text,
+                        help="Specify a prefix to the container ID filename. "
+                             "Final file name will be followed by a timestamp. "
+                             "The default is no prefix.",
+                        default="",
+                        dest="cidfile_prefix")
+
     parser.add_argument("--tmpdir-prefix", type=Text,
                         help="Path prefix for temporary directories",
                         default="tmp")
@@ -120,13 +144,16 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                          default=True, dest="strict")
 
     parser.add_argument("--skip-schemas", action="store_true",
-            help="Skip loading of schemas", default=True, dest="skip_schemas")
+            help="Skip loading of schemas", default=False, dest="skip_schemas")
 
     exgroup = parser.add_mutually_exclusive_group()
     exgroup.add_argument("--verbose", action="store_true", help="Default logging")
     exgroup.add_argument("--quiet", action="store_true", help="Only print warnings and errors.")
     exgroup.add_argument("--debug", action="store_true", help="Print even more logging")
 
+    parser.add_argument("--timestamps", action="store_true", help="Add "
+                        "timestamps to the errors, warnings, and "
+                        "notifications.")
     parser.add_argument("--js-console", action="store_true", help="Enable javascript console output")
     parser.add_argument("--user-space-docker-cmd",
                         help="(Linux/OS X only) Specify a user space docker "
@@ -370,4 +397,4 @@ def generate_parser(toolparser, tool, namemap, records):
         default = inp.get("default", None)
         add_argument(toolparser, name, inptype, records, description, default)
 
-    return toolparser
+    return toolparser

cwltool/executors.py

@@ -0,0 +1,175 @@
+import logging
+import tempfile
+import threading
+
+import os
+from abc import ABCMeta, abstractmethod
+
+from typing import Dict, Text, Any, Tuple, Set, List
+
+from cwltool.builder import Builder
+from cwltool.errors import WorkflowException
+from cwltool.mutation import MutationManager
+from cwltool.job import JobBase
+from cwltool.process import relocateOutputs, cleanIntermediate, Process
+
+
+_logger = logging.getLogger("cwltool")
+
+defaultStreamHandler = logging.StreamHandler()
+_logger.addHandler(defaultStreamHandler)
+_logger.setLevel(logging.INFO)
+
+
+class JobExecutor(object):
+    __metaclass__ = ABCMeta
+
+    def __init__(self):
+        # type: (...) -> None
+        self.final_output = []  # type: List
+        self.final_status = []  # type: List
+        self.output_dirs = set()  # type: Set
+
+    def __call__(self, *args, **kwargs):
+        return self.execute(*args, **kwargs)
+
+    def output_callback(self, out, processStatus):
+        self.final_status.append(processStatus)
+        self.final_output.append(out)
+
+    @abstractmethod
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+        pass
+
+    def execute(self, t,  # type: Process
+                job_order_object,  # type: Dict[Text, Any]
+                logger=_logger,
+                **kwargs  # type: Any
+                ):
+        # type: (...) -> Tuple[Dict[Text, Any], Text]
+
+        if "basedir" not in kwargs:
+            raise WorkflowException("Must provide 'basedir' in kwargs")
+
+        finaloutdir = os.path.abspath(kwargs.get("outdir")) if kwargs.get("outdir") else None
+        kwargs["outdir"] = tempfile.mkdtemp(prefix=kwargs["tmp_outdir_prefix"]) if kwargs.get(
+            "tmp_outdir_prefix") else tempfile.mkdtemp()
+        self.output_dirs.add(kwargs["outdir"])
+        kwargs["mutation_manager"] = MutationManager()
+
+        jobReqs = None
+        if "cwl:requirements" in job_order_object:
+            jobReqs = job_order_object["cwl:requirements"]
+        elif ("cwl:defaults" in t.metadata and "cwl:requirements" in t.metadata["cwl:defaults"]):
+            jobReqs = t.metadata["cwl:defaults"]["cwl:requirements"]
+        if jobReqs:
+            for req in jobReqs:
+                t.requirements.append(req)
+
+        self.run_jobs(t, job_order_object, logger, **kwargs)
+
+        if self.final_output and self.final_output[0] and finaloutdir:
+            self.final_output[0] = relocateOutputs(self.final_output[0], finaloutdir,
+                                                   self.output_dirs, kwargs.get("move_outputs"),
+                                                   kwargs["make_fs_access"](""))
+
+        if kwargs.get("rm_tmpdir"):
+            cleanIntermediate(self.output_dirs)
+
+        if self.final_output and self.final_status:
+            return (self.final_output[0], self.final_status[0])
+        else:
+            return (None, "permanentFail")
+
+
+class SingleJobExecutor(JobExecutor):
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+        jobiter = t.job(job_order_object,
+                        self.output_callback,
+                        **kwargs)
+
+        try:
+            for r in jobiter:
+                if r:
+                    builder = kwargs.get("builder", None)  # type: Builder
+                    if builder is not None:
+                        r.builder = builder
+                    if r.outdir:
+                        self.output_dirs.add(r.outdir)
+                    r.run(**kwargs)
+                else:
+                    logger.error("Workflow cannot make any more progress.")
+                    break
+        except WorkflowException:
+            raise
+        except Exception as e:
+            logger.exception("Got workflow error")
+            raise WorkflowException(Text(e))
+
+
+class MultithreadedJobExecutor(JobExecutor):
+    def __init__(self):
+        super(MultithreadedJobExecutor, self).__init__()
+        self.threads = set()
+        self.exceptions = []
+
+    def run_job(self,
+                job, # type:  JobBase
+                **kwargs  # type:  Any
+                ):
+        # type: (...) -> None
+        def runner():
+            try:
+                job.run(**kwargs)
+            except WorkflowException as e:
+                self.exceptions.append(e)
+            except Exception as e:
+                self.exceptions.append(WorkflowException(Text(e)))
+
+            self.threads.remove(thread)
+
+        thread = threading.Thread(target=runner)
+        thread.daemon = True
+        self.threads.add(thread)
+        thread.start()
+
+    def wait_for_next_completion(self):  # type: () -> None
+        if self.exceptions:
+            raise self.exceptions[0]
+
+    def run_jobs(self,
+                 t,  # type: Process
+                 job_order_object,  # type: Dict[Text, Any]
+                 logger,
+                 **kwargs  # type: Any
+                 ):
+
+        jobiter = t.job(job_order_object, self.output_callback, **kwargs)
+
+        for r in jobiter:
+            if r:
+                builder = kwargs.get("builder", None)  # type: Builder
+                if builder is not None:
+                    r.builder = builder
+                if r.outdir:
+                    self.output_dirs.add(r.outdir)
+                self.run_job(r, **kwargs)
+            else:
+                if len(self.threads):
+                    self.wait_for_next_completion()
+                else:
+                    logger.error("Workflow cannot make any more progress.")
+                    break
+
+        while len(self.threads) > 0:
+            self.wait_for_next_completion()

cwltool/expression.py

@@ -206,7 +206,7 @@ def interpolate(scan, rootvars,
             e = evaluator(scan[w[0] + 1:w[1]], jslib, rootvars, fullJS=fullJS,
                           timeout=timeout, force_docker_pull=force_docker_pull,
                           debug=debug, js_console=js_console)
-            if w[0] == 0 and w[1] == len(scan):
+            if w[0] == 0 and w[1] == len(scan) and len(parts) <= 1:
                 return e
             leaf = json.dumps(e, sort_keys=True)
             if leaf[0] == '"':

cwltool/factory.py

@@ -3,8 +3,9 @@
 from typing import Callable as tCallable
 from typing import Any, Dict, Text, Tuple, Union
 
-from . import load_tool, main, workflow
+from . import load_tool, workflow
 from .argparser import get_default_args
+from .executors import SingleJobExecutor
 from .process import Process
 
 
@@ -36,11 +37,13 @@ class Factory(object):
     def __init__(self,
                  makeTool=workflow.defaultMakeTool,  # type: tCallable[[Any], Process]
                  # should be tCallable[[Dict[Text, Any], Any], Process] ?
-                 executor=main.single_job_executor,  # type: tCallable[...,Tuple[Dict[Text,Any], Text]]
+                 executor=None,  # type: tCallable[...,Tuple[Dict[Text,Any], Text]]
                  **execkwargs  # type: Any
                  ):
         # type: (...) -> None
         self.makeTool = makeTool
+        if executor is None:
+            executor = SingleJobExecutor()
         self.executor = executor
 
         kwargs = get_default_args()

cwltool/job.py

@@ -1,5 +1,7 @@
 from __future__ import absolute_import
+
 import codecs
+import datetime
 import functools
 import io
 import json
@@ -12,25 +14,28 @@
 import sys
 import tempfile
 from io import open
+from threading import Lock
 from typing import (IO, Any, Callable, Dict, Iterable, List, MutableMapping, Text,
-                    Tuple, Union, cast)
+                    Union, cast)
 
 import shellescape
 
-from .utils import copytree_with_merge, docker_windows_path_adjust, onWindows
 from . import docker
 from .builder import Builder
 from .docker_id import docker_vm_id
 from .errors import WorkflowException
 from .pathmapper import PathMapper, ensure_writable
-from .process import (UnsupportedRequirement, empty_subtree, get_feature,
+from .process import (UnsupportedRequirement, get_feature,
                       stageFiles)
 from .utils import bytes2str_in_dicts
+from .utils import copytree_with_merge, docker_windows_path_adjust, onWindows
 
 _logger = logging.getLogger("cwltool")
 
 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
@@ -266,7 +271,8 @@ def _execute(self, runtime, env, rm_tmpdir=True, move_outputs="move"):
         if _logger.isEnabledFor(logging.DEBUG):
             _logger.debug(u"[job %s] %s", self.name, json.dumps(outputs, indent=4))
 
-        self.output_callback(outputs, processStatus)
+        with job_output_lock:
+            self.output_callback(outputs, processStatus)
 
         if self.stagedir and os.path.exists(self.stagedir):
             _logger.debug(u"[job %s] Removing input staging directory %s", self.name, self.stagedir)
@@ -363,8 +369,10 @@ def add_volumes(self, pathmapper, runtime):
                         docker_windows_path_adjust(vol.target)))
 
     def run(self, pull_image=True, rm_container=True,
+            record_container_id=False, cidfile_dir="",
+            cidfile_prefix="",
             rm_tmpdir=True, move_outputs="move", **kwargs):
-        # type: (bool, bool, bool, Text, **Any) -> None
+        # type: (bool, bool, bool, Text, Text, bool, Text, **Any) -> None
 
         (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
 
@@ -463,6 +471,26 @@ def run(self, pull_image=True, rm_container=True,
         # directory." but spec might change to designated temp directory.
         # runtime.append("--env=HOME=/tmp")
         runtime.append(u"--env=HOME=%s" % self.builder.outdir)
+        
+        # add parameters to docker to write a container ID file
+        if record_container_id:
+            if cidfile_dir != "":
+                if not os.path.isdir(cidfile_dir):
+                    _logger.error("--cidfile-dir %s error:\n%s", cidfile_dir,
+                                  cidfile_dir + " is not a directory or "
+                                  "directory doesn't exist, please check it first")
+                    exit(2)
+                if not os.path.exists(cidfile_dir):
+                    _logger.error("--cidfile-dir %s error:\n%s", cidfile_dir,
+                                  "directory doesn't exist, please create it first")
+                    exit(2)
+            else:
+                cidfile_dir = os.getcwd()
+            cidfile_name = datetime.datetime.now().strftime("%Y%m%d%H%M%S-%f")+".cid"
+            if cidfile_prefix != "":
+                cidfile_name = str(cidfile_prefix + "-" + cidfile_name)
+            cidfile_path = os.path.join(cidfile_dir, cidfile_name)
+            runtime.append(u"--cidfile=%s" % cidfile_path)
 
         for t, v in self.environment.items():
             runtime.append(u"--env=%s=%s" % (t, v))