run: provide command-line override of concurrency limit (max_workers)

WDL/CLI.py

@@ -15,7 +15,7 @@
 import docker
 from shlex import quote as shellquote
 from datetime import datetime
-from argparse import ArgumentParser, Action
+from argparse import ArgumentParser, Action, SUPPRESS
 import pkg_resources
 from . import *
 from ._util import (
@@ -299,12 +299,18 @@ def fill_run_subparser(subparsers):
         dest="input_file",
         help="file with Cromwell-style input JSON; command-line inputs will be merged in",
     )
+    run_parser.add_argument(
+        "-t",
+        "--task",
+        metavar="TASK_NAME",
+        help="name of task to run (for WDL documents with multiple tasks & no workflow)",
+    )
     run_parser.add_argument(
         "-j",
         "--json",
         dest="json_only",
         action="store_true",
-        help="just print Cromwell-style input JSON to standard output, then exit",
+        help=SUPPRESS,  # "just print Cromwell-style input JSON to standard output, then exit",
     )
     run_parser.add_argument(
         "-d",
@@ -316,13 +322,15 @@ def fill_run_subparser(subparsers):
     run_parser.add_argument(
         "--copy-input-files",
         action="store_true",
-        help="copy input files for each task (for compatibility with commands assuming write access to them)",
+        help="copy input files for each task (for compatibility with mv/rm/write commands)",
     )
     run_parser.add_argument(
-        "-t",
-        "--task",
-        metavar="TASK_NAME",
-        help="name of task to run (for WDL documents with multiple tasks & no workflow)",
+        "-@",
+        metavar="N",
+        dest="max_workers",
+        type=int,
+        default=None,
+        help="maximum concurrent tasks; defaults to # of processors (limit effectively lower when tasks require multiple processors)",
     )
     run_parser.add_argument(
         "-v",
@@ -346,6 +354,7 @@ def runner(
     rundir=None,
     path=None,
     copy_input_files=False,
+    max_workers=None,
     **kwargs,
 ):
     # load WDL document
@@ -385,7 +394,11 @@ def runner(
             runtime.run_local_task if isinstance(target, Task) else runtime.run_local_workflow
         )
         rundir, output_env = entrypoint(
-            target, input_env, run_dir=rundir, copy_input_files=copy_input_files
+            target,
+            input_env,
+            run_dir=rundir,
+            copy_input_files=copy_input_files,
+            max_workers=max_workers,
         )
     except Exception as exn:
         if isinstance(exn, runtime.task.TaskFailure):

WDL/runtime/task.py

@@ -352,6 +352,7 @@ def run_local_task(
     run_dir: Optional[str] = None,
     copy_input_files: bool = False,
     logger_prefix: str = "wdl:",
+    max_workers: Optional[int] = None,  # unused
 ) -> Tuple[str, Env.Bindings[Value.Base]]:
     """
     Run a task locally.

WDL/runtime/workflow.py

@@ -571,6 +571,7 @@ def run_local_workflow(
     run_dir: Optional[str] = None,
     copy_input_files: bool = False,
     logger_prefix: str = "wdl:",
+    max_workers: Optional[int] = None,
     _test_pickle: bool = False,
 ) -> Tuple[str, Env.Bindings[Value.Base]]:
     """
@@ -584,7 +585,8 @@ def run_local_workflow(
                     exist; if it does, a timestamp-based subdirectory is created and used (defaults
                     to current working directory)
     """
-    max_workers = multiprocessing.cpu_count()
+    if max_workers is None:
+        max_workers = multiprocessing.cpu_count()
     thread_pool = futures.ThreadPoolExecutor(max_workers=max_workers)
     future_task_map = {}
 
@@ -630,6 +632,7 @@ def run_local_workflow(
                             run_id=next_call.id,
                             run_dir=os.path.join(run_dir, next_call.id),
                             copy_input_files=copy_input_files,
+                            max_workers=max_workers,
                             logger_prefix=(logger_id + ":"),
                         )
                         future_task_map[future] = next_call.id