From 7dd74fa15409c2a53effbf0d274d047b48a777a1 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 24 Jul 2026 16:09:35 +0200 Subject: [PATCH 1/7] CI: Stacktraces of `pytest` and `xdist`-workers on `timeout` --- justfile | 4 +- libs/opsqueue_python/tests/conftest.py | 116 ++++++++++++++++++++++++- 2 files changed, 117 insertions(+), 3 deletions(-) diff --git a/justfile b/justfile index 0cba2f3..b1027d8 100644 --- a/justfile +++ b/justfile @@ -53,7 +53,7 @@ test-integration *TEST_ARGS: build-bin build-python cd libs/opsqueue_python source "./.setup_local_venv.sh" - timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} + timeout --signal term --kill-after 1s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest [group('nix')] @@ -68,7 +68,7 @@ nix-test-integration *TEST_ARGS: nix-build export OPSQUEUE_BIN="${nix_build_bin_dir}/bin/opsqueue" export RUST_LOG="opsqueue=debug" - timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} + timeout --signal term --kill-after 1s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Run all linters, fast and slow [group('lint')] diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index 0052bca..c8778a7 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -1,5 +1,9 @@ import cbor2 import pickle +import faulthandler +import signal +import sys +import time from contextlib import contextmanager, ExitStack from typing import Generator, Callable, Any, Iterable import multiprocess # type: ignore[import-untyped] @@ -14,9 +18,119 @@ from opsqueue.consumer import Strategy +_previous_sigterm_handler: Any = None +_sigusr1_dump_file: Any | None = None + + +def _stack_dump_path(pid: int) -> Path: + return Path(f"/tmp/opsqueue-pytest-stack-{pid}.log") + + +def _linux_child_pids(parent_pid: int) -> set[int]: + children_file = Path(f"/proc/{parent_pid}/task/{parent_pid}/children") + try: + raw = children_file.read_text(encoding="utf-8").strip() + except FileNotFoundError: + return set() + + if not raw: + return set() + + return {int(pid_text) for pid_text in raw.split()} + + +def _dump_xdist_worker_traces() -> None: + worker_pids = _linux_child_pids(os.getpid()) + if not worker_pids: + print( + "[opsqueue pytest] No child worker PIDs found at SIGTERM time", + file=sys.stderr, + flush=True, + ) + return + + print( + f"[opsqueue pytest] Requesting stack dump from child workers: {worker_pids}", + file=sys.stderr, + flush=True, + ) + for pid in worker_pids: + try: + os.kill(pid, signal.SIGUSR1) + except ProcessLookupError: + # Worker already exited. + continue + except Exception as exc: # pragma: no cover - defensive diagnostics + print( + f"[opsqueue pytest] Failed sending SIGUSR1 to child pid={pid}: {exc}", + file=sys.stderr, + flush=True, + ) + + # Give workers a short moment to flush their dump files before reading them. + time.sleep(0.15) + for pid in sorted(worker_pids): + dump_path = _stack_dump_path(pid) + if not dump_path.exists(): + print( + f"[opsqueue pytest] No SIGUSR1 stack dump file found for child pid={pid}: {dump_path}", + file=sys.stderr, + flush=True, + ) + continue + + print( + f"[opsqueue pytest] ===== BEGIN CHILD STACK DUMP pid={pid} ({dump_path}) =====", + file=sys.stderr, + flush=True, + ) + try: + print(dump_path.read_text(encoding="utf-8"), file=sys.stderr, flush=True) + except Exception as exc: # pragma: no cover - defensive diagnostics + print( + f"[opsqueue pytest] Failed reading dump file for child pid={pid}: {exc}", + file=sys.stderr, + flush=True, + ) + print( + f"[opsqueue pytest] ===== END CHILD STACK DUMP pid={pid} =====", + file=sys.stderr, + flush=True, + ) + + +def _handle_sigterm(signum: int, frame: object) -> None: + faulthandler.dump_traceback(file=sys.stderr, all_threads=True) + _dump_xdist_worker_traces() + + if callable(_previous_sigterm_handler): + _previous_sigterm_handler(signum, frame) + elif _previous_sigterm_handler == signal.SIG_DFL: + signal.signal(signal.SIGTERM, signal.SIG_DFL) + os.kill(os.getpid(), signal.SIGTERM) + # If previous handler was SIG_IGN, return and keep process alive. + # Timeout will eventually kill the process if it doesn't exit on its own. + + @pytest.hookimpl(tryfirst=True) def pytest_configure(config: pytest.Config) -> None: - print("A") + global _previous_sigterm_handler + global _sigusr1_dump_file + + _sigusr1_dump_file = _stack_dump_path(os.getpid()).open("w", encoding="utf-8") + + # Make SIGUSR1 print a traceback in any process (controller or worker). + faulthandler.register( + signal.SIGUSR1, + file=_sigusr1_dump_file, + all_threads=True, + chain=True, + ) + + # When `timeout` sends SIGTERM to the controller, request worker dumps too. + faulthandler.enable(all_threads=True) + _previous_sigterm_handler = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGTERM, _handle_sigterm) multiprocess.set_start_method("forkserver") From 492107ff5c45b8afbee17ee62528e71736b13e1a Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 12:48:39 +0200 Subject: [PATCH 2/7] fixup! CI: Stacktraces of `pytest` and `xdist`-workers on `timeout` --- justfile | 4 +- libs/opsqueue_python/tests/conftest.py | 78 ++++++++++++++------------ 2 files changed, 44 insertions(+), 38 deletions(-) diff --git a/justfile b/justfile index b1027d8..4daf7a9 100644 --- a/justfile +++ b/justfile @@ -53,7 +53,7 @@ test-integration *TEST_ARGS: build-bin build-python cd libs/opsqueue_python source "./.setup_local_venv.sh" - timeout --signal term --kill-after 1s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} + timeout --signal term --kill-after 2s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest [group('nix')] @@ -68,7 +68,7 @@ nix-test-integration *TEST_ARGS: nix-build export OPSQUEUE_BIN="${nix_build_bin_dir}/bin/opsqueue" export RUST_LOG="opsqueue=debug" - timeout --signal term --kill-after 1s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} + timeout --signal term --kill-after 2s {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Run all linters, fast and slow [group('lint')] diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index c8778a7..afa7f89 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -18,7 +18,6 @@ from opsqueue.consumer import Strategy -_previous_sigterm_handler: Any = None _sigusr1_dump_file: Any | None = None @@ -26,17 +25,24 @@ def _stack_dump_path(pid: int) -> Path: return Path(f"/tmp/opsqueue-pytest-stack-{pid}.log") -def _linux_child_pids(parent_pid: int) -> set[int]: +def _linux_child_pids(parent_pid: int) -> dict[int, tuple[str, str]]: children_file = Path(f"/proc/{parent_pid}/task/{parent_pid}/children") try: raw = children_file.read_text(encoding="utf-8").strip() - except FileNotFoundError: - return set() + except OSError: + return {} if not raw: - return set() + return {} - return {int(pid_text) for pid_text in raw.split()} + def _child_command(pid: int) -> tuple[str, str]: + try: + return (Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip(), Path(f"/proc/{pid}/cmdline").read_text(encoding="utf-8").replace("\x00", " ").strip()) + except OSError: + return ("", "") + + pids = [int(pid_text) for pid_text in raw.split() if pid_text.isdigit()] + return {pid: _child_command(pid) for pid in pids} def _dump_xdist_worker_traces() -> None: @@ -54,7 +60,7 @@ def _dump_xdist_worker_traces() -> None: file=sys.stderr, flush=True, ) - for pid in worker_pids: + for pid in worker_pids.keys(): try: os.kill(pid, signal.SIGUSR1) except ProcessLookupError: @@ -68,19 +74,19 @@ def _dump_xdist_worker_traces() -> None: ) # Give workers a short moment to flush their dump files before reading them. - time.sleep(0.15) - for pid in sorted(worker_pids): + time.sleep(0.10) + for pid in sorted(worker_pids.keys()): dump_path = _stack_dump_path(pid) if not dump_path.exists(): print( - f"[opsqueue pytest] No SIGUSR1 stack dump file found for child pid={pid}: {dump_path}", + f"[opsqueue pytest] No SIGUSR1 stack dump file found for child pid={pid} {worker_pids[pid]}: {dump_path}", file=sys.stderr, flush=True, ) continue print( - f"[opsqueue pytest] ===== BEGIN CHILD STACK DUMP pid={pid} ({dump_path}) =====", + f"[opsqueue pytest] ===== BEGIN CHILD STACK DUMP pid={pid} {worker_pids[pid]} ({dump_path}) =====", file=sys.stderr, flush=True, ) @@ -88,49 +94,49 @@ def _dump_xdist_worker_traces() -> None: print(dump_path.read_text(encoding="utf-8"), file=sys.stderr, flush=True) except Exception as exc: # pragma: no cover - defensive diagnostics print( - f"[opsqueue pytest] Failed reading dump file for child pid={pid}: {exc}", + f"[opsqueue pytest] Failed reading dump file for child pid={pid} {worker_pids[pid]}: {exc}", file=sys.stderr, flush=True, ) print( - f"[opsqueue pytest] ===== END CHILD STACK DUMP pid={pid} =====", + f"[opsqueue pytest] ===== END CHILD STACK DUMP pid={pid} {worker_pids[pid]} =====", file=sys.stderr, flush=True, ) def _handle_sigterm(signum: int, frame: object) -> None: - faulthandler.dump_traceback(file=sys.stderr, all_threads=True) + print( + "[opsqueue pytest] ===== SIGTERM: Start of diagnostic output =====", + file=sys.stderr, + flush=True, + ) _dump_xdist_worker_traces() + faulthandler.dump_traceback(file=sys.stderr) + print( + "[opsqueue pytest] ===== SIGTERM: End of diagnostic output =====", + file=sys.stderr, + flush=True, + ) + os._exit(1) - if callable(_previous_sigterm_handler): - _previous_sigterm_handler(signum, frame) - elif _previous_sigterm_handler == signal.SIG_DFL: - signal.signal(signal.SIGTERM, signal.SIG_DFL) - os.kill(os.getpid(), signal.SIGTERM) # If previous handler was SIG_IGN, return and keep process alive. # Timeout will eventually kill the process if it doesn't exit on its own. +# Register signal handlers at module import time (before pytest_configure hook runs). +_sigusr1_dump_file = _stack_dump_path(os.getpid()).open("w", encoding="utf-8") +# Make SIGUSR1 print a traceback in any process (controller or worker). +faulthandler.register( + signal.SIGUSR1, + file=_sigusr1_dump_file, + all_threads=True, + chain=False, +) +signal.signal(signal.SIGTERM, _handle_sigterm) + @pytest.hookimpl(tryfirst=True) def pytest_configure(config: pytest.Config) -> None: - global _previous_sigterm_handler - global _sigusr1_dump_file - - _sigusr1_dump_file = _stack_dump_path(os.getpid()).open("w", encoding="utf-8") - - # Make SIGUSR1 print a traceback in any process (controller or worker). - faulthandler.register( - signal.SIGUSR1, - file=_sigusr1_dump_file, - all_threads=True, - chain=True, - ) - - # When `timeout` sends SIGTERM to the controller, request worker dumps too. - faulthandler.enable(all_threads=True) - _previous_sigterm_handler = signal.getsignal(signal.SIGTERM) - signal.signal(signal.SIGTERM, _handle_sigterm) multiprocess.set_start_method("forkserver") From 62ef9ad133431286cb4d8ff71d0245bf4eada6ef Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 13:53:17 +0200 Subject: [PATCH 3/7] fixup! fixup! CI: Stacktraces of `pytest` and `xdist`-workers on `timeout` --- libs/opsqueue_python/tests/conftest.py | 70 +++++++++++++------------- 1 file changed, 34 insertions(+), 36 deletions(-) diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index afa7f89..d13f8e9 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -3,7 +3,6 @@ import faulthandler import signal import sys -import time from contextlib import contextmanager, ExitStack from typing import Generator, Callable, Any, Iterable import multiprocess # type: ignore[import-untyped] @@ -37,7 +36,15 @@ def _linux_child_pids(parent_pid: int) -> dict[int, tuple[str, str]]: def _child_command(pid: int) -> tuple[str, str]: try: - return (Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip(), Path(f"/proc/{pid}/cmdline").read_text(encoding="utf-8").replace("\x00", " ").strip()) + return ( + Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip(), + Path(f"/proc/{pid}/cmdline") + .read_text(encoding="utf-8") + .replace( + "\x00", " " + ) # Replace null bytes with spaces before stripping. + .strip(), + ) except OSError: return ("", "") @@ -45,36 +52,38 @@ def _child_command(pid: int) -> tuple[str, str]: return {pid: _child_command(pid) for pid in pids} -def _dump_xdist_worker_traces() -> None: +def _handle_sigterm(signum: int, frame: object) -> None: worker_pids = _linux_child_pids(os.getpid()) - if not worker_pids: + if worker_pids: + print( + f"[opsqueue pytest] Requesting stack dump from child workers: {worker_pids.keys()}", + file=sys.stderr, + flush=True, + ) + for pid in worker_pids.keys(): + try: + os.kill(pid, signal.SIGUSR1) + except ProcessLookupError: + # Worker already exited. + continue + except Exception as exc: # pragma: no cover - defensive diagnostics + print( + f"[opsqueue pytest] Failed sending SIGUSR1 to child pid={pid}: {exc}", + file=sys.stderr, + flush=True, + ) + else: print( "[opsqueue pytest] No child worker PIDs found at SIGTERM time", file=sys.stderr, flush=True, ) - return - print( - f"[opsqueue pytest] Requesting stack dump from child workers: {worker_pids}", - file=sys.stderr, - flush=True, - ) - for pid in worker_pids.keys(): - try: - os.kill(pid, signal.SIGUSR1) - except ProcessLookupError: - # Worker already exited. - continue - except Exception as exc: # pragma: no cover - defensive diagnostics - print( - f"[opsqueue pytest] Failed sending SIGUSR1 to child pid={pid}: {exc}", - file=sys.stderr, - flush=True, - ) + # Dump the stack of the controller process itself, so we can see what it was doing at the time + # of SIGTERM. This also provides the background processes enough time to dump their stacks + # before the controller accesses those. + faulthandler.dump_traceback(file=sys.stderr, all_threads=True) - # Give workers a short moment to flush their dump files before reading them. - time.sleep(0.10) for pid in sorted(worker_pids.keys()): dump_path = _stack_dump_path(pid) if not dump_path.exists(): @@ -104,15 +113,6 @@ def _dump_xdist_worker_traces() -> None: flush=True, ) - -def _handle_sigterm(signum: int, frame: object) -> None: - print( - "[opsqueue pytest] ===== SIGTERM: Start of diagnostic output =====", - file=sys.stderr, - flush=True, - ) - _dump_xdist_worker_traces() - faulthandler.dump_traceback(file=sys.stderr) print( "[opsqueue pytest] ===== SIGTERM: End of diagnostic output =====", file=sys.stderr, @@ -120,9 +120,6 @@ def _handle_sigterm(signum: int, frame: object) -> None: ) os._exit(1) - # If previous handler was SIG_IGN, return and keep process alive. - # Timeout will eventually kill the process if it doesn't exit on its own. - # Register signal handlers at module import time (before pytest_configure hook runs). _sigusr1_dump_file = _stack_dump_path(os.getpid()).open("w", encoding="utf-8") @@ -135,6 +132,7 @@ def _handle_sigterm(signum: int, frame: object) -> None: ) signal.signal(signal.SIGTERM, _handle_sigterm) + @pytest.hookimpl(tryfirst=True) def pytest_configure(config: pytest.Config) -> None: multiprocess.set_start_method("forkserver") From ced1e2bfa13b2e49b63824bfb96de70ee4c914cb Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 14:08:21 +0200 Subject: [PATCH 4/7] fixup! fixup! fixup! CI: Stacktraces of `pytest` and `xdist`-workers on `timeout` --- libs/opsqueue_python/tests/conftest.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index d13f8e9..9aa23ad 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -17,14 +17,11 @@ from opsqueue.consumer import Strategy -_sigusr1_dump_file: Any | None = None - - def _stack_dump_path(pid: int) -> Path: return Path(f"/tmp/opsqueue-pytest-stack-{pid}.log") -def _linux_child_pids(parent_pid: int) -> dict[int, tuple[str, str]]: +def _linux_descendant_pids(parent_pid: int) -> dict[int, tuple[str, str]]: children_file = Path(f"/proc/{parent_pid}/task/{parent_pid}/children") try: raw = children_file.read_text(encoding="utf-8").strip() @@ -49,11 +46,14 @@ def _child_command(pid: int) -> tuple[str, str]: return ("", "") pids = [int(pid_text) for pid_text in raw.split() if pid_text.isdigit()] - return {pid: _child_command(pid) for pid in pids} + result = {pid: _child_command(pid) for pid in pids} + for pid in pids: + result.update(_linux_descendant_pids(pid)) + return result def _handle_sigterm(signum: int, frame: object) -> None: - worker_pids = _linux_child_pids(os.getpid()) + worker_pids = _linux_descendant_pids(os.getpid()) if worker_pids: print( f"[opsqueue pytest] Requesting stack dump from child workers: {worker_pids.keys()}", @@ -122,11 +122,10 @@ def _handle_sigterm(signum: int, frame: object) -> None: # Register signal handlers at module import time (before pytest_configure hook runs). -_sigusr1_dump_file = _stack_dump_path(os.getpid()).open("w", encoding="utf-8") # Make SIGUSR1 print a traceback in any process (controller or worker). faulthandler.register( signal.SIGUSR1, - file=_sigusr1_dump_file, + file=_stack_dump_path(os.getpid()).open("w", encoding="utf-8"), all_threads=True, chain=False, ) From 0e4e5bc19d5bfc4d4bcd9add4735bd46dec0c609 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 14:33:56 +0200 Subject: [PATCH 5/7] fixup! fixup! fixup! fixup! CI: Stacktraces of `pytest` and `xdist`-workers on `timeout` --- libs/opsqueue_python/tests/conftest.py | 56 +++++++++++++++++++------- 1 file changed, 42 insertions(+), 14 deletions(-) diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index 9aa23ad..ed89d1e 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -34,13 +34,15 @@ def _linux_descendant_pids(parent_pid: int) -> dict[int, tuple[str, str]]: def _child_command(pid: int) -> tuple[str, str]: try: return ( - Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip(), + Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip() + or "", Path(f"/proc/{pid}/cmdline") .read_text(encoding="utf-8") .replace( "\x00", " " ) # Replace null bytes with spaces before stripping. - .strip(), + .strip() + or "", ) except OSError: return ("", "") @@ -53,10 +55,21 @@ def _child_command(pid: int) -> tuple[str, str]: def _handle_sigterm(signum: int, frame: object) -> None: - worker_pids = _linux_descendant_pids(os.getpid()) + all_descendant_pids = _linux_descendant_pids(os.getpid()) + + # Only signal processes that registered the SIGUSR1 handler (i.e. have a dump file). + # Internal multiprocess infrastructure processes (resource tracker, forkserver) do not + # register the handler and have no dump file. Sending SIGUSR1 to them with the default + # disposition would kill them, which is undesirable. + worker_pids = { + pid: info + for pid, info in all_descendant_pids.items() + if _stack_dump_path(pid).exists() + } + if worker_pids: print( - f"[opsqueue pytest] Requesting stack dump from child workers: {worker_pids.keys()}", + f"[opsqueue pytest] Requesting stack dump from child workers: {list(worker_pids.keys())}", file=sys.stderr, flush=True, ) @@ -86,14 +99,6 @@ def _handle_sigterm(signum: int, frame: object) -> None: for pid in sorted(worker_pids.keys()): dump_path = _stack_dump_path(pid) - if not dump_path.exists(): - print( - f"[opsqueue pytest] No SIGUSR1 stack dump file found for child pid={pid} {worker_pids[pid]}: {dump_path}", - file=sys.stderr, - flush=True, - ) - continue - print( f"[opsqueue pytest] ===== BEGIN CHILD STACK DUMP pid={pid} {worker_pids[pid]} ({dump_path}) =====", file=sys.stderr, @@ -259,18 +264,41 @@ def read_exact_fd(fd: int, num_bytes: int) -> bytes: return bytes(data) +def _background_main(function: Callable[..., None], args: Iterable[Any]) -> None: + """Entry point for background_processes. + + Registers the SIGUSR1 stack-dump handler before delegating to the real + target function, so that the controller can request a traceback from this + process when a timeout fires. + """ + faulthandler.register( + signal.SIGUSR1, + file=_stack_dump_path(os.getpid()).open("w", encoding="utf-8"), + all_threads=True, + chain=False, + ) + function(*args) + + @contextmanager def background_process( function: Callable[..., None], args: Iterable[Any] = (), ) -> Generator[multiprocess.Process, None, None]: - proc = multiprocess.Process(target=function, args=args) + proc = multiprocess.Process( + target=_background_main, + args=(function, args), + daemon=True, + ) try: - proc.daemon = True proc.start() yield proc finally: proc.terminate() + try: + proc.join(timeout=1.0) + except multiprocess.TimeoutError: + proc.kill() @contextmanager From 9e31f9585c735bfb300057545ea08c8f1c47901b Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 14:34:25 +0200 Subject: [PATCH 6/7] Examples: Mark processes with `daemon=True` --- .../examples/multiprocessing/multiprocessing_consumer.py | 2 +- .../multiprocessing/multiprocessing_consumer_preferdistinct.py | 2 +- .../examples/multiprocessing/multiprocessing_producer_many.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer.py b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer.py index 792ea8a..d99b647 100644 --- a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer.py +++ b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer.py @@ -19,7 +19,7 @@ def my_operation(data: int) -> int: def main() -> None: n_consumers = 16 processes = [ - multiprocessing.Process(target=run_a_consumer, args=(id,)) + multiprocessing.Process(target=run_a_consumer, args=(id,), daemon=True) for id in range(n_consumers) ] for p in processes: diff --git a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer_preferdistinct.py b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer_preferdistinct.py index bc35f7c..94aa479 100644 --- a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer_preferdistinct.py +++ b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_consumer_preferdistinct.py @@ -25,7 +25,7 @@ def my_operation(data: int) -> int: def main() -> None: n_consumers = 4 processes = [ - multiprocessing.Process(target=run_a_consumer, args=(id,)) + multiprocessing.Process(target=run_a_consumer, args=(id,), daemon=True) for id in range(n_consumers) ] for p in processes: diff --git a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_producer_many.py b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_producer_many.py index a91698f..17ae477 100644 --- a/libs/opsqueue_python/examples/multiprocessing/multiprocessing_producer_many.py +++ b/libs/opsqueue_python/examples/multiprocessing/multiprocessing_producer_many.py @@ -22,7 +22,7 @@ def main() -> None: f"Starting {n_producers} producers... When `multiprocessing_consumer_preferdistinct.py` is used, expecting all submissions to finish roughly at the same time" ) processes = [ - multiprocessing.Process(target=run_a_producer, args=(id,)) + multiprocessing.Process(target=run_a_producer, args=(id,), daemon=True) for id in range(n_producers) ] for p in processes: From 1b3126bbddb0009433409a7e75a010ba496d3e70 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Tue, 28 Jul 2026 17:26:24 +0200 Subject: [PATCH 7/7] fixup! Examples: Mark processes with `daemon=True` --- libs/opsqueue_python/tests/conftest.py | 44 +++++++++++++++++--------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index ed89d1e..4a867a0 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -21,33 +21,47 @@ def _stack_dump_path(pid: int) -> Path: return Path(f"/tmp/opsqueue-pytest-stack-{pid}.log") -def _linux_descendant_pids(parent_pid: int) -> dict[int, tuple[str, str]]: - children_file = Path(f"/proc/{parent_pid}/task/{parent_pid}/children") - try: - raw = children_file.read_text(encoding="utf-8").strip() - except OSError: - return {} - - if not raw: - return {} +def _linux_descendant_pids(parent_pid: int = os.getpid()) -> dict[int, tuple[str, str]]: + """ + Returns a dictionary mapping descendant PIDs to their command name and command line. This is a + recursive function that will find all descendants of the given parent PID. The command name and + command line are read from the `/proc` filesystem, which is Linux-specific. + """ + children = ( + Path(f"/proc/{parent_pid}/task/{parent_pid}/children") + .read_text(encoding="utf-8") + .strip() + ) def _child_command(pid: int) -> tuple[str, str]: + """ + Returns the command name and command line of the given process id (pid). + """ try: - return ( + comm = ( Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip() - or "", + or "" + ) + except FileNotFoundError: + # Process might have exited between reading /proc//children and now. + comm = "" + try: + cmdline = ( Path(f"/proc/{pid}/cmdline") .read_text(encoding="utf-8") .replace( "\x00", " " ) # Replace null bytes with spaces before stripping. .strip() - or "", + or "" ) - except OSError: - return ("", "") + except FileNotFoundError: + # Process might have exited between reading /proc//children and now. + cmdline = "" + + return (comm, cmdline) - pids = [int(pid_text) for pid_text in raw.split() if pid_text.isdigit()] + pids = [int(pid_text) for pid_text in children.split()] result = {pid: _child_command(pid) for pid in pids} for pid in pids: result.update(_linux_descendant_pids(pid))