From 87f54b2743efdd18c56027f9ddb877c09ee7775a Mon Sep 17 00:00:00 2001 From: Don Freed Date: Fri, 14 Aug 2026 12:54:53 -0700 Subject: [PATCH 1/5] Fix log-directory setup ordering, id sanitization, and verbosity flags Create the log directory only after validation: setup_logging is now console-only and a new start_run_logs(), called after validate(), derives and prepares the log directory. A typo'd output path again fails fast instead of being silently created, a rerun rejected by validation no longer destroys the previous run's logs, and a log path colliding with a regular file exits cleanly instead of raising. Key job-id counters on the sanitized name (sanitize moved to util.py) so log file paths stay unique even for names that differ only in unsafe characters. Make -v/-q/-d independent flags with explicit precedence: -d -v no longer downgrades DEBUG to INFO, and the new --quiet restores a warnings-only console. Document the flags and the per-run log directory in the README. Co-Authored-By: Claude Fable 5 --- README.md | 8 ++- sentieon_cli/__init__.py | 40 ++++++++++---- sentieon_cli/job.py | 8 ++- sentieon_cli/logging.py | 3 +- sentieon_cli/pipeline.py | 46 ++++++++++------ sentieon_cli/run_logs.py | 10 +--- sentieon_cli/sentieon_pangenome.py | 1 + sentieon_cli/util.py | 7 +++ tests/unit/test_cli_args.py | 40 ++++++++++++++ tests/unit/test_job.py | 12 +++++ tests/unit/test_job_logs.py | 3 +- tests/unit/test_run_logs.py | 87 ++++++++++++++++++++++++++---- 12 files changed, 213 insertions(+), 52 deletions(-) create mode 100644 tests/unit/test_cli_args.py diff --git a/README.md b/README.md index b51d89f..cccd5fc 100644 --- a/README.md +++ b/README.md @@ -55,8 +55,12 @@ sentieon-cli ... ## Global arguments The `sentieon-cli` supports the following global arguments: -- `--verbose`: verbose logging. -- `--debug`: debugging mode for more verbose logging. +- `--verbose` (`-v`): verbose logging. This is the default. +- `--quiet` (`-q`): only log warnings and errors. +- `--debug` (`-d`): debugging mode for more verbose logging. Takes precedence over `--verbose` and `--quiet`. + +## Logging +Each run writes its log files to a directory next to the output VCF, named after the output file with the `.vcf.gz` suffix replaced by `_logs`, so `sample.vcf.gz` produces `sample_logs/`. The directory records the invocation in `command.txt`, the pipeline's own messages in `run.log`, and the output of each tool under `task_logs/`. Every pipeline accepts a `--log_dir` argument to write these files elsewhere. Rerunning a pipeline with the same output overwrites the logs of the previous run. ## Supported pipelines - [**DNAscope**](https://support.sentieon.com/docs/sentieon_cli/#dnascope) - DNAscope pipeline implementation for germline SNV and indel calling from short read data. diff --git a/sentieon_cli/__init__.py b/sentieon_cli/__init__.py index 1de7890..59c9971 100644 --- a/sentieon_cli/__init__.py +++ b/sentieon_cli/__init__.py @@ -1,3 +1,5 @@ +import argparse + from . import argh_parser from .dnascope import DNAscopePipeline from .dnascope_hybrid import DNAscopeHybridPipeline @@ -7,26 +9,41 @@ from .util import __version__ -def main(): - """main entry point for this project""" - parser = argh_parser.CustomArgparseParser() +def add_logging_args(parser: argparse.ArgumentParser) -> None: + """Add the console verbosity flags""" parser.add_argument( "-v", "--verbose", - help="Verbose logging", - action="store_const", - dest="loglevel", - const="INFO", - default="INFO", + help="Verbose logging (the default)", + action="store_true", + ) + parser.add_argument( + "-q", + "--quiet", + help="Only log warnings and errors", + action="store_true", ) parser.add_argument( "-d", "--debug", help="Print debugging info", - action="store_const", - dest="loglevel", - const="DEBUG", + action="store_true", ) + + +def resolve_loglevel(args: argparse.Namespace) -> str: + """The console log level implied by the verbosity flags""" + if args.debug: + return "DEBUG" + if args.quiet and not args.verbose: + return "WARNING" + return "INFO" + + +def main(): + """main entry point for this project""" + parser = argh_parser.CustomArgparseParser() + add_logging_args(parser) parser.add_argument( "--version", action="version", @@ -60,6 +77,7 @@ def main(): dnascope_pangenome_subparser.set_defaults(pipeline=pipeline.main) args = parser.parse_args() + args.loglevel = resolve_loglevel(args) # Job ids must be unique for the whole run, which may execute more than # one DAG, so numbering restarts here rather than per DAG. Job.reset_ids() diff --git a/sentieon_cli/job.py b/sentieon_cli/job.py index bb893bb..a6a15f7 100644 --- a/sentieon_cli/job.py +++ b/sentieon_cli/job.py @@ -5,6 +5,7 @@ from typing import Dict, Optional from .shell_pipeline import Pipeline +from .util import sanitize class Job: @@ -44,8 +45,11 @@ def __init__( self.threads = threads self.resources = {} if resources is None else resources self.task_name = task_name - count = Job._id_counters.get(name, 0) + 1 - Job._id_counters[name] = count + # Log file names are sanitized, so ids must stay unique after + # sanitization; the id itself keeps the readable name. + key = sanitize(name) + count = Job._id_counters.get(key, 0) + 1 + Job._id_counters[key] = count self.job_id = f"{name}-{count}" @classmethod diff --git a/sentieon_cli/logging.py b/sentieon_cli/logging.py index df0a48a..fe12b53 100644 --- a/sentieon_cli/logging.py +++ b/sentieon_cli/logging.py @@ -4,7 +4,8 @@ Handlers live only on the package logger; the loggers returned by ``get_logger`` are plain children that propagate to it. The package logger is always at DEBUG so verbosity can be enforced per handler: the console handler -follows ``-v``/``-d`` while the ``run.log`` file handler always records DEBUG. +follows ``-v``/``-q``/``-d`` while the ``run.log`` file handler always records +DEBUG. """ import logging diff --git a/sentieon_cli/pipeline.py b/sentieon_cli/pipeline.py index 65231aa..55f267a 100644 --- a/sentieon_cli/pipeline.py +++ b/sentieon_cli/pipeline.py @@ -126,30 +126,43 @@ def __init__(self) -> None: self.run_logs: Optional[RunLogs] = None def setup_logging(self, args: argparse.Namespace) -> None: - """Configure the console handler and the run's log directory""" + """Configure console logging""" self.logger = get_logger(__name__) set_console_level(args.loglevel) + def start_run_logs(self) -> None: + """Create the run's log directory and start writing `run.log`. + + Called after `validate`, so a run rejected for an invalid output path + neither creates directories nor clobbers a previous run's logs. + """ # File logging is skipped for dry-runs and when there is nothing to # derive a log directory from (a bare pipeline, as used by tests). - if not self.dry_run and ( - self.log_dir is not None or self.output_vcf is not None - ): - log_dir = self.log_dir - if log_dir is None: - # Check the suffix before deriving the directory name so an - # invalid output path cannot create a garbage-named log dir. - self.validate_output_suffix() - log_dir = pathlib.Path( - str(self.output_vcf).removesuffix(".vcf.gz") + "_logs" - ) - self.run_logs = RunLogs(log_dir) - self.run_logs.setup() + if self.dry_run or (self.log_dir is None and self.output_vcf is None): + self.logger.info("Starting sentieon-cli version: %s", __version__) + return + + log_dir = self.log_dir + if log_dir is None: + # Defensive: `validate` has already checked the output path, but + # not every pipeline's validation covers the suffix. + self.validate_output_suffix() + log_dir = pathlib.Path( + str(self.output_vcf).removesuffix(".vcf.gz") + "_logs" + ) + run_logs = RunLogs(log_dir) + try: + run_logs.setup() + except OSError as exc: + self.logger.error( + "Could not prepare the log directory %s: %s", log_dir, exc + ) + sys.exit(2) + self.run_logs = run_logs # After the file handler is attached, so the banner reaches run.log self.logger.info("Starting sentieon-cli version: %s", __version__) - if self.run_logs: - self.logger.info("Writing logs to: %s", self.run_logs.log_dir) + self.logger.info("Writing logs to: %s", self.run_logs.log_dir) def log_completion(self, success: bool, start_time: float) -> None: """Report the outcome and duration of the run""" @@ -171,6 +184,7 @@ def main(self, args: argparse.Namespace) -> None: success = False try: self.validate() + self.start_run_logs() self.configure() tmp_dir_str = tmp() diff --git a/sentieon_cli/run_logs.py b/sentieon_cli/run_logs.py index 5a7f7c3..adee9db 100644 --- a/sentieon_cli/run_logs.py +++ b/sentieon_cli/run_logs.py @@ -8,30 +8,22 @@ import datetime import logging import pathlib -import re import shlex import shutil import sys from typing import IO, List, Optional, TYPE_CHECKING from .logging import add_file_handler, remove_file_handler -from .util import __version__ +from .util import __version__, sanitize if TYPE_CHECKING: from .job import Job from .shell_pipeline import Command -_UNSAFE = re.compile(r"[^A-Za-z0-9._-]") - _STDERR = "stderr" _STDOUT = "stdout" -def sanitize(component: str) -> str: - """Restrict a path component to filesystem-safe characters""" - return _UNSAFE.sub("-", component) - - @dataclasses.dataclass class _ProcessLog: """A log file opened for one spawned process""" diff --git a/sentieon_cli/sentieon_pangenome.py b/sentieon_cli/sentieon_pangenome.py index 50f23af..bb985db 100644 --- a/sentieon_cli/sentieon_pangenome.py +++ b/sentieon_cli/sentieon_pangenome.py @@ -284,6 +284,7 @@ def main(self, args: argparse.Namespace) -> None: self.logger.debug("VCF contigs are: %s", self.pop_vcf_contigs) self.validate() + self.start_run_logs() self.shards = determine_shards_from_fai( self.fai_data, 10 * 1000 * 1000 ) diff --git a/sentieon_cli/util.py b/sentieon_cli/util.py index 68021ea..fe9cc73 100644 --- a/sentieon_cli/util.py +++ b/sentieon_cli/util.py @@ -31,6 +31,13 @@ NUMA_NODE_PAT = re.compile(r"^NUMA node. CPU\(s\):\s+(?P.*)$") READ_LENGTH_PAT = re.compile(r"SN\taverage length:\t(?P\d*)$") +_UNSAFE = re.compile(r"[^A-Za-z0-9._-]") + + +def sanitize(component: str) -> str: + """Restrict a path component to filesystem-safe characters""" + return _UNSAFE.sub("-", component) + def tmp(): """Create a temporary directory for the current process.""" diff --git a/tests/unit/test_cli_args.py b/tests/unit/test_cli_args.py new file mode 100644 index 0000000..28909ed --- /dev/null +++ b/tests/unit/test_cli_args.py @@ -0,0 +1,40 @@ +""" +Unit tests for the top-level CLI arguments +""" + +import argparse +import os +import sys + +import pytest + +# Add the parent directory to the path so we can import sentieon_cli +sys.path.insert( + 0, + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")), +) + +from sentieon_cli import add_logging_args, resolve_loglevel # noqa: E402 + + +def _loglevel(argv): + parser = argparse.ArgumentParser() + add_logging_args(parser) + return resolve_loglevel(parser.parse_args(argv)) + + +@pytest.mark.parametrize( + "argv,expected", + [ + ([], "INFO"), + (["-v"], "INFO"), + (["-q"], "WARNING"), + (["-d"], "DEBUG"), + # `-d` wins wherever it appears; it used to be undone by a later + # `-v` because both flags wrote to one destination. + (["-d", "-v"], "DEBUG"), + (["-q", "-v"], "INFO"), + ], +) +def test_the_verbosity_flags_resolve_to_a_console_level(argv, expected): + assert _loglevel(argv) == expected diff --git a/tests/unit/test_job.py b/tests/unit/test_job.py index 4b37a58..f77205e 100644 --- a/tests/unit/test_job.py +++ b/tests/unit/test_job.py @@ -16,6 +16,7 @@ from sentieon_cli.dag import DAG # noqa: E402 from sentieon_cli.job import Job # noqa: E402 from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 +from sentieon_cli.util import sanitize # noqa: E402 def _job(name, arg, task_name="test"): @@ -45,6 +46,17 @@ def test_each_name_has_its_own_counter(): assert second.job_id == "dedup-2" +def test_names_differing_only_in_unsafe_characters_share_a_counter(): + # Log file names are sanitized, so ids that only differ in unsafe + # characters would name the same file and truncate each other. + first = _job("a b", "x") + second = _job("a-b", "y") + + assert (first.job_id, second.job_id) == ("a b-1", "a-b-2") + assert sanitize(first.job_id) == "a-b-1" + assert sanitize(second.job_id) == "a-b-2" + + def test_reset_ids_restarts_the_sequence(): assert _job("multiqc", "a").job_id == "multiqc-1" assert _job("multiqc", "b").job_id == "multiqc-2" diff --git a/tests/unit/test_job_logs.py b/tests/unit/test_job_logs.py index e438533..29395bc 100644 --- a/tests/unit/test_job_logs.py +++ b/tests/unit/test_job_logs.py @@ -13,12 +13,13 @@ from sentieon_cli.executor import TAIL_LINES, _log_tail # noqa: E402 from sentieon_cli.job import Job # noqa: E402 -from sentieon_cli.run_logs import RunLogs, sanitize # noqa: E402 +from sentieon_cli.run_logs import RunLogs # noqa: E402 from sentieon_cli.shell_pipeline import ( # noqa: E402 Command, Context, Pipeline, ) +from sentieon_cli.util import sanitize # noqa: E402 def _sink(tmp_path, name: str = "bwa", task_name: str = "alignment"): diff --git a/tests/unit/test_run_logs.py b/tests/unit/test_run_logs.py index 02b8eed..a9c84ff 100644 --- a/tests/unit/test_run_logs.py +++ b/tests/unit/test_run_logs.py @@ -49,6 +49,13 @@ def build_dag(self) -> DAG: raise RuntimeError("boom") +class _ValidatingPipeline(_DummyPipeline): + """A pipeline that checks its output path, as the real ones do.""" + + def validate(self) -> None: + self.validate_output_vcf() + + class _ErroredExecutor: """A stand-in executor that finished with failed jobs.""" @@ -81,6 +88,12 @@ def _args(loglevel: str = "INFO") -> argparse.Namespace: return argparse.Namespace(loglevel=loglevel) +def _start(pipeline: BasePipeline, loglevel: str = "INFO") -> None: + """Set up logging the way `main` does, minus the run itself.""" + pipeline.setup_logging(_args(loglevel)) + pipeline.start_run_logs() + + def test_module_loggers_delegate_to_the_package_logger(): module_logger = cli_logging.get_logger("sentieon_cli.example") @@ -92,7 +105,7 @@ def test_module_loggers_delegate_to_the_package_logger(): def test_default_log_dir_is_derived_from_the_output_vcf(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is not None assert pipeline.run_logs.log_dir == tmp_path / "sample_logs" @@ -102,7 +115,7 @@ def test_default_log_dir_only_strips_the_trailing_suffix(tmp_path): # `str.replace` would mangle a name with '.vcf.gz' in the middle. pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "a.vcf.gz.rerun.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs.log_dir == tmp_path / "a.vcf.gz.rerun_logs" @@ -111,7 +124,7 @@ def test_explicit_log_dir_skips_the_output_suffix_check(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.bcf" pipeline.log_dir = tmp_path / "elsewhere" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs.log_dir == tmp_path / "elsewhere" assert (tmp_path / "elsewhere" / "run.log").is_file() @@ -122,10 +135,64 @@ def test_invalid_output_suffix_exits_before_creating_a_log_dir(tmp_path): pipeline.output_vcf = tmp_path / "sample.bcf" with pytest.raises(SystemExit) as excinfo: - pipeline.setup_logging(_args()) + _start(pipeline) + + assert excinfo.value.code == 2 + assert list(tmp_path.iterdir()) == [] + + +def test_an_invalid_output_path_creates_no_log_dir(tmp_path, messages): + # `validate` runs first, so a typo in the output path is rejected before + # anything is written -- including the log directory next to it. + pipeline = _ValidatingPipeline() + pipeline.output_vcf = tmp_path / "typo" / "sample.vcf.gz" + + with pytest.raises(SystemExit) as excinfo: + pipeline.main(_args()) assert excinfo.value.code == 2 + assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] + assert any("status: failed" in msg for msg in messages) + + +def test_a_rejected_rerun_keeps_the_previous_runs_logs(tmp_path): + log_dir = tmp_path / "logs" + first = _ValidatingPipeline() + first.output_vcf = tmp_path / "sample.vcf.gz" + first.log_dir = log_dir + _start(first) + stale = first.run_logs.task_logs / "alignment" + stale.mkdir(parents=True) + (stale / "bwa-1.0.log").write_text("from the previous run") + first.run_logs.close() + run_log = (log_dir / "run.log").read_text() + + second = _ValidatingPipeline() + second.output_vcf = tmp_path / "typo" / "sample.vcf.gz" + second.log_dir = log_dir + with pytest.raises(SystemExit): + second.main(_args()) + + assert (stale / "bwa-1.0.log").read_text() == "from the previous run" + assert (log_dir / "run.log").read_text() == run_log + + +def test_a_log_dir_colliding_with_a_file_exits_cleanly(tmp_path, messages): + collision = tmp_path / "logs" + collision.write_text("not a directory") + pipeline = _DummyPipeline() + pipeline.log_dir = collision + + with pytest.raises(SystemExit) as excinfo: + _start(pipeline) + + assert excinfo.value.code == 2 + assert pipeline.run_logs is None + assert collision.read_text() == "not a directory" + assert any( + "Could not prepare the log directory" in msg for msg in messages + ) def test_setup_wipes_stale_task_logs_but_keeps_the_log_dir(tmp_path): @@ -160,7 +227,7 @@ def test_command_txt_records_the_invocation(tmp_path, monkeypatch): def test_run_log_captures_debug_while_the_console_stays_at_info(tmp_path): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args("INFO")) + _start(pipeline) logging.getLogger("sentieon_cli.example").debug("a debug record") run_log = pipeline.run_logs.run_log pipeline.run_logs.close() @@ -180,7 +247,7 @@ def test_repeated_setup_does_not_accumulate_handlers(tmp_path): for i in range(3): pipeline = _DummyPipeline() pipeline.output_vcf = tmp_path / f"sample{i}.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert len(package_logger.handlers) == before + 1 pipeline.run_logs.close() @@ -193,7 +260,7 @@ def test_rerun_truncates_the_previous_run_log(tmp_path): for _ in range(2): pipeline = _DummyPipeline() pipeline.log_dir = log_dir - pipeline.setup_logging(_args()) + _start(pipeline) run_log = pipeline.run_logs.run_log pipeline.run_logs.close() @@ -205,7 +272,7 @@ def test_dry_run_skips_file_logging(tmp_path): pipeline = _DummyPipeline() pipeline.dry_run = True pipeline.output_vcf = tmp_path / "sample.vcf.gz" - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] @@ -214,7 +281,7 @@ def test_dry_run_skips_file_logging(tmp_path): def test_bare_pipeline_setup_logging_creates_nothing(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) pipeline = _DummyPipeline() - pipeline.setup_logging(_args()) + _start(pipeline) assert pipeline.run_logs is None assert list(tmp_path.iterdir()) == [] @@ -223,7 +290,7 @@ def test_bare_pipeline_setup_logging_creates_nothing(tmp_path, monkeypatch): def test_check_execution_names_the_task_log_dir(tmp_path): pipeline = _DummyPipeline() pipeline.log_dir = tmp_path / "logs" - pipeline.setup_logging(_args()) + _start(pipeline) job = Job(Pipeline(Command("false")), "boom", task_name="failing") with pytest.raises(DagExecutionError) as excinfo: From 0d4150de2a42787b71a60265bfc89c5695278227 Mon Sep 17 00:00:00 2001 From: Don Freed Date: Fri, 14 Aug 2026 14:45:56 -0700 Subject: [PATCH 2/5] Record per-process rusage by owning the child reap Spawn pipeline processes with RusagePopen, a subprocess.Popen subclass whose _try_wait() reaps with os.wait4() and records the child's rusage, replacing asyncio.create_subprocess_exec (which only provided spawn and reap here: every stdio stream is already a real FD). Waits move to async_wait() (Popen.wait in a worker thread), and the executor raises the loop's default thread pool to 512 lazily-created workers so blocked waits and proc-sub FIFO opens cannot starve each other. Context.cleanup() gains a non-blocking poll() sweep so children nobody awaited are still reaped, as asyncio's watcher used to do. Behavior parity otherwise: negative returncodes on signal death, SIGPIPE tolerance, and the signal/kill paths are unchanged. Metrics emission lands separately. Co-Authored-By: Claude Fable 5 --- sentieon_cli/executor.py | 16 +++++++-- sentieon_cli/shell_pipeline.py | 58 ++++++++++++++++++++++++++----- tests/unit/test_shell_pipeline.py | 24 ++++++------- 3 files changed, 75 insertions(+), 23 deletions(-) diff --git a/sentieon_cli/executor.py b/sentieon_cli/executor.py index 3b9ad58..a84388a 100644 --- a/sentieon_cli/executor.py +++ b/sentieon_cli/executor.py @@ -1,6 +1,7 @@ """Execute jobs""" import asyncio +import concurrent.futures import contextlib import os import pathlib @@ -201,6 +202,15 @@ async def _execute(self) -> None: """Execute jobs from the DAG""" self.jobs_with_errors = [] loop = asyncio.get_running_loop() + # Blocked wait() threads and the proc-sub FIFO-open threads share the + # default pool, so it must be large enough that neither starves the + # other. Threads are created lazily, so the cap costs nothing unless + # it is used, and asyncio.run() shuts the executor down on exit. + loop.set_default_executor( + concurrent.futures.ThreadPoolExecutor( + max_workers=512, thread_name_prefix="sentieon-wait" + ) + ) stop_event = asyncio.Event() restore = self._install_signal_handlers(loop, stop_event) try: @@ -266,7 +276,7 @@ async def run_job(self, job: Job) -> None: ( job, context, - asyncio.create_task(proc.wait()), + asyncio.create_task(proc.async_wait()), start_time, ) ) @@ -346,7 +356,7 @@ async def _terminate_context(self, context: Context) -> None: if not live: return _signal_procs(context, signal.SIGTERM) - waits = [asyncio.create_task(proc.wait()) for proc in live] + waits = [asyncio.create_task(proc.async_wait()) for proc in live] await asyncio.wait(waits, timeout=self.shutdown_grace_period) _kill_survivors(context) await asyncio.wait(waits) @@ -397,7 +407,7 @@ async def _drive(self, stop_event: asyncio.Event) -> None: cmd_failed = True continue ret = ( - await subcommand.proc.wait() + await subcommand.proc.async_wait() ) # Wait on all sub-commands # A subcommand killed by SIGPIPE is not a failure: # it was writing to a pipe whose reader exited early diff --git a/sentieon_cli/shell_pipeline.py b/sentieon_cli/shell_pipeline.py index fa6aa7a..2020954 100644 --- a/sentieon_cli/shell_pipeline.py +++ b/sentieon_cli/shell_pipeline.py @@ -10,10 +10,21 @@ import fcntl import os import pathlib +import resource +import subprocess import tempfile import shlex from abc import ABC, abstractmethod -from typing import Any, Dict, IO, List, Optional, TYPE_CHECKING, Union +from typing import ( + Any, + Dict, + IO, + List, + Optional, + Tuple, + TYPE_CHECKING, + Union, +) from .logging import get_logger @@ -124,6 +135,14 @@ async def cleanup(self) -> None: finally: for fd in unblock_fds: os.close(fd) + # asyncio's child watcher reaped children nobody awaited; Popen does + # not, so sweep up already-exited processes here to avoid zombies for + # callers that only await the final pipeline stage. poll() is + # non-blocking: a process still running is left to its owner. + for command in self.commands: + proc = command.proc + if proc is not None and proc.returncode is None: + proc.poll() for fh in self.file_handles: fh.close() self.temp_dir.cleanup() @@ -135,6 +154,29 @@ async def cleanup(self) -> None: raise exc +class RusagePopen(subprocess.Popen[bytes]): + """A Popen that records the child's rusage when it is reaped.""" + + rusage: Optional[resource.struct_rusage] = None + + def _try_wait(self, wait_flags: int) -> Tuple[int, int]: + # Mirrors the base class, substituting wait4 for waitpid so the + # child's rusage is captured at reap time. Callers hold + # self._waitpid_lock (base-class contract). + try: + pid, sts, res = os.wait4(self.pid, wait_flags) + except ChildProcessError: + pid, sts = self.pid, 0 + else: + if pid == self.pid: + self.rusage = res + return (pid, sts) + + async def async_wait(self) -> int: + """Await the child's exit without blocking the event loop.""" + return await asyncio.to_thread(self.wait) + + class ShellNode(ABC): """Abstract base class for any node in the shell syntax tree.""" @@ -152,7 +194,7 @@ def __init__( self.executable = executable self.args = list(args) self.fail_ok = fail_ok - self.proc: Union[asyncio.subprocess.Process, None] = None + self.proc: Optional[RusagePopen] = None self.exec_kwargs: Dict[str, Any] = exec_kwargs if exec_kwargs else {} async def run( @@ -161,7 +203,7 @@ async def run( stdin: Union[IO[Any], int, None] = None, stdout: Union[IO[Any], int, None] = None, stderr: Union[IO[Any], int, None] = None, - ) -> asyncio.subprocess.Process: + ) -> RusagePopen: # 1. Resolve Arguments (Handle Process Substitutions) final_args = [self.executable] @@ -188,8 +230,8 @@ async def run( # 3. Run the process # We allow the caller to wait for this process - self.proc = await asyncio.create_subprocess_exec( - *final_args, + self.proc = RusagePopen( + final_args, stdin=stdin, stdout=stdout, stderr=stderr, @@ -266,7 +308,7 @@ async def run( stdin: Union[IO[Any], int, None] = None, stdout: Union[IO[Any], int, None] = None, stderr: Union[IO[Any], int, None] = None, - ) -> asyncio.subprocess.Process: + ) -> RusagePopen: # We cannot have handles from both files and to run(). if self.file_input is not None and stdin is not None: raise ValueError( @@ -429,7 +471,7 @@ async def run_inner() -> None: context, stdout=write_handle, ) - await proc.wait() + await proc.async_wait() finally: write_handle.close() @@ -467,7 +509,7 @@ async def run_inner() -> None: if context.closing: return # cleanup unblocked us; do not spawn proc = await self.node.run(context, stdin=read_handle) - await proc.wait() + await proc.async_wait() finally: read_handle.close() diff --git a/tests/unit/test_shell_pipeline.py b/tests/unit/test_shell_pipeline.py index 38ec2bd..c17b9b8 100644 --- a/tests/unit/test_shell_pipeline.py +++ b/tests/unit/test_shell_pipeline.py @@ -34,7 +34,7 @@ async def test_simple_command(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await cmd.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -56,7 +56,7 @@ async def test_simple_pipeline(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await pipeline.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -89,7 +89,7 @@ async def test_pipeline_with_file_io(): ) proc = await pipeline.run(context) - await proc.wait() + await proc.async_wait() with open(outfile_path, "r") as f: output = f.read().strip() @@ -114,7 +114,7 @@ async def test_input_process_substitution(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await cmd.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() stdout_file.seek(0) output = stdout_file.read().strip() @@ -149,7 +149,7 @@ async def test_output_process_substitution(): with tempfile.NamedTemporaryFile(mode="w+", delete=False) as stdout_file: proc = await main_pipeline.run(context, stdout=stdout_file) - await proc.wait() + await proc.async_wait() # Wait for background tasks from process substitution to finish await asyncio.gather(*context.tasks) @@ -198,10 +198,10 @@ async def test_pipe_size_enlarges_internal_pipes(tmp_path): pipe_size=target, ) proc = await pipeline.run(context) - await proc.wait() + await proc.async_wait() for sub in context.commands: if sub.proc: - await sub.proc.wait() + await sub.proc.async_wait() await context.cleanup() assert int(report.read_text()) == target @@ -347,7 +347,7 @@ async def test_cleanup_after_outer_exits_without_opening(): context = Context() cmd = Command("false", InputProcSub(Pipeline(Command("echo", "x")))) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) @@ -364,7 +364,7 @@ async def test_cleanup_unblocks_multiple_procsubs(): OutputProcSub(Pipeline(Command("cat"))), ) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) @@ -380,7 +380,7 @@ async def test_cleanup_still_raises_inner_launch_failure(): "cat", InputProcSub(Pipeline(Command("no_such_cmd_zzz"))) ) proc = await cmd.run(context) - await proc.wait() + await proc.async_wait() with pytest.raises(FileNotFoundError): await asyncio.wait_for(context.cleanup(), timeout=10) @@ -408,7 +408,7 @@ async def test_cleanup_does_not_stall_running_inner_writer(): await asyncio.sleep(0.01) await asyncio.wait_for(context.cleanup(), timeout=10) - assert await proc.wait() == 0 + assert await proc.async_wait() == 0 @pytest.mark.asyncio @@ -417,7 +417,7 @@ async def test_cleanup_twice_is_safe(): context = Context() cmd = Command("cat", InputProcSub(Pipeline(Command("echo", "hello")))) proc = await cmd.run(context, stdout=asyncio.subprocess.DEVNULL) - await proc.wait() + await proc.async_wait() await asyncio.wait_for(context.cleanup(), timeout=10) await asyncio.wait_for(context.cleanup(), timeout=10) From be5b184ddad78e9704f9485dc3697a2d72475350 Mon Sep 17 00:00:00 2001 From: Don Freed Date: Fri, 14 Aug 2026 15:08:13 -0700 Subject: [PATCH 3/5] Report per-process rusage in the run logs Each reaped process now logs a DEBUG line to run.log with its kernel resource usage (utime, stime, peak RSS, page faults, block I/O, context switches), emitted whatever the job's outcome -- signal-killed children included. Successful jobs extend the "Finished command" INFO line with the job's summed user/sys CPU and the per-process peak RSS (a sum would overstate concurrent stages that peak at different times). ru_maxrss is normalized across platforms (bytes on macOS, KiB elsewhere), and the DEBUG line renders it without an internal space (maxrss=33.6MiB) so the key=value line stays space-tokenizable for grep/awk. Co-Authored-By: Claude Fable 5 --- sentieon_cli/executor.py | 63 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/sentieon_cli/executor.py b/sentieon_cli/executor.py index a84388a..0e8f64e 100644 --- a/sentieon_cli/executor.py +++ b/sentieon_cli/executor.py @@ -5,6 +5,7 @@ import contextlib import os import pathlib +import resource import signal import sys import threading @@ -44,6 +45,22 @@ def _log_tail(path: pathlib.Path, lines: int = TAIL_LINES) -> List[str]: return data.decode("utf-8", errors="replace").splitlines()[-lines:] +def _maxrss_bytes(ru: resource.struct_rusage) -> int: + """Return a child's peak resident set size in bytes. + + ``ru_maxrss`` is already bytes on macOS; every other platform we run on + reports kibibytes. + """ + if sys.platform == "darwin": + return int(ru.ru_maxrss) + return int(ru.ru_maxrss) * 1024 + + +def _mib(nbytes: int) -> str: + """Render a byte count as MiB for the human-readable log lines.""" + return f"{nbytes / (1024 * 1024):.1f} MiB" + + def _signal_procs(context: Context, sig: int) -> None: """Send a signal to every live sub-process in a context.""" for subcommand in context.commands: @@ -398,6 +415,14 @@ async def _drive(self, stop_event: asyncio.Event) -> None: # Check if the command failed cmd_failed = False failures: List[Tuple[Command, int]] = [] + # Kernel resource usage, aggregated over the job's + # processes. RSS is the per-process peak rather than a + # sum: concurrent pipeline stages peak at different + # times, so a sum would overstate the job. + have_rusage = False + job_utime = 0.0 + job_stime = 0.0 + job_maxrss = 0 for subcommand in context.commands: if not subcommand.proc: logger.error( @@ -409,6 +434,34 @@ async def _drive(self, stop_event: asyncio.Event) -> None: ret = ( await subcommand.proc.async_wait() ) # Wait on all sub-commands + # Reported for every reaped process, whatever the + # job's outcome; signal-killed children have rusage + # too. + rusage = subcommand.proc.rusage + if rusage is not None: + rss = _maxrss_bytes(rusage) + have_rusage = True + job_utime += rusage.ru_utime + job_stime += rusage.ru_stime + job_maxrss = max(job_maxrss, rss) + logger.debug( + "rusage for %s [pid %d, %s]: utime=%.2fs " + "stime=%.2fs maxrss=%.1fMiB minflt=%d " + "majflt=%d inblock=%d oublock=%d " + "nvcsw=%d nivcsw=%d", + job, + subcommand.proc.pid, + subcommand, + rusage.ru_utime, + rusage.ru_stime, + rss / (1024 * 1024), + rusage.ru_minflt, + rusage.ru_majflt, + rusage.ru_inblock, + rusage.ru_oublock, + rusage.ru_nvcsw, + rusage.ru_nivcsw, + ) # A subcommand killed by SIGPIPE is not a failure: # it was writing to a pipe whose reader exited early # (e.g. `... | head`), which is normal in default @@ -458,6 +511,16 @@ async def _drive(self, stop_event: asyncio.Event) -> None: ) self.jobs_with_errors.append(job) self.start_new_jobs = False + elif have_rusage: + logger.info( + "Finished command in %.2fs (user %.1fs, " + "sys %.1fs, max proc RSS %s): %s", + total_seconds, + job_utime, + job_stime, + _mib(job_maxrss), + job.shell, + ) else: logger.info( f"Finished command in " From fb9fa2c6c5dcebc382fe2d1d5fe7e3ba58f1faeb Mon Sep 17 00:00:00 2001 From: Don Freed Date: Fri, 14 Aug 2026 15:21:03 -0700 Subject: [PATCH 4/5] Test rusage capture and the per-process metrics report test_simple_command now asserts rusage is recorded at reap, and a new signal-death test pins the RusagePopen._try_wait override: if a future CPython changes that private method's shape, the override silently stops being called and rusage stays None, so the test asserts a SIGTERM'd child reports both its negative return code and its rusage. New tests/unit/test_rusage_metrics.py drives a job through LocalExecutor (the first unit-level file to do so) and checks the per-process "rusage for" DEBUG record, the aggregated "Finished command" INFO line, and _maxrss_bytes unit normalization on both platform branches. Co-Authored-By: Claude Fable 5 --- tests/unit/test_rusage_metrics.py | 99 +++++++++++++++++++++++++++++++ tests/unit/test_shell_pipeline.py | 24 ++++++++ 2 files changed, 123 insertions(+) create mode 100644 tests/unit/test_rusage_metrics.py diff --git a/tests/unit/test_rusage_metrics.py b/tests/unit/test_rusage_metrics.py new file mode 100644 index 0000000..a7c0b53 --- /dev/null +++ b/tests/unit/test_rusage_metrics.py @@ -0,0 +1,99 @@ +""" +Unit tests for the per-process resource-usage metrics +""" + +import logging +import os +import pathlib +import resource +import sys +from typing import List + +import pytest + +# Add the parent directory to the path so we can import sentieon_cli +sys.path.insert( + 0, + os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")), +) + +from sentieon_cli.dag import DAG # noqa: E402 +from sentieon_cli.executor import ( # noqa: E402 + LocalExecutor, + _maxrss_bytes, +) +from sentieon_cli.job import Job # noqa: E402 +from sentieon_cli.run_logs import RunLogs # noqa: E402 +from sentieon_cli.scheduler import ThreadScheduler # noqa: E402 +from sentieon_cli.shell_pipeline import Command, Pipeline # noqa: E402 + + +class _RecordingHandler(logging.Handler): + """Collects formatted messages from the package logger.""" + + def __init__(self) -> None: + super().__init__(logging.DEBUG) + self.messages: List[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append(record.getMessage()) + + +@pytest.fixture +def messages(): + """Capture the package logger, which does not propagate to the root.""" + handler = _RecordingHandler() + package_logger = logging.getLogger("sentieon_cli") + package_logger.addHandler(handler) + yield handler.messages + package_logger.removeHandler(handler) + + +def _run_one_job(log_dir: pathlib.Path) -> None: + """Run a single trivial job with its output captured under ``log_dir``""" + dag = DAG() + dag.add_job(Job(Pipeline(Command("true")), "metrics", task_name="metrics")) + run_logs = RunLogs(log_dir) + run_logs.create_dirs() + LocalExecutor(ThreadScheduler(dag, 1), run_logs=run_logs).execute() + + +def test_each_process_of_a_job_reports_its_rusage(tmp_path, messages): + _run_one_job(tmp_path / "logs") + + rusage_lines = [m for m in messages if m.startswith("rusage for ")] + assert len(rusage_lines) == 1 + line = rusage_lines[0] + assert "Job(metrics-1)" in line + for field in ("utime=", "stime=", "maxrss=", "minflt=", "nivcsw="): + assert field in line + + +def test_a_finished_job_reports_its_aggregated_metrics(tmp_path, messages): + _run_one_job(tmp_path / "logs") + + finished = [m for m in messages if m.startswith("Finished command in")] + assert len(finished) == 1 + assert "user " in finished[0] + assert "sys " in finished[0] + assert "max proc RSS" in finished[0] + + +def test_maxrss_is_reported_in_bytes(): + ru = resource.getrusage(resource.RUSAGE_SELF) + + maxrss = _maxrss_bytes(ru) + + assert isinstance(maxrss, int) + assert maxrss > 0 + + +def test_maxrss_units_follow_the_platform(monkeypatch): + """macOS reports bytes; everywhere else reports kibibytes.""" + ru = resource.getrusage(resource.RUSAGE_SELF) + + monkeypatch.setattr(sys, "platform", "darwin") + assert _maxrss_bytes(ru) == ru.ru_maxrss + + monkeypatch.setattr(sys, "platform", "linux") + assert _maxrss_bytes(ru) == ru.ru_maxrss * 1024 diff --git a/tests/unit/test_shell_pipeline.py b/tests/unit/test_shell_pipeline.py index c17b9b8..a6d704c 100644 --- a/tests/unit/test_shell_pipeline.py +++ b/tests/unit/test_shell_pipeline.py @@ -6,6 +6,7 @@ import fcntl import os import pathlib +import signal import sys import tempfile @@ -44,6 +45,29 @@ async def test_simple_command(): assert output == "hello world" assert proc.returncode == 0 + # wait4 captured the child's resource usage as it was reaped + assert proc.rusage is not None + assert proc.rusage.ru_maxrss > 0 + + +@pytest.mark.asyncio +async def test_a_signalled_command_reports_its_signal_and_its_rusage(): + """A child killed by a signal is still reaped through ``wait4``. + + This pins ``RusagePopen._try_wait``: if a future CPython changes that + private method's shape, the override stops being called and ``rusage`` + silently stays None while the negative return code still works. + """ + cmd = Command("sleep", "5") + context = Context() + + proc = await cmd.run(context) + proc.send_signal(signal.SIGTERM) + + assert await proc.async_wait() == -signal.SIGTERM + assert proc.rusage is not None + + await context.cleanup() @pytest.mark.asyncio From 76394ba599ebea9d4b33052fab4f3fcff803c8d8 Mon Sep 17 00:00:00 2001 From: Don Freed Date: Fri, 14 Aug 2026 16:42:43 -0700 Subject: [PATCH 5/5] Tear down children when the run loop fails; make the wait pool tunable Two fixes from a code review of the branch. First, an exception escaping _drive now tears the children down (_shutdown) before propagating: it previously unwound into asyncio.run's shutdown, which joins the pool threads blocked in wait4 on children nobody signalled, hanging the CLI until the children exited on their own. BaseException, so cancellation and KeyboardInterrupt are covered; a teardown failure is logged rather than allowed to mask the original error. Second, the wait/FIFO-open thread pool size is a keyword parameter instead of a hidden constant: None on AsyncExecutor (an executor that spawns nothing leaves the loop's default pool alone) and 512 on LocalExecutor, beside shutdown_grace_period. async_wait() returns immediately for an already-reaped child instead of burning a pool dispatch and a slot. Also document a known limitation honestly: a reap via poll() -- or send_signal/kill, which poll internally -- uses the base Popen machinery and records no rusage. The executor waits every process it reports metrics for, so those paths only collect processes whose rusage is never read. Co-Authored-By: Claude Fable 5 --- sentieon_cli/executor.py | 50 ++++++++++++++++++++++------ sentieon_cli/shell_pipeline.py | 13 +++++++- tests/integration/test_executor.py | 53 ++++++++++++++++++++++++++++++ 3 files changed, 105 insertions(+), 11 deletions(-) diff --git a/sentieon_cli/executor.py b/sentieon_cli/executor.py index 0e8f64e..2fecf6f 100644 --- a/sentieon_cli/executor.py +++ b/sentieon_cli/executor.py @@ -144,6 +144,10 @@ class AsyncExecutor(BaseExecutor, ABC): Set ``install_signal_handlers=True`` to install handlers (on the main thread only) for the duration of a run; the previous handlers are restored afterwards, so a caller's signal state is left untouched. + + ``thread_pool_size`` sizes the loop's default thread pool for the run; + ``None`` leaves the loop's own default alone, which is what an executor + that spawns no processes wants. """ def __init__( @@ -151,9 +155,11 @@ def __init__( scheduler: BaseScheduler, *, install_signal_handlers: bool = False, + thread_pool_size: Optional[int] = None, ) -> None: super().__init__(scheduler) self.install_signal_handlers = install_signal_handlers + self.thread_pool_size = thread_pool_size self.start_new_jobs = True def _install_signal_handlers( @@ -219,19 +225,34 @@ async def _execute(self) -> None: """Execute jobs from the DAG""" self.jobs_with_errors = [] loop = asyncio.get_running_loop() - # Blocked wait() threads and the proc-sub FIFO-open threads share the - # default pool, so it must be large enough that neither starves the - # other. Threads are created lazily, so the cap costs nothing unless - # it is used, and asyncio.run() shuts the executor down on exit. - loop.set_default_executor( - concurrent.futures.ThreadPoolExecutor( - max_workers=512, thread_name_prefix="sentieon-wait" + if self.thread_pool_size is not None: + # Threads are created lazily, so a generous cap costs nothing + # unless it is used, and asyncio.run() shuts the executor down on + # exit. + loop.set_default_executor( + concurrent.futures.ThreadPoolExecutor( + max_workers=self.thread_pool_size, + thread_name_prefix="sentieon-wait", + ) ) - ) stop_event = asyncio.Event() restore = self._install_signal_handlers(loop, stop_event) try: - await self._drive(stop_event) + try: + await self._drive(stop_event) + except BaseException: + # Tear the children down before the exception unwinds into + # asyncio.run's Runner.close, which joins the wait threads -- + # and those threads are blocked in wait4 on live children. + # BaseException so cancellation and KeyboardInterrupt are + # covered too. + self.start_new_jobs = False + try: + await self._shutdown() + except Exception as exc: + # A failed teardown must not mask the original error. + logger.error("teardown after failure also failed: %s", exc) + raise if stop_event.is_set(): await self._shutdown() finally: @@ -247,7 +268,14 @@ async def _shutdown(self) -> None: class LocalExecutor(AsyncExecutor): - """Run jobs locally as async subprocesses.""" + """Run jobs locally as sub-processes, driven by an asyncio loop. + + ``thread_pool_size`` defaults high because the loop's default pool serves + two kinds of blocking work at once: one thread per in-flight ``wait()`` + and the proc-sub FIFO opens. A FIFO open queued behind saturated wait + threads would deadlock the run, so the pool must comfortably exceed the + number of processes a run keeps in flight. + """ def __init__( self, @@ -256,10 +284,12 @@ def __init__( install_signal_handlers: bool = False, shutdown_grace_period: float = 10.0, run_logs: Optional[RunLogs] = None, + thread_pool_size: int = 512, ) -> None: super().__init__( scheduler, install_signal_handlers=install_signal_handlers, + thread_pool_size=thread_pool_size, ) self.shutdown_grace_period = shutdown_grace_period self.run_logs = run_logs diff --git a/sentieon_cli/shell_pipeline.py b/sentieon_cli/shell_pipeline.py index 2020954..d2d1277 100644 --- a/sentieon_cli/shell_pipeline.py +++ b/sentieon_cli/shell_pipeline.py @@ -155,7 +155,15 @@ async def cleanup(self) -> None: class RusagePopen(subprocess.Popen[bytes]): - """A Popen that records the child's rusage when it is reaped.""" + """A Popen that records the child's rusage when it is reaped. + + Only the ``wait``/``async_wait`` path reaps through ``_try_wait`` and + sets ``rusage``; a reap via ``poll`` -- or ``send_signal``/``kill``, + which poll internally -- goes through the base class's machinery and + leaves ``rusage`` None. The executor waits every process it reports + metrics for, so those paths only collect processes whose rusage is + never read. + """ rusage: Optional[resource.struct_rusage] = None @@ -174,6 +182,9 @@ def _try_wait(self, wait_flags: int) -> Tuple[int, int]: async def async_wait(self) -> int: """Await the child's exit without blocking the event loop.""" + if self.returncode is not None: + # Already reaped; no need for a pool thread. + return self.returncode return await asyncio.to_thread(self.wait) diff --git a/tests/integration/test_executor.py b/tests/integration/test_executor.py index 41ce529..d0ea7f5 100644 --- a/tests/integration/test_executor.py +++ b/tests/integration/test_executor.py @@ -8,6 +8,7 @@ import sys import tempfile import threading +import time import pytest @@ -475,6 +476,58 @@ def __init__(self) -> None: assert proc.returncode is not None +def test_error_in_the_run_loop_terminates_running_jobs(monkeypatch): + """An exception escaping the run loop must tear the children down first. + + Otherwise it unwinds into asyncio.run's shutdown, which joins the wait + threads -- and those are blocked in wait4 on children nobody signalled, + so the run hangs until they exit on their own (regression: a `sleep 30` + held the CLI for 30s and was never terminated).""" + import sentieon_cli.executor as executor_mod + + contexts = [] + real_context = executor_mod.Context + + class RecordingContext(real_context): # type: ignore[valid-type,misc] + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + contexts.append(self) + + monkeypatch.setattr(executor_mod, "Context", RecordingContext) + + dag = DAG() + dag.add_job(Job(Pipeline(Command("sleep", "30")), "slow", task_name="t")) + dag.add_job(Job(Pipeline(Command("true")), "quick", task_name="t")) + + scheduler = ThreadScheduler(dag, 2) + + def boom(job): + # Fires when the quick job finishes, with the slow one still running + raise RuntimeError("scheduler exploded") + + monkeypatch.setattr(scheduler, "job_finished", boom) + + executor = LocalExecutor(scheduler, shutdown_grace_period=0.5) + start = time.monotonic() + with pytest.raises(RuntimeError): + executor.execute() + elapsed = time.monotonic() - start + + assert elapsed < 15, "the run waited for the sleep instead of killing it" + spawned = [sub.proc for c in contexts for sub in c.commands if sub.proc] + assert spawned, "expected both jobs to have spawned" + for proc in spawned: + assert proc.returncode is not None + + +def test_thread_pool_size_is_configurable(): + """The wait/FIFO-open pool cap is a parameter, not a hidden constant.""" + scheduler = ThreadScheduler(DAG(), 1) + + assert LocalExecutor(scheduler).thread_pool_size == 512 + assert LocalExecutor(scheduler, thread_pool_size=8).thread_pool_size == 8 + + def test_sigpipe_producer_does_not_fail_job(tmp_path): """A producer killed by SIGPIPE when a consumer exits early (e.g. `seq | head -1`) produces the right output and must not fail the job