Skip to content

Commit 6e88927

Browse files
[3.14] gh-159449: Fix free-threaded asyncio race that corrupts the task lists (GH-159483) (#161380)
(cherry picked from commit aaaf61c) Co-authored-by: Jean-Baptiste Lespiau <jeanbaptiste.lespiau@gmail.com>
1 parent 9f36d85 commit 6e88927

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
@@ -67,8 +67,8 @@ typedef struct TaskObj {
6767
PyObject *task_context;
6868
struct llist_node task_node;
6969
#ifdef Py_GIL_DISABLED
70-
// thread id of the thread where this task was created
71-
uintptr_t task_tid;
70+
// unique id of the thread state where the task was registered
71+
uint64_t task_tstate_id;
7272
#endif
7373
} TaskObj;
7474

@@ -2209,6 +2209,9 @@ register_task(TaskObj *task)
22092209
return;
22102210
}
22112211
_PyThreadStateImpl *tstate = (_PyThreadStateImpl *) _PyThreadState_GET();
2212+
#ifdef Py_GIL_DISABLED
2213+
task->task_tstate_id = tstate->base.id;
2214+
#endif
22122215
struct llist_node *head = &tstate->asyncio_tasks_head;
22132216
llist_insert_tail(head, &task->task_node);
22142217
}
@@ -2228,15 +2231,15 @@ static void
22282231
unregister_task(TaskObj *task)
22292232
{
22302233
#ifdef Py_GIL_DISABLED
2231-
// check if we are in the same thread
2232-
// if so, we can avoid locking
2233-
if (task->task_tid == _Py_ThreadId()) {
2234+
PyThreadState *tstate = _PyThreadState_GET();
2235+
// The task can only be on this thread's list if this thread state
2236+
// registered it, in which case no locking is needed.
2237+
if (task->task_tstate_id == tstate->id) {
22342238
unregister_task_safe(task);
22352239
}
22362240
else {
2237-
// we are in a different thread
2238-
// stop the world then check and remove the task
2239-
PyThreadState *tstate = _PyThreadState_GET();
2241+
// The task is on another thread's list or on the interpreter's
2242+
// list of lingering tasks: stop the world to remove it.
22402243
_PyEval_StopTheWorld(tstate->interp);
22412244
unregister_task_safe(task);
22422245
_PyEval_StartTheWorld(tstate->interp);
@@ -2370,9 +2373,6 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
23702373
}
23712374

23722375
Py_CLEAR(self->task_fut_waiter);
2373-
#ifdef Py_GIL_DISABLED
2374-
self->task_tid = _Py_ThreadId();
2375-
#endif
23762376
self->task_must_cancel = 0;
23772377
self->task_log_destroy_pending = 1;
23782378
self->task_num_cancels_requested = 0;
@@ -4009,7 +4009,6 @@ add_tasks_llist(struct llist_node *head, PyListObject *tasks)
40094009
struct llist_node *node;
40104010
llist_for_each_safe(node, head) {
40114011
TaskObj *task = llist_data(node, TaskObj, task_node);
4012-
assert(task->task_state == STATE_PENDING);
40134012
// The linked list holds borrowed references to task
40144013
// as such it is possible that the task is concurrently
40154014
// deallocated while added to this list.

0 commit comments

Comments
 (0)