code, type cleanup

Author: mr-c

cwltool/builder.py

@@ -113,7 +113,7 @@ def __init__(self,
                  hints=None,                # type: List[Dict[Text, Any]]
                  timeout=None,              # type: float
                  debug=False,               # type: bool
-                 resources=None,            # type: Dict[Text, int]
+                 resources=None,            # type: Dict[str, int]
                  js_console=False,          # type: bool
                  mutation_manager=None,     # type: Optional[MutationManager]
                  formatgraph=None,          # type: Optional[Graph]
@@ -154,7 +154,7 @@ def __init__(self,
         self.tmpdir = tmpdir
 
         if resources is None:
-            self.resources = {}  # type: Dict[Text, int]
+            self.resources = {}  # type: Dict[str, int]
         else:
             self.resources = resources
 

cwltool/context.py

@@ -1,5 +1,5 @@
 import copy
-import threading
+import threading  # pylint: disable=unused-import
 
 from .utils import DEFAULT_TMP_PREFIX
 from .stdfsaccess import StdFsAccess
@@ -70,6 +70,8 @@ def copy(self):
 class RuntimeContext(ContextBase):
     def __init__(self, kwargs=None):
         # type: (Optional[Dict[str, Any]]) -> None
+        select_resources_callable = Callable[  # pylint: disable=unused-variable
+            [Dict[str, int], RuntimeContext], Dict[str, int]]
         self.user_space_docker_cmd = "" # type: Text
         self.secret_store = None        # type: Optional[SecretStore]
         self.no_read_only = False       # type: bool
@@ -109,7 +111,7 @@ def __init__(self, kwargs=None):
         self.docker_stagedir = ""       # type: Text
         self.js_console = False         # type: bool
         self.job_script_provider = None  # type: Optional[DependenciesConfiguration]
-        self.select_resources = None    # type: Optional[Callable[[Dict[Text, int], RuntimeContext], Dict[Text, int]]]
+        self.select_resources = None    # type: Optional[select_resources_callable]
         self.eval_timeout = 20          # type: float
         self.postScatterEval = None     # type: Optional[Callable[[Dict[Text, Any]], Dict[Text, Any]]]
         self.on_error = "stop"          # type: Text

cwltool/executors.py

@@ -1,3 +1,5 @@
+# -*- coding: utf-8 -*-
+""" Single and multi-threaded executors."""
 import os
 import tempfile
 import threading
@@ -46,33 +48,33 @@ def run_jobs(self,
                  process,           # type: Process
                  job_order_object,  # type: Dict[Text, Any]
                  logger,
-                 runtimeContext     # type: RuntimeContext
+                 runtime_context     # type: RuntimeContext
                 ):  # type: (...) -> None
         """ Execute the jobs for the given Process. """
         pass
 
     def execute(self,
                 process,           # type: Process
                 job_order_object,  # type: Dict[Text, Any]
-                runtimeContext,    # type: RuntimeContext
+                runtime_context,    # type: RuntimeContext
                 logger=_logger,
                ):  # type: (...) -> Tuple[Optional[Dict[Text, Any]], Text]
         """ Execute the process. """
 
-        if not runtimeContext.basedir:
+        if not runtime_context.basedir:
             raise WorkflowException("Must provide 'basedir' in runtimeContext")
 
         finaloutdir = None  # Type: Optional[Text]
-        original_outdir = runtimeContext.outdir
+        original_outdir = runtime_context.outdir
         if isinstance(original_outdir, string_types):
             finaloutdir = os.path.abspath(original_outdir)
-        runtimeContext = runtimeContext.copy()
-        runtimeContext.outdir = tempfile.mkdtemp(
-            prefix=getdefault(runtimeContext.tmp_outdir_prefix, DEFAULT_TMP_PREFIX))
-        self.output_dirs.add(runtimeContext.outdir)
-        runtimeContext.mutation_manager = MutationManager()
-        runtimeContext.toplevel = True
-        runtimeContext.workflow_eval_lock = threading.Condition(threading.RLock())
+        runtime_context = runtime_context.copy()
+        runtime_context.outdir = tempfile.mkdtemp(
+            prefix=getdefault(runtime_context.tmp_outdir_prefix, DEFAULT_TMP_PREFIX))
+        self.output_dirs.add(runtime_context.outdir)
+        runtime_context.mutation_manager = MutationManager()
+        runtime_context.toplevel = True
+        runtime_context.workflow_eval_lock = threading.Condition(threading.RLock())
 
         job_reqs = None
         if "cwl:requirements" in job_order_object:
@@ -84,20 +86,20 @@ def execute(self,
             for req in job_reqs:
                 process.requirements.append(req)
 
-        self.run_jobs(process, job_order_object, logger, runtimeContext)
+        self.run_jobs(process, job_order_object, logger, runtime_context)
 
         if self.final_output and self.final_output[0] and finaloutdir:
             self.final_output[0] = relocateOutputs(
                 self.final_output[0], finaloutdir, self.output_dirs,
-                runtimeContext.move_outputs, runtimeContext.make_fs_access(""),
-                getdefault(runtimeContext.compute_checksum, True))
+                runtime_context.move_outputs, runtime_context.make_fs_access(""),
+                getdefault(runtime_context.compute_checksum, True))
 
-        if runtimeContext.rm_tmpdir:
+        if runtime_context.rm_tmpdir:
             cleanIntermediate(self.output_dirs)
 
         if self.final_output and self.final_status:
 
-            if runtimeContext.research_obj is not None and \
+            if runtime_context.research_obj is not None and \
                     isinstance(process, (JobBase, Process, WorkflowJobStep,
                                          WorkflowJob)) and process.parent_wf:
                 process_run_id = None
@@ -118,45 +120,46 @@ def run_jobs(self,
                  process,           # type: Process
                  job_order_object,  # type: Dict[Text, Any]
                  logger,
-                 runtimeContext     # type: RuntimeContext
+                 runtime_context     # type: RuntimeContext
                 ):  # type: (...) -> None
 
         process_run_id = None  # type: Optional[str]
         reference_locations = {}  # type: Dict[Text,Text]
 
         # define provenance profile for single commandline tool
         if not isinstance(process, Workflow) \
-                and runtimeContext.research_obj is not None:
-            orcid = runtimeContext.orcid
-            full_name = runtimeContext.cwl_full_name
+                and runtime_context.research_obj is not None:
+            orcid = runtime_context.orcid
+            full_name = runtime_context.cwl_full_name
             process.provenance_object = CreateProvProfile(
-                runtimeContext.research_obj, orcid, full_name)
+                runtime_context.research_obj, orcid, full_name)
             process.parent_wf = process.provenance_object
-        jobiter = process.job(job_order_object, self.output_callback, runtimeContext)
+        jobiter = process.job(job_order_object, self.output_callback,
+                              runtime_context)
 
         try:
             for job in jobiter:
                 if job:
-                    if runtimeContext.builder is not None:
-                        job.builder = runtimeContext.builder
+                    if runtime_context.builder is not None:
+                        job.builder = runtime_context.builder
                     if job.outdir:
                         self.output_dirs.add(job.outdir)
-                    if runtimeContext.research_obj is not None:
+                    if runtime_context.research_obj is not None:
                         if not isinstance(process, Workflow):
-                            runtimeContext.prov_obj = process.provenance_object
+                            runtime_context.prov_obj = process.provenance_object
                         else:
-                            runtimeContext.prov_obj = job.prov_obj
-                        assert runtimeContext.prov_obj
+                            runtime_context.prov_obj = job.prov_obj
+                        assert runtime_context.prov_obj
                         process_run_id, reference_locations = \
-                                runtimeContext.prov_obj.evaluate(
-                                        process, job, job_order_object,
-                                        runtimeContext.make_fs_access,
-                                        runtimeContext)
-                        runtimeContext = runtimeContext.copy()
-                        runtimeContext.process_run_id = process_run_id
-                        runtimeContext.reference_locations = \
+                                runtime_context.prov_obj.evaluate(
+                                    process, job, job_order_object,
+                                    runtime_context.make_fs_access,
+                                    runtime_context)
+                        runtime_context = runtime_context.copy()
+                        runtime_context.process_run_id = process_run_id
+                        runtime_context.reference_locations = \
                             reference_locations
-                    job.run(runtimeContext)
+                    job.run(runtime_context)
                 else:
                     logger.error("Workflow cannot make any more progress.")
                     break
@@ -182,20 +185,24 @@ def __init__(self):  # type: () -> None
         self.exceptions = []  # type: List[WorkflowException]
         self.pending_jobs = []  # type: List[JobBase]
 
-        self.max_ram = psutil.virtual_memory().total / 2**20
+        self.max_ram = psutil.virtual_memory().available / 2**20
         self.max_cores = psutil.cpu_count()
         self.allocated_ram = 0
         self.allocated_cores = 0
 
-    def select_resources(self, request, runtimeContext):
-        result = {}
+    def select_resources(self, request, runtime_context):  # pylint: disable=unused-argument
+        # type: (Dict[str, int], RuntimeContext) -> Dict[str, int]
+        """ Naïve check for available cpu cores and memory. """
+        result = {}  # type: Dict[str, int]
         maxrsc = {
             "cores": self.max_cores,
             "ram": self.max_ram
         }
         for rsc in ("cores", "ram"):
             if request[rsc+"Min"] > maxrsc[rsc]:
-                raise WorkflowException("Requested at least %d %s but only %d available", request[rsc+"Min"], rsc, maxrsc[rsc])
+                raise WorkflowException(
+                    "Requested at least %d %s but only %d available" %
+                    (request[rsc+"Min"], rsc, maxrsc[rsc]))
             if request[rsc+"Max"] < maxrsc[rsc]:
                 result[rsc] = request[rsc+"Max"]
             else:
@@ -204,8 +211,8 @@ def select_resources(self, request, runtimeContext):
         return result
 
     def run_job(self,
-                job,      # type: JobBase
-                runtimeContext  # type: RuntimeContext
+                job,             # type: JobBase
+                runtime_context  # type: RuntimeContext
                ):  # type: (...) -> None
         """ Execute a single Job in a seperate thread. """
 
@@ -216,28 +223,28 @@ def run_job(self,
             job = self.pending_jobs[0]
             if isinstance(job, JobBase):
                 if ((self.allocated_ram + job.builder.resources["ram"]) > self.max_ram or
-                    (self.allocated_cores + job.builder.resources["cores"]) > self.max_cores):
+                        (self.allocated_cores + job.builder.resources["cores"]) > self.max_cores):
                     return
 
             self.pending_jobs.pop(0)
 
             def runner():
                 """ Job running thread. """
                 try:
-                    job.run(runtimeContext)
+                    job.run(runtime_context)
                 except WorkflowException as err:
                     _logger.exception("Got workflow error")
                     self.exceptions.append(err)
-                except Exception as err:
+                except Exception as err:  # pylint: disable=broad-except
                     _logger.exception("Got workflow error")
                     self.exceptions.append(WorkflowException(Text(err)))
                 finally:
-                    with runtimeContext.workflow_eval_lock:
+                    with runtime_context.workflow_eval_lock:
                         self.threads.remove(thread)
                         if isinstance(job, JobBase):
                             self.allocated_ram -= job.builder.resources["ram"]
                             self.allocated_cores -= job.builder.resources["cores"]
-                        runtimeContext.workflow_eval_lock.notifyAll()
+                        runtime_context.workflow_eval_lock.notifyAll()
 
             thread = threading.Thread(target=runner)
             thread.daemon = True
@@ -258,30 +265,32 @@ def run_jobs(self,
                  process,           # type: Process
                  job_order_object,  # type: Dict[Text, Any]
                  logger,
-                 runtimeContext     # type: RuntimeContext
+                 runtime_context     # type: RuntimeContext
                 ):  # type: (...) -> None
 
-        jobiter = process.job(job_order_object, self.output_callback, runtimeContext)
+        jobiter = process.job(job_order_object, self.output_callback,
+                              runtime_context)
 
-        if runtimeContext.workflow_eval_lock is None:
-            raise WorkflowException("runtimeContext.workflow_eval_lock must not be None")
+        if runtime_context.workflow_eval_lock is None:
+            raise WorkflowException(
+                "runtimeContext.workflow_eval_lock must not be None")
 
-        runtimeContext.workflow_eval_lock.acquire()
+        runtime_context.workflow_eval_lock.acquire()
         for job in jobiter:
             if job is not None:
-                if runtimeContext.builder is not None:
-                    job.builder = runtimeContext.builder
+                if runtime_context.builder is not None:
+                    job.builder = runtime_context.builder
                 if job.outdir:
                     self.output_dirs.add(job.outdir)
 
-            self.run_job(job, runtimeContext)
+            self.run_job(job, runtime_context)
 
             if job is None:
                 if self.threads:
-                    self.wait_for_next_completion(runtimeContext)
+                    self.wait_for_next_completion(runtime_context)
                 else:
                     logger.error("Workflow cannot make any more progress.")
                     break
 
         while self.threads:
-            self.wait_for_next_completion(runtimeContext)
+            self.wait_for_next_completion(runtime_context)

cwltool/expression.py

@@ -253,7 +253,7 @@ def do_eval(ex,                       # type: Union[Text, Dict]
             requirements,             # type: List[Dict[Text, Any]]
             outdir,                   # type: Optional[Text]
             tmpdir,                   # type: Optional[Text]
-            resources,                # type: Dict[Text, int]
+            resources,                # type: Dict[str, int]
             context=None,             # type: Any
             timeout=None,             # type: float
             force_docker_pull=False,  # type: bool
@@ -262,7 +262,7 @@ def do_eval(ex,                       # type: Union[Text, Dict]
             strip_whitespace=True     # type: bool
            ):  # type: (...) -> Any
 
-    runtime = copy.copy(resources)  # type: Dict[Text, Any]
+    runtime = copy.copy(resources)  # type: Dict[str, Any]
     runtime["tmpdir"] = docker_windows_path_adjust(tmpdir)
     runtime["outdir"] = docker_windows_path_adjust(outdir)
 

cwltool/process.py

@@ -571,20 +571,6 @@ def __init__(self,
 
     def _init_job(self, joborder, runtimeContext):
         # type: (Dict[Text, Text], RuntimeContext) -> Builder
-        """
-        kwargs:
-
-        use_container: do/don't use Docker when DockerRequirement hint provided
-        make_fs_access: make an FsAccess() object with given basedir
-        docker_outdir: output directory inside docker for this job
-        docker_tmpdir: tmpdir inside docker for this job
-        docker_stagedir: stagedir inside docker for this job
-        outdir: outdir on host for this job
-        tmpdir: tmpdir on host for this job
-        stagedir: stagedir on host for this job
-        select_resources: callback to select compute resources
-        tmp_outdir_prefix: Path prefix for intermediate output directories
-        """
 
         job = cast(Dict[Text, Union[Dict[Text, Any], List,
                                     Text]], copy.deepcopy(joborder))
@@ -717,7 +703,7 @@ def _init_job(self, joborder, runtimeContext):
         return builder
 
     def evalResources(self, builder, runtimeContext):
-        # type: (Builder, RuntimeContext) -> Dict[Text, int]
+        # type: (Builder, RuntimeContext) -> Dict[str, int]
         resourceReq, _ = self.get_requirement("ResourceRequirement")
         if resourceReq is None:
             resourceReq = {}
@@ -730,7 +716,7 @@ def evalResources(self, builder, runtimeContext):
             "tmpdirMax": 1024,
             "outdirMin": 1024,
             "outdirMax": 1024
-        }  # type: Dict[Text, int]
+        }  # type: Dict[str, int]
         for a in ("cores", "ram", "tmpdir", "outdir"):
             mn = None
             mx = None