From 83265fc3b819fcb61da193080b1474f27bf4bc94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Thu, 15 Oct 2026 19:02:26 +0800 Subject: [PATCH] Stop interrupted benchmark containers before collecting diagnostics --- deploy/benchmark_docker/README.md | 14 ++- .../direct_docker_workspace.py | 36 +++++- deploy/benchmark_docker/docker_sandbox.py | 67 +++++++++-- deploy/benchmark_docker/pool.py | 8 +- tests/integration/test_benchmark_docker.py | 39 +++++++ tests/unit/test_benchmark_docker.py | 106 +++++++++++++++++- tests/unit/test_benchmark_docker_workspace.py | 38 ++++++- 7 files changed, 281 insertions(+), 27 deletions(-) diff --git a/deploy/benchmark_docker/README.md b/deploy/benchmark_docker/README.md index 8b74bdf1..438eb0e7 100644 --- a/deploy/benchmark_docker/README.md +++ b/deploy/benchmark_docker/README.md @@ -93,9 +93,17 @@ wrapper but no manifest are rejected; to reuse one, extract it and supply its `export_evaluation_to` saves the workspace and query artifacts for both build and query-only cases; it includes `build_log` when that phase ran. Set `export_build_log_to` to retain the construction log separately. On a failed -build, the pool also returns its text in `CaseResult.build_log`. A timed-out -AgentScope shell command makes its container unusable: the pool closes it after -retaining the log and creates a replacement when more cases need a worker. +build, the pool also returns its text in `CaseResult.build_log`. `command_timeout` +(default 1800 seconds) bounds the entire shell command, including worker startup, +jobs, and Application shutdown. On timeout, cancellation, or a shell backend +failure, the sandbox force-stops the container without candidate or plugin +cooperation, attempts to cache the build log for up to 5 seconds, and removes the +container before propagating the error. Docker stop and removal each have a +30-second timeout; cleanup failures include the container ID in exception notes. +Only the cached build log remains available after removal. The pool creates a +replacement when more cases need a worker. +Custom `workspace_builder` implementations must provide an async `stop()` that +force-stops execution while retaining files for diagnostics, in addition to `close()`. Replacement requires successful cleanup of the old worker. Cleanup failures are reported in `CaseResult.error` and make `success` false for cases that used that worker; failures for an unused worker apply to the whole run. Cancellation diff --git a/deploy/benchmark_docker/direct_docker_workspace.py b/deploy/benchmark_docker/direct_docker_workspace.py index 5a6f8827..ea62749c 100644 --- a/deploy/benchmark_docker/direct_docker_workspace.py +++ b/deploy/benchmark_docker/direct_docker_workspace.py @@ -15,6 +15,8 @@ from typing import Any from aiodocker.exceptions import DockerError from agentscope.workspace import DockerWorkspace +DOCKER_CLEANUP_TIMEOUT = 30.0 + class DirectDockerWorkspace(DockerWorkspace): """Use the supplied image directly and skip the unused MCP gateway.""" @@ -28,6 +30,7 @@ class DirectDockerWorkspace(DockerWorkspace): def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) self._lifecycle_lock = asyncio.Lock() + self._stopped = False async def _build_or_reuse_image(self) -> None: """Select the caller-provided image instead of deriving a gateway image.""" @@ -56,13 +59,36 @@ class DirectDockerWorkspace(DockerWorkspace): exc.add_note(f"workspace close failed: {type(close_exc).__name__}: {close_exc}") raise self.is_alive = True + self._stopped = False + + async def stop(self) -> None: + """Kill all container processes without asking the candidate to shut down.""" + async with self._lifecycle_lock: + self.is_alive = False + if self._container is None or self._stopped: + return + async with asyncio.timeout(DOCKER_CLEANUP_TIMEOUT): + try: + await self._container.kill(signal="SIGKILL") + await self._container.wait(condition="not-running") + except DockerError as exc: + if exc.status != 404: + # Docker returns 409 for an already stopped container. + if exc.status != 409 or (await self._container.show())["State"]["Running"]: + raise + self._stopped = True async def _teardown_backend(self) -> None: """Remove the container, preserving handles until cleanup is confirmed.""" self.is_alive = False if self._container is not None: # Preserve AgentScope's bind-mount ownership repair on native Linux. - if self.host_workdir is not None and sys.platform == "linux" and self._backend is not None: + if ( + not self._stopped + and self.host_workdir is not None + and sys.platform == "linux" + and self._backend is not None + ): try: await self._backend.exec_shell( ["chown", "-R", f"{os.getuid()}:{os.getgid()}", self.workdir], @@ -71,16 +97,16 @@ class DirectDockerWorkspace(DockerWorkspace): except Exception: pass try: - # Force deletion also stops a running container; a separate kill - # would add a failure point without proving that removal succeeded. - await self._container.delete(force=True) + # Also stop processes if an earlier explicit stop failed, and + # retain handles unless Docker confirms removal. + await asyncio.wait_for(self._container.delete(force=True), timeout=DOCKER_CLEANUP_TIMEOUT) except DockerError as exc: if exc.status != 404: raise self._container = None self._backend = None if self._client is not None: - await self._client.close() + await asyncio.wait_for(self._client.close(), timeout=DOCKER_CLEANUP_TIMEOUT) self._client = None self._backend = None diff --git a/deploy/benchmark_docker/docker_sandbox.py b/deploy/benchmark_docker/docker_sandbox.py index 92a4b632..7b3c48ce 100644 --- a/deploy/benchmark_docker/docker_sandbox.py +++ b/deploy/benchmark_docker/docker_sandbox.py @@ -30,6 +30,8 @@ WORKSPACE_EXPORT_ARCHIVE = "/tmp/reme-workspace-export.tar.gz" RUNTIME_LAYOUT = f"{CASE_ROOT}/runtime-layout.json" _CASE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") _CASE_ARCHIVE_FORMAT = "reme-case-v1" +_CLEANUP_TIMEOUT = 30.0 +_LOG_TIMEOUT = 5.0 _PRIVATE_EVALUATION_FIELDS = frozenset( {"query", "question", "golden_answer", "question_type", "rubric", "judge_arguments"}, ) @@ -68,6 +70,10 @@ class Workspace(Protocol): """Close the workspace.""" raise NotImplementedError + async def stop(self) -> None: + """Force-stop container processes while keeping files available.""" + raise NotImplementedError + def get_backend(self) -> Backend: """Return the execution backend.""" raise NotImplementedError @@ -81,7 +87,7 @@ class SandboxCommandError(RuntimeError): class SandboxCommandTimeoutError(SandboxCommandError): - """A command timed out and may still be running inside the container.""" + """A command timed out; cleanup failures are attached as exception notes.""" def _validate_case_id(case_id: str) -> None: @@ -379,6 +385,7 @@ class DockerReMeSandbox: self._closed = False self._reusable = True self._operation_lock = asyncio.Lock() + self._build_log: bytes | None = None @property def reusable(self) -> bool: @@ -524,13 +531,15 @@ class DockerReMeSandbox: raise SandboxCommandError("sandbox container is no longer reusable") started = _utc_now() try: - result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout) + async with asyncio.timeout(self.command_timeout): + result = await self.backend.exec_shell(command, cwd=cwd, timeout=self.command_timeout) except TimeoutError as exc: - self._reusable = False - raise SandboxCommandTimeoutError(f"sandbox action {name!r} timed out; container must be destroyed") from exc - except (Exception, asyncio.CancelledError): + error = SandboxCommandTimeoutError(f"sandbox action {name!r} timed out") + await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], error) + raise error from exc + except (Exception, asyncio.CancelledError) as exc: # A backend failure may occur after Docker started the command. - self._reusable = False + await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], exc) raise self.actions.append( ActionRecord( @@ -543,10 +552,36 @@ class DockerReMeSandbox: ), ) if result.exit_code == -1: - self._reusable = False - raise SandboxCommandTimeoutError(f"sandbox action {name!r} timed out; container must be destroyed") + error = SandboxCommandTimeoutError(f"sandbox action {name!r} timed out") + await cleanup_on_error([(f"container {self.container_id!r}", self._abort())], error) + raise error return result + async def _abort(self) -> None: + """Retire an uncertain command while its caller holds the operation lock.""" + self._reusable = False + error = None + try: + await asyncio.wait_for(self.workspace.stop(), timeout=_CLEANUP_TIMEOUT) + # Read files only after the candidate has stopped. Diagnostics must + # never delay termination or prevent container removal. + try: + assert self.backend is not None + self._build_log = await asyncio.wait_for( + self.backend.read_file(f"{CASE_ROOT}/build_log/build.log"), + timeout=_LOG_TIMEOUT, + ) + except Exception: + pass + except BaseException as exc: + error = exc + raise + finally: + if error is None: + await self._close_workspace() + else: + await cleanup_on_error([("container removal", self._close_workspace())], error) + async def _exec_checked(self, name: str, command: list[str], *, cwd: str | None = None) -> Any: result = await self._exec(name, command, cwd=cwd) if not result.ok(): @@ -776,6 +811,10 @@ class DockerReMeSandbox: return (await self._read_build_log()).decode("utf-8", errors="replace") async def _read_build_log(self) -> bytes: + if self._build_log is not None: + return self._build_log + if not self.reusable: + raise SandboxCommandError("build log unavailable after interrupted command") return await self._require_artifact(f"{CASE_ROOT}/build_log/build.log", "build log") async def _require_artifact(self, path: str, description: str) -> bytes: @@ -1078,10 +1117,14 @@ class DockerReMeSandbox: async def close(self) -> None: """Destroy the case container. Export before calling this method.""" async with self._operation_lock: - if not self._closed: - self._reusable = False - await self.workspace.close() - self._closed = True + await self._close_workspace() + + async def _close_workspace(self) -> None: + """Close without acquiring the operation lock a second time.""" + if not self._closed: + self._reusable = False + await asyncio.wait_for(self.workspace.close(), timeout=_CLEANUP_TIMEOUT) + self._closed = True async def __aenter__(self) -> "DockerReMeSandbox": return self diff --git a/deploy/benchmark_docker/pool.py b/deploy/benchmark_docker/pool.py index 92b9bca0..740cc058 100644 --- a/deploy/benchmark_docker/pool.py +++ b/deploy/benchmark_docker/pool.py @@ -140,7 +140,7 @@ class BenchmarkDockerPool: try: result.build = await worker.run_build(list(plan.build_jobs)) except Exception as exc: - result.error = f"{type(exc).__name__}: {exc}" + result.error = format_error(exc) await self._retain_failed_build_log(worker, plan, result) return result if not result.build.get("success"): @@ -162,14 +162,14 @@ class BenchmarkDockerPool: await worker.export_evaluation(plan.export_evaluation_to) return result except Exception as exc: # Keep other cases running and release this worker. - result.error = f"{type(exc).__name__}: {exc}" + result.error = format_error(exc) return result finally: if worker.reusable: available.put_nowait(worker) else: - # A timed-out command may still be running in the container. - # Retire it before allowing another case into this slot. + # Confirm cleanup (or retry a failed removal) before + # allowing another case into this slot. closed = False try: await worker.close() diff --git a/tests/integration/test_benchmark_docker.py b/tests/integration/test_benchmark_docker.py index a22f7497..82a59fbd 100644 --- a/tests/integration/test_benchmark_docker.py +++ b/tests/integration/test_benchmark_docker.py @@ -102,6 +102,45 @@ def _query(query_id: str) -> EvaluationQuery: ) +@pytest.mark.asyncio +async def test_real_docker_timeout_kills_blocked_thread_pool(base_image): + """A host deadline must terminate a worker blocked in executor shutdown.""" + from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspace + from deploy.benchmark_docker.docker_sandbox import DockerReMeSandbox, SandboxCommandTimeoutError + + container_id = f"timeout-{uuid.uuid4().hex[:12]}" + workspace = DirectDockerWorkspace(workspace_id=container_id, base_image=base_image) + case = DockerReMeSandbox( + case_id="timeout", + container_id=container_id, + candidate=ImageCandidate(base_image), + workspace=workspace, + config="unused", + plugins=(), + command_timeout=2, + environment_names=[], + ) + try: + await workspace.initialize() + case.backend = workspace.get_backend() + # Reproduce the shutdown condition without importing ReMe or a plugin. + script = ( + "import threading; from pathlib import Path; from concurrent.futures import ThreadPoolExecutor; " + "pool = ThreadPoolExecutor(1); pool.submit(threading.Event().wait); " + "log = Path('/workspace/case/build_log/build.log'); log.parent.mkdir(parents=True); " + "log.write_text('waiting for thread pool shutdown'); pool.shutdown(wait=True)" + ) + # pylint: disable=protected-access + async with case._operation_lock: + with pytest.raises(SandboxCommandTimeoutError): + await asyncio.wait_for(case._exec("blocked-close", ["python3", "-c", script]), timeout=75) + assert not _container_exists(f"as_ws_{container_id}") + assert not case.reusable + assert await case.read_build_log() == "waiting for thread pool shutdown" + finally: + await case.close() + + @pytest.mark.asyncio async def test_real_docker_source_pool_workspace_reuse(base_image, tmp_path): """Run build, query, workspace reuse, and query-only export in a real container.""" diff --git a/tests/unit/test_benchmark_docker.py b/tests/unit/test_benchmark_docker.py index e997076c..788a6711 100644 --- a/tests/unit/test_benchmark_docker.py +++ b/tests/unit/test_benchmark_docker.py @@ -803,6 +803,7 @@ class FakeWorkspace: def __init__(self, backend: FakeBackend) -> None: self.backend = backend self.closed = False + self.stopped = False async def initialize(self) -> None: pass @@ -813,6 +814,9 @@ class FakeWorkspace: async def close(self) -> None: self.closed = True + async def stop(self) -> None: + self.stopped = True + def _fake_docker_factory(tmp_path, workspace_builder, *, install_projects=(".",)): (tmp_path / "reme").mkdir() @@ -833,6 +837,7 @@ async def test_backend_timeout_prevents_direct_case_reset(tmp_path): with pytest.raises(SandboxCommandTimeoutError): await case.run_build([JobRequest("build")]) assert not case.reusable + assert workspace.stopped and workspace.closed assert case.actions[-1].exit_code == -1 assert await case.read_build_log() == "build still running" with pytest.raises(SandboxCommandError, match="no longer reusable"): @@ -863,6 +868,7 @@ async def test_external_timeout_prevents_case_reset(tmp_path): with pytest.raises(TimeoutError): await asyncio.wait_for(build, timeout=0.01) assert not case.reusable + assert workspace.stopped and workspace.closed with pytest.raises(SandboxCommandError, match="no longer reusable"): await case.reset_case("next") inbox_count = sum("/inbox/" in path for path in workspace.backend.files) @@ -874,6 +880,98 @@ async def test_external_timeout_prevents_case_reset(tmp_path): assert workspace.closed +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["deadline", "backend-timeout", "cancel"]) +async def test_interrupted_worker_stops_before_reading_logs_and_removal(tmp_path, failure): + events = [] + started = asyncio.Event() + stopping = asyncio.Event() + release_stop = asyncio.Event() + + class BlockingBackend(FakeBackend): + async def exec_shell(self, command, *, cwd=None, timeout=None): + if "/workspace/harness/worker.py" in command: + self.files["/workspace/case/build_log/build.log"] = b"closing application" + started.set() + if failure == "backend-timeout": + raise TimeoutError("backend timed out") + await asyncio.Event().wait() # Includes a worker stuck in Application.close(). + return await super().exec_shell(command, cwd=cwd, timeout=timeout) + + async def read_file(self, path): + events.append("read") + assert workspace.stopped + return await super().read_file(path) + + class OrderedWorkspace(FakeWorkspace): + async def stop(self): + stopping.set() + await release_stop.wait() + events.append("stop") + await super().stop() + + async def close(self): + events.append("close") + await super().close() + + workspace = OrderedWorkspace(BlockingBackend()) + factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace) + case = await factory.create_case("interrupted") + case.command_timeout = 0.01 if failure == "deadline" else 10 + task = asyncio.create_task(case.run_build([JobRequest("build")])) + await asyncio.wait_for(started.wait(), timeout=2) + if failure == "cancel": + task.cancel() + await asyncio.wait_for(stopping.wait(), timeout=2) + if failure == "cancel": + task.cancel() # Repeated cancellation must not interrupt container cleanup. + release_stop.set() + expected = asyncio.CancelledError if failure == "cancel" else SandboxCommandTimeoutError + with pytest.raises(expected): + await asyncio.wait_for(task, timeout=2) + assert events == ["stop", "read", "close"] + assert workspace.closed and not case.reusable + assert await case.read_build_log() == "closing application" + await case.close() + assert events == ["stop", "read", "close"] + + +@pytest.mark.asyncio +async def test_failed_stop_still_removes_container_and_preserves_timeout(tmp_path): + class BrokenStopWorkspace(FakeWorkspace): + async def stop(self): + raise RuntimeError("kill failed") + + workspace = BrokenStopWorkspace(TimedOutBackend()) + factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace) + case = await factory.create_case("failed-stop") + with pytest.raises(SandboxCommandTimeoutError) as failure: + await case.run_build([JobRequest("build")]) + assert workspace.closed + assert any("kill failed" in note for note in failure.value.__notes__) + with pytest.raises(SandboxCommandError, match="log unavailable"): + await case.read_build_log() + + +@pytest.mark.asyncio +async def test_blocked_diagnostics_cannot_prevent_removal(tmp_path, monkeypatch): + from deploy.benchmark_docker import docker_sandbox + + monkeypatch.setattr(docker_sandbox, "_LOG_TIMEOUT", 0.01) + + class BlockedLogBackend(TimedOutBackend): + async def read_file(self, path): + assert workspace.stopped + await asyncio.Event().wait() + + workspace = FakeWorkspace(BlockedLogBackend()) + factory = _fake_docker_factory(tmp_path, lambda _id, _image, _env: workspace) + case = await factory.create_case("blocked-log") + with pytest.raises(SandboxCommandTimeoutError): + await asyncio.wait_for(case.run_build([JobRequest("build")]), timeout=2) + assert workspace.closed + + @pytest.mark.asyncio async def test_cancelled_request_upload_prevents_case_reset(tmp_path): started = asyncio.Event() @@ -1340,7 +1438,11 @@ async def test_pool_retries_strict_delete_without_replacing_uncertain_container( from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspace workspaces = [] - container = SimpleNamespace(delete=AsyncMock(side_effect=[RuntimeError("delete failed"), None])) + container = SimpleNamespace( + kill=AsyncMock(), + wait=AsyncMock(), + delete=AsyncMock(side_effect=[RuntimeError("delete failed"), RuntimeError("delete failed"), None]), + ) client = SimpleNamespace(close=AsyncMock()) class StrictWorkspace(DirectDockerWorkspace): @@ -1361,6 +1463,6 @@ async def test_pool_retries_strict_delete_without_replacing_uncertain_container( assert "worker close failed" in results[0].error assert "delete failed" in results[1].error assert len(workspaces) == 1 - assert container.delete.await_count == 2 + assert container.delete.await_count == 3 assert workspaces[0]._container is workspaces[0]._client is None client.close.assert_awaited_once() diff --git a/tests/unit/test_benchmark_docker_workspace.py b/tests/unit/test_benchmark_docker_workspace.py index a30521bc..09f8ee33 100644 --- a/tests/unit/test_benchmark_docker_workspace.py +++ b/tests/unit/test_benchmark_docker_workspace.py @@ -14,13 +14,49 @@ from deploy.benchmark_docker.direct_docker_workspace import DirectDockerWorkspac def make_workspace(): workspace = DirectDockerWorkspace(workspace_id="cleanup-test") - workspace._container = SimpleNamespace(delete=AsyncMock()) + workspace._container = SimpleNamespace(kill=AsyncMock(), wait=AsyncMock(), delete=AsyncMock()) workspace._client = SimpleNamespace(close=AsyncMock()) workspace._backend = SimpleNamespace(exec_shell=AsyncMock()) workspace.is_alive = True return workspace +@pytest.mark.asyncio +async def test_stop_kills_and_confirms_exit_before_removal(): + workspace = make_workspace() + container = workspace._container + await workspace.stop() + container.kill.assert_awaited_once_with(signal="SIGKILL") + container.wait.assert_awaited_once_with(condition="not-running") + container.delete.assert_not_awaited() + assert workspace._backend is not None + assert not workspace.is_alive + await workspace.stop() + assert container.kill.await_count == 1 + await workspace.close() + container.delete.assert_awaited_once_with(force=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["kill", "delete"]) +async def test_docker_cleanup_timeout_retains_handles(monkeypatch, operation): + from deploy.benchmark_docker import direct_docker_workspace + + monkeypatch.setattr(direct_docker_workspace, "DOCKER_CLEANUP_TIMEOUT", 0.01) + workspace = make_workspace() + container = workspace._container + + async def blocked(**_kwargs): + await asyncio.Event().wait() + + getattr(container, operation).side_effect = blocked + with pytest.raises(TimeoutError): + await (workspace.stop() if operation == "kill" else workspace.close()) + assert workspace._container is container + getattr(container, operation).side_effect = None + await workspace.close() + + @pytest.mark.asyncio async def test_delete_failure_retains_handles_and_can_be_retried(): workspace = make_workspace()