Skip to content

Commit aaaf61c

Browse files
gh-159449: Fix free-threaded asyncio race that corrupts the task lists (#159483)
Co-authored-by: Kumar Aditya <kumaraditya@python.org>
1 parent 4b7b581 commit aaaf61c

3 files changed

Lines changed: 65 additions & 12 deletions

File tree

‎Lib/test/test_asyncio/test_free_threading.py‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,59 @@ def runner():
116116
support.gc_collect()
117117
self.assertIsNone(wr())
118118

119+
def test_unregister_lingering_tasks_of_exited_thread(self) -> None:
120+
# gh-159449: a new thread can get the same OS thread id as
121+
# an exited thread whose tasks linger on the interpreter's list.
122+
class InertLoop:
123+
def get_debug(self):
124+
return False
125+
126+
def is_running(self):
127+
return False
128+
129+
def call_soon(self, *args, **kwargs):
130+
pass
131+
132+
loop = InertLoop()
133+
done = threading.Event()
134+
135+
async def coro():
136+
pass
137+
138+
def worker(handoff):
139+
while handoff:
140+
for task in handoff.pop():
141+
# complete the task right away, unlike Task.cancel()
142+
asyncio.Future.cancel(task)
143+
tasks = []
144+
for _ in range(32):
145+
c = coro()
146+
task = self.factory(loop, c)
147+
c.close()
148+
task._log_destroy_pending = False
149+
tasks.append(task)
150+
handoff.append(tasks)
151+
152+
def spawner():
153+
handoff = []
154+
for _ in range(100):
155+
thread = Thread(target=worker, args=(handoff,))
156+
thread.start()
157+
thread.join()
158+
159+
def inspector():
160+
while not done.is_set():
161+
asyncio.all_tasks(loop)
162+
163+
inspectors = [Thread(target=inspector) for _ in range(2)]
164+
with threading_helper.start_threads(inspectors):
165+
try:
166+
spawners = [Thread(target=spawner) for _ in range(4)]
167+
with threading_helper.start_threads(spawners):
168+
pass
169+
finally:
170+
done.set()
171+
119172
def test_run_coroutine_threadsafe(self) -> None:
120173
results = []
121174

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Fix a free-threaded asyncio race leading to segfaults.

‎Modules/_asynciomodule.c‎

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -68,8 +68,8 @@ typedef struct TaskObj {
6868
PyObject *task_context;
6969
struct llist_node task_node;
7070
#ifdef Py_GIL_DISABLED
71-
// thread id of the thread where this task was created
72-
uintptr_t task_tid;
71+
// unique id of the thread state where the task was registered
72+
uint64_t task_tstate_id;
7373
#endif
7474
} TaskObj;
7575

@@ -2185,6 +2185,9 @@ register_task(_PyThreadStateImpl *ts, TaskObj *task)
21852185
assert(task->task_node.prev != NULL);
21862186
return;
21872187
}
2188+
#ifdef Py_GIL_DISABLED
2189+
task->task_tstate_id = ts->base.id;
2190+
#endif
21882191
struct llist_node *head = &ts->asyncio_tasks_head;
21892192
llist_insert_tail(head, &task->task_node);
21902193
}
@@ -2204,15 +2207,15 @@ static void
22042207
unregister_task(TaskObj *task)
22052208
{
22062209
#ifdef Py_GIL_DISABLED
2207-
// check if we are in the same thread
2208-
// if so, we can avoid locking
2209-
if (task->task_tid == _Py_ThreadId()) {
2210+
PyThreadState *tstate = _PyThreadState_GET();
2211+
// The task can only be on this thread's list if this thread state
2212+
// registered it, in which case no locking is needed.
2213+
if (task->task_tstate_id == tstate->id) {
22102214
unregister_task_safe(task);
22112215
}
22122216
else {
2213-
// we are in a different thread
2214-
// stop the world then check and remove the task
2215-
PyThreadState *tstate = _PyThreadState_GET();
2217+
// The task is on another thread's list or on the interpreter's
2218+
// list of lingering tasks: stop the world to remove it.
22162219
_PyEval_StopTheWorld(tstate->interp);
22172220
unregister_task_safe(task);
22182221
_PyEval_StartTheWorld(tstate->interp);
@@ -2340,9 +2343,6 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
23402343
}
23412344

23422345
Py_CLEAR(self->task_fut_waiter);
2343-
#ifdef Py_GIL_DISABLED
2344-
self->task_tid = _Py_ThreadId();
2345-
#endif
23462346
self->task_must_cancel = 0;
23472347
self->task_log_destroy_pending = 1;
23482348
self->task_num_cancels_requested = 0;
@@ -3987,7 +3987,6 @@ add_tasks_llist(struct llist_node *head, PyListObject *tasks)
39873987
struct llist_node *node;
39883988
llist_for_each_safe(node, head) {
39893989
TaskObj *task = llist_data(node, TaskObj, task_node);
3990-
assert(task->task_state == STATE_PENDING);
39913990
// The linked list holds borrowed references to task
39923991
// as such it is possible that the task is concurrently
39933992
// deallocated while added to this list.

0 commit comments

Comments
 (0)