From 71e5876aaea0e137d5670ed73a19cf5e2af7a4ea Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Jul 2026 17:55:37 +0000 Subject: [PATCH] Sweep untracked tasks and close thread_loop on ServiceThread shutdown Fixes #54. ServiceThread._shutdown_thread() only cancels/awaits futures registered via this Service's own add_future()/@Service.task tracking. A driver library running inside the thread (e.g. a Kafka client's internal consumer poll/heartbeat loop) can schedule tasks directly on thread_loop, bypassing that tracking entirely. Left unswept, those tasks were simply abandoned the moment _serve() returned and thread_loop stopped running -- producing "coroutine ... was never awaited" / "Task was destroyed but it is pending" warnings at some later, unpredictable point (e.g. interpreter shutdown), since the abandoned task/coroutine object is only reported once garbage collected. mode.Worker already has an equivalent safety net for the main loop: _gather_all() sweeps every task on self.loop via all_tasks(), tracked or not, before closing it. ServiceThread had no equivalent for thread_loop, and never explicitly closed thread_loop at all. Add ServiceThread._gather_remaining_tasks(), called at the end of _serve()'s finally block (after _shutdown_thread(), while thread_loop is still running): cancel and await every task still on thread_loop, tracked or not, mirroring Worker._gather_all(). Also close thread_loop explicitly once _start_thread() returns -- but only when ServiceThread created it itself (thread_loop=None, the default); a caller who passed their own thread_loop= may expect to keep managing its lifecycle, so that case is left untouched. Verified with a minimal repro (no aiokafka/Kafka broker required): a subclass that schedules an untracked asyncio.ensure_future(...) task in on_thread_started() no longer leaves it pending after stop() -- confirmed by checking the task's own .done()/.cancelled() state directly, which is deterministic (a warnings-capture-based version of the same test was tried first and found unreliable, since the GC-triggered warning can fire outside any reasonably-sized capture window). Full suite passes on Python 3.11 and 3.13, ruff clean, --random-order stable across repeated runs, and a full faust-streaming/faust cross-check (tests/unit + tests/functional, 2128 passed/32 skipped/0 failed, unchanged from before this change) confirms no regression in the AIOKafkaConsumerThread-based driver that originally reported this issue. Assisted-by: Claude Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01HgnKFtXZbCjXNoVNWa5JLd --- mode/threads.py | 33 ++++++++++++++++ tests/functional/test_threads.py | 65 ++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) create mode 100644 tests/functional/test_threads.py diff --git a/mode/threads.py b/mode/threads.py index 1cff5fe..5c1573a 100644 --- a/mode/threads.py +++ b/mode/threads.py @@ -20,6 +20,7 @@ from .services import Service from .utils.futures import ( + all_tasks, maybe_async, maybe_set_exception, maybe_set_result, @@ -109,6 +110,10 @@ def __init__( raise NotImplementedError("executor argument no longer supported") self.parent_loop = loop or get_event_loop() self.thread_loop = thread_loop or asyncio.new_event_loop() + # Only close the loop on shutdown if we created it ourselves -- + # a caller who passed their own thread_loop= may expect to keep + # managing its lifecycle. + self._close_thread_loop_on_stop = thread_loop is None self._thread_started = Event(loop=self.parent_loop) if Worker is not None: self.Worker = Worker @@ -209,6 +214,9 @@ def _start_thread(self) -> None: # shutdown here, since _shutdown_thread will not execute. self.set_shutdown() raise + finally: + if self._close_thread_loop_on_stop: + self.thread_loop.close() async def stop(self) -> None: if self._started.is_set(): @@ -241,6 +249,30 @@ async def _shutdown_thread(self) -> None: await self._default_stop_futures() await self._default_stop_exit_stacks() + async def _gather_remaining_tasks(self) -> None: + # _default_stop_futures() above only cancels/awaits futures that + # were registered with *this* Service via add_future()/@Service.task. + # A driver library running inside this thread (e.g. a Kafka client's + # internal consumer poll/heartbeat loop) may schedule tasks directly + # on thread_loop, bypassing that tracking entirely. Left unswept, + # those tasks are simply abandoned the moment this coroutine returns + # and thread_loop stops running -- producing "coroutine ... was + # never awaited" / "Task was destroyed but it is pending" warnings + # at some later, unpredictable point (e.g. interpreter shutdown). + # Mirrors Worker._gather_all(), which does the same sweep for the + # main loop. + current = asyncio.current_task() + tasks = { + task + for task in all_tasks(loop=self.thread_loop) + if task is not current and not task.done() + } + if not tasks: + return + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + async def _serve(self) -> None: try: # start the service @@ -260,6 +292,7 @@ async def _serve(self) -> None: raise finally: await self._shutdown_thread() + await self._gather_remaining_tasks() @Service.task async def _thread_keepalive(self) -> None: diff --git a/tests/functional/test_threads.py b/tests/functional/test_threads.py new file mode 100644 index 0000000..6c0a2ea --- /dev/null +++ b/tests/functional/test_threads.py @@ -0,0 +1,65 @@ +import asyncio + +import pytest + +from mode.threads import ServiceThread + + +class UntrackedTaskThread(ServiceThread): + """Mimics a driver library (e.g. a Kafka client) scheduling a task + directly on ``thread_loop``, bypassing Service's own + add_future()/@Service.task tracking entirely. + """ + + debug_task: asyncio.Task + + async def on_thread_started(self) -> None: + self.debug_task = asyncio.ensure_future(self._untracked_sleep()) + + async def _untracked_sleep(self) -> None: + await asyncio.sleep(5.0) + + +@pytest.mark.asyncio +async def test_ServiceThread__sweeps_untracked_tasks_on_stop(): + # Regression test for #54: a task scheduled on thread_loop outside of + # Service's own future-tracking must still be cancelled and awaited to + # completion when the thread stops, instead of being silently + # abandoned -- which later triggers "coroutine ... was never awaited" / + # "Task was destroyed but it is pending" warnings at some + # unpredictable later point (e.g. interpreter shutdown), whenever the + # GC happens to finalize the orphaned task/coroutine object. Asserting + # directly on the task's own state (rather than trying to catch that + # GC-timing-dependent warning) is what actually proves the sweep ran. + thread = UntrackedTaskThread() + await thread.start() + await asyncio.sleep(0.2) # let on_thread_started's task actually begin + assert not thread.debug_task.done() + + await thread.stop() + + assert thread.debug_task.done() + assert thread.debug_task.cancelled() + + +@pytest.mark.asyncio +async def test_ServiceThread__closes_self_created_thread_loop(): + thread = ServiceThread() + await thread.start() + loop = thread.thread_loop + await thread.stop() + + assert loop.is_closed() + + +@pytest.mark.asyncio +async def test_ServiceThread__does_not_close_externally_provided_thread_loop(): + external_loop = asyncio.new_event_loop() + try: + thread = ServiceThread(thread_loop=external_loop) + await thread.start() + await thread.stop() + + assert not external_loop.is_closed() + finally: + external_loop.close()