cpython/Modules/_testinternalcapi/test_critical_sections.c

/*
 * C Extension module to test pycore_critical_section.h API.
 */

#include "parts.h"

#include "pycore_critical_section.h"

#ifdef Py_GIL_DISABLED
#define assert_nogil …
#define assert_gil …
#else
#define assert_gil …
#define assert_nogil(x) …
#endif


static PyObject *
test_critical_sections(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

static void
lock_unlock_object(PyObject *obj, int recurse_depth)
{ … }

static void
lock_unlock_two_objects(PyObject *a, PyObject *b, int recurse_depth)
{ … }


// Test that nested critical sections do not deadlock if they attempt to lock
// the same object.
static PyObject *
test_critical_sections_nest(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

// Test that a critical section is suspended by a Py_BEGIN_ALLOW_THREADS and
// resumed by a Py_END_ALLOW_THREADS.
static PyObject *
test_critical_sections_suspend(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

#ifdef Py_CAN_START_THREADS
struct test_data { … };

static void
thread_critical_sections(void *arg)
{ … }

static PyObject *
test_critical_sections_threads(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

static void
pysleep(int ms)
{ … }

struct test_data_gc { … };

static void
thread_gc(void *arg)
{ … }

static PyObject *
test_critical_sections_gc(PyObject *self, PyObject *Py_UNUSED(args))
{ … }

#endif

static PyMethodDef test_methods[] = …;

int
_PyTestInternalCapi_Init_CriticalSection(PyObject *mod)
{ … }