| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -150,6 +150,23 @@ extern void _PyThreadState_Detach(PyThreadState *tstate); | |||
| 150 | 150 | // to the "detached" state. | |
| 151 | 151 | extern void _PyThreadState_Suspend(PyThreadState *tstate); | |
| 152 | 152 | ||
| 153 | + #ifdef Py_GIL_DISABLED | ||
| 154 | + // Try to atomically transition a *different* thread's state from "detached" | ||
| 155 | + // to "suspended". On success, the target thread cannot attach until | ||
| 156 | + // _PyThreadState_ResumeDetached() is called, and the caller may safely | ||
| 157 | + // perform operations that are normally only permitted for the owning thread | ||
| 158 | + // (such as merging the biased reference counts of objects it owns). | ||
| 159 | + // | ||
| 160 | + // The caller must not run arbitrary Python code, allocate GC objects, or | ||
| 161 | + // stop the world while holding the thread in the suspended state. | ||
| 162 | + // Returns 1 on success, 0 if the thread was not in the "detached" state. | ||
| 163 | + extern int _PyThreadState_TrySuspendDetached(PyThreadState *tstate); | ||
| 164 | + | ||
| 165 | + // Undo a successful _PyThreadState_TrySuspendDetached(): switch the thread | ||
| 166 | + // back to "detached" and wake it if it is waiting to attach. | ||
| 167 | + extern void _PyThreadState_ResumeDetached(PyThreadState *tstate); | ||
| 168 | + #endif | ||
| 169 | + | ||
| 153 | 170 | // Mark the thread state as "shutting down". This is used during interpreter | |
| 154 | 171 | // and runtime finalization. The thread may no longer attach to the | |
| 155 | 172 | // interpreter and will instead block via _PyThreadState_HangThread(). | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -5,7 +5,9 @@ | |||
| 5 | 5 | import time | |
| 6 | 6 | from unittest import TestCase | |
| 7 | 7 | import gc | |
| 8 | + import weakref | ||
| 8 | 9 | ||
| 10 | + from test import support | ||
| 9 | 11 | from test.support import threading_helper | |
| 10 | 12 | ||
| 11 | 13 | ||
@@ -95,6 +97,39 @@ def evil(): | |||
| 95 | 97 | thread.start() | |
| 96 | 98 | thread.join() | |
| 97 | 99 | ||
| 100 | + def test_merge_brc_queue_of_detached_thread(self): | ||
| 101 | + # GH-157838: objects queued for merging by a thread that is detached | ||
| 102 | + # (blocked in a lock acquire, sleep, etc.) are merged and freed on its | ||
| 103 | + # behalf instead of staying alive until it runs Python code again. | ||
| 104 | + lock = threading.Lock() | ||
| 105 | + lock.acquire() | ||
| 106 | + ready = threading.Event() | ||
| 107 | + objs = [] | ||
| 108 | + | ||
| 109 | + def worker(): | ||
| 110 | + # Objects owned by this thread; only the list holds a reference. | ||
| 111 | + objs.extend(MyObj() for _ in range(100)) | ||
| 112 | + ready.set() | ||
| 113 | + lock.acquire() # block while detached | ||
| 114 | + | ||
| 115 | + thread = Thread(target=worker) | ||
| 116 | + thread.start() | ||
| 117 | + try: | ||
| 118 | + ready.wait() | ||
| 119 | + # The worker may not have detached yet when the first objects | ||
| 120 | + # are dropped; keep trying until one is freed immediately. | ||
| 121 | + for _ in support.sleeping_retry(support.SHORT_TIMEOUT, error=False): | ||
| 122 | + obj = objs.pop() | ||
| 123 | + wr = weakref.ref(obj) | ||
| 124 | + del obj | ||
| 125 | + if wr() is None: | ||
| 126 | + break | ||
| 127 | + else: | ||
| 128 | + self.fail("object not freed while owning thread was detached") | ||
| 129 | + finally: | ||
| 130 | + lock.release() | ||
| 131 | + thread.join() | ||
| 132 | + | ||
| 98 | 133 | def test_gc_callbacks_race_with_mutation(self): | |
| 99 | 134 | def collect(): | |
| 100 | 135 | b.wait() | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1 @@ | |||
| 1 | + Merge biased reference counts on behalf of threads that are detached instead of waiting for them to attach again, in the free-threaded build. | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -48,6 +48,28 @@ find_thread_state(struct _brc_bucket *bucket, uintptr_t thread_id) | |||
| 48 | 48 | return NULL; | |
| 49 | 49 | } | |
| 50 | 50 | ||
| 51 | + // Merge the refcounts of all objects in `stack`, keeping the queue's reference. | ||
| 52 | + static void | ||
| 53 | + merge_queued_refcounts(_PyObjectStack *stack) | ||
| 54 | + { | ||
| 55 | + for (_PyObjectStackChunk *buf = stack->head; buf != NULL; buf = buf->prev) { | ||
| 56 | + for (Py_ssize_t i = 0; i < buf->n; i++) { | ||
| 57 | + _Py_ExplicitMergeRefcount(buf->objs[i], 0); | ||
| 58 | + } | ||
| 59 | + } | ||
| 60 | + } | ||
| 61 | + | ||
| 62 | + // Release the queue's reference to each merged object. This may run | ||
| 63 | + // destructors, so the bucket mutex must not be held. | ||
| 64 | + static void | ||
| 65 | + decref_merged_objects(_PyObjectStack *stack) | ||
| 66 | + { | ||
| 67 | + PyObject *ob; | ||
| 68 | + while ((ob = _PyObjectStack_Pop(stack)) != NULL) { | ||
| 69 | + Py_DECREF(ob); | ||
| 70 | + } | ||
| 71 | + } | ||
| 72 | + | ||
| 51 | 73 | // Enqueue an object to be merged by the owning thread. This steals a | |
| 52 | 74 | // reference to the object. | |
| 53 | 75 | void | |
@@ -93,6 +115,22 @@ _Py_brc_queue_object(PyObject *ob) | |||
| 93 | 115 | return; | |
| 94 | 116 | } | |
| 95 | 117 | ||
| 118 | + if (_PyThreadState_TrySuspendDetached(&tstate->base)) { | ||
| 119 | + // The owning thread is detached (e.g. blocked on a lock or in a | ||
| 120 | + // system call) and may not run Python code again for a long time, | ||
| 121 | + // so merge its queue on its behalf instead of waiting for it. While | ||
| 122 | + // it is held in the "suspended" state it cannot attach and therefore | ||
| 123 | + // cannot touch ob_ref_local or ob_tid. | ||
| 124 | + _PyObjectStack merged = {0}; | ||
| 125 | + _PyObjectStack_Merge(&merged, &tstate->brc.objects_to_merge); | ||
| 126 | + merge_queued_refcounts(&merged); | ||
| 127 | + _PyThreadState_ResumeDetached(&tstate->base); | ||
| 128 | + PyMutex_Unlock(&bucket->mutex); | ||
| 129 | + | ||
| 130 | + decref_merged_objects(&merged); | ||
| 131 | + return; | ||
| 132 | + } | ||
| 133 | + | ||
| 96 | 134 | // Notify owning thread | |
| 97 | 135 | _Py_set_eval_breaker_bit(&tstate->base, _PY_EVAL_EXPLICIT_MERGE_BIT); | |
| 98 | 136 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -2380,6 +2380,27 @@ _PyThreadState_SetShuttingDown(PyThreadState *tstate) | |||
| 2380 | 2380 | #endif | |
| 2381 | 2381 | } | |
| 2382 | 2382 | ||
| 2383 | + #ifdef Py_GIL_DISABLED | ||
| 2384 | + int | ||
| 2385 | + _PyThreadState_TrySuspendDetached(PyThreadState *tstate) | ||
| 2386 | + { | ||
| 2387 | + assert(tstate != _PyThreadState_GET()); | ||
| 2388 | + int expected = _Py_THREAD_DETACHED; | ||
| 2389 | + return _Py_atomic_compare_exchange_int(&tstate->state, &expected, | ||
| 2390 | + _Py_THREAD_SUSPENDED); | ||
| 2391 | + } | ||
| 2392 | + | ||
| 2393 | + void | ||
| 2394 | + _PyThreadState_ResumeDetached(PyThreadState *tstate) | ||
| 2395 | + { | ||
| 2396 | + assert(tstate != _PyThreadState_GET()); | ||
| 2397 | + assert(_Py_atomic_load_int_relaxed(&tstate->state) == _Py_THREAD_SUSPENDED); | ||
| 2398 | + _Py_atomic_store_int(&tstate->state, _Py_THREAD_DETACHED); | ||
| 2399 | + // Wake the thread if it is parked in tstate_wait_attach(). | ||
| 2400 | + _PyParkingLot_UnparkAll(&tstate->state); | ||
| 2401 | + } | ||
| 2402 | + #endif | ||
| 2403 | + | ||
| 2383 | 2404 | // Decrease stop-the-world counter of remaining number of threads that need to | |
| 2384 | 2405 | // pause. If we are the final thread to pause, notify the requesting thread. | |
| 2385 | 2406 | static void | |
| Back | FazBrowse Home | New Git URL |
0 commit comments