mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-04 08:28:36 +00:00
fix(shell): reap zombie processes on all subprocess exit paths
The previous fix (dbcc7cb5) only added os.waitpid() to _kill_process(), covering the timeout/cancel path of one-shot exec. Zombies continued to accumulate because several other exit paths never reaped children: - _ExecSession.kill(): sent SIGKILL + process.wait(5s) but had no os.waitpid() fallback if the wait timed out - ExecTool.execute() generic exception handler: leaked the subprocess if communicate() raised an unexpected error - Normal completion paths: relied entirely on asyncio's child-watcher, which can miss exits inside Docker containers (pidfd/SIGCHLD gaps) Changes: - Extract _reap_pid() helper for consistent, safe os.waitpid(WNOHANG) - Add _reap_pid() fallback to _ExecSession.kill() via try/finally - Add _reap_pid() safety-net after normal process exit in both ExecTool.execute() and _ExecSession.poll() - Kill + reap subprocess in the generic except Exception handler - Add periodic zombie reaper background task (every 30s) in the gateway as a last line of defense
This commit is contained in:
parent
d32f8961b1
commit
c9e014fdea
@ -3,6 +3,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import os
|
||||||
import time
|
import time
|
||||||
import uuid
|
import uuid
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
@ -148,6 +149,9 @@ class _ExecSession:
|
|||||||
asyncio.gather(self._stdout_task, self._stderr_task),
|
asyncio.gather(self._stdout_task, self._stderr_task),
|
||||||
timeout=2.0,
|
timeout=2.0,
|
||||||
)
|
)
|
||||||
|
# Safety-net reap after normal exit.
|
||||||
|
from nanobot.agent.tools.shell import _reap_pid
|
||||||
|
_reap_pid(self.process.pid)
|
||||||
elif yield_time_ms > 0:
|
elif yield_time_ms > 0:
|
||||||
await self._wait_for_buffered_output()
|
await self._wait_for_buffered_output()
|
||||||
|
|
||||||
@ -171,8 +175,14 @@ class _ExecSession:
|
|||||||
if self.process.returncode is not None:
|
if self.process.returncode is not None:
|
||||||
return
|
return
|
||||||
self.process.kill()
|
self.process.kill()
|
||||||
with suppress(asyncio.TimeoutError):
|
try:
|
||||||
await asyncio.wait_for(self.process.wait(), timeout=5.0)
|
with suppress(asyncio.TimeoutError):
|
||||||
|
await asyncio.wait_for(self.process.wait(), timeout=5.0)
|
||||||
|
finally:
|
||||||
|
# Safety-net waitpid — prevent zombie if asyncio's child watcher
|
||||||
|
# did not reap the process (common in containers).
|
||||||
|
from nanobot.agent.tools.shell import _reap_pid
|
||||||
|
_reap_pid(self.process.pid)
|
||||||
|
|
||||||
async def _wait_for_buffered_output(self) -> None:
|
async def _wait_for_buffered_output(self) -> None:
|
||||||
deadline = time.monotonic() + OUTPUT_DRAIN_GRACE_S
|
deadline = time.monotonic() + OUTPUT_DRAIN_GRACE_S
|
||||||
|
|||||||
@ -41,6 +41,24 @@ from nanobot.security.workspace_policy import is_path_within
|
|||||||
_IS_WINDOWS = sys.platform == "win32"
|
_IS_WINDOWS = sys.platform == "win32"
|
||||||
|
|
||||||
|
|
||||||
|
def _reap_pid(pid: int) -> None:
|
||||||
|
"""Best-effort ``waitpid`` to reap a child and prevent zombies.
|
||||||
|
|
||||||
|
Call this after killing or after normal completion of any subprocess
|
||||||
|
as a safety net — asyncio's child-watcher *should* have reaped it,
|
||||||
|
but in containers / edge-cases it sometimes doesn't.
|
||||||
|
"""
|
||||||
|
if _IS_WINDOWS:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
os.waitpid(pid, os.WNOHANG)
|
||||||
|
except (ProcessLookupError, ChildProcessError):
|
||||||
|
# Already reaped, or not our child — both are fine.
|
||||||
|
pass
|
||||||
|
except OSError as exc:
|
||||||
|
logger.debug("_reap_pid({}): {}", pid, exc)
|
||||||
|
|
||||||
|
|
||||||
# Policy note appended to recoverable workspace-boundary guard errors.
|
# Policy note appended to recoverable workspace-boundary guard errors.
|
||||||
_WORKSPACE_BOUNDARY_NOTE = (
|
_WORKSPACE_BOUNDARY_NOTE = (
|
||||||
"\n\nNote: this is a hard policy boundary, not a transient failure. "
|
"\n\nNote: this is a hard policy boundary, not a transient failure. "
|
||||||
@ -283,6 +301,7 @@ class ExecTool(Tool):
|
|||||||
if yield_time_ms is not None:
|
if yield_time_ms is not None:
|
||||||
return await self._execute_session(prepared, yield_time_ms, max_output_chars)
|
return await self._execute_session(prepared, yield_time_ms, max_output_chars)
|
||||||
|
|
||||||
|
process: asyncio.subprocess.Process | None = None
|
||||||
try:
|
try:
|
||||||
process = await self._spawn(
|
process = await self._spawn(
|
||||||
prepared.command,
|
prepared.command,
|
||||||
@ -304,6 +323,11 @@ class ExecTool(Tool):
|
|||||||
await self._kill_process(process)
|
await self._kill_process(process)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
|
# Safety-net reap: asyncio *should* have reaped the child via
|
||||||
|
# communicate(), but in containers the child-watcher sometimes
|
||||||
|
# misses it, leaving a zombie.
|
||||||
|
_reap_pid(process.pid)
|
||||||
|
|
||||||
output_parts = []
|
output_parts = []
|
||||||
|
|
||||||
if stdout:
|
if stdout:
|
||||||
@ -330,6 +354,10 @@ class ExecTool(Tool):
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
# Kill and reap the child if it was spawned but an unexpected
|
||||||
|
# error prevented communicate() from completing.
|
||||||
|
if process is not None:
|
||||||
|
await self._kill_process(process)
|
||||||
return ToolResult.error(f"Error executing command: {str(e)}")
|
return ToolResult.error(f"Error executing command: {str(e)}")
|
||||||
|
|
||||||
async def _execute_session(
|
async def _execute_session(
|
||||||
@ -604,11 +632,7 @@ class ExecTool(Tool):
|
|||||||
with suppress(asyncio.TimeoutError):
|
with suppress(asyncio.TimeoutError):
|
||||||
await asyncio.wait_for(process.wait(), timeout=5.0)
|
await asyncio.wait_for(process.wait(), timeout=5.0)
|
||||||
finally:
|
finally:
|
||||||
if not _IS_WINDOWS:
|
_reap_pid(process.pid)
|
||||||
try:
|
|
||||||
os.waitpid(process.pid, os.WNOHANG)
|
|
||||||
except (ProcessLookupError, ChildProcessError) as e:
|
|
||||||
logger.debug("Process already reaped or not found: {}", e)
|
|
||||||
|
|
||||||
def _build_env(self) -> dict[str, str]:
|
def _build_env(self) -> dict[str, str]:
|
||||||
"""Build a minimal environment for subprocess execution.
|
"""Build a minimal environment for subprocess execution.
|
||||||
|
|||||||
@ -201,6 +201,7 @@ app = typer.Typer(
|
|||||||
)
|
)
|
||||||
|
|
||||||
console = Console()
|
console = Console()
|
||||||
|
_IS_WINDOWS = sys.platform == "win32"
|
||||||
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
|
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
|
||||||
_REASONING_SENTENCE_ENDINGS = (".", "!", "?", "。", "!", "?")
|
_REASONING_SENTENCE_ENDINGS = (".", "!", "?", "。", "!", "?")
|
||||||
_REASONING_FLUSH_CHARS = 60
|
_REASONING_FLUSH_CHARS = 60
|
||||||
@ -1695,6 +1696,29 @@ def _run_gateway(
|
|||||||
else:
|
else:
|
||||||
console.print("[yellow]✗[/yellow] Heartbeat: disabled")
|
console.print("[yellow]✗[/yellow] Heartbeat: disabled")
|
||||||
|
|
||||||
|
async def _zombie_reaper() -> None:
|
||||||
|
"""Periodically reap zombie child processes.
|
||||||
|
|
||||||
|
asyncio's child-watcher *should* reap all children, but inside
|
||||||
|
Docker containers the pidfd / SIGCHLD mechanism can miss exits.
|
||||||
|
This task runs every 30 s and calls ``os.waitpid(-1, WNOHANG)``
|
||||||
|
in a loop to collect any zombies that slipped through.
|
||||||
|
"""
|
||||||
|
_INTERVAL = 30
|
||||||
|
while True:
|
||||||
|
await asyncio.sleep(_INTERVAL)
|
||||||
|
reaped = 0
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
pid, _ = os.waitpid(-1, os.WNOHANG)
|
||||||
|
if pid == 0:
|
||||||
|
break # no more zombie children
|
||||||
|
reaped += 1
|
||||||
|
except ChildProcessError:
|
||||||
|
break # no child processes at all
|
||||||
|
if reaped:
|
||||||
|
logger.info("Zombie reaper: reaped {} defunct child process(es)", reaped)
|
||||||
|
|
||||||
async def _health_server(host: str, health_port: int):
|
async def _health_server(host: str, health_port: int):
|
||||||
"""Lightweight HTTP health endpoint on the gateway port."""
|
"""Lightweight HTTP health endpoint on the gateway port."""
|
||||||
import json as _json
|
import json as _json
|
||||||
@ -1820,6 +1844,11 @@ def _run_gateway(
|
|||||||
name="nanobot-local-triggers",
|
name="nanobot-local-triggers",
|
||||||
),
|
),
|
||||||
]
|
]
|
||||||
|
if not _IS_WINDOWS:
|
||||||
|
tasks.append(asyncio.create_task(
|
||||||
|
_zombie_reaper(),
|
||||||
|
name="nanobot-zombie-reaper",
|
||||||
|
))
|
||||||
if health_server_enabled:
|
if health_server_enabled:
|
||||||
tasks.append(asyncio.create_task(
|
tasks.append(asyncio.create_task(
|
||||||
_health_server(config.gateway.host, port),
|
_health_server(config.gateway.host, port),
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user