From bec28b5050a22459a24e70f574a3c537b846f049 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 5 Jul 2026 01:13:47 -0400 Subject: [PATCH] fix: Reap JobRunner subprocesses before os._exit to prevent orphans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main_hcs.py exits via os._exit(), which skips multiprocessing's atexit hook — the one that would normally terminate daemon children. Any JobRunner subprocess still alive at that point (its shutdown event never set because the exit-time abort join timed out, the worker died before _finish_jobs, or the non-blocking shutdown threads lost the race) was orphaned to PPID 1 and leaked its queue/event semaphores (the "resource_tracker: leaked semaphore objects" warning users see). - Add shutdown_all_job_runners(): finds live JobRunner children via multiprocessing.active_children(), runs their full shutdown() in parallel, then terminates any survivor. Called in main_hcs.py right before os._exit(). - JobRunner.shutdown() now also clears _ready_event so its semaphore is released like the other primitives. - Fix the _finish_jobs comment that claimed "the OS will terminate subprocesses anyway" — untrue under os._exit. Verified: unit tests reap a runner whose shutdown event was never set; GUI driver runs (simulation) close normally and mid-acquisition with zero orphaned children and no resource_tracker warning. Co-Authored-By: Claude Fable 5 --- software/control/core/job_processing.py | 37 ++++++++++++++ software/control/core/multi_point_worker.py | 4 +- software/main_hcs.py | 10 ++++ .../control/core/test_job_runner_teardown.py | 49 +++++++++++++++++++ 4 files changed, 99 insertions(+), 1 deletion(-) create mode 100644 software/tests/control/core/test_job_runner_teardown.py diff --git a/software/control/core/job_processing.py b/software/control/core/job_processing.py index 189b1796f..5c5049736 100644 --- a/software/control/core/job_processing.py +++ b/software/control/core/job_processing.py @@ -2,6 +2,7 @@ import multiprocessing import queue import os +import threading import time import json from datetime import datetime @@ -925,6 +926,7 @@ def shutdown(self, timeout_s=1.0): self._output_queue = None self._shutdown_event = None self._pending_count = None + self._ready_event = None def run(self): import logging @@ -1034,3 +1036,38 @@ def run(self): log_memory("WORKER_SHUTDOWN", include_children=False) stop_worker_monitoring() self._log.info("Shutdown request received, exiting run.") + + +def shutdown_all_job_runners(timeout_s: float = 5.0) -> None: + """Best-effort teardown of any still-alive JobRunner subprocesses. + + Called before the application exits. main_hcs.py exits via os._exit(), which + skips multiprocessing's atexit cleanup, so daemon JobRunner children that the + per-acquisition (non-blocking) shutdown threads have not finished with would + otherwise survive as orphans and leak their queue/event semaphores. + """ + runners = [p for p in multiprocessing.active_children() if isinstance(p, JobRunner)] + if not runners: + return + log = squid.logging.get_logger("shutdown_all_job_runners") + log.info(f"Shutting down {len(runners)} job runner subprocess(es) before exit...") + + def safe_shutdown(runner): + try: + runner.shutdown(timeout_s=max(1.0, timeout_s - 1.0)) + except Exception as e: + # A concurrent per-acquisition shutdown thread may already be tearing + # this runner down; termination below still guarantees the process dies. + log.debug(f"Job runner shutdown raced or failed: {e}") + + threads = [threading.Thread(target=safe_shutdown, args=(r,), daemon=True) for r in runners] + for t in threads: + t.start() + deadline = time.time() + timeout_s + for t in threads: + t.join(timeout=max(0.1, deadline - time.time())) + for runner in runners: + if runner.is_alive(): + log.warning(f"Job runner {runner.pid} still alive after {timeout_s}s; terminating.") + runner.terminate() + runner.join(timeout=1.0) diff --git a/software/control/core/multi_point_worker.py b/software/control/core/multi_point_worker.py index 7fd973452..2c03b1b1f 100644 --- a/software/control/core/multi_point_worker.py +++ b/software/control/core/multi_point_worker.py @@ -697,7 +697,9 @@ def time_left(): # Using daemon threads is safe here because: # 1. All jobs are complete and results are already drained # 2. The subprocess termination is best-effort cleanup only - # 3. If app exits before threads complete, OS will terminate subprocesses anyway + # 3. If the app exits before these threads complete, main_hcs.py calls + # shutdown_all_job_runners() before os._exit() (os._exit skips the + # multiprocessing atexit hook that would normally reap daemon children) # 4. This prevents slow subprocess termination from blocking acquisition completion log = self._log # Capture for closure diff --git a/software/main_hcs.py b/software/main_hcs.py index dddc81816..ad22ea14e 100644 --- a/software/main_hcs.py +++ b/software/main_hcs.py @@ -453,5 +453,15 @@ def launch_claude_code(): except Exception as e: log.warning(f"Error during shutdown abort handling: {e}") + # os._exit() below skips multiprocessing's atexit hook, so daemon JobRunner + # children still alive (e.g. their non-blocking shutdown threads lost the + # race) would be orphaned and leak their queue/event semaphores. + try: + from control.core.job_processing import shutdown_all_job_runners + + shutdown_all_job_runners(timeout_s=5.0) + except Exception as e: + log.warning(f"Error shutting down job runners: {e}") + logging.shutdown() # Flush log handlers before os._exit() bypasses Python cleanup os._exit(exit_code) diff --git a/software/tests/control/core/test_job_runner_teardown.py b/software/tests/control/core/test_job_runner_teardown.py new file mode 100644 index 000000000..a641640e6 --- /dev/null +++ b/software/tests/control/core/test_job_runner_teardown.py @@ -0,0 +1,49 @@ +"""Tests for shutdown_all_job_runners (exit-time reaping of JobRunner children). + +main_hcs.py exits via os._exit(), which skips multiprocessing's atexit hook, so +any JobRunner subprocess still alive at that point (its shutdown event never +set — e.g. the exit-time abort join timed out, or the worker died before +_finish_jobs) would be orphaned and leak its queue/event semaphores. +shutdown_all_job_runners() is called right before os._exit() to close that hole. +""" + +import multiprocessing +import time + +from control.core.job_processing import JobRunner, shutdown_all_job_runners + + +def _wait_dead(runner, timeout_s=10.0): + deadline = time.time() + timeout_s + while time.time() < deadline: + if not runner.is_alive(): + return True + time.sleep(0.1) + return not runner.is_alive() + + +def test_reaps_runner_that_was_never_told_to_stop(): + runner = JobRunner() + runner.start() + assert runner.wait_ready(timeout_s=15.0), "job runner subprocess never became ready" + assert runner.is_alive() + + shutdown_all_job_runners(timeout_s=5.0) + + assert _wait_dead(runner), "job runner still alive after shutdown_all_job_runners()" + assert not any(isinstance(p, JobRunner) for p in multiprocessing.active_children()) + + +def test_noop_when_no_runners(): + shutdown_all_job_runners(timeout_s=1.0) + + +def test_safe_after_runner_already_shut_down(): + runner = JobRunner() + runner.start() + assert runner.wait_ready(timeout_s=15.0) + runner.shutdown(timeout_s=5.0) + assert _wait_dead(runner) + + shutdown_all_job_runners(timeout_s=2.0) + assert not any(isinstance(p, JobRunner) for p in multiprocessing.active_children())