Skip to content
Open
53 changes: 53 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,59 @@ def runner():
support.gc_collect()
self.assertIsNone(wr())

def test_unregister_lingering_tasks_of_exited_thread(self) -> None:
# gh-159449: a new thread can get the same OS thread id as
# an exited thread whose tasks linger on the interpreter's list.
class InertLoop:
def get_debug(self):
return False

def is_running(self):
return False

def call_soon(self, *args, **kwargs):
pass

loop = InertLoop()
done = threading.Event()

async def coro():
pass

def worker(handoff):
while handoff:
for task in handoff.pop():
# complete the task right away, unlike Task.cancel()
asyncio.Future.cancel(task)
tasks = []
for _ in range(32):
c = coro()
task = self.factory(loop, c)
c.close()
task._log_destroy_pending = False
tasks.append(task)
handoff.append(tasks)

def spawner():
handoff = []
for _ in range(100):
thread = Thread(target=worker, args=(handoff,))
thread.start()
thread.join()

def inspector():
while not done.is_set():
asyncio.all_tasks(loop)

inspectors = [Thread(target=inspector) for _ in range(2)]
with threading_helper.start_threads(inspectors):
try:
spawners = [Thread(target=spawner) for _ in range(4)]
with threading_helper.start_threads(spawners):
pass
finally:
done.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.
24 changes: 12 additions & 12 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;
// id of the thread state whose asyncio_tasks_head the task was
// registered on; unlike _Py_ThreadId() it is never reused
uint64_t task_tstate_id;
#endif
} TaskObj;

Expand Down Expand Up @@ -2185,6 +2186,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 @@ -2204,15 +2208,15 @@ 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 task can only be on this thread's list if this thread state
// registered it, in which case no locking is needed.
if (task->task_tstate_id == tstate->id) {
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 is on another thread's list or on the interpreter's
// list of lingering tasks: stop the world to remove it.
_PyEval_StopTheWorld(tstate->interp);
unregister_task_safe(task);
_PyEval_StartTheWorld(tstate->interp);
Expand Down Expand Up @@ -2340,9 +2344,6 @@ _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();
#endif
self->task_must_cancel = 0;
self->task_log_destroy_pending = 1;
self->task_num_cancels_requested = 0;
Expand Down Expand Up @@ -3987,7 +3988,6 @@ 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 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