diff --git a/litellm/litellm_core_utils/logging_worker.py b/litellm/litellm_core_utils/logging_worker.py index 7aa51068099..c0a30015627 100644 --- a/litellm/litellm_core_utils/logging_worker.py +++ b/litellm/litellm_core_utils/logging_worker.py @@ -73,9 +73,8 @@ class LoggingWorker: ) # Cancel orphaned tasks bound to the old (likely closed) event # loop and suppress "Task was destroyed but it is pending!" - # warnings. We cannot await these tasks because their event loop - # is no longer running, so we mark them to skip the __del__ - # warning instead. + # warnings. We cannot await these tasks because their event loop + # is no longer running. self._discard_orphaned_tasks() # Clear old state - these are bound to the old loop self._queue = None @@ -418,25 +417,25 @@ class LoggingWorker: break def _discard_orphaned_tasks(self) -> None: - """Cancel orphaned tasks and suppress their destroy warnings. + """Cancel orphaned tasks and suppress destroy warnings. - When the event loop changes or the process is exiting, pending tasks - bound to the old (likely closed) loop cannot be properly awaited. - We attempt to cancel them and, regardless of whether cancel() - succeeds (it may raise ``RuntimeError`` if the loop is already - closed), suppress the ``"Task was destroyed but it is pending!"`` - warning by setting ``_log_destroy_pending = False``. + When the event loop changes or the process exits, pending tasks bound to + the old loop cannot be awaited. We still cancel them best-effort and + disable asyncio's pending-task destroy warning. """ all_tasks = list(self._running_tasks) if self._worker_task is not None: all_tasks.append(self._worker_task) - for t in all_tasks: - if not t.done(): - try: - t.cancel() - except RuntimeError: - pass # Event loop is already closed - t._log_destroy_pending = False + + for task in all_tasks: + if task.done(): + continue + try: + task.cancel() + except RuntimeError: + pass # Event loop is already closed. + task._log_destroy_pending = False + self._worker_task = None self._running_tasks.clear() @@ -495,9 +494,8 @@ class LoggingWorker: Note: All logging in this method is wrapped to handle cases where logging handlers are closed during shutdown. """ - # Cancel the old worker task — its event loop is already closed. - # Suppress "Task was destroyed but it is pending!" warnings since - # the closed loop cannot process cancellation. + # The original worker loop is bound to the old event loop, which is + # often already closed by the time atexit runs. self._discard_orphaned_tasks() if self._queue is None: diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index db71aa4c03b..9ae5c420e9e 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -363,61 +363,36 @@ class TestLoggingWorker: @pytest.mark.asyncio async def test_event_loop_change_cancels_orphaned_tasks(self): - """Test that switching event loops cancels old tasks and suppresses warnings. - - When the event loop changes (e.g. between asyncio.run() calls or in - test suites), _ensure_queue() must cancel the old worker task and set - _log_destroy_pending = False so that garbage-collecting the orphaned - task does not emit "Task was destroyed but it is pending!" warnings. - """ + """Restarting on a new loop should cancel and discard the old worker.""" worker = LoggingWorker(timeout=1.0, max_queue_size=10) - - # Start the worker — creates _worker_task on the current loop worker.start() await asyncio.sleep(0.05) old_task = worker._worker_task - assert old_task is not None and not old_task.done() + assert old_task is not None + assert old_task.done() is False - # Simulate an event loop change: bind worker to a different loop - # so that the next _ensure_queue() call detects the mismatch. worker._bound_loop = asyncio.new_event_loop() - - # Re-start triggers _ensure_queue() which should cancel the old task worker.start() await asyncio.sleep(0.05) - # The old task should have been cancelled and marked to suppress - # the "Task was destroyed but it is pending!" warning. - assert old_task.cancelled() or old_task.done() + assert old_task.done() or old_task.cancelled() assert old_task._log_destroy_pending is False - - # The worker should be running on the current loop now assert worker._worker_task is not None assert worker._worker_task is not old_task await worker.stop() def test_flush_on_exit_cancels_worker_task(self): - """Test that _flush_on_exit cancels the worker task to avoid warnings. - - When the atexit handler fires, the original event loop is closed. - _flush_on_exit must cancel the orphaned worker task and suppress - its destroy warning before creating a new loop to drain the queue. - """ + """_flush_on_exit should discard the old worker bound to a closed loop.""" loop = asyncio.new_event_loop() worker = LoggingWorker(timeout=1.0, max_queue_size=10) - # Start the worker on a loop, then close it (simulating process exit) loop.run_until_complete(self._start_worker(worker)) old_task = worker._worker_task assert old_task is not None - # Close the loop — this is what happens before atexit fires loop.close() - - # Now _flush_on_exit should clean up the orphaned task without - # raising RuntimeError from the closed event loop. worker._flush_on_exit() assert old_task._log_destroy_pending is False