Skip to content

Commit 3619efa

Browse files
miss-islingtonjblespiaukumaraditya303
authored
[3.15] gh-159449: Fix free-threaded asyncio race that corrupts the task lists (GH-159483) (#161379)
gh-159449: Fix free-threaded asyncio race that corrupts the task lists (GH-159483) (cherry picked from commit aaaf61c) Co-authored-by: Jean-Baptiste Lespiau <jeanbaptiste.lespiau@gmail.com> Co-authored-by: Kumar Aditya <kumaraditya@python.org>
1 parent 0d4914c commit 3619efa

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

@@ -2195,6 +2195,9 @@ register_task(_PyThreadStateImpl *ts, TaskObj *task)
21952195
assert(task->task_node.prev != NULL);
21962196
return;
21972197
}
2198+
#ifdef Py_GIL_DISABLED
2199+
task->task_tstate_id = ts->base.id;
2200+
#endif
21982201
struct llist_node *head = &ts->asyncio_tasks_head;
21992202
llist_insert_tail(head, &task->task_node);
22002203
}
@@ -2214,15 +2217,15 @@ static void
22142217
unregister_task(TaskObj *task)
22152218
{
22162219
#ifdef Py_GIL_DISABLED
2217-
// check if we are in the same thread
2218-
// if so, we can avoid locking
2219-
if (task->task_tid == _Py_ThreadId()) {
2220+
PyThreadState *tstate = _PyThreadState_GET();
2221+
// The task can only be on this thread's list if this thread state
2222+
// registered it, in which case no locking is needed.
2223+
if (task->task_tstate_id == tstate->id) {
22202224
unregister_task_safe(task);
22212225
}
22222226
else {
2223-
// we are in a different thread
2224-
// stop the world then check and remove the task
2225-
PyThreadState *tstate = _PyThreadState_GET();
2227+
// The task is on another thread's list or on the interpreter's
2228+
// list of lingering tasks: stop the world to remove it.
22262229
_PyEval_StopTheWorld(tstate->interp);
22272230
unregister_task_safe(task);
22282231
_PyEval_StartTheWorld(tstate->interp);
@@ -2350,9 +2353,6 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
23502353
}
23512354

23522355
Py_CLEAR(self->task_fut_waiter);
2353-
#ifdef Py_GIL_DISABLED
2354-
self->task_tid = _Py_ThreadId();
2355-
#endif
23562356
self->task_must_cancel = 0;
23572357
self->task_log_destroy_pending = 1;
23582358
self->task_num_cancels_requested = 0;
@@ -3994,7 +3994,6 @@ add_tasks_llist(struct llist_node *head, PyListObject *tasks)
39943994
struct llist_node *node;
39953995
llist_for_each_safe(node, head) {
39963996
TaskObj *task = llist_data(node, TaskObj, task_node);
3997-
assert(task->task_state == STATE_PENDING);
39983997
// The linked list holds borrowed references to task
39993998
// as such it is possible that the task is concurrently
40003999
// deallocated while added to this list.

0 commit comments

Comments
 (0)