/* interpreters module */

/* low-level access to interpreter primitives */

#include "Python.h"

#include "frameobject.h"

#include "internal/pystate.h"

static char *

_copy_raw_string(PyObject *strobj)

{

const char *str = PyUnicode_AsUTF8(strobj);

if (str == NULL) {

return NULL;

}

char *copied = PyMem_Malloc(strlen(str)+1);

if (str == NULL) {

PyErr_NoMemory();

return NULL;

}

strcpy(copied, str);

return copied;

}

static PyInterpreterState *

_get_current(void)

{

PyThreadState *tstate = PyThreadState_Get();

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

// to check the result for NULL.

return tstate->interp;

}

static int64_t

_coerce_id(PyObject *id)

{

id = PyNumber_Long(id);

if (id == NULL) {

if (PyErr_ExceptionMatches(PyExc_TypeError)) {

PyErr_SetString(PyExc_TypeError,

"'id' must be a non-negative int");

}

else {

PyErr_SetString(PyExc_ValueError,

"'id' must be a non-negative int");

}

return -1;

}

int64_t cid = PyLong_AsLongLong(id);

Py_DECREF(id);

if (cid == -1 && PyErr_Occurred() != NULL) {

if (!PyErr_ExceptionMatches(PyExc_OverflowError)) {

PyErr_SetString(PyExc_ValueError,

"'id' must be a non-negative int");

}

return -1;

}

if (cid < 0) {

PyErr_SetString(PyExc_ValueError,

"'id' must be a non-negative int");

return -1;

}

return cid;

}

/* data-sharing-specific code ***********************************************/

struct _sharednsitem {

char *name;

_PyCrossInterpreterData data;

};

static int

_sharednsitem_init(struct _sharednsitem *item, PyObject *key, PyObject *value)

{

item->name = _copy_raw_string(key);

if (item->name == NULL) {

return -1;

}

if (_PyObject_GetCrossInterpreterData(value, &item->data) != 0) {

return -1;

}

return 0;

}

static void

_sharednsitem_clear(struct _sharednsitem *item)

{

if (item->name != NULL) {

PyMem_Free(item->name);

}

_PyCrossInterpreterData_Release(&item->data);

}

static int

_sharednsitem_apply(struct _sharednsitem *item, PyObject *ns)

{

PyObject *name = PyUnicode_FromString(item->name);

if (name == NULL) {

return -1;

}

PyObject *value = _PyCrossInterpreterData_NewObject(&item->data);

if (value == NULL) {

Py_DECREF(name);

return -1;

}

int res = PyDict_SetItem(ns, name, value);

Py_DECREF(name);

Py_DECREF(value);

return res;

}

typedef struct _sharedns {

Py_ssize_t len;

struct _sharednsitem* items;

} _sharedns;

static _sharedns *

_sharedns_new(Py_ssize_t len)

{

_sharedns *shared = PyMem_NEW(_sharedns, 1);

if (shared == NULL) {

PyErr_NoMemory();

return NULL;

}

shared->len = len;

shared->items = PyMem_NEW(struct _sharednsitem, len);

if (shared->items == NULL) {

PyErr_NoMemory();

PyMem_Free(shared);

return NULL;

}

return shared;

}

static void

_sharedns_free(_sharedns *shared)

{

for (Py_ssize_t i=0; i < shared->len; i++) {

_sharednsitem_clear(&shared->items[i]);

}

PyMem_Free(shared->items);

PyMem_Free(shared);

}

static _sharedns *

_get_shared_ns(PyObject *shareable)

{

if (shareable == NULL || shareable == Py_None) {

return NULL;

}

Py_ssize_t len = PyDict_Size(shareable);

if (len == 0) {

return NULL;

}

_sharedns *shared = _sharedns_new(len);

if (shared == NULL) {

return NULL;

}

Py_ssize_t pos = 0;

for (Py_ssize_t i=0; i < len; i++) {

PyObject *key, *value;

if (PyDict_Next(shareable, &pos, &key, &value) == 0) {

break;

}

if (_sharednsitem_init(&shared->items[i], key, value) != 0) {

break;

}

}

if (PyErr_Occurred()) {

_sharedns_free(shared);

return NULL;

}

return shared;

}

static int

_sharedns_apply(_sharedns *shared, PyObject *ns)

{

for (Py_ssize_t i=0; i < shared->len; i++) {

if (_sharednsitem_apply(&shared->items[i], ns) != 0) {

return -1;

}

}

return 0;

}

// Ultimately we'd like to preserve enough information about the

// exception and traceback that we could re-constitute (or at least

// simulate, a la traceback.TracebackException), and even chain, a copy

// of the exception in the calling interpreter.

typedef struct _sharedexception {

char *name;

char *msg;

} _sharedexception;

static _sharedexception *

_sharedexception_new(void)

{

_sharedexception *err = PyMem_NEW(_sharedexception, 1);

if (err == NULL) {

PyErr_NoMemory();

return NULL;

}

err->name = NULL;

err->msg = NULL;

return err;

}

static void

_sharedexception_clear(_sharedexception *exc)

{

if (exc->name != NULL) {

PyMem_Free(exc->name);

}

if (exc->msg != NULL) {

PyMem_Free(exc->msg);

}

}

static void

_sharedexception_free(_sharedexception *exc)

{

_sharedexception_clear(exc);

PyMem_Free(exc);

}

static _sharedexception *

_sharedexception_bind(PyObject *exctype, PyObject *exc, PyObject *tb)

{

assert(exctype != NULL);

char *failure = NULL;

_sharedexception *err = _sharedexception_new();

if (err == NULL) {

goto finally;

}

PyObject *name = PyUnicode_FromFormat("%S", exctype);

if (name == NULL) {

failure = "unable to format exception type name";

goto finally;

}

err->name = _copy_raw_string(name);

Py_DECREF(name);

if (err->name == NULL) {

if (PyErr_ExceptionMatches(PyExc_MemoryError)) {

failure = "out of memory copying exception type name";

}

failure = "unable to encode and copy exception type name";

goto finally;

}

if (exc != NULL) {

PyObject *msg = PyUnicode_FromFormat("%S", exc);

if (msg == NULL) {

failure = "unable to format exception message";

goto finally;

}

err->msg = _copy_raw_string(msg);

Py_DECREF(msg);

if (err->msg == NULL) {

if (PyErr_ExceptionMatches(PyExc_MemoryError)) {

failure = "out of memory copying exception message";

}

failure = "unable to encode and copy exception message";

goto finally;

}

}

finally:

if (failure != NULL) {

PyErr_Clear();

if (err->name != NULL) {

PyMem_Free(err->name);

err->name = NULL;

}

err->msg = failure;

}

return err;

}

static void

_sharedexception_apply(_sharedexception *exc, PyObject *wrapperclass)

{

if (exc->name != NULL) {

if (exc->msg != NULL) {

PyErr_Format(wrapperclass, "%s: %s", exc->name, exc->msg);

}

else {

PyErr_SetString(wrapperclass, exc->name);

}

}

else if (exc->msg != NULL) {

PyErr_SetString(wrapperclass, exc->msg);

}

else {

PyErr_SetNone(wrapperclass);

}

}

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

static PyObject *ChannelError;

static PyObject *ChannelNotFoundError;

static PyObject *ChannelClosedError;

static PyObject *ChannelEmptyError;

static int

channel_exceptions_init(PyObject *ns)

{

// XXX Move the exceptions into per-module memory?

// A channel-related operation failed.

ChannelError = PyErr_NewException("_xxsubinterpreters.ChannelError",

PyExc_RuntimeError, NULL);

if (ChannelError == NULL) {

return -1;

}

if (PyDict_SetItemString(ns, "ChannelError", ChannelError) != 0) {

return -1;

}

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

ChannelNotFoundError = PyErr_NewException(

"_xxsubinterpreters.ChannelNotFoundError", ChannelError, NULL);

if (ChannelNotFoundError == NULL) {

return -1;

}

if (PyDict_SetItemString(ns, "ChannelNotFoundError", ChannelNotFoundError) != 0) {

return -1;

}

// An operation tried to use a closed channel.

ChannelClosedError = PyErr_NewException(

"_xxsubinterpreters.ChannelClosedError", ChannelError, NULL);

if (ChannelClosedError == NULL) {

return -1;

}

if (PyDict_SetItemString(ns, "ChannelClosedError", ChannelClosedError) != 0) {

return -1;

}

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

ChannelEmptyError = PyErr_NewException(

"_xxsubinterpreters.ChannelEmptyError", ChannelError, NULL);

if (ChannelEmptyError == NULL) {

return -1;

}

if (PyDict_SetItemString(ns, "ChannelEmptyError", ChannelEmptyError) != 0) {

return -1;

}

return 0;

}

/* the channel queue */

struct _channelitem;

typedef struct _channelitem {

_PyCrossInterpreterData *data;

struct _channelitem *next;

} _channelitem;

static _channelitem *

_channelitem_new(void)

{

_channelitem *item = PyMem_NEW(_channelitem, 1);

if (item == NULL) {

PyErr_NoMemory();

return NULL;

}

item->data = NULL;

item->next = NULL;

return item;

}

static void

_channelitem_clear(_channelitem *item)

{

if (item->data != NULL) {

_PyCrossInterpreterData_Release(item->data);

PyMem_Free(item->data);

item->data = NULL;

}

item->next = NULL;

}

static void

_channelitem_free(_channelitem *item)

{

_channelitem_clear(item);

PyMem_Free(item);

}

static void

_channelitem_free_all(_channelitem *item)

{

while (item != NULL) {

_channelitem *last = item;

item = item->next;

_channelitem_free(last);

}

}

static _PyCrossInterpreterData *

_channelitem_popped(_channelitem *item)

{

_PyCrossInterpreterData *data = item->data;

item->data = NULL;

_channelitem_free(item);

return data;

}

typedef struct _channelqueue {

int64_t count;

_channelitem *first;

_channelitem *last;

} _channelqueue;

static _channelqueue *

_channelqueue_new(void)

{

_channelqueue *queue = PyMem_NEW(_channelqueue, 1);

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

PyMem_Free(queue);

}

static int

_channelqueue_put(_channelqueue *queue, _PyCrossInterpreterData *data)

{

_channelitem *item = _channelitem_new();

if (item == NULL) {

return -1;

}

item->data = data;

queue->count += 1;

if (queue->first == NULL) {

queue->first = item;

}

else {

queue->last->next = item;

}

queue->last = item;

return 0;

}

static _PyCrossInterpreterData *

_channelqueue_get(_channelqueue *queue)

{

_channelitem *item = queue->first;

if (item == NULL) {

return NULL;

}

queue->first = item->next;

if (queue->last == item) {

queue->last = NULL;

}

queue->count -= 1;

return _channelitem_popped(item);

}

/* channel-interpreter associations */

struct _channelend;

typedef struct _channelend {

struct _channelend *next;

int64_t interp;

int open;

} _channelend;

static _channelend *

_channelend_new(int64_t interp)

{

_channelend *end = PyMem_NEW(_channelend, 1);

if (end == NULL) {

PyErr_NoMemory();

return NULL;

}

end->next = NULL;

end->interp = interp;

end->open = 1;

return end;

}

static void

_channelend_free(_channelend *end)

{

PyMem_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 interp, _channelend **pprev)

{

_channelend *prev = NULL;

_channelend *end = first;

while (end != NULL) {

if (end->interp == interp) {

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 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 = PyMem_NEW(_channelends, 1);

if (ends== NULL) {

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

PyMem_Free(ends);

}

static _channelend *

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

int send)

{

_channelend *end = _channelend_new(interp);

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 interp, int send)

{

_channelend *prev;

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

interp, &prev);

if (end != NULL) {

if (!end->open) {

PyErr_SetString(ChannelClosedError, "channel already closed");

return -1;

}

// already associated

return 0;

}

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

return -1;

}

return 0;

}

static int

_channelends_is_open(_channelends *ends)

{

if (ends->numsendopen != 0 || ends->numrecvopen != 0) {

return 1;

}

if (ends->send == NULL && ends->recv == NULL) {

return 1;

}

return 0;

}

static void

_channelends_close_end(_channelends *ends, _channelend *end, int send)

{

end->open = 0;

if (send) {

ends->numsendopen -= 1;

}

else {

ends->numrecvopen -= 1;

}

}

static int

_channelends_close_interpreter(_channelends *ends, int64_t interp, int which)

{

_channelend *prev;

_channelend *end;

if (which >= 0) { // send/both

end = _channelend_find(ends->send, interp, &prev);

if (end == NULL) {

// never associated so add it

end = _channelends_add(ends, prev, interp, 1);

if (end == NULL) {

return -1;

}

}

_channelends_close_end(ends, end, 1);

}

if (which <= 0) { // recv/both

end = _channelend_find(ends->recv, interp, &prev);

if (end == NULL) {

// never associated so add it

end = _channelends_add(ends, prev, interp, 0);

if (end == NULL) {

return -1;

}

}

_channelends_close_end(ends, end, 0);

}

return 0;

}

static void

_channelends_close_all(_channelends *ends)

{

// Ensure all the "send"-associated interpreters are closed.

_channelend *end;

for (end = ends->send; end != NULL; end = end->next) {

_channelends_close_end(ends, end, 1);

}

// Ensure all the "recv"-associated interpreters are closed.

for (end = ends->recv; end != NULL; end = end->next) {

_channelends_close_end(ends, end, 0);

}

}

/* channels */

struct _channel;

typedef struct _channel {

PyThread_type_lock mutex;

_channelqueue *queue;

_channelends *ends;

int open;

} _PyChannelState;

static _PyChannelState *

_channel_new(void)

{

_PyChannelState *chan = PyMem_NEW(_PyChannelState, 1);

if (chan == NULL) {

return NULL;

}

chan->mutex = PyThread_allocate_lock();

if (chan->mutex == NULL) {

PyMem_Free(chan);

PyErr_SetString(ChannelError,

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

return NULL;

}

chan->queue = _channelqueue_new();

if (chan->queue == NULL) {

PyMem_Free(chan);

return NULL;

}

chan->ends = _channelends_new();

if (chan->ends == NULL) {

_channelqueue_free(chan->queue);

PyMem_Free(chan);

return NULL;

}

chan->open = 1;

return chan;

}

static void

_channel_free(_PyChannelState *chan)

{

PyThread_acquire_lock(chan->mutex, WAIT_LOCK);

_channelqueue_free(chan->queue);

_channelends_free(chan->ends);

PyThread_release_lock(chan->mutex);

PyThread_free_lock(chan->mutex);

PyMem_Free(chan);

}

static int

_channel_add(_PyChannelState *chan, int64_t interp,

_PyCrossInterpreterData *data)

{

int res = -1;

PyThread_acquire_lock(chan->mutex, WAIT_LOCK);

if (!chan->open) {

PyErr_SetString(ChannelClosedError, "channel closed");

goto done;

}

if (_channelends_associate(chan->ends, interp, 1) != 0) {

goto done;

}

if (_channelqueue_put(chan->queue, data) != 0) {

goto done;

}

res = 0;

done:

PyThread_release_lock(chan->mutex);

return res;

}

static _PyCrossInterpreterData *

_channel_next(_PyChannelState *chan, int64_t interp)

{

_PyCrossInterpreterData *data = NULL;

PyThread_acquire_lock(chan->mutex, WAIT_LOCK);

if (!chan->open) {

PyErr_SetString(ChannelClosedError, "channel closed");

goto done;

}

if (_channelends_associate(chan->ends, interp, 0) != 0) {

goto done;

}

data = _channelqueue_get(chan->queue);

done:

PyThread_release_lock(chan->mutex);

return data;

}

static int

_channel_close_interpreter(_PyChannelState *chan, int64_t interp, int which)

{

PyThread_acquire_lock(chan->mutex, WAIT_LOCK);

int res = -1;

if (!chan->open) {

PyErr_SetString(ChannelClosedError, "channel already closed");

goto done;

}

if (_channelends_close_interpreter(chan->ends, interp, which) != 0) {

goto done;

}

chan->open = _channelends_is_open(chan->ends);

res = 0;

done:

PyThread_release_lock(chan->mutex);

return res;

}

static int

_channel_close_all(_PyChannelState *chan)

{

int res = -1;

PyThread_acquire_lock(chan->mutex, WAIT_LOCK);

if (!chan->open) {

PyErr_SetString(ChannelClosedError, "channel already closed");

goto done;

}

chan->open = 0;

// We *could* also just leave these in place, since we've marked

// the channel as closed already.

_channelends_close_all(chan->ends);

res = 0;

done:

PyThread_release_lock(chan->mutex);

return res;

}

/* the set of channels */

struct _channelref;

typedef struct _channelref {

int64_t id;

_PyChannelState *chan;

struct _channelref *next;

Py_ssize_t objcount;

} _channelref;

static _channelref *

_channelref_new(int64_t id, _PyChannelState *chan)

{

_channelref *ref = PyMem_NEW(_channelref, 1);

if (ref == NULL) {

return NULL;

}

ref->id = id;

ref->chan = chan;

ref->next = NULL;

ref->objcount = 0;

return ref;

}

//static void

//_channelref_clear(_channelref *ref)

//{

// ref->id = -1;

// ref->chan = NULL;

// ref->next = NULL;

// ref->objcount = 0;

//}

static void

_channelref_free(_channelref *ref)

{

//_channelref_clear(ref);

PyMem_Free(ref);

}

static _channelref *

_channelref_find(_channelref *first, int64_t id, _channelref **pprev)

{

_channelref *prev = NULL;

_channelref *ref = first;

while (ref != NULL) {

if (ref->id == id) {

break;

}

prev = ref;

ref = ref->next;

}

if (pprev != NULL) {

*pprev = prev;

}

return ref;

}

typedef struct _channels {

PyThread_type_lock mutex;

_channelref *head;

int64_t numopen;

int64_t next_id;

} _channels;

static int

_channels_init(_channels *channels)

{

if (channels->mutex == NULL) {

channels->mutex = PyThread_allocate_lock();

if (channels->mutex == NULL) {

PyErr_SetString(ChannelError,

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

return -1;

}

}

channels->head = NULL;

channels->numopen = 0;

channels->next_id = 0;

return 0;

}

static int64_t

_channels_next_id(_channels *channels) // needs lock

{

int64_t id = channels->next_id;

if (id < 0) {

/* overflow */

PyErr_SetString(ChannelError,

"failed to get a channel ID");

return -1;

}

channels->next_id += 1;

return id;

}

static _PyChannelState *

_channels_lookup(_channels *channels, int64_t id, PyThread_type_lock *pmutex)

{

_PyChannelState *chan = NULL;

PyThread_acquire_lock(channels->mutex, WAIT_LOCK);

if (pmutex != NULL) {

*pmutex = NULL;

}

_channelref *ref = _channelref_find(channels->head, id, NULL);

if (ref == NULL) {

PyErr_Format(ChannelNotFoundError, "channel %d not found", id);

goto done;

}

if (ref->chan == NULL || !ref->chan->open) {

PyErr_Format(ChannelClosedError, "channel %d closed", id);

goto done;

}

if (pmutex != NULL) {

// The mutex will be closed by the caller.

*pmutex = channels->mutex;

}

chan = ref->chan;

done:

if (pmutex == NULL || *pmutex == NULL) {

PyThread_release_lock(channels->mutex);

}

return chan;

}

static int64_t

_channels_add(_channels *channels, _PyChannelState *chan)

{

int64_t cid = -1;

PyThread_acquire_lock(channels->mutex, WAIT_LOCK);

// Create a new ref.

int64_t id = _channels_next_id(channels);

if (id < 0) {

goto done;

}

_channelref *ref = _channelref_new(id, chan);

if (ref == NULL) {

goto done;

}

// Add it to the list.