27707168593

Committed 17 Jun 2026 05:23PM UTC coverage: 95.701% (-0.1%) from 95.806%

Build # 27707168593

Build Type

Pull #887

github

Committed by

web-flow

Commit Message

Merge a648fe80d into 24eb628b0

Pull Request Pull Request #887: [WDL 1.2] narrow FileCoercion lint criteria

Coverage Stats

51 of 61 new or added lines in 2 files covered. (83.61%)

2 existing lines in 2 files now uncovered.

8704 of 9095 relevant lines covered (95.7%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.09

/WDL/runtime/workflow.py

"""
Workflow runner building blocks & local driver

--------
Overview
--------

Workflow execution proceeds according to the AST's ``WorkflowNode`` graph, in which each Decl,
Call, Scatter, Conditional, and (implicit) Gather operation has its own node which advertises its
dependencies on the other nodes.

Abstractly, we plan to "visit" each node after visiting all of its dependencies. The node's type
prescribes some job to do upon visitation, such as evaluating a Decl's WDL expression, running a
task on some inputs, or scheduling additional jobs to scatter over an array. Named WDL values
(``WDL.Env.Bindings[WDL.Value.Base]``) are transmitted along each dependency edge, and WDL
expressions in each node are evaluated in the environment formed from the union of the node's
incoming dependency edges.

Scatter sections contain a body, which provides a template for the job subgraph to be scheduled for
each element of the runtime-evaluated scatter array. They also contain template Gather nodes, each
dependent on a body subgraph node. Once all the body subgraph jobs have been scheduled, the Gather
jobs can be scheduled as well, with their dependencies multiplexed to the corresponding subgraph
jobs. Nodes outside of the scatter section depend on the Gather nodes rather than reaching into the
body subgraph directly.

Conditional sections are treated similarly, but only zero or one instance of its body subgraph will
be launched. Scatter and Conditional sections may be nested, inducing a multi-level tree of Gather
operations.
"""

import logging
import multiprocessing
import os
import json
import signal
import traceback
import pickle
import threading
from concurrent import futures
from typing import Optional, List, Callable, Tuple, Dict, NamedTuple, Set, Union
from contextlib import ExitStack
from .. import Env, Value, Tree, Error
from .task import run_local_task
from ._io_helpers import (
    _add_downloadable_defaults,
    _warn_output_basename_collisions,
    link_outputs,
)
from .download import able as downloadable, run_cached as download
from ._stdlib import WorkflowStdLib
from ._workflow_state import StateMachine
from .._util import (
    write_atomic,
    write_values_json,
    provision_run_dir,
    TerminationSignalFlag,
    LoggingFileHandler,
    compose_coroutines,
    pathsize,
)
from .._util import StructuredLogMessage as _
from . import config, _statusbar
from .cache import CallCache, CallCacheAddPaths, call_cache_key, new as new_call_cache
from .error import RunFailed, Terminated, error_json


class WorkflowMainLoopResult(NamedTuple):
    outputs: Env.Bindings[Value.Base]
    cache_add_paths: CallCacheAddPaths


class _ThreadPools:
    # Singleton managing the thread pools for concurrent task and subworkflow execution
    #
    # All tasks run on one thread pool.
    #
    # Each subworkflow call runs on a thread pool reserved for its nested call depth level.
    # (If we kept just one subworkflow thread pool, then nested calls could deadlock when no thread
    # is available to run a subworkflow whilst the caller blocks its own thread.)
    _lock: threading.Lock
    _cleanup: ExitStack
    _task_pool: futures.ThreadPoolExecutor
    _subworkflow_pools: List[futures.ThreadPoolExecutor]
    _subworkflow_concurrency: int
    _logger: logging.Logger

    def __init__(self, cfg: config.Loader, cleanup: ExitStack, logger: logging.Logger) -> None:
        self._logger = logger
        self._lock = threading.Lock()
        self._cleanup = cleanup.enter_context(ExitStack())

        task_concurrency = (
            cfg["scheduler"].get_int("task_concurrency")
            or (
                cfg.get_int("scheduler", "call_concurrency")  # pre-v1.5.4 legacy
                if cfg.has_option("scheduler", "call_concurrency")
                else 0
            )
            or multiprocessing.cpu_count()
        )

        self._task_pool = futures.ThreadPoolExecutor(max_workers=task_concurrency)
        self._cleanup.callback(futures.ThreadPoolExecutor.shutdown, self._task_pool)
        self._logger.info(_("task thread pool initialized", task_concurrency=task_concurrency))

        self._subworkflow_concurrency = cfg.get_int("scheduler", "subworkflow_concurrency") or max(
            task_concurrency, multiprocessing.cpu_count()
        )
        self._subworkflow_pools = []

    def submit_task(self, *args, **kwargs):
        with self._lock:
            return self._task_pool.submit(*args, **kwargs)

    def submit_subworkflow(self, call_depth: int, *args, **kwargs):
        with self._lock:
            if call_depth >= len(self._subworkflow_pools):
                # First time at this call depth -- initialize a thread pool for it
                assert call_depth == len(self._subworkflow_pools)
                pool = futures.ThreadPoolExecutor(self._subworkflow_concurrency)
                self._cleanup.callback(futures.ThreadPoolExecutor.shutdown, pool)
                self._subworkflow_pools.append(pool)
                self._logger.info(
                    _(
                        "subworkflow thread pool initialized",
                        subworkflow_concurrency=self._subworkflow_concurrency,
                        call_depth=call_depth,
                    )
                )
            return self._subworkflow_pools[call_depth].submit(*args, **kwargs)


def _download_workflow_input_files(
    cfg: config.Loader,
    logger: logging.Logger,
    logger_prefix: List[str],
    run_dir: str,
    inputs: Env.Bindings[Value.Base],
    thread_pools: _ThreadPools,
    cache: CallCache,
) -> None:
    """
    Find all File & Directory input values that are downloadable URIs (including any nested within
    compound values), and ensure the cache is "primed" with them. Workflow inputs are not modified:
    future task input localization will resolve each URI from the cache.
    """

    # scan inputs for URIs
    uris: Set[Tuple[str, bool]] = set()

    def scan_uri(v: Union[Value.File, Value.Directory]) -> str:
        nonlocal uris
        directory = isinstance(v, Value.Directory)
        uri = v.value
        if (uri, directory) not in uris and downloadable(cfg, uri, directory=directory):
            uris.add((uri, directory))
        return uri

    Value.rewrite_env_paths(inputs, scan_uri)
    if not uris:
        return
    logger.notice(_("downloading input URIs", count=len(uris)))

    # download them on the thread pool (but possibly further limiting concurrency)
    download_concurrency = cfg.get_int("scheduler", "download_concurrency")
    if download_concurrency <= 0:
        download_concurrency = 999999
    ops: Dict[futures.Future[Tuple[bool, str]], str] = {}
    incomplete = len(uris)
    outstanding: Set[futures.Future[Tuple[bool, str]]] = set()
    downloaded_bytes = 0
    cached_hits = 0
    exn = None

    while incomplete and not exn:
        assert len(outstanding) <= incomplete

        # top up thread pool's queue (up to download_concurrency)
        while uris and len(outstanding) < download_concurrency:
            (uri, directory) = uris.pop()
            logger.info(
                _(f"schedule input {'directory' if directory else 'file'} download", uri=uri)
            )
            future = thread_pools.submit_task(
                download,
                cfg,
                logger,
                cache,
                uri,
                directory=directory,
                run_dir=os.path.join(run_dir, "download", str(len(ops)), "."),
                logger_prefix=logger_prefix + [f"download{len(ops)}"],
            )
            ops[future] = uri
            outstanding.add(future)
        assert outstanding

        # wait for one or more oustanding downloads to finish
        just_finished, still_outstanding = futures.wait(
            outstanding, return_when=futures.FIRST_COMPLETED
        )
        outstanding = still_outstanding
        for future in just_finished:
            # check results
            try:
                future_exn = future.exception()
            except futures.CancelledError:
                future_exn = Terminated()
            if not future_exn:
                uri = ops[future]
                cached, filename = future.result()
                if cached:
                    cached_hits += 1
                else:
                    sz = pathsize(filename)
                    logger.info(_("downloaded input", uri=uri, path=filename, bytes=sz))
                    downloaded_bytes += sz
            elif not exn:
                # cancel pending ops and signal running ones to abort
                for outsfut in outstanding:
                    outsfut.cancel()
                os.kill(os.getpid(), signal.SIGUSR1)
                exn = future_exn
            incomplete -= 1

    if exn:
        raise exn
    logger.notice(
        _(
            "processed input URIs",
            cached=cached_hits,
            downloaded=len(ops) - cached_hits,
            downloaded_bytes=downloaded_bytes,
        )
    )


def run_local_workflow(
    cfg: config.Loader,
    workflow: Tree.Workflow,
    inputs: Env.Bindings[Value.Base],
    run_id: Optional[str] = None,
    run_dir: Optional[str] = None,
    logger_prefix: Optional[List[str]] = None,
    _thread_pools: Optional[_ThreadPools] = None,
    _cache: Optional[CallCache] = None,
    _test_pickle: bool = False,
    _run_id_stack: Optional[List[str]] = None,
) -> Tuple[str, Env.Bindings[Value.Base]]:
    """
    Run a workflow locally.

    Inputs shall have been typechecked already. File inputs are presumed to be local POSIX file
    paths that can be mounted into a container.

    :param run_id: unique ID for the run, defaults to workflow name
    :param run_dir: directory under which to create a timestamp-named subdirectory for this run
                    (defaults to current working directory).
                    If the final path component is ".", then operate in run_dir directly.
    """

    # provision run directory and log file
    run_id = run_id or workflow.name
    _run_id_stack = _run_id_stack or []
    run_dir = provision_run_dir(workflow.name, run_dir, last_link=not _run_id_stack)

    logger_prefix = logger_prefix or ["wdl"]
    logger_id = logger_prefix + ["w:" + run_id]
    logger = logging.getLogger(".".join(logger_id))
    logfile = os.path.join(run_dir, "workflow.log")
    with ExitStack() as cleanup:
        cleanup.enter_context(
            LoggingFileHandler(
                logger,
                logfile,
            )
        )
        if cfg.has_option("logging", "json") and cfg["logging"].get_bool("json"):
            cleanup.enter_context(
                LoggingFileHandler(
                    logger,
                    logfile + ".json",
                    json=True,
                )
            )
        logger.notice(
            _(
                "workflow start",
                name=workflow.name,
                source=workflow.pos.uri,
                line=workflow.pos.line,
                column=workflow.pos.column,
                dir=run_dir,
            )
        )
        logger.debug(_("thread", ident=threading.get_ident()))
        terminating = cleanup.enter_context(TerminationSignalFlag(logger))
        cache = _cache
        if not cache:
            cache = cleanup.enter_context(new_call_cache(cfg, logger))
            assert cache and _thread_pools is None
        if not _thread_pools:
            cache.flock(logfile, exclusive=True)  # flock top-level workflow.log
        write_values_json(inputs, os.path.join(run_dir, "inputs.json"), namespace=workflow.name)

        # query call cache
        cache_inputs = inputs
        cache_key = call_cache_key(workflow.name, workflow.digest, cache_inputs)
        cached = cache.get(cache_key, cache_inputs, workflow.effective_outputs)
        if cached is not None:
            for outp in workflow.effective_outputs:
                v = cached[outp.name]
                vj = json.dumps(v.json)
                logger.info(
                    _(
                        "cached output",
                        name=outp.name,
                        value=(v.json if len(vj) < 4096 else "(((large)))"),
                    )
                )
            _outputs = link_outputs(
                cache,
                cached,
                run_dir,
                hardlinks=cfg["file_io"].get_bool("output_hardlinks"),
                use_relative_output_paths=cfg["file_io"].get_bool("use_relative_output_paths"),
            )
            write_values_json(
                cached, os.path.join(run_dir, "outputs.json"), namespace=workflow.name
            )
            logger.notice("done (cached)")
            # returning `cached`, not the rewritten `_outputs`, to retain opportunity to find
            # cached downstream inputs
            return (run_dir, cached)

        # if we're the top-level workflow, provision thread pools
        if not _thread_pools:
            # delayed heavy imports -- load .task_container now to work around python issue41567
            import importlib_metadata
            from .task_container import new as _new_task_container  # noqa: F401

            assert not _run_id_stack
            try:
                # log version into workflow.log
                version = "v" + importlib_metadata.version("miniwdl")
            except importlib_metadata.PackageNotFoundError:
                version = "UNKNOWN"
            logger.notice(_("miniwdl", version=version, uname=" ".join(os.uname())))

            thread_pools = _ThreadPools(cfg, cleanup, logger)
        else:
            assert _run_id_stack and _cache
            thread_pools = _thread_pools

        try:
            # run workflow state machine
            main_loop_result = _workflow_main_loop(
                cfg,
                workflow,
                inputs,
                _run_id_stack + [run_id],
                run_dir,
                logger,
                logger_id,
                thread_pools,
                cache,
                terminating,
                _test_pickle,
            )
            outputs = main_loop_result.outputs
            cache_add_paths = main_loop_result.cache_add_paths
        except:
            _statusbar.abort()
            if not _run_id_stack and cfg["scheduler"].get_bool("fail_fast"):
                # if we're the top-level worfklow, signal abort to anything still running
                # concurrently on the thread pools (SIGUSR1 will be picked up by
                # TerminationSignalFlag)
                os.kill(os.getpid(), signal.SIGUSR1)
            raise

        cache.put(
            cache_key,
            outputs,
            run_dir=run_dir,
            inputs=cache_inputs,
            add_paths=cache_add_paths,
        )

    return (run_dir, outputs)


def _workflow_main_loop(
    cfg: config.Loader,
    workflow: Tree.Workflow,
    inputs: Env.Bindings[Value.Base],
    run_id_stack: List[str],
    run_dir: str,
    logger: logging.Logger,
    logger_id: List[str],
    thread_pools: _ThreadPools,
    cache: CallCache,
    terminating: Callable[[], bool],
    _test_pickle: bool,
) -> WorkflowMainLoopResult:
    assert isinstance(cfg, config.Loader)
    call_futures = {}
    try:
        # start plugin coroutines and process inputs through them
        with compose_coroutines(
            [
                (
                    lambda kwargs, cor=cor: cor(  # type: ignore
                        cfg, logger, run_id_stack, run_dir, workflow, **kwargs
                    )
                )
                for cor in [cor2 for _, cor2 in sorted(config.load_plugins(cfg, "workflow"))]
            ],
            {"inputs": inputs},
        ) as plugins:
            recv = next(plugins)
            inputs = recv["inputs"]

            # download input files, if needed
            _download_workflow_input_files(
                cfg,
                logger,
                logger_id,
                run_dir,
                _add_downloadable_defaults(cfg, workflow.available_inputs, inputs),
                thread_pools,
                cache,
            )

            # run workflow state machine to completion
            state = StateMachine(".".join(logger_id), run_dir, workflow, inputs)
            while state.outputs is None:
                if _test_pickle:
                    state = pickle.loads(pickle.dumps(state))
                if terminating():
                    raise Terminated()
                # schedule all runnable calls
                stdlib = WorkflowStdLib(cfg, workflow.effective_wdl_version, state, cache)
                next_call = state.step(cfg, stdlib)
                while next_call:
                    call_dir = os.path.join(run_dir, next_call.id)
                    if os.path.exists(call_dir):
                        logger.warning(
                            _("call subdirectory already exists, conflict likely", dir=call_dir)
                        )
                    sub_args = (cfg, next_call.callee, next_call.inputs)
                    sub_kwargs = {
                        "run_id": next_call.id,
                        "run_dir": os.path.join(call_dir, "."),
                        "logger_prefix": logger_id,
                        "_cache": cache,
                        "_run_id_stack": run_id_stack,
                    }
                    # submit to appropriate thread pool
                    if isinstance(next_call.callee, Tree.Task):
                        _statusbar.task_backlogged()
                        future = thread_pools.submit_task(run_local_task, *sub_args, **sub_kwargs)
                    elif isinstance(next_call.callee, Tree.Workflow):
                        future = thread_pools.submit_subworkflow(
                            len(run_id_stack) - 1,
                            run_local_workflow,
                            *sub_args,
                            **sub_kwargs,
                            _thread_pools=thread_pools,
                        )
                    else:
                        assert False
                    child_key = call_cache_key(
                        next_call.callee.name, next_call.callee.digest, next_call.inputs
                    )
                    call_futures[future] = (next_call.id, child_key)
                    next_call = state.step(cfg, stdlib)
                # no more calls to launch right now; wait for an outstanding call to finish
                future = next(futures.as_completed(call_futures), None)
                if future:
                    __, outputs = future.result()
                    call_id, child_key = call_futures[future]
                    # Fold child task/subworkflow add_paths into the parent workflow add_paths.
                    # This makes a workflow cache hit sensitive to source-relative paths used
                    # inside its calls, even when the parent workflow doesn't read those files.
                    state.cache_add_paths.update(cache.get_add_paths(child_key))
                    state.call_finished(call_id, outputs)
                    call_futures.pop(future)
                else:
                    assert state.outputs is not None

            # create output_links
            outputs = link_outputs(
                cache,
                state.outputs,
                run_dir,
                hardlinks=cfg["file_io"].get_bool("output_hardlinks"),
                # Relative output paths only make sense at the top level, and hence is only used here.
                use_relative_output_paths=cfg["file_io"].get_bool("use_relative_output_paths"),
            )

            # process outputs through plugins
            recv = plugins.send({"outputs": outputs})
            outputs = recv["outputs"]
            _warn_output_basename_collisions(logger, outputs)

            # write outputs.json
            write_values_json(
                outputs, os.path.join(run_dir, "outputs.json"), namespace=workflow.name
            )
            logger.notice("done")
            return WorkflowMainLoopResult(outputs, state.cache_add_paths)
    except Exception as exn:
        tbtxt = traceback.format_exc()
        logger.debug(tbtxt)
        cause: BaseException = exn
        while isinstance(cause, RunFailed) and cause.__cause__:
            cause = cause.__cause__
        wrapper = RunFailed(workflow, run_id_stack[-1], run_dir)
        try:
            write_atomic(
                json.dumps(
                    error_json(
                        wrapper,
                        cause=exn,
                        traceback=tbtxt if not isinstance(exn, Error.RuntimeError) else None,
                    ),
                    indent=2,
                ),
                os.path.join(run_dir, "error.json"),
            )
        except Exception as exn2:
            logger.debug(traceback.format_exc())
            logger.critical(_("failed to write error.json", dir=run_dir, message=str(exn2)))
        if not isinstance(exn, RunFailed):
            logger.error(
                _(
                    str(wrapper),
                    dir=run_dir,
                    **error_json(
                        exn, traceback=tbtxt if not isinstance(exn, Error.RuntimeError) else None
                    ),
                )
            )
        elif not isinstance(exn.__cause__, Terminated):
            logger.error(
                _("call failure propagating", **{"from": getattr(exn, "run_id"), "dir": run_dir})
            )
        # Cancel all future tasks that havent started
        for key in call_futures:
            key.cancel()
        raise wrapper from exn

1	"""
2	Workflow runner building blocks & local driver
3
4	--------
5	Overview
6	--------
7
8	Workflow execution proceeds according to the AST's ``WorkflowNode`` graph, in which each Decl,
9	Call, Scatter, Conditional, and (implicit) Gather operation has its own node which advertises its
10	dependencies on the other nodes.
11
12	Abstractly, we plan to "visit" each node after visiting all of its dependencies. The node's type
13	prescribes some job to do upon visitation, such as evaluating a Decl's WDL expression, running a
14	task on some inputs, or scheduling additional jobs to scatter over an array. Named WDL values
15	(``WDL.Env.Bindings[WDL.Value.Base]``) are transmitted along each dependency edge, and WDL
16	expressions in each node are evaluated in the environment formed from the union of the node's
17	incoming dependency edges.
18
19	Scatter sections contain a body, which provides a template for the job subgraph to be scheduled for
20	each element of the runtime-evaluated scatter array. They also contain template Gather nodes, each
21	dependent on a body subgraph node. Once all the body subgraph jobs have been scheduled, the Gather
22	jobs can be scheduled as well, with their dependencies multiplexed to the corresponding subgraph
23	jobs. Nodes outside of the scatter section depend on the Gather nodes rather than reaching into the
24	body subgraph directly.
25
26	Conditional sections are treated similarly, but only zero or one instance of its body subgraph will
27	be launched. Scatter and Conditional sections may be nested, inducing a multi-level tree of Gather
28	operations.
29	"""
30
31	import logging	1✔
32	import multiprocessing	1✔
33	import os	1✔
34	import json	1✔
35	import signal	1✔
36	import traceback	1✔
37	import pickle	1✔
38	import threading	1✔
39	from concurrent import futures	1✔
40	from typing import Optional, List, Callable, Tuple, Dict, NamedTuple, Set, Union	1✔
41	from contextlib import ExitStack	1✔
42	from .. import Env, Value, Tree, Error	1✔
43	from .task import run_local_task	1✔
44	from ._io_helpers import (	1✔
45	_add_downloadable_defaults,
46	_warn_output_basename_collisions,
47	link_outputs,
48	)
49	from .download import able as downloadable, run_cached as download	1✔
50	from ._stdlib import WorkflowStdLib	1✔
51	from ._workflow_state import StateMachine	1✔
52	from .._util import (	1✔
53	write_atomic,
54	write_values_json,
55	provision_run_dir,
56	TerminationSignalFlag,
57	LoggingFileHandler,
58	compose_coroutines,
59	pathsize,
60	)
61	from .._util import StructuredLogMessage as _	1✔
62	from . import config, _statusbar	1✔
63	from .cache import CallCache, CallCacheAddPaths, call_cache_key, new as new_call_cache	1✔
64	from .error import RunFailed, Terminated, error_json	1✔
65
66
67	class WorkflowMainLoopResult(NamedTuple):	1✔
68	outputs: Env.Bindings[Value.Base]	1✔
69	cache_add_paths: CallCacheAddPaths	1✔
70
71
72	class _ThreadPools:	1✔
73	# Singleton managing the thread pools for concurrent task and subworkflow execution
74	#
75	# All tasks run on one thread pool.
76	#
77	# Each subworkflow call runs on a thread pool reserved for its nested call depth level.
78	# (If we kept just one subworkflow thread pool, then nested calls could deadlock when no thread
79	# is available to run a subworkflow whilst the caller blocks its own thread.)
80	_lock: threading.Lock	1✔
81	_cleanup: ExitStack	1✔
82	_task_pool: futures.ThreadPoolExecutor	1✔
83	_subworkflow_pools: List[futures.ThreadPoolExecutor]	1✔
84	_subworkflow_concurrency: int	1✔
85	_logger: logging.Logger	1✔
86
87	def __init__(self, cfg: config.Loader, cleanup: ExitStack, logger: logging.Logger) -> None:	1✔
88	self._logger = logger	1✔
89	self._lock = threading.Lock()	1✔
90	self._cleanup = cleanup.enter_context(ExitStack())	1✔
91
92	task_concurrency = (	1✔
93	cfg["scheduler"].get_int("task_concurrency")
94	or (
95	cfg.get_int("scheduler", "call_concurrency") # pre-v1.5.4 legacy
96	if cfg.has_option("scheduler", "call_concurrency")
97	else 0
98	)
99	or multiprocessing.cpu_count()
100	)
101
102	self._task_pool = futures.ThreadPoolExecutor(max_workers=task_concurrency)	1✔
103	self._cleanup.callback(futures.ThreadPoolExecutor.shutdown, self._task_pool)	1✔
104	self._logger.info(_("task thread pool initialized", task_concurrency=task_concurrency))	1✔
105
106	self._subworkflow_concurrency = cfg.get_int("scheduler", "subworkflow_concurrency") or max(	1✔
107	task_concurrency, multiprocessing.cpu_count()
108	)
109	self._subworkflow_pools = []	1✔
110
111	def submit_task(self, args, *kwargs):	1✔
112	with self._lock:	1✔
113	return self._task_pool.submit(args, *kwargs)	1✔
114
115	def submit_subworkflow(self, call_depth: int, args, *kwargs):	1✔
116	with self._lock:	1✔
117	if call_depth >= len(self._subworkflow_pools):	1✔
118	# First time at this call depth -- initialize a thread pool for it
119	assert call_depth == len(self._subworkflow_pools)	1✔
120	pool = futures.ThreadPoolExecutor(self._subworkflow_concurrency)	1✔
121	self._cleanup.callback(futures.ThreadPoolExecutor.shutdown, pool)	1✔
122	self._subworkflow_pools.append(pool)	1✔
123	self._logger.info(	1✔
124	_(
125	"subworkflow thread pool initialized",
126	subworkflow_concurrency=self._subworkflow_concurrency,
127	call_depth=call_depth,
128	)
129	)
130	return self._subworkflow_pools[call_depth].submit(args, *kwargs)	1✔
131
132
133	def _download_workflow_input_files(	1✔
134	cfg: config.Loader,
135	logger: logging.Logger,
136	logger_prefix: List[str],
137	run_dir: str,
138	inputs: Env.Bindings[Value.Base],
139	thread_pools: _ThreadPools,
140	cache: CallCache,
141	) -> None:
142	"""
143	Find all File & Directory input values that are downloadable URIs (including any nested within
144	compound values), and ensure the cache is "primed" with them. Workflow inputs are not modified:
145	future task input localization will resolve each URI from the cache.
146	"""
147
148	# scan inputs for URIs
149	uris: Set[Tuple[str, bool]] = set()	1✔
150
151	def scan_uri(v: Union[Value.File, Value.Directory]) -> str:	1✔
152	nonlocal uris
153	directory = isinstance(v, Value.Directory)	1✔
154	uri = v.value	1✔
155	if (uri, directory) not in uris and downloadable(cfg, uri, directory=directory):	1✔
156	uris.add((uri, directory))	1✔
157	return uri	1✔
158
159	Value.rewrite_env_paths(inputs, scan_uri)	1✔
160	if not uris:	1✔
161	return	1✔
162	logger.notice(_("downloading input URIs", count=len(uris)))	1✔
163
164	# download them on the thread pool (but possibly further limiting concurrency)
165	download_concurrency = cfg.get_int("scheduler", "download_concurrency")	1✔
166	if download_concurrency <= 0:	1✔
167	download_concurrency = 999999	1✔
168	ops: Dict[futures.Future[Tuple[bool, str]], str] = {}	1✔
169	incomplete = len(uris)	1✔
170	outstanding: Set[futures.Future[Tuple[bool, str]]] = set()	1✔
171	downloaded_bytes = 0	1✔
172	cached_hits = 0	1✔
173	exn = None	1✔
174
175	while incomplete and not exn:	1✔
176	assert len(outstanding) <= incomplete	1✔
177
178	# top up thread pool's queue (up to download_concurrency)
179	while uris and len(outstanding) < download_concurrency:	1✔
180	(uri, directory) = uris.pop()	1✔
181	logger.info(	1✔
182	_(f"schedule input {'directory' if directory else 'file'} download", uri=uri)
183	)
184	future = thread_pools.submit_task(	1✔
185	download,
186	cfg,
187	logger,
188	cache,
189	uri,
190	directory=directory,
191	run_dir=os.path.join(run_dir, "download", str(len(ops)), "."),
192	logger_prefix=logger_prefix + [f"download{len(ops)}"],
193	)
194	ops[future] = uri	1✔
195	outstanding.add(future)	1✔
196	assert outstanding	1✔
197
198	# wait for one or more oustanding downloads to finish
199	just_finished, still_outstanding = futures.wait(	1✔
200	outstanding, return_when=futures.FIRST_COMPLETED
201	)
202	outstanding = still_outstanding	1✔
203	for future in just_finished:	1✔
204	# check results
205	try:	1✔
206	future_exn = future.exception()	1✔
207	except futures.CancelledError:	×
208	future_exn = Terminated()	×
209	if not future_exn:	1✔
210	uri = ops[future]	1✔
211	cached, filename = future.result()	1✔
212	if cached:	1✔
213	cached_hits += 1	1✔
214	else:
215	sz = pathsize(filename)	1✔
216	logger.info(_("downloaded input", uri=uri, path=filename, bytes=sz))	1✔
217	downloaded_bytes += sz	1✔
218	elif not exn:	1✔
219	# cancel pending ops and signal running ones to abort
220	for outsfut in outstanding:	1✔
UNCOV 221	outsfut.cancel()	×
222	os.kill(os.getpid(), signal.SIGUSR1)	1✔
223	exn = future_exn	1✔
224	incomplete -= 1	1✔
225
226	if exn:	1✔
227	raise exn	1✔
228	logger.notice(	1✔
229	_(
230	"processed input URIs",
231	cached=cached_hits,
232	downloaded=len(ops) - cached_hits,
233	downloaded_bytes=downloaded_bytes,
234	)
235	)
236
237
238	def run_local_workflow(	1✔
239	cfg: config.Loader,
240	workflow: Tree.Workflow,
241	inputs: Env.Bindings[Value.Base],
242	run_id: Optional[str] = None,
243	run_dir: Optional[str] = None,
244	logger_prefix: Optional[List[str]] = None,
245	_thread_pools: Optional[_ThreadPools] = None,
246	_cache: Optional[CallCache] = None,
247	_test_pickle: bool = False,
248	_run_id_stack: Optional[List[str]] = None,
249	) -> Tuple[str, Env.Bindings[Value.Base]]:
250	"""
251	Run a workflow locally.
252
253	Inputs shall have been typechecked already. File inputs are presumed to be local POSIX file
254	paths that can be mounted into a container.
255
256	:param run_id: unique ID for the run, defaults to workflow name
257	:param run_dir: directory under which to create a timestamp-named subdirectory for this run
258	(defaults to current working directory).
259	If the final path component is ".", then operate in run_dir directly.
260	"""
261
262	# provision run directory and log file
263	run_id = run_id or workflow.name	1✔
264	_run_id_stack = _run_id_stack or []	1✔
265	run_dir = provision_run_dir(workflow.name, run_dir, last_link=not _run_id_stack)	1✔
266
267	logger_prefix = logger_prefix or ["wdl"]	1✔
268	logger_id = logger_prefix + ["w:" + run_id]	1✔
269	logger = logging.getLogger(".".join(logger_id))	1✔
270	logfile = os.path.join(run_dir, "workflow.log")	1✔
271	with ExitStack() as cleanup:	1✔
272	cleanup.enter_context(	1✔
273	LoggingFileHandler(
274	logger,
275	logfile,
276	)
277	)
278	if cfg.has_option("logging", "json") and cfg["logging"].get_bool("json"):	1✔
279	cleanup.enter_context(	1✔
280	LoggingFileHandler(
281	logger,
282	logfile + ".json",
283	json=True,
284	)
285	)
286	logger.notice(	1✔
287	_(
288	"workflow start",
289	name=workflow.name,
290	source=workflow.pos.uri,
291	line=workflow.pos.line,
292	column=workflow.pos.column,
293	dir=run_dir,
294	)
295	)
296	logger.debug(_("thread", ident=threading.get_ident()))	1✔
297	terminating = cleanup.enter_context(TerminationSignalFlag(logger))	1✔
298	cache = _cache	1✔
299	if not cache:	1✔
300	cache = cleanup.enter_context(new_call_cache(cfg, logger))	1✔
301	assert cache and _thread_pools is None	1✔
302	if not _thread_pools:	1✔
303	cache.flock(logfile, exclusive=True) # flock top-level workflow.log	1✔
304	write_values_json(inputs, os.path.join(run_dir, "inputs.json"), namespace=workflow.name)	1✔
305
306	# query call cache
307	cache_inputs = inputs	1✔
308	cache_key = call_cache_key(workflow.name, workflow.digest, cache_inputs)	1✔
309	cached = cache.get(cache_key, cache_inputs, workflow.effective_outputs)	1✔
310	if cached is not None:	1✔
311	for outp in workflow.effective_outputs:	1✔
312	v = cached[outp.name]	1✔
313	vj = json.dumps(v.json)	1✔
314	logger.info(	1✔
315	_(
316	"cached output",
317	name=outp.name,
318	value=(v.json if len(vj) < 4096 else "(((large)))"),
319	)
320	)
321	_outputs = link_outputs(	1✔
322	cache,
323	cached,
324	run_dir,
325	hardlinks=cfg["file_io"].get_bool("output_hardlinks"),
326	use_relative_output_paths=cfg["file_io"].get_bool("use_relative_output_paths"),
327	)
328	write_values_json(	1✔
329	cached, os.path.join(run_dir, "outputs.json"), namespace=workflow.name
330	)
331	logger.notice("done (cached)")	1✔
332	# returning `cached`, not the rewritten `_outputs`, to retain opportunity to find
333	# cached downstream inputs
334	return (run_dir, cached)	1✔
335
336	# if we're the top-level workflow, provision thread pools
337	if not _thread_pools:	1✔
338	# delayed heavy imports -- load .task_container now to work around python issue41567
339	import importlib_metadata	1✔
340	from .task_container import new as _new_task_container # noqa: F401	1✔
341
342	assert not _run_id_stack	1✔
343	try:	1✔
344	# log version into workflow.log
345	version = "v" + importlib_metadata.version("miniwdl")	1✔
346	except importlib_metadata.PackageNotFoundError:	1✔
347	version = "UNKNOWN"	1✔
348	logger.notice(_("miniwdl", version=version, uname=" ".join(os.uname())))	1✔
349
350	thread_pools = _ThreadPools(cfg, cleanup, logger)	1✔
351	else:
352	assert _run_id_stack and _cache	1✔
353	thread_pools = _thread_pools	1✔
354
355	try:	1✔
356	# run workflow state machine
357	main_loop_result = _workflow_main_loop(	1✔
358	cfg,
359	workflow,
360	inputs,
361	_run_id_stack + [run_id],
362	run_dir,
363	logger,
364	logger_id,
365	thread_pools,
366	cache,
367	terminating,
368	_test_pickle,
369	)
370	outputs = main_loop_result.outputs	1✔
371	cache_add_paths = main_loop_result.cache_add_paths	1✔
372	except:	1✔
373	_statusbar.abort()	1✔
374	if not _run_id_stack and cfg["scheduler"].get_bool("fail_fast"):	1✔
375	# if we're the top-level worfklow, signal abort to anything still running
376	# concurrently on the thread pools (SIGUSR1 will be picked up by
377	# TerminationSignalFlag)
378	os.kill(os.getpid(), signal.SIGUSR1)	1✔
379	raise	1✔
380
381	cache.put(	1✔
382	cache_key,
383	outputs,
384	run_dir=run_dir,
385	inputs=cache_inputs,
386	add_paths=cache_add_paths,
387	)
388
389	return (run_dir, outputs)	1✔
390
391
392	def _workflow_main_loop(	1✔
393	cfg: config.Loader,
394	workflow: Tree.Workflow,
395	inputs: Env.Bindings[Value.Base],
396	run_id_stack: List[str],
397	run_dir: str,
398	logger: logging.Logger,
399	logger_id: List[str],
400	thread_pools: _ThreadPools,
401	cache: CallCache,
402	terminating: Callable[[], bool],
403	_test_pickle: bool,
404	) -> WorkflowMainLoopResult:
405	assert isinstance(cfg, config.Loader)	1✔
406	call_futures = {}	1✔
407	try:	1✔
408	# start plugin coroutines and process inputs through them
409	with compose_coroutines(	1✔
410	[
411	(
412	lambda kwargs, cor=cor: cor( # type: ignore
413	cfg, logger, run_id_stack, run_dir, workflow, **kwargs
414	)
415	)
416	for cor in [cor2 for _, cor2 in sorted(config.load_plugins(cfg, "workflow"))]
417	],
418	{"inputs": inputs},
419	) as plugins:
420	recv = next(plugins)	1✔
421	inputs = recv["inputs"]	1✔
422
423	# download input files, if needed
424	_download_workflow_input_files(	1✔
425	cfg,
426	logger,
427	logger_id,
428	run_dir,
429	_add_downloadable_defaults(cfg, workflow.available_inputs, inputs),
430	thread_pools,
431	cache,
432	)
433
434	# run workflow state machine to completion
435	state = StateMachine(".".join(logger_id), run_dir, workflow, inputs)	1✔
436	while state.outputs is None:	1✔
437	if _test_pickle:	1✔
438	state = pickle.loads(pickle.dumps(state))	1✔
439	if terminating():	1✔
440	raise Terminated()	×
441	# schedule all runnable calls
442	stdlib = WorkflowStdLib(cfg, workflow.effective_wdl_version, state, cache)	1✔
443	next_call = state.step(cfg, stdlib)	1✔
444	while next_call:	1✔
445	call_dir = os.path.join(run_dir, next_call.id)	1✔
446	if os.path.exists(call_dir):	1✔
447	logger.warning(	×
448	_("call subdirectory already exists, conflict likely", dir=call_dir)
449	)
450	sub_args = (cfg, next_call.callee, next_call.inputs)	1✔
451	sub_kwargs = {	1✔
452	"run_id": next_call.id,
453	"run_dir": os.path.join(call_dir, "."),
454	"logger_prefix": logger_id,
455	"_cache": cache,
456	"_run_id_stack": run_id_stack,
457	}
458	# submit to appropriate thread pool
459	if isinstance(next_call.callee, Tree.Task):	1✔
460	_statusbar.task_backlogged()	1✔
461	future = thread_pools.submit_task(run_local_task, sub_args, *sub_kwargs)	1✔
462	elif isinstance(next_call.callee, Tree.Workflow):	1✔
463	future = thread_pools.submit_subworkflow(	1✔
464	len(run_id_stack) - 1,
465	run_local_workflow,
466	*sub_args,
467	**sub_kwargs,
468	_thread_pools=thread_pools,
469	)
470	else:
471	assert False	×
472	child_key = call_cache_key(	1✔
473	next_call.callee.name, next_call.callee.digest, next_call.inputs
474	)
475	call_futures[future] = (next_call.id, child_key)	1✔
476	next_call = state.step(cfg, stdlib)	1✔
477	# no more calls to launch right now; wait for an outstanding call to finish
478	future = next(futures.as_completed(call_futures), None)	1✔
479	if future:	1✔
480	__, outputs = future.result()	1✔
481	call_id, child_key = call_futures[future]	1✔
482	# Fold child task/subworkflow add_paths into the parent workflow add_paths.
483	# This makes a workflow cache hit sensitive to source-relative paths used
484	# inside its calls, even when the parent workflow doesn't read those files.
485	state.cache_add_paths.update(cache.get_add_paths(child_key))	1✔
486	state.call_finished(call_id, outputs)	1✔
487	call_futures.pop(future)	1✔
488	else:
489	assert state.outputs is not None	1✔
490
491	# create output_links
492	outputs = link_outputs(	1✔
493	cache,
494	state.outputs,
495	run_dir,
496	hardlinks=cfg["file_io"].get_bool("output_hardlinks"),
497	# Relative output paths only make sense at the top level, and hence is only used here.
498	use_relative_output_paths=cfg["file_io"].get_bool("use_relative_output_paths"),
499	)
500
501	# process outputs through plugins
502	recv = plugins.send({"outputs": outputs})	1✔
503	outputs = recv["outputs"]	1✔
504	_warn_output_basename_collisions(logger, outputs)	1✔
505
506	# write outputs.json
507	write_values_json(	1✔
508	outputs, os.path.join(run_dir, "outputs.json"), namespace=workflow.name
509	)
510	logger.notice("done")	1✔
511	return WorkflowMainLoopResult(outputs, state.cache_add_paths)	1✔
512	except Exception as exn:	1✔
513	tbtxt = traceback.format_exc()	1✔
514	logger.debug(tbtxt)	1✔
515	cause: BaseException = exn	1✔
516	while isinstance(cause, RunFailed) and cause.__cause__:	1✔
517	cause = cause.__cause__	1✔
518	wrapper = RunFailed(workflow, run_id_stack[-1], run_dir)	1✔
519	try:	1✔
520	write_atomic(	1✔
521	json.dumps(
522	error_json(
523	wrapper,
524	cause=exn,
525	traceback=tbtxt if not isinstance(exn, Error.RuntimeError) else None,
526	),
527	indent=2,
528	),
529	os.path.join(run_dir, "error.json"),
530	)
531	except Exception as exn2:	×
532	logger.debug(traceback.format_exc())	×
533	logger.critical(_("failed to write error.json", dir=run_dir, message=str(exn2)))	×
534	if not isinstance(exn, RunFailed):	1✔
535	logger.error(	1✔
536	_(
537	str(wrapper),
538	dir=run_dir,
539	**error_json(
540	exn, traceback=tbtxt if not isinstance(exn, Error.RuntimeError) else None
541	),
542	)
543	)
544	elif not isinstance(exn.__cause__, Terminated):	1✔
545	logger.error(	1✔
546	_("call failure propagating", **{"from": getattr(exn, "run_id"), "dir": run_dir})
547	)
548	# Cancel all future tasks that havent started
549	for key in call_futures:	1✔
550	key.cancel()	1✔
551	raise wrapper from exn	1✔

chanzuckerberg / miniwdl / 27707168593

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous