Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions Lib/test/test_asyncio/test_free_threading.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,72 @@ def runner():
support.gc_collect()
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():
await asyncio.Future()

def make_pending_tasks(loop, count):
tasks = []
for _ in range(count):
task = loop.create_task(never_finishes())
task._log_destroy_pending = False
tasks.append(task)
return tasks

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()

def inspector():
loop = asyncio.new_event_loop()
try:
while not stop.is_set():
asyncio.all_tasks(loop)
finally:
loop.close()

inspectors = [Thread(target=inspector) for _ in range(2)]
with threading_helper.start_threads(inspectors):
try:
for _ in range(15):
batches = []
workers = [Thread(target=worker, args=(batches,)) for _ in range(4)]
with threading_helper.start_threads(workers):
pass

def cleanup(batch):
loop, tasks = batch
for task in tasks:
task.cancel()
task.get_coro().close()
tasks.clear()
loop.close()

cleaners = [Thread(target=cleanup, args=(b,)) for b in batches]
with threading_helper.start_threads(cleaners):
pass
support.gc_collect()
finally:
stop.set()

def test_run_coroutine_threadsafe(self) -> None:
results = []

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix a free-threaded asyncio race leading to segfaults.
33 changes: 23 additions & 10 deletions Modules/_asynciomodule.c
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,9 @@ typedef struct TaskObj {
PyObject *task_context;
struct llist_node task_node;
#ifdef Py_GIL_DISABLED
// thread id of the thread where this task was created
uintptr_t task_tid;
// unique id (PyThreadState.id) of the thread state that created this
// task, see unregister_task()
uint64_t task_tstate_id;
#endif
} TaskObj;

Expand Down Expand Up @@ -2151,6 +2152,9 @@ 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);
}
Expand All @@ -2170,15 +2174,19 @@ static void
unregister_task(TaskObj *task)
{
#ifdef Py_GIL_DISABLED
// check if we are in the same thread
// if so, we can avoid locking
if (task->task_tid == _Py_ThreadId()) {
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) {
unregister_task_safe(task);
}
else {
// we are in a different thread
// stop the world then check and remove the task
PyThreadState *tstate = _PyThreadState_GET();
// 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
_PyEval_StopTheWorld(tstate->interp);
unregister_task_safe(task);
_PyEval_StartTheWorld(tstate->interp);
Expand Down Expand Up @@ -2307,7 +2315,9 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,

Py_CLEAR(self->task_fut_waiter);
#ifdef Py_GIL_DISABLED
self->task_tid = _Py_ThreadId();
if (self->task_node.next == NULL) {
self->task_tstate_id = _PyThreadState_GET()->id;
}
#endif
self->task_must_cancel = 0;
self->task_log_destroy_pending = 1;
Expand Down Expand Up @@ -3948,7 +3958,10 @@ 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);
assert(task->task_state == STATE_PENDING);
// 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.
// The linked list holds borrowed references to task
// as such it is possible that the task is concurrently
// deallocated while added to this list.
Expand Down
Loading