parking_lot.c

#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>


typedefstruct {
// The mutex protects the waiter queue and the num_waiters counter.
_PyRawMutexmutex;

// Linked list of `struct wait_entry` waiters in this bucket.
structllist_noderoot;
size_tnum_waiters;
} Bucket;

structwait_entry {
void*park_arg;
uintptr_taddr;
_PySemaphoresema;
structllist_nodenode;
boolis_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.
#defineNUM_BUCKETS 257

#defineBUCKET_INIT(b, i) [i] = { .root = LLIST_INIT(b[i].root) }
#defineBUCKET_INIT_2(b, i) BUCKET_INIT(b, i), BUCKET_INIT(b, i+1)
#defineBUCKET_INIT_4(b, i) BUCKET_INIT_2(b, i), BUCKET_INIT_2(b, i+2)
#defineBUCKET_INIT_8(b, i) BUCKET_INIT_4(b, i), BUCKET_INIT_4(b, i+4)
#defineBUCKET_INIT_16(b, i) BUCKET_INIT_8(b, i), BUCKET_INIT_8(b, i+8)
#defineBUCKET_INIT_32(b, i) BUCKET_INIT_16(b, i), BUCKET_INIT_16(b, i+16)
#defineBUCKET_INIT_64(b, i) BUCKET_INIT_32(b, i), BUCKET_INIT_32(b, i+32)
#defineBUCKET_INIT_128(b, i) BUCKET_INIT_64(b, i), BUCKET_INIT_64(b, i+64)
#defineBUCKET_INIT_256(b, i) BUCKET_INIT_128(b, i), BUCKET_INIT_128(b, i+128)

// Table of waiters (hashed by address)
staticBucketbuckets[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
}

staticint
_PySemaphore_PlatformWait(_PySemaphore*sema, PyTime_ttimeout)
{
intres;
#if defined(MS_WINDOWS)
DWORDwait;
DWORDmillis=0;
if (timeout<0) {
millis=INFINITE;
 }
else {
PyTime_tdiv=_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;
 }
 }
wait=WaitForSingleObjectEx(sema->platform_sem, millis, FALSE);
if (wait==WAIT_OBJECT_0) {
res=Py_PARK_OK;
 }
elseif (wait==WAIT_TIMEOUT) {
res=Py_PARK_TIMEOUT;
 }
else {
res=Py_PARK_INTR;
 }
#elif defined(_Py_USE_SEMAPHORES)
interr;
if (timeout >= 0) {
structtimespects;

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

err=sem_clockwait(&sema->platform_sem, CLOCK_MONOTONIC, &ts);
#else
PyTime_tnow;
// silently ignore error: cannot report error to the caller
 (void)PyTime_TimeRaw(&now);
PyTime_tdeadline=_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;
 }
elseif (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);
interr=0;
if (sema->counter==0) {
if (timeout >= 0) {
structtimespects;
#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_tnow;
 (void)PyTime_TimeRaw(&now);
PyTime_tdeadline=_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;
 }
elseif (err) {
res=Py_PARK_TIMEOUT;
 }
else {
res=Py_PARK_INTR;
 }
pthread_mutex_unlock(&sema->mutex);
#endif
returnres;
}

int
_PySemaphore_Wait(_PySemaphore*sema, PyTime_ttimeout, intdetach)
{
PyThreadState*tstate=NULL;
if (detach) {
tstate=_PyThreadState_GET();
if (tstate&&_Py_atomic_load_int_relaxed(&tstate->state) ==
_Py_THREAD_ATTACHED) {
// Only detach if we are attached
PyEval_ReleaseThread(tstate);
 }
else {
tstate=NULL;
 }
 }
intres=_PySemaphore_PlatformWait(sema, timeout);
if (tstate) {
PyEval_AcquireThread(tstate);
 }
returnres;
}

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)
interr=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
}

staticvoid
enqueue(Bucket*bucket, constvoid*address, structwait_entry*wait)
{
llist_insert_tail(&bucket->root, &wait->node);
++bucket->num_waiters;
}

staticstructwait_entry*
dequeue(Bucket*bucket, constvoid*address)
{
// find the first waiter that is waiting on `address`
structllist_node*root=&bucket->root;
structllist_node*node;
llist_for_each(node, root) {
structwait_entry*wait=llist_data(node, structwait_entry, node);
if (wait->addr== (uintptr_t)address) {
llist_remove(node);
--bucket->num_waiters;
wait->is_unparking= true;
returnwait;
 }
 }
returnNULL;
}

staticvoid
dequeue_all(Bucket*bucket, constvoid*address, structllist_node*dst)
{
// remove and append all matching waiters to dst
structllist_node*root=&bucket->root;
structllist_node*node;
llist_for_each_safe(node, root) {
structwait_entry*wait=llist_data(node, structwait_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)
staticint
atomic_memcmp(constvoid*addr, constvoid*expected, size_taddr_size)
{
switch (addr_size) {
case1: return_Py_atomic_load_uint8(addr) ==*(constuint8_t*)expected;
case2: return_Py_atomic_load_uint16(addr) ==*(constuint16_t*)expected;
case4: return_Py_atomic_load_uint32(addr) ==*(constuint32_t*)expected;
case8: return_Py_atomic_load_uint64(addr) ==*(constuint64_t*)expected;
default: Py_UNREACHABLE();
 }
}

int
_PyParkingLot_Park(constvoid*addr, constvoid*expected, size_tsize,
PyTime_ttimeout_ns, void*park_arg, intdetach)
{
structwait_entrywait= {
 .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);
returnPy_PARK_AGAIN;
 }
_PySemaphore_Init(&wait.sema);
enqueue(bucket, addr, &wait);
_PyRawMutex_Unlock(&bucket->mutex);

intres=_PySemaphore_Wait(&wait.sema, timeout_ns, detach);
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, detach);
 } while (res!=Py_PARK_OK);
 goto done;
 }
else {
llist_remove(&wait.node);
--bucket->num_waiters;
 }
_PyRawMutex_Unlock(&bucket->mutex);

done:
_PySemaphore_Destroy(&wait.sema);
returnres;

}

void
_PyParkingLot_Unpark(constvoid*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);
structwait_entry*waiter=dequeue(bucket, addr);
if (waiter) {
inthas_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(constvoid*addr)
{
structllist_nodehead=LLIST_INIT(head);
Bucket*bucket=&buckets[((uintptr_t)addr) % NUM_BUCKETS];

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

structllist_node*node;
llist_for_each_safe(node, &head) {
structwait_entry*waiter=llist_data(node, structwait_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_ti=0; i<NUM_BUCKETS; i++) {
llist_init(&buckets[i].root);
 }
}