From 3d7614f0bd719b5e644791cfb2935be54c3b3943 Mon Sep 17 00:00:00 2001 From: daiyongyuan <1533208939@qq.com> Date: Thu, 24 Sep 2026 23:13:10 +0800 Subject: [PATCH 1/2] fix: terminate detached descendants on bbdev cancellation --- api/tests/test_process_cleanup.py | 72 +++++++++++++++++++++++++++++++ api/utils/process.py | 23 +--------- api/utils/process_registry.py | 7 ++- api/utils/stream_run.py | 4 +- 4 files changed, 79 insertions(+), 27 deletions(-) create mode 100644 api/tests/test_process_cleanup.py diff --git a/api/tests/test_process_cleanup.py b/api/tests/test_process_cleanup.py new file mode 100644 index 0000000..7fe5f82 --- /dev/null +++ b/api/tests/test_process_cleanup.py @@ -0,0 +1,72 @@ +import os +import shlex +import signal +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from utils.process import _exists, terminate_group +from utils.process_registry import cancel_process, register_process, unregister_process +from utils.stream_run import stream_run + + +class ProcessCleanupTest(unittest.TestCase): + def test_nested_session_exits_on_shutdown_and_cancel(self): + for cancel in (False, True): + with self.subTest(cancel=cancel): + parent = subprocess.Popen( + [ + sys.executable, + "-c", + "import subprocess, sys, time; " + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'], start_new_session=True); " + "print(child.pid, flush=True); time.sleep(300)", + ], + stdout=subprocess.PIPE, + text=True, + start_new_session=True, + ) + child_pid = int(parent.stdout.readline()) + try: + if cancel: + register_process("nested-session-test", parent.pid) + self.assertTrue(cancel_process("nested-session-test")) + parent.wait(timeout=2) + else: + terminate_group(parent, term_timeout=1) + self.assertFalse(_exists(child_pid)) + finally: + unregister_process("nested-session-test") + if _exists(child_pid): + os.kill(child_pid, signal.SIGKILL) + if parent.poll() is None: + parent.kill() + parent.wait() + parent.stdout.close() + + def test_timeout_exits_nested_session(self): + with tempfile.TemporaryDirectory() as directory: + child_pid_file = Path(directory) / "child.pid" + command = shlex.join( + [ + sys.executable, + "-c", + "import subprocess, sys, time; " + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'], start_new_session=True); " + "open(sys.argv[1], 'w').write(str(child.pid)); time.sleep(300)", + str(child_pid_file), + ] + ) + try: + result = stream_run(command, timeout=1) + self.assertNotEqual(result.returncode, 0) + self.assertFalse(_exists(int(child_pid_file.read_text()))) + finally: + if child_pid_file.exists(): + child_pid = int(child_pid_file.read_text()) + if _exists(child_pid): + os.kill(child_pid, signal.SIGKILL) diff --git a/api/utils/process.py b/api/utils/process.py index 0105f34..f003965 100644 --- a/api/utils/process.py +++ b/api/utils/process.py @@ -113,27 +113,8 @@ def kill_tree(pid: int, term_timeout: float = 5.0): def terminate_group(proc: subprocess.Popen, term_timeout: float = 5.0): if proc.poll() is not None: return - - try: - os.killpg(proc.pid, signal.SIGTERM) - except (ProcessLookupError, PermissionError, OSError): - kill_tree(proc.pid, term_timeout=term_timeout) - - try: - proc.wait(timeout=term_timeout) - return - except subprocess.TimeoutExpired: - pass - - try: - os.killpg(proc.pid, signal.SIGKILL) - except (ProcessLookupError, PermissionError, OSError): - kill_tree(proc.pid, term_timeout=1.0) - - try: - proc.wait(timeout=1) - except subprocess.TimeoutExpired: - pass + kill_tree(proc.pid, term_timeout=term_timeout) + proc.wait(timeout=1) def _is_workspace_server(pid: int, workflow_dir: str, worker_url: str | None = None) -> bool: diff --git a/api/utils/process_registry.py b/api/utils/process_registry.py index 27ec1a4..f2acd25 100644 --- a/api/utils/process_registry.py +++ b/api/utils/process_registry.py @@ -6,6 +6,8 @@ from contextvars import ContextVar from typing import Dict +from utils.process import kill_tree + _lock = threading.Lock() _processes: Dict[str, int] = {} @@ -38,10 +40,7 @@ def cancel_process(scope: str) -> bool: _cancelled.add(scope) if pid is None: return False - try: - os.killpg(pid, signal.SIGTERM) - except ProcessLookupError: - return False + kill_tree(pid) return True diff --git a/api/utils/stream_run.py b/api/utils/stream_run.py index 699efa8..0fb2a02 100644 --- a/api/utils/stream_run.py +++ b/api/utils/stream_run.py @@ -3,6 +3,7 @@ import threading from typing import Optional, List, Callable +from utils.process import kill_tree from utils.process_registry import current_task_scope, register_process, unregister_process @@ -133,8 +134,7 @@ def emit(): # Wait for process to finish (with timeout) process.wait(timeout=timeout) except subprocess.TimeoutExpired: - # Kill process on timeout - process.kill() + kill_tree(process.pid) process.wait() # Wait for threads to finish From 33126549b0221cf5f3eda5424aa41458aa0642fb Mon Sep 17 00:00:00 2001 From: daiyongyuan <1533208939@qq.com> Date: Fri, 25 Sep 2026 07:48:23 +0800 Subject: [PATCH 2/2] test: move process cleanup unit tests out of API examples --- api/{tests => unit_tests}/test_process_cleanup.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename api/{tests => unit_tests}/test_process_cleanup.py (100%) diff --git a/api/tests/test_process_cleanup.py b/api/unit_tests/test_process_cleanup.py similarity index 100% rename from api/tests/test_process_cleanup.py rename to api/unit_tests/test_process_cleanup.py