improve

Author: itajaja

cwltool/factory.py

@@ -35,7 +35,7 @@ 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=main.job_executor,  # type: tCallable[...,Tuple[Dict[Text,Any], Text]]
                  **execkwargs  # type: Any
                  ):
         # type: (...) -> None

cwltool/job.py

@@ -10,6 +10,7 @@
 import subprocess
 import sys
 import tempfile
+from threading import Lock
 from io import open
 from typing import (IO, Any, Callable, Dict, Iterable, List, MutableMapping, Text,
                     Tuple, Union, cast)
@@ -30,6 +31,8 @@
 
 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
@@ -265,7 +268,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)

cwltool/main.py

@@ -245,10 +245,10 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     return parser
 
 
-def single_job_executor(t,  # type: Process
-                        job_order_object,  # type: Dict[Text, Any]
-                        **kwargs  # type: Any
-                        ):
+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 = []
@@ -721,7 +721,7 @@ def supportedCWLversions(enable_dev):
 
 def main(argsl=None,  # type: List[str]
          args=None,  # type: argparse.Namespace
-         executor=single_job_executor,  # type: Callable[..., Tuple[Dict[Text, Any], Text]]
+         executor=job_executor,  # 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]