Skip to content

Commit fd61525

Browse files
colesburycolesbury
committed
pythongh-111964: Add _PyRWMutex a "readers-writer" lock
This adds `_PyRWMutex`, a "readers-writer" lock, which wil be used to serialize global stop-the-world pauses with per-interpreter pauses.
1 parent cf6110b commit fd61525

3 files changed

Lines changed: 203 additions & 0 deletions

File tree

‎Include/internal/pycore_lock.h‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,30 @@ _PyOnceFlag_CallOnce(_PyOnceFlag *flag, _Py_once_fn_t *fn, void *arg)
213213
return _PyOnceFlag_CallOnceSlow(flag, fn, arg);
214214
}
215215

216+
// A readers-writer (RW) lock. The lock supports multiple concurrent readers or
217+
// a single writer. The lock is write-preferring: if a writer is waiting, then
218+
// new readers will be blocked. This avoids starvation of writers.
219+
//
220+
// The low two bits store whether the lock is write-locked (_Py_LOCKED) and
221+
// whether there are parked threads (_Py_HAS_PARKED). The remaining bits are
222+
// used to store the number of readers.
223+
//
224+
// The design is optimized for simplicity of the implementation. The lock is
225+
// not fair: if fairness is desired, use an additional PyMutex to serialize
226+
// writers. The lock is also not reentrant.
227+
typedef struct {
228+
uintptr_t bits;
229+
} _PyRWMutex;
230+
231+
// Read lock
232+
PyAPI_FUNC(void) _PyRWMutex_RLock(_PyRWMutex *rwmutex);
233+
PyAPI_FUNC(void) _PyRWMutex_RUnlock(_PyRWMutex *rwmutex);
234+
235+
// Write lock
236+
PyAPI_FUNC(void) _PyRWMutex_Lock(_PyRWMutex *rwmutex);
237+
PyAPI_FUNC(void) _PyRWMutex_Unlock(_PyRWMutex *rwmutex);
238+
239+
216240
#ifdef __cplusplus
217241
}
218242
#endif

‎Modules/_testinternalcapi/test_lock.c‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -372,6 +372,98 @@ test_lock_once(PyObject *self, PyObject *obj)
372372
Py_RETURN_NONE;
373373
}
374374

375+
struct test_rwlock_data {
376+
Py_ssize_t nthreads;
377+
_PyRWMutex rw;
378+
PyEvent step1;
379+
PyEvent step2;
380+
PyEvent step3;
381+
PyEvent done;
382+
};
383+
384+
static void
385+
rdlock_thread(void *arg)
386+
{
387+
struct test_rwlock_data *test_data = arg;
388+
389+
// Acquire the lock in read mode
390+
_PyRWMutex_RLock(&test_data->rw);
391+
PyEvent_Wait(&test_data->step1);
392+
_PyRWMutex_RUnlock(&test_data->rw);
393+
394+
_PyRWMutex_RLock(&test_data->rw);
395+
PyEvent_Wait(&test_data->step3);
396+
_PyRWMutex_RUnlock(&test_data->rw);
397+
398+
if (_Py_atomic_add_ssize(&test_data->nthreads, -1) == 1) {
399+
_PyEvent_Notify(&test_data->done);
400+
}
401+
}
402+
static void
403+
wrlock_thread(void *arg)
404+
{
405+
struct test_rwlock_data *test_data = arg;
406+
407+
// First acquire the lock in write mode
408+
_PyRWMutex_Lock(&test_data->rw);
409+
PyEvent_Wait(&test_data->step2);
410+
_PyRWMutex_Unlock(&test_data->rw);
411+
412+
if (_Py_atomic_add_ssize(&test_data->nthreads, -1) == 1) {
413+
_PyEvent_Notify(&test_data->done);
414+
}
415+
}
416+
417+
static void
418+
wait_until(uintptr_t *ptr, uintptr_t value)
419+
{
420+
// wait up to two seconds for *ptr == value
421+
int iters = 0;
422+
uintptr_t bits;
423+
do {
424+
pysleep(10);
425+
bits = _Py_atomic_load_uintptr(ptr);
426+
iters++;
427+
} while (bits != value && iters < 200);
428+
}
429+
430+
static PyObject *
431+
test_lock_rwlock(PyObject *self, PyObject *obj)
432+
{
433+
struct test_rwlock_data test_data = {.nthreads = 3};
434+
435+
// Start two readers
436+
PyThread_start_new_thread(rdlock_thread, &test_data);
437+
PyThread_start_new_thread(rdlock_thread, &test_data);
438+
439+
// wait up to two seconds for the threads to attempt to read-lock "rw"
440+
wait_until(&test_data.rw.bits, 8);
441+
assert(test_data.rw.bits == 8);
442+
443+
// start writer (while readers hold lock)
444+
PyThread_start_new_thread(wrlock_thread, &test_data);
445+
wait_until(&test_data.rw.bits, 10);
446+
assert(test_data.rw.bits == 10);
447+
448+
// readers release lock, writer should acquire it
449+
_PyEvent_Notify(&test_data.step1);
450+
wait_until(&test_data.rw.bits, 3);
451+
assert(test_data.rw.bits == 3);
452+
453+
// writer releases lock, readers acquire it
454+
_PyEvent_Notify(&test_data.step2);
455+
wait_until(&test_data.rw.bits, 8);
456+
assert(test_data.rw.bits == 8);
457+
458+
// readers release lock again
459+
_PyEvent_Notify(&test_data.step3);
460+
wait_until(&test_data.rw.bits, 0);
461+
assert(test_data.rw.bits == 0);
462+
463+
PyEvent_Wait(&test_data.done);
464+
Py_RETURN_NONE;
465+
}
466+
375467
static PyMethodDef test_methods[] = {
376468
{"test_lock_basic", test_lock_basic, METH_NOARGS},
377469
{"test_lock_two_threads", test_lock_two_threads, METH_NOARGS},
@@ -380,6 +472,7 @@ static PyMethodDef test_methods[] = {
380472
_TESTINTERNALCAPI_BENCHMARK_LOCKS_METHODDEF
381473
{"test_lock_benchmark", test_lock_benchmark, METH_NOARGS},
382474
{"test_lock_once", test_lock_once, METH_NOARGS},
475+
{"test_lock_rwlock", test_lock_rwlock, METH_NOARGS},
383476
{NULL, NULL} /* sentinel */
384477
};
385478

‎Python/lock.c‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -353,3 +353,89 @@ _PyOnceFlag_CallOnceSlow(_PyOnceFlag *flag, _Py_once_fn_t *fn, void *arg)
353353
v = _Py_atomic_load_uint8(&flag->v);
354354
}
355355
}
356+
357+
#define _PyRWMutex_READER_SHIFT 2
358+
359+
void
360+
_PyRWMutex_RLock(_PyRWMutex *rwmutex)
361+
{
362+
uintptr_t bits = _Py_atomic_load_uintptr_relaxed(&rwmutex->bits);
363+
for (;;) {
364+
// If the lock is not write-locked and there is no writer waiting, then
365+
// we can increment the reader count.
366+
if ((bits & (_Py_LOCKED|_Py_HAS_PARKED)) == 0) {
367+
uintptr_t newval = bits + (1 << _PyRWMutex_READER_SHIFT);
368+
if (!_Py_atomic_compare_exchange_uintptr(&rwmutex->bits,
369+
&bits, newval)) {
370+
continue;
371+
}
372+
return;
373+
}
374+
375+
// Set _Py_HAS_PARKED if it's not already set.
376+
if ((bits & _Py_HAS_PARKED) == 0) {
377+
uintptr_t newval = bits | _Py_HAS_PARKED;
378+
if (!_Py_atomic_compare_exchange_uintptr(&rwmutex->bits,
379+
&bits, newval)) {
380+
continue;
381+
}
382+
bits = newval;
383+
}
384+
385+
_PyParkingLot_Park(&rwmutex->bits, &bits, sizeof(bits), -1, NULL, 1);
386+
bits = _Py_atomic_load_uintptr_relaxed(&rwmutex->bits);
387+
}
388+
}
389+
390+
void
391+
_PyRWMutex_RUnlock(_PyRWMutex *rwmutex)
392+
{
393+
uintptr_t bits = _Py_atomic_add_uintptr(&rwmutex->bits, -(1 << _PyRWMutex_READER_SHIFT));
394+
bits -= (1 << _PyRWMutex_READER_SHIFT);
395+
396+
if ((bits >> _PyRWMutex_READER_SHIFT) == 0 && (bits & _Py_HAS_PARKED)) {
397+
_PyParkingLot_UnparkAll(&rwmutex->bits);
398+
return;
399+
}
400+
}
401+
402+
void
403+
_PyRWMutex_Lock(_PyRWMutex *rwmutex)
404+
{
405+
uintptr_t bits = _Py_atomic_load_uintptr_relaxed(&rwmutex->bits);
406+
for (;;) {
407+
// If there are no active readers and it's not already write-locked,
408+
// then we can grab the lock.
409+
if ((bits & ~_Py_HAS_PARKED) == 0) {
410+
if (!_Py_atomic_compare_exchange_uintptr(&rwmutex->bits,
411+
&bits,
412+
bits | _Py_LOCKED)) {
413+
return;
414+
}
415+
continue;
416+
}
417+
418+
if (!(bits & _Py_HAS_PARKED)) {
419+
if (!_Py_atomic_compare_exchange_uintptr(&rwmutex->bits,
420+
&bits,
421+
bits | _Py_HAS_PARKED)) {
422+
continue;
423+
}
424+
bits |= _Py_HAS_PARKED;
425+
}
426+
427+
_PyParkingLot_Park(&rwmutex->bits, &bits, sizeof(bits), -1, NULL, 1);
428+
bits = _Py_atomic_load_uintptr_relaxed(&rwmutex->bits);
429+
}
430+
}
431+
432+
void
433+
_PyRWMutex_Unlock(_PyRWMutex *rwmutex)
434+
{
435+
uintptr_t old_bits = _Py_atomic_exchange_uintptr(&rwmutex->bits, 0);
436+
assert(old_bits >> _PyRWMutex_READER_SHIFT == 0);
437+
438+
if ((old_bits & _Py_HAS_PARKED) != 0) {
439+
_PyParkingLot_UnparkAll(&rwmutex->bits);
440+
}
441+
}

0 commit comments

Comments
 (0)