/src/cpython/Python/parking_lot.c

Source
#include "Python.h"

#include "pycore_llist.h"
#include "pycore_lock.h"          // _PyRawMutex
#include "pycore_parking_lot.h"
#include "pycore_pyerrors.h"      // _Py_FatalErrorFormat
#include "pycore_pystate.h"       // _PyThreadState_GET
#include "pycore_semaphore.h"     // _PySemaphore
#include "pycore_time.h"          // _PyTime_Add()

#include <stdbool.h>


typedef struct {
    // The mutex protects the waiter queue and the num_waiters counter.
    _PyRawMutex mutex;

    // Linked list of `struct wait_entry` waiters in this bucket.
    struct llist_node root;
    size_t num_waiters;
} Bucket;

struct wait_entry {
    void *park_arg;
    uintptr_t addr;
    _PySemaphore sema;
    struct llist_node node;
    bool is_unparking;
};

// Prime number to avoid correlations with memory addresses.
// We want this to be roughly proportional to the number of CPU cores
// to minimize contention on the bucket locks, but not too big to avoid
// wasting memory. The exact choice does not matter much.
#define NUM_BUCKETS 257

#define BUCKET_INIT(b, i) [i] = { .root = LLIST_INIT(b[i].root) }
#define BUCKET_INIT_2(b, i)   BUCKET_INIT(b, i),     BUCKET_INIT(b, i+1)
#define BUCKET_INIT_4(b, i)   BUCKET_INIT_2(b, i),   BUCKET_INIT_2(b, i+2)
#define BUCKET_INIT_8(b, i)   BUCKET_INIT_4(b, i),   BUCKET_INIT_4(b, i+4)
#define BUCKET_INIT_16(b, i)  BUCKET_INIT_8(b, i),   BUCKET_INIT_8(b, i+8)
#define BUCKET_INIT_32(b, i)  BUCKET_INIT_16(b, i),  BUCKET_INIT_16(b, i+16)
#define BUCKET_INIT_64(b, i)  BUCKET_INIT_32(b, i),  BUCKET_INIT_32(b, i+32)
#define BUCKET_INIT_128(b, i) BUCKET_INIT_64(b, i),  BUCKET_INIT_64(b, i+64)
#define BUCKET_INIT_256(b, i) BUCKET_INIT_128(b, i), BUCKET_INIT_128(b, i+128)

// Table of waiters (hashed by address)
static Bucket buckets[NUM_BUCKETS] = {
    BUCKET_INIT_256(buckets, 0),
    BUCKET_INIT(buckets, 256),
};

void
_PySemaphore_Init(_PySemaphore *sema)
{
#if defined(MS_WINDOWS)
    sema->platform_sem = CreateSemaphore(
        NULL,   //  attributes
        0,      //  initial count
        10,     //  maximum count
        NULL    //  unnamed
    );
    if (!sema->platform_sem) {
        Py_FatalError("parking_lot: CreateSemaphore failed");
    }
#elif defined(_Py_USE_SEMAPHORES)
    if (sem_init(&sema->platform_sem, /*pshared=*/0, /*value=*/0) < 0) {
        Py_FatalError("parking_lot: sem_init failed");
    }
#else
    if (pthread_mutex_init(&sema->mutex, NULL) != 0) {
        Py_FatalError("parking_lot: pthread_mutex_init failed");
    }
    if (pthread_cond_init(&sema->cond, NULL)) {
        Py_FatalError("parking_lot: pthread_cond_init failed");
    }
    sema->counter = 0;
#endif
}

void
_PySemaphore_Destroy(_PySemaphore *sema)
{
#if defined(MS_WINDOWS)
    CloseHandle(sema->platform_sem);
#elif defined(_Py_USE_SEMAPHORES)
    sem_destroy(&sema->platform_sem);
#else
    pthread_mutex_destroy(&sema->mutex);
    pthread_cond_destroy(&sema->cond);
#endif
}

int
_PySemaphore_Wait(_PySemaphore *sema, PyTime_t timeout)
{
    int res;
#if defined(MS_WINDOWS)
    DWORD wait;
    DWORD millis = 0;
    if (timeout < 0) {
        millis = INFINITE;
    }
    else {
        PyTime_t div = _PyTime_AsMilliseconds(timeout, _PyTime_ROUND_TIMEOUT);
        // Prevent overflow with clamping the result
        if ((PyTime_t)PY_DWORD_MAX < div) {
            millis = PY_DWORD_MAX;
        }
        else {
            millis = (DWORD) div;
        }
    }

    HANDLE handles[2] = { sema->platform_sem, NULL };
    HANDLE sigint_event = NULL;
    DWORD count = 1;
    if (_Py_IsMainThread()) {
        // gh-135099: Wait on the SIGINT event only in the main thread. Other
        // threads would ignore the result anyways, and accessing
        // `_PyOS_SigintEvent()` from non-main threads may race with
        // interpreter shutdown, which closes the event handle. Note that
        // non-main interpreters will ignore the result.
        sigint_event = _PyOS_SigintEvent();
        if (sigint_event != NULL) {
            handles[1] = sigint_event;
            count = 2;
        }
    }
    wait = WaitForMultipleObjects(count, handles, FALSE, millis);
    if (wait == WAIT_OBJECT_0) {
        res = Py_PARK_OK;
    }
    else if (wait == WAIT_OBJECT_0 + 1) {
        assert(sigint_event != NULL);
        ResetEvent(sigint_event);
        res = Py_PARK_INTR;
    }
    else if (wait == WAIT_TIMEOUT) {
        res = Py_PARK_TIMEOUT;
    }
    else {
        _Py_FatalErrorFormat(__func__,
            "unexpected error from semaphore: %u (error: %u)",
            wait, GetLastError());
    }
#elif defined(_Py_USE_SEMAPHORES)
    int err;
    if (timeout >= 0) {
        struct timespec ts;

#if defined(CLOCK_MONOTONIC) && defined(HAVE_SEM_CLOCKWAIT) && !defined(_Py_THREAD_SANITIZER)
        PyTime_t now;
        // silently ignore error: cannot report error to the caller
        (void)PyTime_MonotonicRaw(&now);
        PyTime_t deadline = _PyTime_Add(now, timeout);
        _PyTime_AsTimespec_clamp(deadline, &ts);

        err = sem_clockwait(&sema->platform_sem, CLOCK_MONOTONIC, &ts);
#else
        PyTime_t now;
        // silently ignore error: cannot report error to the caller
        (void)PyTime_TimeRaw(&now);
        PyTime_t deadline = _PyTime_Add(now, timeout);

        _PyTime_AsTimespec_clamp(deadline, &ts);

        err = sem_timedwait(&sema->platform_sem, &ts);
#endif
    }
    else {
        err = sem_wait(&sema->platform_sem);
    }
    if (err == -1) {
        err = errno;
        if (err == EINTR) {
            res = Py_PARK_INTR;
        }
        else if (err == ETIMEDOUT) {
            res = Py_PARK_TIMEOUT;
        }
        else {
            _Py_FatalErrorFormat(__func__,
                "unexpected error from semaphore: %d",
                err);
        }
    }
    else {
        res = Py_PARK_OK;
    }
#else
    pthread_mutex_lock(&sema->mutex);
    int err = 0;
    if (sema->counter == 0) {
        if (timeout >= 0) {
            struct timespec ts;
#if defined(HAVE_PTHREAD_COND_TIMEDWAIT_RELATIVE_NP)
            _PyTime_AsTimespec_clamp(timeout, &ts);
            err = pthread_cond_timedwait_relative_np(&sema->cond, &sema->mutex, &ts);
#else
            PyTime_t now;
            (void)PyTime_TimeRaw(&now);
            PyTime_t deadline = _PyTime_Add(now, timeout);
            _PyTime_AsTimespec_clamp(deadline, &ts);

            err = pthread_cond_timedwait(&sema->cond, &sema->mutex, &ts);
#endif // HAVE_PTHREAD_COND_TIMEDWAIT_RELATIVE_NP
        }
        else {
            err = pthread_cond_wait(&sema->cond, &sema->mutex);
        }
    }
    if (sema->counter > 0) {
        sema->counter--;
        res = Py_PARK_OK;
    }
    else if (err) {
        res = Py_PARK_TIMEOUT;
    }
    else {
        res = Py_PARK_INTR;
    }
    pthread_mutex_unlock(&sema->mutex);
#endif
    return res;
}

void
_PySemaphore_Wakeup(_PySemaphore *sema)
{
#if defined(MS_WINDOWS)
    if (!ReleaseSemaphore(sema->platform_sem, 1, NULL)) {
        Py_FatalError("parking_lot: ReleaseSemaphore failed");
    }
#elif defined(_Py_USE_SEMAPHORES)
    int err = sem_post(&sema->platform_sem);
    if (err != 0) {
        Py_FatalError("parking_lot: sem_post failed");
    }
#else
    pthread_mutex_lock(&sema->mutex);
    sema->counter++;
    pthread_cond_signal(&sema->cond);
    pthread_mutex_unlock(&sema->mutex);
#endif
}

static void
enqueue(Bucket *bucket, const void *address, struct wait_entry *wait)
{
    llist_insert_tail(&bucket->root, &wait->node);
    ++bucket->num_waiters;
}

static struct wait_entry *
dequeue(Bucket *bucket, const void *address)
{
    // find the first waiter that is waiting on `address`
    struct llist_node *root = &bucket->root;
    struct llist_node *node;
    llist_for_each(node, root) {
        struct wait_entry *wait = llist_data(node, struct wait_entry, node);
        if (wait->addr == (uintptr_t)address) {
            llist_remove(node);
            --bucket->num_waiters;
            wait->is_unparking = true;
            return wait;
        }
    }
    return NULL;
}

static void
dequeue_all(Bucket *bucket, const void *address, struct llist_node *dst)
{
    // remove and append all matching waiters to dst
    struct llist_node *root = &bucket->root;
    struct llist_node *node;
    llist_for_each_safe(node, root) {
        struct wait_entry *wait = llist_data(node, struct wait_entry, node);
        if (wait->addr == (uintptr_t)address) {
            llist_remove(node);
            llist_insert_tail(dst, node);
            --bucket->num_waiters;
            wait->is_unparking = true;
        }
    }
}

// Checks that `*addr == *expected` (only works for 1, 2, 4, or 8 bytes)
static int
atomic_memcmp(const void *addr, const void *expected, size_t addr_size)
{
    switch (addr_size) {
    case 1: return _Py_atomic_load_uint8(addr) == *(const uint8_t *)expected;
    case 2: return _Py_atomic_load_uint16(addr) == *(const uint16_t *)expected;
    case 4: return _Py_atomic_load_uint32(addr) == *(const uint32_t *)expected;
    case 8: return _Py_atomic_load_uint64(addr) == *(const uint64_t *)expected;
    default: Py_UNREACHABLE();
    }
}

int
_PyParkingLot_Park(const void *addr, const void *expected, size_t size,
                   PyTime_t timeout_ns, void *park_arg, int detach)
{
    struct wait_entry wait = {
        .park_arg = park_arg,
        .addr = (uintptr_t)addr,
        .is_unparking = false,
    };

    Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];

    _PyRawMutex_Lock(&bucket->mutex);
    if (!atomic_memcmp(addr, expected, size)) {
        _PyRawMutex_Unlock(&bucket->mutex);
        return Py_PARK_AGAIN;
    }
    _PySemaphore_Init(&wait.sema);
    enqueue(bucket, addr, &wait);
    _PyRawMutex_Unlock(&bucket->mutex);

    PyThreadState *tstate = NULL;
    if (detach) {
        tstate = _PyThreadState_GET();
        if (tstate && _PyThreadState_IsAttached(tstate)) {
            // Only detach if we are attached
            PyEval_ReleaseThread(tstate);
        }
        else {
            tstate = NULL;
        }
    }

    int res = _PySemaphore_Wait(&wait.sema, timeout_ns);
    if (res == Py_PARK_OK) {
        goto done;
    }

    // timeout or interrupt
    _PyRawMutex_Lock(&bucket->mutex);
    if (wait.is_unparking) {
        _PyRawMutex_Unlock(&bucket->mutex);
        // Another thread has started to unpark us. Wait until we process the
        // wakeup signal.
        do {
            res = _PySemaphore_Wait(&wait.sema, -1);
        } while (res != Py_PARK_OK);
        goto done;
    }
    else {
        llist_remove(&wait.node);
        --bucket->num_waiters;
    }
    _PyRawMutex_Unlock(&bucket->mutex);

done:
    _PySemaphore_Destroy(&wait.sema);
    if (tstate) {
        PyEval_AcquireThread(tstate);
    }
    return res;

}

void
_PyParkingLot_Unpark(const void *addr, _Py_unpark_fn_t *fn, void *arg)
{
    Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];

    // Find the first waiter that is waiting on `addr`
    _PyRawMutex_Lock(&bucket->mutex);
    struct wait_entry *waiter = dequeue(bucket, addr);
    if (waiter) {
        int has_more_waiters = (bucket->num_waiters > 0);
        fn(arg, waiter->park_arg, has_more_waiters);
    }
    else {
        fn(arg, NULL, 0);
    }
    _PyRawMutex_Unlock(&bucket->mutex);

    if (waiter) {
        // Wakeup the waiter outside of the bucket lock
        _PySemaphore_Wakeup(&waiter->sema);
    }
}

void
_PyParkingLot_UnparkAll(const void *addr)
{
    struct llist_node head = LLIST_INIT(head);
    Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];

    _PyRawMutex_Lock(&bucket->mutex);
    dequeue_all(bucket, addr, &head);
    _PyRawMutex_Unlock(&bucket->mutex);

    struct llist_node *node;
    llist_for_each_safe(node, &head) {
        struct wait_entry *waiter = llist_data(node, struct wait_entry, node);
        llist_remove(node);
        _PySemaphore_Wakeup(&waiter->sema);
    }
}

void
_PyParkingLot_AfterFork(void)
{
    // After a fork only one thread remains. That thread cannot be blocked
    // so all entries in the parking lot are for dead threads.
    memset(buckets, 0, sizeof(buckets));
    for (Py_ssize_t i = 0; i < NUM_BUCKETS; i++) {
        llist_init(&buckets[i].root);
    }
}

Coverage Report

Created: 2025-10-10 06:33

Line	Count	Source
1		#include "Python.h"
2
3		#include "pycore_llist.h"
4		#include "pycore_lock.h" // _PyRawMutex
5		#include "pycore_parking_lot.h"
6		#include "pycore_pyerrors.h" // _Py_FatalErrorFormat
7		#include "pycore_pystate.h" // _PyThreadState_GET
8		#include "pycore_semaphore.h" // _PySemaphore
9		#include "pycore_time.h" // _PyTime_Add()
10
11		#include <stdbool.h>
12
13
14		typedef struct {
15		// The mutex protects the waiter queue and the num_waiters counter.
16		_PyRawMutex mutex;
17
18		// Linked list of `struct wait_entry` waiters in this bucket.
19		struct llist_node root;
20		size_t num_waiters;
21		} Bucket;
22
23		struct wait_entry {
24		void *park_arg;
25		uintptr_t addr;
26		_PySemaphore sema;
27		struct llist_node node;
28		bool is_unparking;
29		};
30
31		// Prime number to avoid correlations with memory addresses.
32		// We want this to be roughly proportional to the number of CPU cores
33		// to minimize contention on the bucket locks, but not too big to avoid
34		// wasting memory. The exact choice does not matter much.
35	0	#define NUM_BUCKETS 257
36
37		#define BUCKET_INIT(b, i) [i] = { .root = LLIST_INIT(b[i].root) }
38		#define BUCKET_INIT_2(b, i) BUCKET_INIT(b, i), BUCKET_INIT(b, i+1)
39		#define BUCKET_INIT_4(b, i) BUCKET_INIT_2(b, i), BUCKET_INIT_2(b, i+2)
40		#define BUCKET_INIT_8(b, i) BUCKET_INIT_4(b, i), BUCKET_INIT_4(b, i+4)
41		#define BUCKET_INIT_16(b, i) BUCKET_INIT_8(b, i), BUCKET_INIT_8(b, i+8)
42		#define BUCKET_INIT_32(b, i) BUCKET_INIT_16(b, i), BUCKET_INIT_16(b, i+16)
43		#define BUCKET_INIT_64(b, i) BUCKET_INIT_32(b, i), BUCKET_INIT_32(b, i+32)
44		#define BUCKET_INIT_128(b, i) BUCKET_INIT_64(b, i), BUCKET_INIT_64(b, i+64)
45		#define BUCKET_INIT_256(b, i) BUCKET_INIT_128(b, i), BUCKET_INIT_128(b, i+128)
46
47		// Table of waiters (hashed by address)
48		static Bucket buckets[NUM_BUCKETS] = {
49		BUCKET_INIT_256(buckets, 0),
50		BUCKET_INIT(buckets, 256),
51		};
52
53		void
54		_PySemaphore_Init(_PySemaphore *sema)
55	0	{
56		#if defined(MS_WINDOWS)
57		sema->platform_sem = CreateSemaphore(
58		NULL, // attributes
59		0, // initial count
60		10, // maximum count
61		NULL // unnamed
62		);
63		if (!sema->platform_sem) {
64		Py_FatalError("parking_lot: CreateSemaphore failed");
65		}
66		#elif defined(_Py_USE_SEMAPHORES)
67	0	if (sem_init(&sema->platform_sem, /pshared=/0, /value=/0) < 0) {
68	0	Py_FatalError("parking_lot: sem_init failed");
69	0	}
70		#else
71		if (pthread_mutex_init(&sema->mutex, NULL) != 0) {
72		Py_FatalError("parking_lot: pthread_mutex_init failed");
73		}
74		if (pthread_cond_init(&sema->cond, NULL)) {
75		Py_FatalError("parking_lot: pthread_cond_init failed");
76		}
77		sema->counter = 0;
78		#endif
79	0	}
80
81		void
82		_PySemaphore_Destroy(_PySemaphore *sema)
83	0	{
84		#if defined(MS_WINDOWS)
85		CloseHandle(sema->platform_sem);
86		#elif defined(_Py_USE_SEMAPHORES)
87		sem_destroy(&sema->platform_sem);
88		#else
89		pthread_mutex_destroy(&sema->mutex);
90		pthread_cond_destroy(&sema->cond);
91		#endif
92	0	}
93
94		int
95		_PySemaphore_Wait(_PySemaphore *sema, PyTime_t timeout)
96	0	{
97	0	int res;
98		#if defined(MS_WINDOWS)
99		DWORD wait;
100		DWORD millis = 0;
101		if (timeout < 0) {
102		millis = INFINITE;
103		}
104		else {
105		PyTime_t div = _PyTime_AsMilliseconds(timeout, _PyTime_ROUND_TIMEOUT);
106		// Prevent overflow with clamping the result
107		if ((PyTime_t)PY_DWORD_MAX < div) {
108		millis = PY_DWORD_MAX;
109		}
110		else {
111		millis = (DWORD) div;
112		}
113		}
114
115		HANDLE handles[2] = { sema->platform_sem, NULL };
116		HANDLE sigint_event = NULL;
117		DWORD count = 1;
118		if (_Py_IsMainThread()) {
119		// gh-135099: Wait on the SIGINT event only in the main thread. Other
120		// threads would ignore the result anyways, and accessing
121		// `_PyOS_SigintEvent()` from non-main threads may race with
122		// interpreter shutdown, which closes the event handle. Note that
123		// non-main interpreters will ignore the result.
124		sigint_event = _PyOS_SigintEvent();
125		if (sigint_event != NULL) {
126		handles[1] = sigint_event;
127		count = 2;
128		}
129		}
130		wait = WaitForMultipleObjects(count, handles, FALSE, millis);
131		if (wait == WAIT_OBJECT_0) {
132		res = Py_PARK_OK;
133		}
134		else if (wait == WAIT_OBJECT_0 + 1) {
135		assert(sigint_event != NULL);
136		ResetEvent(sigint_event);
137		res = Py_PARK_INTR;
138		}
139		else if (wait == WAIT_TIMEOUT) {
140		res = Py_PARK_TIMEOUT;
141		}
142		else {
143		_Py_FatalErrorFormat(__func__,
144		"unexpected error from semaphore: %u (error: %u)",
145		wait, GetLastError());
146		}
147		#elif defined(_Py_USE_SEMAPHORES)
148		int err;
149	0	if (timeout >= 0) {
150	0	struct timespec ts;
151
152	0	#if defined(CLOCK_MONOTONIC) && defined(HAVE_SEM_CLOCKWAIT) && !defined(_Py_THREAD_SANITIZER)
153	0	PyTime_t now;
154		// silently ignore error: cannot report error to the caller
155	0	(void)PyTime_MonotonicRaw(&now);
156	0	PyTime_t deadline = _PyTime_Add(now, timeout);
157	0	_PyTime_AsTimespec_clamp(deadline, &ts);
158
159	0	err = sem_clockwait(&sema->platform_sem, CLOCK_MONOTONIC, &ts);
160		#else
161		PyTime_t now;
162		// silently ignore error: cannot report error to the caller
163		(void)PyTime_TimeRaw(&now);
164		PyTime_t deadline = _PyTime_Add(now, timeout);
165
166		_PyTime_AsTimespec_clamp(deadline, &ts);
167
168		err = sem_timedwait(&sema->platform_sem, &ts);
169		#endif
170	0	}
171	0	else {
172	0	err = sem_wait(&sema->platform_sem);
173	0	}
174	0	if (err == -1) {
175	0	err = errno;
176	0	if (err == EINTR) {
177	0	res = Py_PARK_INTR;
178	0	}
179	0	else if (err == ETIMEDOUT) {
180	0	res = Py_PARK_TIMEOUT;
181	0	}
182	0	else {
183	0	_Py_FatalErrorFormat(__func__,
184	0	"unexpected error from semaphore: %d",
185	0	err);
186	0	}
187	0	}
188	0	else {
189	0	res = Py_PARK_OK;
190	0	}
191		#else
192		pthread_mutex_lock(&sema->mutex);
193		int err = 0;
194		if (sema->counter == 0) {
195		if (timeout >= 0) {
196		struct timespec ts;
197		#if defined(HAVE_PTHREAD_COND_TIMEDWAIT_RELATIVE_NP)
198		_PyTime_AsTimespec_clamp(timeout, &ts);
199		err = pthread_cond_timedwait_relative_np(&sema->cond, &sema->mutex, &ts);
200		#else
201		PyTime_t now;
202		(void)PyTime_TimeRaw(&now);
203		PyTime_t deadline = _PyTime_Add(now, timeout);
204		_PyTime_AsTimespec_clamp(deadline, &ts);
205
206		err = pthread_cond_timedwait(&sema->cond, &sema->mutex, &ts);
207		#endif // HAVE_PTHREAD_COND_TIMEDWAIT_RELATIVE_NP
208		}
209		else {
210		err = pthread_cond_wait(&sema->cond, &sema->mutex);
211		}
212		}
213		if (sema->counter > 0) {
214		sema->counter--;
215		res = Py_PARK_OK;
216		}
217		else if (err) {
218		res = Py_PARK_TIMEOUT;
219		}
220		else {
221		res = Py_PARK_INTR;
222		}
223		pthread_mutex_unlock(&sema->mutex);
224		#endif
225	0	return res;
226	0	}
227
228		void
229		_PySemaphore_Wakeup(_PySemaphore *sema)
230	0	{
231		#if defined(MS_WINDOWS)
232		if (!ReleaseSemaphore(sema->platform_sem, 1, NULL)) {
233		Py_FatalError("parking_lot: ReleaseSemaphore failed");
234		}
235		#elif defined(_Py_USE_SEMAPHORES)
236		int err = sem_post(&sema->platform_sem);
237	0	if (err != 0) {
238	0	Py_FatalError("parking_lot: sem_post failed");
239	0	}
240		#else
241		pthread_mutex_lock(&sema->mutex);
242		sema->counter++;
243		pthread_cond_signal(&sema->cond);
244		pthread_mutex_unlock(&sema->mutex);
245		#endif
246	0	}
247
248		static void
249		enqueue(Bucket bucket, const void address, struct wait_entry *wait)
250	0	{
251	0	llist_insert_tail(&bucket->root, &wait->node);
252	0	++bucket->num_waiters;
253	0	}
254
255		static struct wait_entry *
256		dequeue(Bucket bucket, const void address)
257	0	{
258		// find the first waiter that is waiting on `address`
259	0	struct llist_node *root = &bucket->root;
260	0	struct llist_node *node;
261	0	llist_for_each(node, root) {
262	0	struct wait_entry *wait = llist_data(node, struct wait_entry, node);
263	0	if (wait->addr == (uintptr_t)address) {
264	0	llist_remove(node);
265	0	--bucket->num_waiters;
266	0	wait->is_unparking = true;
267	0	return wait;
268	0	}
269	0	}
270	0	return NULL;
271	0	}
272
273		static void
274		dequeue_all(Bucket bucket, const void address, struct llist_node *dst)
275	0	{
276		// remove and append all matching waiters to dst
277	0	struct llist_node *root = &bucket->root;
278	0	struct llist_node *node;
279	0	llist_for_each_safe(node, root) {
280	0	struct wait_entry *wait = llist_data(node, struct wait_entry, node);
281	0	if (wait->addr == (uintptr_t)address) {
282	0	llist_remove(node);
283	0	llist_insert_tail(dst, node);
284	0	--bucket->num_waiters;
285	0	wait->is_unparking = true;
286	0	}
287	0	}
288	0	}
289
290		// Checks that `addr == expected` (only works for 1, 2, 4, or 8 bytes)
291		static int
292		atomic_memcmp(const void addr, const void expected, size_t addr_size)
293	0	{
294	0	switch (addr_size) {
295	0	case 1: return _Py_atomic_load_uint8(addr) == (const uint8_t )expected;
296	0	case 2: return _Py_atomic_load_uint16(addr) == (const uint16_t )expected;
297	0	case 4: return _Py_atomic_load_uint32(addr) == (const uint32_t )expected;
298	0	case 8: return _Py_atomic_load_uint64(addr) == (const uint64_t )expected;
299	0	default: Py_UNREACHABLE();
300	0	}
301	0	}
302
303		int
304		_PyParkingLot_Park(const void addr, const void expected, size_t size,
305		PyTime_t timeout_ns, void *park_arg, int detach)
306	0	{
307	0	struct wait_entry wait = {
308	0	.park_arg = park_arg,
309	0	.addr = (uintptr_t)addr,
310	0	.is_unparking = false,
311	0	};
312
313	0	Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];
314
315	0	_PyRawMutex_Lock(&bucket->mutex);
316	0	if (!atomic_memcmp(addr, expected, size)) {
317	0	_PyRawMutex_Unlock(&bucket->mutex);
318	0	return Py_PARK_AGAIN;
319	0	}
320	0	_PySemaphore_Init(&wait.sema);
321	0	enqueue(bucket, addr, &wait);
322	0	_PyRawMutex_Unlock(&bucket->mutex);
323
324	0	PyThreadState *tstate = NULL;
325	0	if (detach) {
326	0	tstate = _PyThreadState_GET();
327	0	if (tstate && _PyThreadState_IsAttached(tstate)) {
328		// Only detach if we are attached
329	0	PyEval_ReleaseThread(tstate);
330	0	}
331	0	else {
332	0	tstate = NULL;
333	0	}
334	0	}
335
336	0	int res = _PySemaphore_Wait(&wait.sema, timeout_ns);
337	0	if (res == Py_PARK_OK) {
338	0	goto done;
339	0	}
340
341		// timeout or interrupt
342	0	_PyRawMutex_Lock(&bucket->mutex);
343	0	if (wait.is_unparking) {
344	0	_PyRawMutex_Unlock(&bucket->mutex);
345		// Another thread has started to unpark us. Wait until we process the
346		// wakeup signal.
347	0	do {
348	0	res = _PySemaphore_Wait(&wait.sema, -1);
349	0	} while (res != Py_PARK_OK);
350	0	goto done;
351	0	}
352	0	else {
353	0	llist_remove(&wait.node);
354	0	--bucket->num_waiters;
355	0	}
356	0	_PyRawMutex_Unlock(&bucket->mutex);
357
358	0	done:
359	0	_PySemaphore_Destroy(&wait.sema);
360	0	if (tstate) {
361	0	PyEval_AcquireThread(tstate);
362	0	}
363	0	return res;
364
365	0	}
366
367		void
368		_PyParkingLot_Unpark(const void addr, _Py_unpark_fn_t fn, void *arg)
369	0	{
370	0	Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];
371
372		// Find the first waiter that is waiting on `addr`
373	0	_PyRawMutex_Lock(&bucket->mutex);
374	0	struct wait_entry *waiter = dequeue(bucket, addr);
375	0	if (waiter) {
376	0	int has_more_waiters = (bucket->num_waiters > 0);
377	0	fn(arg, waiter->park_arg, has_more_waiters);
378	0	}
379	0	else {
380	0	fn(arg, NULL, 0);
381	0	}
382	0	_PyRawMutex_Unlock(&bucket->mutex);
383
384	0	if (waiter) {
385		// Wakeup the waiter outside of the bucket lock
386	0	_PySemaphore_Wakeup(&waiter->sema);
387	0	}
388	0	}
389
390		void
391		_PyParkingLot_UnparkAll(const void *addr)
392	0	{
393	0	struct llist_node head = LLIST_INIT(head);
394	0	Bucket *bucket = &buckets[((uintptr_t)addr) % NUM_BUCKETS];
395
396	0	_PyRawMutex_Lock(&bucket->mutex);
397	0	dequeue_all(bucket, addr, &head);
398	0	_PyRawMutex_Unlock(&bucket->mutex);
399
400	0	struct llist_node *node;
401	0	llist_for_each_safe(node, &head) {
402	0	struct wait_entry *waiter = llist_data(node, struct wait_entry, node);
403	0	llist_remove(node);
404	0	_PySemaphore_Wakeup(&waiter->sema);
405	0	}
406	0	}
407
408		void
409		_PyParkingLot_AfterFork(void)
410	0	{
411		// After a fork only one thread remains. That thread cannot be blocked
412		// so all entries in the parking lot are for dead threads.
413	0	memset(buckets, 0, sizeof(buckets));
414	0	for (Py_ssize_t i = 0; i < NUM_BUCKETS; i++) {
415	0	llist_init(&buckets[i].root);
416	0	}
417	0	}