gh-129289: fix crash when task finalizer is not called in asyncio (#129840)

Include/internal/pycore_object.h

@@ -121,7 +121,7 @@ PyAPI_DATA(Py_ssize_t) _Py_RefTotal;
 
 extern void _Py_AddRefTotal(PyThreadState *, Py_ssize_t);
 extern PyAPI_FUNC(void) _Py_IncRefTotal(PyThreadState *);
-extern void _Py_DecRefTotal(PyThreadState *);
+extern PyAPI_FUNC(void) _Py_DecRefTotal(PyThreadState *);
 
 #  define _Py_DEC_REFTOTAL(interp) \
     interp->object_state.reftotal--
@@ -710,7 +710,7 @@ _PyObject_SetMaybeWeakref(PyObject *op)
     }
 }
 
-extern int _PyObject_ResurrectEndSlow(PyObject *op);
+extern PyAPI_FUNC(int) _PyObject_ResurrectEndSlow(PyObject *op);
 #endif
 
 // Temporarily resurrects an object during deallocation. The refcount is set

Lib/test/test_asyncio/test_tasks.py

@@ -2296,6 +2296,22 @@ async def kill_me(loop):
 
         self.assertEqual(self.all_tasks(loop=self.loop), set())
 
+    def test_task_not_crash_without_finalization(self):
+        Task = self.__class__.Task
+
+        class Subclass(Task):
+            def __del__(self):
+                pass
+
+        async def coro():
+            await asyncio.sleep(0.01)
+
+        task = Subclass(coro(), loop = self.loop)
+        task._log_destroy_pending = False
+
+        del task
+
+        support.gc_collect()
 
     @mock.patch('asyncio.base_events.logger')
     def test_tb_logger_not_called_after_cancel(self, m_log):

Modules/_asynciomodule.c

@@ -2938,15 +2938,6 @@ _asyncio_Task_set_name_impl(TaskObj *self, PyObject *value)
 static void
 TaskObj_finalize(TaskObj *task)
 {
-    asyncio_state *state = get_asyncio_state_by_def((PyObject *)task);
-    // Unregister the task from the linked list of tasks.
-    // Since task is a native task, we directly call the
-    // unregister_task function. Third party event loops
-    // should use the asyncio._unregister_task function.
-    // See https://docs.python.org/3/library/asyncio-extending.html#task-lifetime-support
-
-    unregister_task(state, task);
-
     PyObject *context;
     PyObject *message = NULL;
     PyObject *func;
@@ -3071,8 +3062,15 @@ TaskObj_dealloc(PyObject *self)
 {
     TaskObj *task = (TaskObj *)self;
 
-    if (PyObject_CallFinalizerFromDealloc(self) < 0) {
-        // resurrected.
+    _PyObject_ResurrectStart(self);
+    // Unregister the task here so that even if any subclass of Task
+    // which doesn't end up calling TaskObj_finalize not crashes.
+    asyncio_state *state = get_asyncio_state_by_def(self);
+    unregister_task(state, task);
+
+    PyObject_CallFinalizer(self);
+
+    if (_PyObject_ResurrectEnd(self)) {
         return;
     }