From 4be3a8f6d5c7733d4794e16377b469404f116192 Mon Sep 17 00:00:00 2001 From: openhands Date: Sun, 23 Aug 2026 18:02:45 +0000 Subject: [PATCH 1/2] test: demonstrate cancellation can't free blocking sync code Two failing tests showing the same root cause: asyncio cancellation (CancelledError) cannot be delivered to tasks blocked in synchronous code (C-level locks, blocking I/O, futexes). The cleanup code tries to cancel but gives up after a tiny timeout, leaving zombie threads. Test 1: AsyncExecutor.close() hangs forever when a task is blocked in a worker thread (anyio.to_thread). close() waits indefinitely for the task to finish on its own. Test 2: bubus EventBus handler timeout fires after 1s and logs the error, but the handler's worker thread (via run_in_executor) stays alive forever because cancelling the asyncio Future doesn't cancel the underlying thread. The finally block waits only 0.1s then abandons the task. In production, both accumulate zombie threads until the thread pool is exhausted and new conversation creation stalls. --- tests/sdk/utils/test_cancellation_deadlock.py | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 tests/sdk/utils/test_cancellation_deadlock.py diff --git a/tests/sdk/utils/test_cancellation_deadlock.py b/tests/sdk/utils/test_cancellation_deadlock.py new file mode 100644 index 0000000000..564e9e7a7f --- /dev/null +++ b/tests/sdk/utils/test_cancellation_deadlock.py @@ -0,0 +1,170 @@ +"""Failing tests that demonstrate cancellation doesn't work on blocking sync code. + +Two manifestations of the same underlying problem — a task blocked inside a +synchronous call (futex / blocking I/O / C-level lock) cannot be cancelled by +asyncio, because ``CancelledError`` can only be delivered at an ``await`` point. +The cleanup code knows this and *tries* to cancel, but gives up after a tiny +timeout, leaving zombie threads that accumulate until the process is killed. + +Test 1 — ``AsyncExecutor.close()`` hangs forever on a task blocked in a worker + thread (the bug addressed by PR #4548). + +Test 2 — ``bubus`` EventBus handler timeout logs the error after the timeout + but the handler's worker thread stays alive forever, because + ``asyncio.wait_for`` cancels the coroutine but cannot interrupt the + blocking sync call underneath. The ``finally`` block only waits 0.1s + for the cancellation to take effect, then abandons the task. + +Both tests fail on ``main`` and should pass once the underlying cancellation +issue is fixed. +""" + +import asyncio +import threading +import time + +import pytest + +from openhands.sdk.utils.async_executor import AsyncExecutor + + +# ── Test 1: AsyncExecutor.close() hangs forever ─────────────────────────── + + +def test_async_executor_close_hangs_on_blocking_task(): + """AsyncExecutor.close() must not hang when a task is blocked in a worker + thread that ignores cancellation. + + Without the fix (PR #4548): ``close()`` calls + ``portal_cm.__exit__(None, None, None)`` which takes anyio's graceful path + (``cancel_remaining=False``) and waits forever for the task to finish on + its own. + + With the fix: ``close()`` cancels remaining tasks and bounds the thread + join with a timeout, so it returns even when the task ignores cancellation. + """ + import anyio + from anyio.to_thread import run_sync + + thread_alive = threading.Event() + + async def blocked_in_worker_thread(): + thread_alive.set() + # anyio cannot deliver cancellation until the worker thread returns. + await run_sync(lambda: time.sleep(300)) + + executor = AsyncExecutor() + executor.portal.start_task_soon(blocked_in_worker_thread) + time.sleep(0.2) # let the task start + + done = threading.Event() + + def _close(): + try: + # Try with timeout (available if PR #4548 is applied); fall back + # to the unpatched close() signature. + try: + executor.close(timeout=2.0) + except TypeError: + executor.close() + finally: + done.set() + + threading.Thread(target=_close, daemon=True).start() + + # Without the fix, close() hangs forever — fail after 5s rather than hang. + assert done.wait(timeout=5), ( + "AsyncExecutor.close() hung >5s on a task blocked in a worker thread. " + "The task ignores cancellation (blocked in sync code), and close() " + "does not bound the wait." + ) + assert thread_alive.is_set(), "Task thread should have started" + + +# ── Test 2: bubus EventBus timeout leaves zombie threads ────────────────── + + +def test_bubus_timeout_does_not_free_blocked_handler_thread(): + """bubus EventBus handler timeout must actually free the thread, not just + log a warning and abandon it. + + When a handler is blocked in synchronous code (e.g. a browser launch that + hangs on a C-level lock), ``asyncio.wait_for`` cancels the asyncio task + but cannot interrupt the underlying blocking call. The ``finally`` block + only waits 0.1s for the cancellation to take effect, then moves on — + leaving the thread stuck forever. + + In production this accumulates zombie threads (one per browser launch + failure) until the thread pool is exhausted and new conversation creation + stalls. + + This test registers a handler that blocks in a worker thread, dispatches an + event with a short timeout, and asserts that no zombie threads remain + after the timeout fires and cleanup runs. + """ + pytest.importorskip("bubus") + from bubus import EventBus, BaseEvent + + thread_started = threading.Event() + thread_should_stop = threading.Event() + + class TestEvent(BaseEvent): + pass + + def _block_sync(): + # Simulate a handler blocked in synchronous C-level code (e.g. a + # browser launch stuck on a C-level lock / futex inside playwright). + # Unlike anyio.to_thread.run_sync, a raw ThreadPoolExecutor thread + # cannot be cancelled by asyncio — CancelledError can only be + # delivered at an await point, and run_in_executor doesn't cancel + # the underlying thread. + thread_started.set() + thread_should_stop.wait(timeout=30) + return "done" + + async def blocking_handler(event): + # run_in_executor returns a Future; cancelling the asyncio task + # cancels the Future's await but NOT the underlying thread. + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _block_sync) + + bus = EventBus(name="test_bus") + bus.on(TestEvent, blocking_handler) + bus._start() + + async def _run(): + # Dispatch the event — the handler will block for 30s + bus.dispatch(TestEvent(event_timeout=1.0)) + # step() will time out after the event's timeout (1s) + await bus.step(timeout=1.0) + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(asyncio.wait_for(_run(), timeout=5)) + except (asyncio.TimeoutError, TimeoutError, Exception): + pass # Expected — the handler timed out + + time.sleep(1) # give cleanup code time to run + + # Signal the thread to stop so it can exit if it's still alive + thread_should_stop.set() + time.sleep(0.5) + + bus.stop(timeout=1) + + # Count active non-daemon threads — if bubus properly cancelled the handler, + # there should be no leftover threads from the blocking handler. + active_worker_threads = [ + t for t in threading.enumerate() + if t is not threading.main_thread() and t.is_alive() and not t.daemon + ] + + # This assertion FAILS because bubus's timeout can't cancel blocking sync code: + assert len(active_worker_threads) == 0, ( + f"bubus timeout fired but {len(active_worker_threads)} thread(s) are still " + "alive. The handler was blocked in synchronous code and " + "asyncio.wait_for could not deliver cancellation. The finally block " + "waited only 0.1s then abandoned the task, leaving zombie threads " + "that accumulate until the thread pool is exhausted." + ) From 557a4d369ca14a2774d2d02dfa9c78571743a86f Mon Sep 17 00:00:00 2001 From: neubig Date: Thu, 27 Aug 2026 01:06:19 +0000 Subject: [PATCH 2/2] test: mark known cancellation deadlocks as expected failures Co-authored-by: openhands --- tests/sdk/utils/test_cancellation_deadlock.py | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/tests/sdk/utils/test_cancellation_deadlock.py b/tests/sdk/utils/test_cancellation_deadlock.py index 564e9e7a7f..e26a0c4121 100644 --- a/tests/sdk/utils/test_cancellation_deadlock.py +++ b/tests/sdk/utils/test_cancellation_deadlock.py @@ -31,6 +31,10 @@ # ── Test 1: AsyncExecutor.close() hangs forever ─────────────────────────── +@pytest.mark.xfail( + strict=False, + reason="AsyncExecutor cannot cancel synchronous work blocked in a worker thread", +) def test_async_executor_close_hangs_on_blocking_task(): """AsyncExecutor.close() must not hang when a task is blocked in a worker thread that ignores cancellation. @@ -43,7 +47,6 @@ def test_async_executor_close_hangs_on_blocking_task(): With the fix: ``close()`` cancels remaining tasks and bounds the thread join with a timeout, so it returns even when the task ignores cancellation. """ - import anyio from anyio.to_thread import run_sync thread_alive = threading.Event() @@ -63,10 +66,7 @@ def _close(): try: # Try with timeout (available if PR #4548 is applied); fall back # to the unpatched close() signature. - try: - executor.close(timeout=2.0) - except TypeError: - executor.close() + executor.close() finally: done.set() @@ -84,6 +84,10 @@ def _close(): # ── Test 2: bubus EventBus timeout leaves zombie threads ────────────────── +@pytest.mark.xfail( + strict=False, + reason="bubus cannot cancel synchronous work blocked in a worker thread", +) def test_bubus_timeout_does_not_free_blocked_handler_thread(): """bubus EventBus handler timeout must actually free the thread, not just log a warning and abandon it. @@ -103,7 +107,7 @@ def test_bubus_timeout_does_not_free_blocked_handler_thread(): after the timeout fires and cleanup runs. """ pytest.importorskip("bubus") - from bubus import EventBus, BaseEvent + from bubus import BaseEvent, EventBus thread_started = threading.Event() thread_should_stop = threading.Event() @@ -142,7 +146,7 @@ async def _run(): asyncio.set_event_loop(loop) try: loop.run_until_complete(asyncio.wait_for(_run(), timeout=5)) - except (asyncio.TimeoutError, TimeoutError, Exception): + except (TimeoutError, Exception): pass # Expected — the handler timed out time.sleep(1) # give cleanup code time to run @@ -151,12 +155,14 @@ async def _run(): thread_should_stop.set() time.sleep(0.5) - bus.stop(timeout=1) + loop.run_until_complete(bus.stop(timeout=1)) + loop.close() # Count active non-daemon threads — if bubus properly cancelled the handler, # there should be no leftover threads from the blocking handler. active_worker_threads = [ - t for t in threading.enumerate() + t + for t in threading.enumerate() if t is not threading.main_thread() and t.is_alive() and not t.daemon ]