gh-91887: Store strong references to pending tasks

Include/internal/pycore_global_strings.h

@@ -228,6 +228,7 @@ struct _Py_global_strings {
         STRUCT_FOR_ID(_abc_impl)
         STRUCT_FOR_ID(_abstract_)
         STRUCT_FOR_ID(_active)
+        STRUCT_FOR_ID(_add_pending_task)
         STRUCT_FOR_ID(_align_)
         STRUCT_FOR_ID(_anonymous_)
         STRUCT_FOR_ID(_argtypes_)
@@ -237,11 +238,13 @@ struct _Py_global_strings {
         STRUCT_FOR_ID(_bootstrap)
         STRUCT_FOR_ID(_check_retval_)
         STRUCT_FOR_ID(_dealloc_warn)
+        STRUCT_FOR_ID(_del_pending_task)
         STRUCT_FOR_ID(_feature_version)
         STRUCT_FOR_ID(_field_types)
         STRUCT_FOR_ID(_fields_)
         STRUCT_FOR_ID(_finalizing)
         STRUCT_FOR_ID(_find_and_load)
+        STRUCT_FOR_ID(_finish_execution)
         STRUCT_FOR_ID(_fix_up_module)
         STRUCT_FOR_ID(_flags_)
         STRUCT_FOR_ID(_get_sourcefile)
@@ -257,6 +260,7 @@ struct _Py_global_strings {
         STRUCT_FOR_ID(_needs_com_addref_)
         STRUCT_FOR_ID(_only_immortal)
         STRUCT_FOR_ID(_pack_)
+        STRUCT_FOR_ID(_pending_tasks)
         STRUCT_FOR_ID(_restype_)
         STRUCT_FOR_ID(_showwarnmsg)
         STRUCT_FOR_ID(_shutdown)

Lib/asyncio/base_events.py

@@ -449,6 +449,10 @@ def __init__(self):
         # Set to True when `loop.shutdown_default_executor` is called.
         self._executor_shutdown_called = False
 
+        # Holds references to all pending tasks to avoid garbage
+        # collection (see #91887)
+        self._pending_tasks = set()
+
     def __repr__(self):
         return (
             f'<{self.__class__.__name__} running={self.is_running()} '
@@ -2043,6 +2047,16 @@ def _set_coroutine_origin_tracking(self, enabled):
 
         self._coroutine_origin_tracking_enabled = enabled
 
+    def _add_pending_task(self, task):
+        """Add task to the _pending_tasks set.
+
+        This avoids garbage collection as long as the loop is alive.
+        """
+        self._pending_tasks.add(task)
+
+    def _del_pending_task(self, task):
+        self._pending_tasks.discard(task)
+
     def get_debug(self):
         return self._debug
 

Lib/asyncio/futures.py

@@ -152,11 +152,11 @@ def cancel(self, msg=None):
             return False
         self._state = _CANCELLED
         self._cancel_message = msg
-        self.__schedule_callbacks()
+        self._finish_execution()
         return True
 
-    def __schedule_callbacks(self):
-        """Internal: Ask the event loop to call all callbacks.
+    def _finish_execution(self):
+        """Ask the event loop to call all callbacks.
 
         The callbacks are scheduled to be called as soon as possible. Also
         clears the callback list.
@@ -257,7 +257,7 @@ def set_result(self, result):
             raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
         self._result = result
         self._state = _FINISHED
-        self.__schedule_callbacks()
+        self._finish_execution()
 
     def set_exception(self, exception):
         """Mark the future done and set an exception.
@@ -279,7 +279,7 @@ def set_exception(self, exception):
         self._exception = exception
         self._exception_tb = exception.__traceback__
         self._state = _FINISHED
-        self.__schedule_callbacks()
+        self._finish_execution()
         self.__log_traceback = True
 
     def __await__(self):

Lib/asyncio/tasks.py

@@ -127,6 +127,10 @@ def __init__(self, coro, *, loop=None, name=None, context=None,
             self._loop.call_soon(self.__step, context=self._context)
             _register_task(self)
 
+    def _finish_execution(self):
+        super()._finish_execution()
+        _unregister_task(self)
+
     def __del__(self):
         if self._state == futures._PENDING and self._log_destroy_pending:
             context = {
@@ -1047,6 +1051,8 @@ def factory(loop, coro, *, name=None, context=None):
 def _register_task(task):
     """Register an asyncio Task scheduled to run on an event loop."""
     _scheduled_tasks.add(task)
+    loop = futures._get_loop(task)
+    loop._add_pending_task(task)
 
 
 def _register_eager_task(task):
@@ -1082,6 +1088,8 @@ def _swap_current_task(loop, task):
 def _unregister_task(task):
     """Unregister a completed, scheduled Task."""
     _scheduled_tasks.discard(task)
+    loop = futures._get_loop(task)
+    loop._del_pending_task(task)
 
 
 def _unregister_eager_task(task):

Lib/test/test_asyncio/test_tasks.py

@@ -2278,6 +2278,17 @@ async def kill_me(loop):
         coro = None
         source_traceback = task._source_traceback
         task = None
+
+        # no more reference to kill_me() task in user code. the task
+        # should be kept alive as long as it is pending and we hold a
+        # reference to the event loop (#91887)
+        support.gc_collect()
+
+        self.assertEqual(len(asyncio.all_tasks(loop=self.loop)), 1)
+        mock_handler.assert_not_called()
+
+        # remove strong reference held by the event loop
+        self.loop._pending_tasks.clear()
 
         # no more reference to kill_me() task: the task is destroyed by the GC
         support.gc_collect()

Misc/ACKS

@@ -723,6 +723,7 @@ Derek Harland
 Jason Harper
 David Harrigan
 Brian Harring
+Alexander Hartl
 Jonathan Hartley
 Travis B. Hartwell
 Henrik Harutyunyan

Misc/NEWS.d/next/Library/2024-07-02-14-07-32.gh-issue-91887.eWzc5E.rst

@@ -0,0 +1,2 @@
+Retain strong references for pending :class:`asyncio.Task` objects to avoid rare crashes due
+to background tasks. Patch by Alexander Hartl.