cpython/Modules/_testinternalcapi/test_lock.c

// C Extension module to test pycore_lock.h API

#include "parts.h"
#include "pycore_lock.h"
#include "pycore_pythread.h"      // PyThread_get_thread_ident_ex()

#include "clinic/test_lock.c.h"

#ifdef MS_WINDOWS
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#else
#include <unistd.h>         // usleep()
#endif

/*[clinic input]
module _testinternalcapi
[clinic start generated code]*/
/*[clinic end generated code: output=da39a3ee5e6b4b0d input=7bb583d8c9eb9a78]*/


static void
pysleep(int ms)
{ … }

static PyObject *
test_lock_basic(PyObject *self, PyObject *obj)
{ … }

struct test_lock2_data { … };

static void
lock_thread(void *arg)
{ … }

static PyObject *
test_lock_two_threads(PyObject *self, PyObject *obj)
{ … }

#define COUNTER_THREADS …
#define COUNTER_ITERS …

struct test_data_counter { … };

struct thread_data_counter { … };

static void
counter_thread(void *arg)
{ … }

static PyObject *
test_lock_counter(PyObject *self, PyObject *obj)
{ … }

#define SLOW_COUNTER_ITERS …

static void
slow_counter_thread(void *arg)
{ … }

static PyObject *
test_lock_counter_slow(PyObject *self, PyObject *obj)
{ … }

struct bench_data_locks { … };

struct bench_thread_data { … };

static void
thread_benchmark_locks(void *arg)
{ … }

/*[clinic input]
_testinternalcapi.benchmark_locks

    num_threads: Py_ssize_t
    use_pymutex: bool = True
    critical_section_length: int = 1
    time_ms: int = 1000
    /

[clinic start generated code]*/

static PyObject *
_testinternalcapi_benchmark_locks_impl(PyObject *module,
                                       Py_ssize_t num_threads,
                                       int use_pymutex,
                                       int critical_section_length,
                                       int time_ms)
/*[clinic end generated code: output=381df8d7e9a74f18 input=f3aeaf688738c121]*/
{ … }

static PyObject *
test_lock_benchmark(PyObject *module, PyObject *obj)
{ … }

static int
init_maybe_fail(void *arg)
{ … }

static PyObject *
test_lock_once(PyObject *self, PyObject *obj)
{ … }

struct test_rwlock_data { … };

static void
rdlock_thread(void *arg)
{ … }
static void
wrlock_thread(void *arg)
{ … }

static void
wait_until(uintptr_t *ptr, uintptr_t value)
{ … }

static PyObject *
test_lock_rwlock(PyObject *self, PyObject *obj)
{ … }

static PyObject *
test_lock_recursive(PyObject *self, PyObject *obj)
{ … }

static PyMethodDef test_methods[] = …;

int
_PyTestInternalCapi_Init_Lock(PyObject *mod)
{ … }