GH-107803: use circular double linked list for tasks in `_asyncio` (#126577)

Modules/_asynciomodule.c

@@ -138,47 +138,18 @@ typedef struct {
     /* Counter for autogenerated Task names */
     uint64_t task_name_counter;
 
-    /* Linked-list of all tasks which are instances of asyncio.Task or subclasses
+    /* Circular linked-list of all tasks which are instances of asyncio.Task or subclasses
        of it. Third party tasks implementations which don't inherit from
        asyncio.Task are tracked separately using the 'non_asyncio_tasks' WeakSet.
-       `tail` is used as a sentinel to mark the end of the linked-list. It avoids one
+       `first` is used as a sentinel to mark the end of the linked-list. It avoids one
        branch in checking for empty list when adding a new task, the list is
-       initialized with `head` pointing to `tail` to mark an empty list.
+       initialized with `head`, `head->next` and `head->prev` pointing to `first`
+       to mark an empty list.
 
-       Invariants:
-        * When the list is empty:
-        - asyncio_tasks.head == &asyncio_tasks.tail
-        - asyncio_tasks.head->prev == NULL
-        - asyncio_tasks.head->next == NULL
-
-        * After adding the first task 'task1':
-        - asyncio_tasks.head == task1
-        - task1->next == &asyncio_tasks.tail
-        - task1->prev == NULL
-        - asyncio_tasks.tail.prev == task1
-
-        * After adding a second task 'task2':
-        - asyncio_tasks.head == task2
-        - task2->next == task1
-        - task2->prev == NULL
-        - task1->prev == task2
-        - asyncio_tasks.tail.prev == task1
-
-        * After removing task 'task1':
-        - asyncio_tasks.head == task2
-        - task2->next == &asyncio_tasks.tail
-        - task2->prev == NULL
-        - asyncio_tasks.tail.prev == task2
-
-        * After removing task 'task2', the list is empty:
-        - asyncio_tasks.head == &asyncio_tasks.tail
-        - asyncio_tasks.head->prev == NULL
-        - asyncio_tasks.tail.prev == NULL
-        - asyncio_tasks.tail.next == NULL
     */
 
     struct {
-        TaskObj tail;
+        TaskObj first;
         TaskObj *head;
     } asyncio_tasks;
 
@@ -1925,7 +1896,7 @@ register_task(asyncio_state *state, TaskObj *task)
 {
     ASYNCIO_STATE_LOCK(state);
     assert(Task_Check(state, task));
-    assert(task != &state->asyncio_tasks.tail);
+    assert(task != &state->asyncio_tasks.first);
     if (task->next != NULL) {
         // already registered
         goto exit;
@@ -1934,8 +1905,10 @@ register_task(asyncio_state *state, TaskObj *task)
     assert(state->asyncio_tasks.head != NULL);
 
     task->next = state->asyncio_tasks.head;
+    task->prev = state->asyncio_tasks.head->prev;
+    state->asyncio_tasks.head->prev->next = task;
     state->asyncio_tasks.head->prev = task;
-    state->asyncio_tasks.head = task;
+
 exit:
     ASYNCIO_STATE_UNLOCK(state);
 }
@@ -1951,7 +1924,7 @@ unregister_task(asyncio_state *state, TaskObj *task)
 {
     ASYNCIO_STATE_LOCK(state);
     assert(Task_Check(state, task));
-    assert(task != &state->asyncio_tasks.tail);
+    assert(task != &state->asyncio_tasks.first);
     if (task->next == NULL) {
         // not registered
         assert(task->prev == NULL);
@@ -1959,12 +1932,7 @@ unregister_task(asyncio_state *state, TaskObj *task)
         goto exit;
     }
     task->next->prev = task->prev;
-    if (task->prev == NULL) {
-        assert(state->asyncio_tasks.head == task);
-        state->asyncio_tasks.head = task->next;
-    } else {
-        task->prev->next = task->next;
-    }
+    task->prev->next = task->next;
     task->next = NULL;
     task->prev = NULL;
     assert(state->asyncio_tasks.head != task);
@@ -3657,12 +3625,10 @@ _asyncio_all_tasks_impl(PyObject *module, PyObject *loop)
     Py_DECREF(eager_iter);
     int err = 0;
     ASYNCIO_STATE_LOCK(state);
-    TaskObj *head = state->asyncio_tasks.head;
+    TaskObj *first = &state->asyncio_tasks.first;
+    TaskObj *head = state->asyncio_tasks.head->next;
     Py_INCREF(head);
-    assert(head != NULL);
-    assert(head->prev == NULL);
-    TaskObj *tail = &state->asyncio_tasks.tail;
-    while (head != tail)
+    while (head != first)
     {
         if (add_one_task(state, tasks, (PyObject *)head, loop) < 0) {
             Py_DECREF(tasks);
@@ -3875,9 +3841,12 @@ static int
 module_exec(PyObject *mod)
 {
     asyncio_state *state = get_asyncio_state(mod);
-    Py_SET_TYPE(&state->asyncio_tasks.tail, state->TaskType);
-    _Py_SetImmortalUntracked((PyObject *)&state->asyncio_tasks.tail);
-    state->asyncio_tasks.head = &state->asyncio_tasks.tail;
+
+    Py_SET_TYPE(&state->asyncio_tasks.first, state->TaskType);
+    _Py_SetImmortalUntracked((PyObject *)&state->asyncio_tasks.first);
+    state->asyncio_tasks.head = &state->asyncio_tasks.first;
+    state->asyncio_tasks.head->next = &state->asyncio_tasks.first;
+    state->asyncio_tasks.head->prev = &state->asyncio_tasks.first;
 
 #define CREATE_TYPE(m, tp, spec, base)                                  \
     do {                                                                \