Refactor executors & use OOP approach to executors

* add an option for choosing executors: sequential or multithreaded
* use class-based approach to executors and move them to a separate file

Author: Anton Khodak

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
+
+from cwltool.builder import Builder
+from cwltool.errors import WorkflowException
+from cwltool.mutation import MutationManager
+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):
+        self.final_output = []
+        self.final_status = []
+        self.output_dirs = 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
+                ):
+
+        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.fetch_iter_lock = threading.Lock()
+        self.threads = set()
+        self.exceptions = []
+
+    def run_job(self, job, **kwargs):
+        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)
+
+            if self.fetch_iter_lock.locked():
+                self.fetch_iter_lock.release()
+
+        thread = threading.Thread(target=runner)
+        thread.daemon = True
+        self.threads.add(thread)
+        thread.start()
+
+    def wait_for_next_completion(self):
+        self.fetch_iter_lock.acquire()
+        self.fetch_iter_lock.acquire()
+        self.fetch_iter_lock.release()
+        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/factory.py

@@ -3,7 +3,8 @@
 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 .executors import SingleJobExecutor
 from .process import Process
 
 
@@ -35,7 +36,7 @@ class Factory(object):
     def __init__(self,
                  makeTool=workflow.defaultMakeTool,  # type: tCallable[[Any], Process]
                  # should be tCallable[[Dict[Text, Any], Any], Process] ?
-                 executor=main.job_executor,  # type: tCallable[...,Tuple[Dict[Text,Any], Text]]
+                 executor=SingleJobExecutor(),  # type: tCallable[...,Tuple[Dict[Text,Any], Text]]
                  **execkwargs  # type: Any
                  ):
         # type: (...) -> None

cwltool/main.py

@@ -28,6 +28,7 @@
 from .builder import Builder
 from .cwlrdf import printdot, printrdf
 from .errors import UnsupportedRequirement, WorkflowException
+from .executors import SingleJobExecutor, MultithreadedJobExecutor
 from .load_tool import (FetcherConstructorType, resolve_tool_uri,
         fetch_document, make_tool, validate_document, jobloaderctx,
         resolve_overrides, load_overrides)
@@ -64,7 +65,8 @@ 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("--executor", type=Text, choices={"single", "parallel"}, default="single",
+                        help="Workflow executor type: sequential/multithreaded. Default: sequential")
     parser.add_argument("--preserve-environment", type=Text, action="append",
                         help="Preserve specific environment variable when running CommandLineTools.  May be provided multiple times.",
                         metavar="ENVVAR",
@@ -253,100 +255,6 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     return parser
 
 
-def job_executor(t,  # type: Process
-                 job_order_object,  # type: Dict[Text, Any]
-                 **kwargs  # type: Any
-                 ):
-    # type: (...) -> Tuple[Dict[Text, Any], Text]
-    final_output = []
-    final_status = []
-
-    def output_callback(out, processStatus):
-        final_status.append(processStatus)
-        final_output.append(out)
-
-    if "basedir" not in kwargs:
-        raise WorkflowException("Must provide 'basedir' in kwargs")
-
-    output_dirs = set()
-    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()
-    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)
-
-    fetch_iter_lock = threading.Lock()
-    threads = set()
-    exceptions = []
-
-    def run_job(job):
-        def runner():
-            try:
-                job.run(**kwargs)
-            except WorkflowException as e:
-                exceptions.append(e)
-            except Exception as e:
-                exceptions.append(WorkflowException(Text(e)))
-
-            threads.remove(thread)
-
-            if fetch_iter_lock.locked():
-                fetch_iter_lock.release()
-
-        thread = threading.Thread(target=runner)
-        thread.daemon = True
-        threads.add(thread)
-        thread.start()
-
-    def wait_for_next_completion():
-        fetch_iter_lock.acquire()
-        fetch_iter_lock.acquire()
-        fetch_iter_lock.release()
-        if exceptions:
-            raise exceptions[0]
-
-    jobiter = t.job(job_order_object, 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:
-                output_dirs.add(r.outdir)
-            run_job(r)
-        else:
-            if len(threads):
-                wait_for_next_completion()
-            else:
-                _logger.error("Workflow cannot make any more progress.")
-                break
-
-    while len(threads) > 0:
-        wait_for_next_completion()
-
-    if final_output and final_output[0] and finaloutdir:
-        final_output[0] = relocateOutputs(final_output[0], finaloutdir,
-                                          output_dirs, kwargs.get("move_outputs"),
-                                          kwargs["make_fs_access"](""))
-
-    if kwargs.get("rm_tmpdir"):
-        cleanIntermediate(output_dirs)
-
-    if final_output and final_status:
-        return (final_output[0], final_status[0])
-    else:
-        return (None, "permanentFail")
-
 
 class FSAction(argparse.Action):
     objclass = None  # type: Text
@@ -756,7 +664,7 @@ def supportedCWLversions(enable_dev):
 
 def main(argsl=None,  # type: List[str]
          args=None,  # type: argparse.Namespace
-         executor=job_executor,  # type: Callable[..., Tuple[Dict[Text, Any], Text]]
+         executor=None,  # type: Callable[..., Tuple[Dict[Text, Any], Text]]
          makeTool=workflow.defaultMakeTool,  # type: Callable[..., Process]
          selectResources=None,  # type: Callable[[Dict[Text, int]], Dict[Text, int]]
          stdin=sys.stdin,  # type: IO[Any]
@@ -991,6 +899,17 @@ def main(argsl=None,  # type: List[str]
         except SystemExit as e:
             return e.code
 
+        if not executor:
+            if args.executor == "single":
+                 executor = SingleJobExecutor()
+            elif args.executor == "parallel":
+                executor = MultithreadedJobExecutor()
+            else:
+                _logger.error("Unknow type of executor: {0}".format(args.executor))
+                arg_parser().print_help()
+                return 1
+
+
         if isinstance(job_order_object, int):
             return job_order_object
 
@@ -999,6 +918,7 @@ def main(argsl=None,  # type: List[str]
             del args.workflow
             del args.job_order
             (out, status) = executor(tool, job_order_object,
+                                     logger=_logger,
                                      makeTool=makeTool,
                                      select_resources=selectResources,
                                      make_fs_access=make_fs_access,