#ifndef Py_BUILD_CORE_BUILTIN

# define Py_BUILD_CORE_MODULE 1

#include "Python.h"

#include "interpreteridobject.h"

#include "pycore_pybuffer.h" // _PyBuffer_ReleaseInInterpreterAndRawFree()

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

#ifdef MS_WINDOWS

#define WIN32_LEAN_AND_MEAN

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

#elif defined(HAVE_SCHED_H)

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

#define MODULE_NAME "_xxinterpchannels"

#define GLOBAL_MALLOC(TYPE) \

PyMem_RawMalloc(sizeof(TYPE))

#define GLOBAL_FREE(VAR) \

PyMem_RawFree(VAR)

struct xid_class_registry {

};

static int

register_xid_class(PyTypeObject *cls, crossinterpdatafunc shared,

struct xid_class_registry *classes)

{

int res = _PyCrossInterpreterData_RegisterClass(cls, shared);

if (res == 0) {

assert(classes->count < MAX_XID_CLASSES);

// The class has refs elsewhere, so we need to incref here.

classes->added[classes->count].cls = cls;

classes->count += 1;

}

return res;

}

static void

clear_xid_class_registry(struct xid_class_registry *classes)

{

while (classes->count > 0) {

classes->count -= 1;

PyTypeObject *cls = classes->added[classes->count].cls;

_PyCrossInterpreterData_UnregisterClass(cls);

}

}

#define XID_IGNORE_EXC 1

#define XID_FREE 2

static int

_release_xid_data(_PyCrossInterpreterData *data, int flags)

{

int ignoreexc = flags & XID_IGNORE_EXC;

PyObject *exc;

if (ignoreexc) {

exc = PyErr_GetRaisedException();

}

int res;

if (flags & XID_FREE) {

res = _PyCrossInterpreterData_ReleaseAndRawFree(data);

}

else {

res = _PyCrossInterpreterData_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);

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 "." Py_STRINGIFY(NAME), BASE)

static PyTypeObject *

add_new_type(PyObject *mod, PyType_Spec *spec, crossinterpdatafunc shared,

struct xid_class_registry *classes)

{

PyTypeObject *cls = (PyTypeObject *)PyType_FromModuleAndSpec(

mod, spec, NULL);

if (cls == NULL) {

return NULL;

}

if (PyModule_AddType(mod, cls) < 0) {

Py_DECREF(cls);

return NULL;

}

if (shared != NULL) {

if (register_xid_class(cls, shared, classes)) {

Py_DECREF(cls);

return NULL;

}

}

return cls;

}

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;

}

typedef struct {

PyObject_HEAD

Py_buffer *view;

int64_t interpid;

} XIBufferViewObject;

static PyObject *

xibufferview_from_xid(PyTypeObject *cls, _PyCrossInterpreterData *data)

{

assert(data->data != NULL);

assert(data->obj == NULL);

assert(data->interpid >= 0);

XIBufferViewObject *self = PyObject_Malloc(sizeof(XIBufferViewObject));

if (self == NULL) {

return NULL;

}

PyObject_Init((PyObject *)self, cls);

self->view = (Py_buffer *)data->data;

self->interpid = data->interpid;

return (PyObject *)self;

}

static void

xibufferview_dealloc(XIBufferViewObject *self)

{

PyInterpreterState *interp = _PyInterpreterState_LookUpID(self->interpid);

/* If the interpreter is no longer alive then we have problems,

assert(interp != NULL);

if (_PyBuffer_ReleaseInInterpreterAndRawFree(interp, self->view) < 0) {

// XXX Emit a warning?

PyErr_Clear();

}

PyTypeObject *tp = Py_TYPE(self);

tp->tp_free(self);

/* "Instances of heap-allocated types hold a reference to their type."

// XXX Why don't we implement Py_TPFLAGS_HAVE_GC, e.g. Py_tp_traverse,

// like we do for _abc._abc_data?

Py_DECREF(tp);

}

static int

xibufferview_getbuf(XIBufferViewObject *self, Py_buffer *view, int flags)

{

/* Only PyMemoryView_FromObject() should ever call this,

*view = *self->view;

view->obj = (PyObject *)self;

// XXX Should we leave it alone?

view->internal = NULL;

return 0;

}

static PyType_Slot XIBufferViewType_slots[] = {

{Py_tp_dealloc, (destructor)xibufferview_dealloc},

{Py_bf_getbuffer, (getbufferproc)xibufferview_getbuf},

// We don't bother with Py_bf_releasebuffer since we don't need it.

{0, NULL},

};

static PyType_Spec XIBufferViewType_spec = {

.name = MODULE_NAME ".CrossInterpreterBufferView",

.basicsize = sizeof(XIBufferViewObject),

.flags = (Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE |

Py_TPFLAGS_DISALLOW_INSTANTIATION | Py_TPFLAGS_IMMUTABLETYPE),

.slots = XIBufferViewType_slots,

};

static PyTypeObject * _get_current_xibufferview_type(void);

static PyObject *

_memoryview_from_xid(_PyCrossInterpreterData *data)

{

PyTypeObject *cls = _get_current_xibufferview_type();

if (cls == NULL) {

return NULL;

}

PyObject *obj = xibufferview_from_xid(cls, data);

if (obj == NULL) {

return NULL;

}

return PyMemoryView_FromObject(obj);

}

static int

_memoryview_shared(PyThreadState *tstate, PyObject *obj,

_PyCrossInterpreterData *data)

{

Py_buffer *view = PyMem_RawMalloc(sizeof(Py_buffer));

if (view == NULL) {

return -1;

}

if (PyObject_GetBuffer(obj, view, PyBUF_FULL_RO) < 0) {

PyMem_RawFree(view);

return -1;

}

_PyCrossInterpreterData_Init(data, tstate->interp, view, NULL,

_memoryview_from_xid);

return 0;

}

static int

register_builtin_xid_types(struct xid_class_registry *classes)

{

PyTypeObject *cls;

crossinterpdatafunc func;

// builtin memoryview

cls = &PyMemoryView_Type;

func = _memoryview_shared;

if (register_xid_class(cls, func, classes)) {

return -1;

}

return 0;

}

typedef struct {

struct xid_class_registry xid_classes;

/* Added at runtime by interpreters module. */

PyTypeObject *send_channel_type;

PyTypeObject *recv_channel_type;

/* heap types */

PyTypeObject *ChannelInfoType;

PyTypeObject *ChannelIDType;

PyTypeObject *XIBufferViewType;

/* 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) {

// XXX import it?

PyErr_SetString(PyExc_RuntimeError,

MODULE_NAME " module not imported yet");

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);

Py_VISIT(state->XIBufferViewType);

/* exceptions */

Py_VISIT(state->ChannelError);

Py_VISIT(state->ChannelNotFoundError);

Py_VISIT(state->ChannelClosedError);

Py_VISIT(state->ChannelEmptyError);

Py_VISIT(state->ChannelNotEmptyError);

return 0;

}

static int

clear_module_state(module_state *state)

{

/* external types */

Py_CLEAR(state->send_channel_type);

Py_CLEAR(state->recv_channel_type);

/* heap types */

Py_CLEAR(state->ChannelInfoType);

if (state->ChannelIDType != NULL) {

(void)_PyCrossInterpreterData_UnregisterClass(state->ChannelIDType);

}

Py_CLEAR(state->ChannelIDType);

Py_CLEAR(state->XIBufferViewType);

/* exceptions */

Py_CLEAR(state->ChannelError);

Py_CLEAR(state->ChannelNotFoundError);

Py_CLEAR(state->ChannelClosedError);

Py_CLEAR(state->ChannelEmptyError);

Py_CLEAR(state->ChannelNotEmptyError);

return 0;

}

static PyTypeObject *

_get_current_xibufferview_type(void)

{

module_state *state = _get_current_module_state();

if (state == NULL) {

return NULL;

}

return state->XIBufferViewType;

}

#define CHANNEL_SEND 1

#define CHANNEL_BOTH 0

#define CHANNEL_RECV -1

#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

#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_CHANNEL_MUTEX_INIT) {

PyErr_SetString(state->ChannelError,

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

}

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;

}

typedef uintptr_t _channelitem_id_t;

typedef struct wait_info {

} _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;

PyThread_release_lock(waiting->mutex);

if (waiting->received != received) {

assert(received == 1);

waiting->received = received;

}

waiting->status = WAITING_RELEASED;

}

static void

_waiting_finish_releasing(_waiting_t *waiting)

{

while (waiting->status == WAITING_RELEASING) {

#ifdef MS_WINDOWS

SwitchToThread();

#elif defined(HAVE_SCHED_H)

sched_yield();

}

}

struct _channelitem;

typedef struct _channelitem {

} _channelitem;

static inline _channelitem_id_t

_channelitem_ID(_channelitem *item)

{

return (_channelitem_id_t)item;

}

static void

_channelitem_init(_channelitem *item,

_PyCrossInterpreterData *data, _waiting_t *waiting)

{

*item = (_channelitem){

.data = data,

.waiting = waiting,

};

if (waiting != NULL) {

waiting->itemid = _channelitem_ID(item);

}

}

static void

_channelitem_clear(_channelitem *item)

{

item->next = NULL;

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) {

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

_waiting_release(item->waiting, 0);

}

item->waiting = NULL;

}

}

static _channelitem *

_channelitem_new(_PyCrossInterpreterData *data, _waiting_t *waiting)

{

_channelitem *item = GLOBAL_MALLOC(_channelitem);

if (item == NULL) {

PyErr_NoMemory();

return NULL;

}

_channelitem_init(item, data, waiting);

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,

_PyCrossInterpreterData **p_data, _waiting_t **p_waiting)

{

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

*p_data = item->data;

*p_waiting = item->waiting;

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

item->data = NULL;

item->waiting = NULL;

_channelitem_free(item);

}

typedef struct _channelqueue {

} _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,

_PyCrossInterpreterData *data, _waiting_t *waiting)

{

_channelitem *item = _channelitem_new(data, waiting);

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,

_PyCrossInterpreterData **p_data, _waiting_t **p_waiting)

{

_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);

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,

_PyCrossInterpreterData **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;

_channelitem_popped(item, p_data, p_waiting);

}

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;

if (item->data->interpid == interpid) {

if (prev == NULL) {

queue->first = item->next;

}

else {

prev->next = item->next;

}

_channelitem_free(item);

queue->count -= 1;

}

else {

prev = item;

}

}

}

struct _channelend;

typedef struct _channelend {

} _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) {