@@ -117,11 +117,6 @@ def runner():
         self.assertIsNone(wr())
 
     def test_lingering_task_different_thread_finalized(self) -> None:
-        # When a thread exits with pending tasks, PyThreadState_Clear() moves
-        # them onto interp->asyncio_tasks_head. Unregistering those tasks from
-        # later threads (which may reuse the exited thread's OS thread ID) or
-        # during PyThreadState_Clear() while all_tasks() walks the task lists
-        # must not corrupt the shared list.
         tls = threading.local()
 
         async def never_finishes():
@@ -138,14 +133,12 @@ def make_pending_tasks(loop, count):
         def worker(batches):
             loop = asyncio.new_event_loop()
             loop.set_task_factory(self.factory)
-            # Tasks held in thread-local storage are cleared by PyThreadState_Clear
-            # after llist_concat moves tstate->asyncio_tasks_head to interp.
+
             tls_tasks = make_pending_tasks(loop, 8)
             for task in tls_tasks:
                 task.get_coro().close()
             tls.tasks = tls_tasks
-            # Tasks returned in batches stay alive until a subsequent worker
-            # thread (often reusing this thread's OS thread ID) unregisters them.
+
             batches.append((loop, make_pending_tasks(loop, 8)))
 
         stop = threading.Event()
 
@@ -71,6 +71,7 @@ typedef struct TaskObj {
     // unique id (PyThreadState.id) of the thread state that created this
     // task, see unregister_task()
     uint64_t task_tstate_id;
+
 #endif
 } TaskObj;
 
@@ -1796,9 +1797,10 @@ FutureIter_dealloc(PyObject *it)
 }
 
 static PySendResult
-FutureIter_am_send_lock_held(FutureObj *fut, PyObject **result)
+FutureIter_am_send_lock_held(futureiterobject *it, PyObject **result)
 {
     PyObject *res;
+    FutureObj *fut = it->future;
     _Py_CRITICAL_SECTION_ASSERT_OBJECT_LOCKED(fut);
 
     *result = NULL;
@@ -1829,24 +1831,11 @@ FutureIter_am_send(PyObject *op,
                    PyObject **result)
 {
     futureiterobject *it = (futureiterobject*)op;
-
-    FutureObj *fut;
-    Py_BEGIN_CRITICAL_SECTION(op);
-    fut = (FutureObj*)Py_XNewRef(it->future);
-    Py_END_CRITICAL_SECTION();
-
-    if (fut == NULL) {
-        PyErr_SetNone(PyExc_StopIteration);
-        *result = NULL;
-        return PYGEN_ERROR;
-    }
-
     /* arg is unused, see the comment on FutureIter_send for clarification */
     PySendResult res;
-    Py_BEGIN_CRITICAL_SECTION(fut);
-    res = FutureIter_am_send_lock_held(fut, result);
+    Py_BEGIN_CRITICAL_SECTION(it->future);
+    res = FutureIter_am_send_lock_held(it, result);
     Py_END_CRITICAL_SECTION();
-    Py_DECREF(fut);
     return res;
 }
 
@@ -1940,9 +1929,7 @@ FutureIter_throw(PyObject *op, PyObject *const *args, Py_ssize_t nargs)
         goto fail;
     }
 
-    Py_BEGIN_CRITICAL_SECTION(self);
     Py_CLEAR(self->future);
-    Py_END_CRITICAL_SECTION();
 
     PyErr_Restore(type, val, tb);
 
@@ -1955,16 +1942,6 @@ FutureIter_throw(PyObject *op, PyObject *const *args, Py_ssize_t nargs)
     return NULL;
 }
 
-static PyObject *
-FutureIter_close(PyObject *self, PyObject *arg)
-{
-    futureiterobject *it = (futureiterobject*)self;
-    Py_BEGIN_CRITICAL_SECTION(self);
-    Py_CLEAR(it->future);
-    Py_END_CRITICAL_SECTION();
-    Py_RETURN_NONE;
-}
-
 static int
 FutureIter_clear(PyObject *op)
 {
@@ -1973,6 +1950,13 @@ FutureIter_clear(PyObject *op)
     return 0;
 }
 
+static PyObject *
+FutureIter_close(PyObject *self, PyObject *arg)
+{
+    (void)FutureIter_clear(self);
+    Py_RETURN_NONE;
+}
+
 static int
 FutureIter_traverse(PyObject *op, visitproc visit, void *arg)
 {
@@ -1989,22 +1973,6 @@ static PyMethodDef FutureIter_methods[] = {
     {NULL, NULL}        /* Sentinel */
 };
 
-static PyObject *
-FutureIter_get_fi_future(PyObject *op, void *Py_UNUSED(closure))
-{
-    futureiterobject *it = (futureiterobject*)op;
-    PyObject *fut;
-    Py_BEGIN_CRITICAL_SECTION(op);
-    fut = it->future ? Py_NewRef(it->future) : Py_None;
-    Py_END_CRITICAL_SECTION();
-    return fut;
-}
-
-static PyGetSetDef FutureIter_getsetlist[] = {
-    {"fi_future", FutureIter_get_fi_future, NULL, NULL},
-    {NULL} /* Sentinel */
-};
-
 static PyType_Slot FutureIter_slots[] = {
     {Py_tp_dealloc, FutureIter_dealloc},
     {Py_tp_getattro, PyObject_GenericGetAttr},
@@ -2013,7 +1981,6 @@ static PyType_Slot FutureIter_slots[] = {
     {Py_tp_iter, PyObject_SelfIter},
     {Py_tp_iternext, FutureIter_iternext},
     {Py_tp_methods, FutureIter_methods},
-    {Py_tp_getset, FutureIter_getsetlist},
 
     // async methods
     {Py_am_send, FutureIter_am_send},
@@ -2186,9 +2153,6 @@ register_task(_PyThreadStateImpl *ts, TaskObj *task)
         assert(task->task_node.prev != NULL);
         return;
     }
-#ifdef Py_GIL_DISABLED
-    task->task_tstate_id = ts->base.id;
-#endif
     struct llist_node *head = &ts->asyncio_tasks_head;
     llist_insert_tail(head, &task->task_node);
 }
@@ -2208,19 +2172,15 @@ static void
 unregister_task(TaskObj *task)
 {
 #ifdef Py_GIL_DISABLED
-    PyThreadState *tstate = _PyThreadState_GET();
-    // The lock-free path is only safe while the task is still linked into the
-    // current thread's own list. That holds if this thread state created the
-    // task and has not started PyThreadState_Clear(), which moves lingering
-    // tasks onto the shared interp->asyncio_tasks_head list. Compare thread
-    // state ids rather than _Py_ThreadId(), which is reused by threads that
-    // start after the creating thread has exited.
-    if (task->task_tstate_id == tstate->id && !tstate->_status.finalizing) {
+    // check if we are in the same thread
+    // if so, we can avoid locking
+    if (task->task_tid == _Py_ThreadId()) {
         unregister_task_safe(task);
     }
     else {
-        // the task may be linked into another thread's list or into the
-        // interpreter's list: stop the world then check and remove the task
+        // we are in a different thread
+        // stop the world then check and remove the task
+        PyThreadState *tstate = _PyThreadState_GET();
         _PyEval_StopTheWorld(tstate->interp);
         unregister_task_safe(task);
         _PyEval_StartTheWorld(tstate->interp);
@@ -2349,9 +2309,7 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
 
     Py_CLEAR(self->task_fut_waiter);
 #ifdef Py_GIL_DISABLED
-    if (self->task_node.next == NULL) {
-        self->task_tstate_id = _PyThreadState_GET()->id;
-    }
+    self->task_tid = _Py_ThreadId();
 #endif
     self->task_must_cancel = 0;
     self->task_log_destroy_pending = 1;
@@ -2813,12 +2771,7 @@ _asyncio_Task_get_name_impl(TaskObj *self)
 {
     if (self->task_name) {
         if (PyLong_CheckExact(self->task_name)) {
-            // Formatting can run arbitrary code (e.g. a GC finalizer that logs
-            // and re-enters get_name()), which may replace self->task_name.
-            // Hold a strong reference to the counter while formatting it.
-            PyObject *counter = Py_NewRef(self->task_name);
-            PyObject *name = PyUnicode_FromFormat("Task-%S", counter);
-            Py_DECREF(counter);
+            PyObject *name = PyUnicode_FromFormat("Task-%S", self->task_name);
             if (name == NULL) {
                 return NULL;
             }
@@ -3997,10 +3950,7 @@ add_tasks_llist(struct llist_node *head, PyListObject *tasks)
     struct llist_node *node;
     llist_for_each_safe(node, head) {
         TaskObj *task = llist_data(node, TaskObj, task_node);
-        // The task may already be done but not yet unlinked: its state is set
-        // before unregister_task() removes it, and that removal may be
-        // waiting for a stop-the-world pause. add_one_task() filters out
-        // tasks that are done.
+        assert(task->task_state == STATE_PENDING);
         // The linked list holds borrowed references to task
         // as such it is possible that the task is concurrently
         // deallocated while added to this list.