/* interpreters module */

/* low-level access to interpreter primitives */

#ifndef Py_BUILD_CORE_BUILTIN

# define Py_BUILD_CORE_MODULE 1

#endif

#include "Python.h"

#include "pycore_crossinterp.h" // _PyXIData_t

#include "pycore_interp.h" // _PyInterpreterState_LookUpID()

#include "pycore_pystate.h" // _PyInterpreterState_GetIDObject()

#ifdef MS_WINDOWS

#ifndef WIN32_LEAN_AND_MEAN

# define WIN32_LEAN_AND_MEAN

#endif

#include <windows.h> // SwitchToThread()

#elif defined(HAVE_SCHED_H)

#include <sched.h> // sched_yield()

#endif

#define REGISTERS_HEAP_TYPES

#define HAS_FALLBACK

#define HAS_UNBOUND_ITEMS

#include "_interpreters_common.h"

#undef HAS_UNBOUND_ITEMS

#undef HAS_FALLBACK

#undef REGISTERS_HEAP_TYPES

/*

This module has the following process-global state:

_globals (static struct globals):

mutex (PyMutex)

module_count (int)

channels (struct _channels):

numopen (int64_t)

next_id; (int64_t)

mutex (PyThread_type_lock)

head (linked list of struct _channelref *):

cid (int64_t)

objcount (Py_ssize_t)

next (struct _channelref *):

...

chan (struct _channel *):

open (int)

mutex (PyThread_type_lock)

closing (struct _channel_closing *):

ref (struct _channelref *):

...

ends (struct _channelends *):

numsendopen (int64_t)

numrecvopen (int64_t)

send (struct _channelend *):

interpid (int64_t)

open (int)

next (struct _channelend *)

recv (struct _channelend *):

...

queue (struct _channelqueue *):

count (int64_t)

first (struct _channelitem *):

next (struct _channelitem *):

...

data (_PyXIData_t *):

data (void *)

obj (PyObject *)

interpid (int64_t)

new_object (xid_newobjfunc)

free (xid_freefunc)

last (struct _channelitem *):

...

The above state includes the following allocations by the module:

* 1 top-level mutex (to protect the rest of the state)

* for each channel:

* 1 struct _channelref

* 1 struct _channel

* 0-1 struct _channel_closing

* 1 struct _channelends

* 2 struct _channelend

* 1 struct _channelqueue

* for each item in each channel:

* 1 struct _channelitem

* 1 _PyXIData_t

The only objects in that global state are the references held by each

channel's queue, which are safely managed via the _PyXIData_*()

API.. The module does not create any objects that are shared globally.

*/

#define MODULE_NAME _interpchannels

#define MODULE_NAME_STR Py_STRINGIFY(MODULE_NAME)

#define MODINIT_FUNC_NAME RESOLVE_MODINIT_FUNC_NAME(MODULE_NAME)

#define GLOBAL_MALLOC(TYPE) \

PyMem_RawMalloc(sizeof(TYPE))

#define GLOBAL_FREE(VAR) \

PyMem_RawFree(VAR)

#define XID_IGNORE_EXC 1

#define XID_FREE 2

static int

_release_xid_data(_PyXIData_t *data, int flags)

{

int ignoreexc = flags & XID_IGNORE_EXC;

PyObject *exc;

if (ignoreexc) {

exc = PyErr_GetRaisedException();

}

int res;

if (flags & XID_FREE) {

res = _PyXIData_ReleaseAndRawFree(data);

}

else {

res = _PyXIData_Release(data);

}

if (res < 0) {

/* The owning interpreter is already destroyed. */

if (ignoreexc) {

// XXX Emit a warning?

PyErr_Clear();

}

}

if (flags & XID_FREE) {

/* Either way, we free the data. */

}

if (ignoreexc) {

PyErr_SetRaisedException(exc);

}

return res;

}

static PyInterpreterState *

_get_current_interp(void)

{

// PyInterpreterState_Get() aborts if lookup fails, so don't need

// to check the result for NULL.

return PyInterpreterState_Get();

}

static PyObject *

_get_current_module(void)

{

PyObject *name = PyUnicode_FromString(MODULE_NAME_STR);

if (name == NULL) {

return NULL;

}

PyObject *mod = PyImport_GetModule(name);

Py_DECREF(name);

if (mod == NULL) {

return NULL;

}

assert(mod != Py_None);

return mod;

}

static PyObject *

get_module_from_owned_type(PyTypeObject *cls)

{

assert(cls != NULL);

return _get_current_module();

// XXX Use the more efficient API now that we use heap types:

//return PyType_GetModule(cls);

}

static struct PyModuleDef moduledef;

static PyObject *

get_module_from_type(PyTypeObject *cls)

{

assert(cls != NULL);

return _get_current_module();

// XXX Use the more efficient API now that we use heap types:

//return PyType_GetModuleByDef(cls, &moduledef);

}

static PyObject *

add_new_exception(PyObject *mod, const char *name, PyObject *base)

{

assert(!PyObject_HasAttrStringWithError(mod, name));

PyObject *exctype = PyErr_NewException(name, base, NULL);

if (exctype == NULL) {

return NULL;

}

int res = PyModule_AddType(mod, (PyTypeObject *)exctype);

if (res < 0) {

Py_DECREF(exctype);

return NULL;

}

return exctype;

}

#define ADD_NEW_EXCEPTION(MOD, NAME, BASE) \

add_new_exception(MOD, MODULE_NAME_STR "." Py_STRINGIFY(NAME), BASE)

static int

wait_for_lock(PyThread_type_lock mutex, PY_TIMEOUT_T timeout)

{

PyLockStatus res = PyThread_acquire_lock_timed_with_retries(mutex, timeout);

if (res == PY_LOCK_INTR) {

/* KeyboardInterrupt, etc. */

assert(PyErr_Occurred());

return -1;

}

else if (res == PY_LOCK_FAILURE) {

assert(!PyErr_Occurred());

assert(timeout > 0);

PyErr_SetString(PyExc_TimeoutError, "timed out");

return -1;

}

assert(res == PY_LOCK_ACQUIRED);

PyThread_release_lock(mutex);

return 0;

}

static int

ensure_highlevel_module_loaded(void)

{

PyObject *highlevel =

PyImport_ImportModule("concurrent.interpreters._channels");

if (highlevel == NULL) {

PyErr_Clear();

highlevel = PyImport_ImportModule("test.support.channels");

if (highlevel == NULL) {

return -1;

}

}

Py_DECREF(highlevel);

return 0;

}

/* module state *************************************************************/

typedef struct {

/* Added at runtime by interpreters module. */

PyTypeObject *send_channel_type;

PyTypeObject *recv_channel_type;

/* heap types */

PyTypeObject *ChannelInfoType;

PyTypeObject *ChannelIDType;

/* exceptions */

PyObject *ChannelError;

PyObject *ChannelNotFoundError;

PyObject *ChannelClosedError;

PyObject *ChannelEmptyError;

PyObject *ChannelNotEmptyError;

} module_state;

static inline module_state *

get_module_state(PyObject *mod)

{

assert(mod != NULL);

module_state *state = PyModule_GetState(mod);

assert(state != NULL);

return state;

}

static module_state *

_get_current_module_state(void)

{

PyObject *mod = _get_current_module();

if (mod == NULL) {

mod = PyImport_ImportModule(MODULE_NAME_STR);

if (mod == NULL) {

return NULL;

}

}

module_state *state = get_module_state(mod);

Py_DECREF(mod);

return state;

}

static int

traverse_module_state(module_state *state, visitproc visit, void *arg)

{

/* external types */

Py_VISIT(state->send_channel_type);

Py_VISIT(state->recv_channel_type);

/* heap types */

Py_VISIT(state->ChannelInfoType);

Py_VISIT(state->ChannelIDType);

/* exceptions */

Py_VISIT(state->ChannelError);

Py_VISIT(state->ChannelNotFoundError);

Py_VISIT(state->ChannelClosedError);

Py_VISIT(state->ChannelEmptyError);

Py_VISIT(state->ChannelNotEmptyError);

return 0;

}

static void

clear_xid_types(module_state *state)

{

/* external types */

if (state->send_channel_type != NULL) {

(void)clear_xid_class(state->send_channel_type);

Py_CLEAR(state->send_channel_type);

}

if (state->recv_channel_type != NULL) {

(void)clear_xid_class(state->recv_channel_type);

Py_CLEAR(state->recv_channel_type);

}

/* heap types */

if (state->ChannelIDType != NULL) {

(void)clear_xid_class(state->ChannelIDType);

Py_CLEAR(state->ChannelIDType);

}

}

static int

clear_module_state(module_state *state)

{

clear_xid_types(state);

/* heap types */

Py_CLEAR(state->ChannelInfoType);

/* exceptions */

Py_CLEAR(state->ChannelError);

Py_CLEAR(state->ChannelNotFoundError);

Py_CLEAR(state->ChannelClosedError);

Py_CLEAR(state->ChannelEmptyError);

Py_CLEAR(state->ChannelNotEmptyError);

return 0;

}

/* channel-specific code ****************************************************/

#define CHANNEL_SEND 1

#define CHANNEL_BOTH 0

#define CHANNEL_RECV -1

/* channel errors */

#define ERR_CHANNEL_NOT_FOUND -2

#define ERR_CHANNEL_CLOSED -3

#define ERR_CHANNEL_INTERP_CLOSED -4

#define ERR_CHANNEL_EMPTY -5

#define ERR_CHANNEL_NOT_EMPTY -6

#define ERR_CHANNEL_MUTEX_INIT -7 // currently unused

#define ERR_CHANNELS_MUTEX_INIT -8

#define ERR_NO_NEXT_CHANNEL_ID -9

#define ERR_CHANNEL_CLOSED_WAITING -10

static int

exceptions_init(PyObject *mod)

{

module_state *state = get_module_state(mod);

if (state == NULL) {

return -1;

}

#define ADD(NAME, BASE) \

do { \

assert(state->NAME == NULL); \

state->NAME = ADD_NEW_EXCEPTION(mod, NAME, BASE); \

if (state->NAME == NULL) { \

return -1; \

} \

} while (0)

// A channel-related operation failed.

ADD(ChannelError, PyExc_RuntimeError);

// An operation tried to use a channel that doesn't exist.

ADD(ChannelNotFoundError, state->ChannelError);

// An operation tried to use a closed channel.

ADD(ChannelClosedError, state->ChannelError);

// An operation tried to pop from an empty channel.

ADD(ChannelEmptyError, state->ChannelError);

// An operation tried to close a non-empty channel.

ADD(ChannelNotEmptyError, state->ChannelError);

#undef ADD

return 0;

}

static int

handle_channel_error(int err, PyObject *mod, int64_t cid)

{

if (err == 0) {

assert(!PyErr_Occurred());

return 0;

}

assert(err < 0);

module_state *state = get_module_state(mod);

assert(state != NULL);

if (err == ERR_CHANNEL_NOT_FOUND) {

PyErr_Format(state->ChannelNotFoundError,

"channel %" PRId64 " not found", cid);

}

else if (err == ERR_CHANNEL_CLOSED) {

PyErr_Format(state->ChannelClosedError,

"channel %" PRId64 " is closed", cid);

}

else if (err == ERR_CHANNEL_CLOSED_WAITING) {

PyErr_Format(state->ChannelClosedError,

"channel %" PRId64 " has closed", cid);

}

else if (err == ERR_CHANNEL_INTERP_CLOSED) {

PyErr_Format(state->ChannelClosedError,

"channel %" PRId64 " is already closed", cid);

}

else if (err == ERR_CHANNEL_EMPTY) {

PyErr_Format(state->ChannelEmptyError,

"channel %" PRId64 " is empty", cid);

}

else if (err == ERR_CHANNEL_NOT_EMPTY) {

PyErr_Format(state->ChannelNotEmptyError,

"channel %" PRId64 " may not be closed "

"if not empty (try force=True)",

cid);

}

else if (err == ERR_CHANNELS_MUTEX_INIT) {

PyErr_SetString(state->ChannelError,

"can't initialize mutex for channel management");

}

else if (err == ERR_NO_NEXT_CHANNEL_ID) {

PyErr_SetString(state->ChannelError,

"failed to get a channel ID");

}

else {

assert(PyErr_Occurred());

}

return 1;

}

/* the channel queue */

typedef uintptr_t _channelitem_id_t;

typedef struct wait_info {

PyThread_type_lock mutex;

enum {

WAITING_NO_STATUS = 0,

WAITING_ACQUIRED = 1,

WAITING_RELEASING = 2,

WAITING_RELEASED = 3,

} status;

int received;

_channelitem_id_t itemid;

} _waiting_t;

static int

_waiting_init(_waiting_t *waiting)

{

PyThread_type_lock mutex = PyThread_allocate_lock();

if (mutex == NULL) {

PyErr_NoMemory();

return -1;

}

*waiting = (_waiting_t){

.mutex = mutex,

.status = WAITING_NO_STATUS,

};

return 0;

}

static void

_waiting_clear(_waiting_t *waiting)

{

assert(waiting->status != WAITING_ACQUIRED

&& waiting->status != WAITING_RELEASING);

if (waiting->mutex != NULL) {

PyThread_free_lock(waiting->mutex);

waiting->mutex = NULL;

}

}

static _channelitem_id_t

_waiting_get_itemid(_waiting_t *waiting)

{

return waiting->itemid;

}

static void

_waiting_acquire(_waiting_t *waiting)

{

assert(waiting->status == WAITING_NO_STATUS);

PyThread_acquire_lock(waiting->mutex, NOWAIT_LOCK);

waiting->status = WAITING_ACQUIRED;

}

static void

_waiting_release(_waiting_t *waiting, int received)

{

assert(waiting->mutex != NULL);

assert(waiting->status == WAITING_ACQUIRED);

assert(!waiting->received);

waiting->status = WAITING_RELEASING;

if (waiting->received != received) {

assert(received == 1);

waiting->received = received;

}

waiting->status = WAITING_RELEASED;

PyThread_release_lock(waiting->mutex);

}

static void

_waiting_finish_releasing(_waiting_t *waiting)

{

while (waiting->status == WAITING_RELEASING) {

#ifdef MS_WINDOWS

SwitchToThread();

#elif defined(HAVE_SCHED_H)

sched_yield();

#endif

}

}

struct _channelitem;

typedef struct _channelitem {

/* The interpreter that added the item to the queue.

The actual bound interpid is found in item->data.

This is necessary because item->data might be NULL,

meaning the interpreter has been destroyed. */

int64_t interpid;

_PyXIData_t *data;

_waiting_t *waiting;

unboundop_t unboundop;

struct _channelitem *next;

} _channelitem;

static inline _channelitem_id_t

_channelitem_ID(_channelitem *item)

{

return (_channelitem_id_t)item;

}

static void

_channelitem_init(_channelitem *item,

int64_t interpid, _PyXIData_t *data,

_waiting_t *waiting, unboundop_t unboundop)

{

if (interpid < 0) {

interpid = _get_interpid(data);

}

else {

assert(data == NULL

|| _PyXIData_INTERPID(data) < 0

|| interpid == _PyXIData_INTERPID(data));

}

*item = (_channelitem){

.interpid = interpid,

.data = data,

.waiting = waiting,

.unboundop = unboundop,

};

if (waiting != NULL) {

waiting->itemid = _channelitem_ID(item);

}

}

static void

_channelitem_clear_data(_channelitem *item, int removed)

{

if (item->data != NULL) {

// It was allocated in channel_send().

(void)_release_xid_data(item->data, XID_IGNORE_EXC | XID_FREE);

item->data = NULL;

}

if (item->waiting != NULL && removed) {

if (item->waiting->status == WAITING_ACQUIRED) {

_waiting_release(item->waiting, 0);

}

item->waiting = NULL;

}

}

static void

_channelitem_clear(_channelitem *item)

{

item->next = NULL;

_channelitem_clear_data(item, 1);

}

static _channelitem *

_channelitem_new(int64_t interpid, _PyXIData_t *data,

_waiting_t *waiting, unboundop_t unboundop)

{

_channelitem *item = GLOBAL_MALLOC(_channelitem);

if (item == NULL) {

PyErr_NoMemory();

return NULL;

}

_channelitem_init(item, interpid, data, waiting, unboundop);

return item;

}

static void

_channelitem_free(_channelitem *item)

{

_channelitem_clear(item);

GLOBAL_FREE(item);

}

static void

_channelitem_free_all(_channelitem *item)

{

while (item != NULL) {

_channelitem *last = item;

item = item->next;

_channelitem_free(last);

}

}

static void

_channelitem_popped(_channelitem *item,

_PyXIData_t **p_data, _waiting_t **p_waiting,

int *p_unboundop)

{

assert(item->waiting == NULL || item->waiting->status == WAITING_ACQUIRED);

*p_data = item->data;

*p_waiting = item->waiting;

*p_unboundop = item->unboundop;

// We clear them here, so they won't be released in _channelitem_clear().

item->data = NULL;

item->waiting = NULL;

_channelitem_free(item);

}

static int

_channelitem_clear_interpreter(_channelitem *item)

{

assert(item->interpid >= 0);

if (item->data == NULL) {

// Its interpreter was already cleared (or it was never bound).

// For UNBOUND_REMOVE it should have been freed at that time.

assert(item->unboundop != UNBOUND_REMOVE);

return 0;

}

assert(_PyXIData_INTERPID(item->data) == item->interpid);

switch (item->unboundop) {

case UNBOUND_REMOVE:

// The caller must free/clear it.

return 1;

case UNBOUND_ERROR:

case UNBOUND_REPLACE:

// We won't need the cross-interpreter data later

// so we completely throw it away.

_channelitem_clear_data(item, 0);

return 0;

default:

Py_FatalError("not reachable");

return -1;

}

}

typedef struct _channelqueue {

int64_t count;

_channelitem *first;

_channelitem *last;

} _channelqueue;

static _channelqueue *

_channelqueue_new(void)

{

_channelqueue *queue = GLOBAL_MALLOC(_channelqueue);

if (queue == NULL) {

PyErr_NoMemory();

return NULL;

}

queue->count = 0;

queue->first = NULL;

queue->last = NULL;

return queue;

}

static void

_channelqueue_clear(_channelqueue *queue)

{

_channelitem_free_all(queue->first);

queue->count = 0;

queue->first = NULL;

queue->last = NULL;

}

static void

_channelqueue_free(_channelqueue *queue)

{

_channelqueue_clear(queue);

GLOBAL_FREE(queue);

}

static int

_channelqueue_put(_channelqueue *queue,

int64_t interpid, _PyXIData_t *data,

_waiting_t *waiting, unboundop_t unboundop)

{

_channelitem *item = _channelitem_new(interpid, data, waiting, unboundop);

if (item == NULL) {

return -1;

}

queue->count += 1;

if (queue->first == NULL) {

queue->first = item;

}

else {

queue->last->next = item;

}

queue->last = item;

if (waiting != NULL) {

_waiting_acquire(waiting);

}

return 0;

}

static int

_channelqueue_get(_channelqueue *queue,

_PyXIData_t **p_data, _waiting_t **p_waiting,

int *p_unboundop)

{

_channelitem *item = queue->first;

if (item == NULL) {

return ERR_CHANNEL_EMPTY;

}

queue->first = item->next;

if (queue->last == item) {

queue->last = NULL;

}

queue->count -= 1;

_channelitem_popped(item, p_data, p_waiting, p_unboundop);

return 0;

}

static int

_channelqueue_find(_channelqueue *queue, _channelitem_id_t itemid,

_channelitem **p_item, _channelitem **p_prev)

{

_channelitem *prev = NULL;

_channelitem *item = NULL;

if (queue->first != NULL) {

if (_channelitem_ID(queue->first) == itemid) {

item = queue->first;

}

else {

prev = queue->first;

while (prev->next != NULL) {

if (_channelitem_ID(prev->next) == itemid) {

item = prev->next;

break;

}

prev = prev->next;

}

if (item == NULL) {

prev = NULL;

}

}

}

if (p_item != NULL) {

*p_item = item;

}

if (p_prev != NULL) {

*p_prev = prev;

}

return (item != NULL);

}

static void

_channelqueue_remove(_channelqueue *queue, _channelitem_id_t itemid,

_PyXIData_t **p_data, _waiting_t **p_waiting)

{

_channelitem *prev = NULL;

_channelitem *item = NULL;

int found = _channelqueue_find(queue, itemid, &item, &prev);

if (!found) {

return;

}

assert(item->waiting != NULL);

assert(!item->waiting->received);

if (prev == NULL) {

assert(queue->first == item);

queue->first = item->next;

}

else {

assert(queue->first != item);

assert(prev->next == item);

prev->next = item->next;

}

item->next = NULL;

if (queue->last == item) {

queue->last = prev;

}

queue->count -= 1;

unboundop_t unboundop;

_channelitem_popped(item, p_data, p_waiting, &unboundop);

}

static void

_channelqueue_clear_interpreter(_channelqueue *queue, int64_t interpid)

{

_channelitem *prev = NULL;

_channelitem *next = queue->first;

while (next != NULL) {

_channelitem *item = next;

next = item->next;

int remove = (item->interpid == interpid)

? _channelitem_clear_interpreter(item)

: 0;

if (remove) {

_channelitem_free(item);

if (prev == NULL) {

queue->first = next;

}

else {

prev->next = next;

}

queue->count -= 1;

}

else {

prev = item;

}

}

}

/* channel-interpreter associations */

struct _channelend;

typedef struct _channelend {

struct _channelend *next;

int64_t interpid;

int open;

} _channelend;

static _channelend *

_channelend_new(int64_t interpid)

{

_channelend *end = GLOBAL_MALLOC(_channelend);

if (end == NULL) {

PyErr_NoMemory();

return NULL;

}

end->next = NULL;

end->interpid = interpid;

end->open = 1;

return end;

}

static void

_channelend_free(_channelend *end)

{

GLOBAL_FREE(end);

}

static void

_channelend_free_all(_channelend *end)

{

while (end != NULL) {

_channelend *last = end;

end = end->next;

_channelend_free(last);

}

}

static _channelend *

_channelend_find(_channelend *first, int64_t interpid, _channelend **pprev)

{

_channelend *prev = NULL;

_channelend *end = first;

while (end != NULL) {

if (end->interpid == interpid) {

break;

}

prev = end;

end = end->next;

}

if (pprev != NULL) {

*pprev = prev;

}

return end;

}

typedef struct _channelassociations {

// Note that the list entries are never removed for interpreter

// for which the channel is closed. This should not be a problem in

// practice. Also, a channel isn't automatically closed when an

// interpreter is destroyed.

int64_t numsendopen;

int64_t numrecvopen;

_channelend *send;

_channelend *recv;

} _channelends;

static _channelends *

_channelends_new(void)

{

_channelends *ends = GLOBAL_MALLOC(_channelends);

if (ends == NULL) {

PyErr_NoMemory();

return NULL;

}

ends->numsendopen = 0;

ends->numrecvopen = 0;

ends->send = NULL;

ends->recv = NULL;

return ends;

}

static void

_channelends_clear(_channelends *ends)

{

_channelend_free_all(ends->send);

ends->send = NULL;

ends->numsendopen = 0;

_channelend_free_all(ends->recv);

ends->recv = NULL;

ends->numrecvopen = 0;

}

static void

_channelends_free(_channelends *ends)

{

_channelends_clear(ends);

GLOBAL_FREE(ends);

}

static _channelend *

_channelends_add(_channelends *ends, _channelend *prev, int64_t interpid,

int send)

{

_channelend *end = _channelend_new(interpid);

if (end == NULL) {

return NULL;

}

if (prev == NULL) {

if (send) {

ends->send = end;

}

else {

ends->recv = end;

}

}

else {

prev->next = end;

}

if (send) {

ends->numsendopen += 1;

}

else {

ends->numrecvopen += 1;

}

return end;

}

static int

_channelends_associate(_channelends *ends, int64_t interpid, int send)

{

_channelend *prev;

_channelend *end = _channelend_find(send ? ends->send : ends->recv,

interpid, &prev);

if (end != NULL) {

if (!end->open) {

return ERR_CHANNEL_CLOSED;

}

// already associated

return 0;

}

if (_channelends_add(ends, prev, interpid, send) == NULL) {

return -1;

}

return 0;

}

static int

_channelends_is_open(_channelends *ends)

{