bpo-32604: [_xxsubinterpreters] Add channel_send_wait(). by ericsnowcurrently · Pull Request #19829 · python/cpython
#include "Python.h" #include "pythread.h" #include "frameobject.h" #include "interpreteridobject.h"
/* locks */
typedef struct _lockobj { PyObject_HEAD PyThread_type_lock lock; PyInterpreterState *owner; int done; } _lockobj;
static int _lockobj_init(_lockobj *lock) { lock->lock = PyThread_allocate_lock(); if (lock->lock == NULL) { return -1; } lock->done = 0; lock->owner = _get_current(); return 0; }
static void _lockobj_dealloc(_lockobj *lock) { PyThread_free_lock(lock->lock); PyObject_Del(lock); }
// This is cross-interpreter safe. static int _lockobj_acquire(_lockobj *lock) { // Do not wait. return PyThread_acquire_lock(lock->lock, 0); }
// This is cross-interpreter safe. static void _lockobj_release(_lockobj *lock) { PyThread_release_lock(lock->lock); }
static PyObject * _lockobj_call(PyObject *self, PyObject *args, PyObject *kwargs) { static char *kwlist[] = {"timeout", NULL}; PY_TIMEOUT_T timeout = _PyThread_TIMEOUT_NOT_SET; if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|O&:__call__", kwlist, _PyThread_timeout_arg_converter, &timeout)) { return NULL; } _lockobj *lock = (_lockobj *)self;
if (lock->done) { Py_RETURN_TRUE; }
// Wait for the lock to be released. _PyTime_t end = timeout > 0 ? _PyTime_GetMonotonicClock() + timeout : 0; PyLockStatus r = PY_LOCK_FAILURE; do { _PyTime_t microseconds = _PyTime_AsMicroseconds(timeout, _PyTime_ROUND_CEILING);
/* first a simple non-blocking try without releasing the GIL */ r = PyThread_acquire_lock_timed(lock->lock, 0, 0); if (r == PY_LOCK_FAILURE && microseconds != 0) { Py_BEGIN_ALLOW_THREADS r = PyThread_acquire_lock_timed(lock, microseconds, 0); Py_END_ALLOW_THREADS }
if (r == PY_LOCK_INTR) { /* Run signal handlers if we were interrupted. Propagate * exceptions from signal handlers, such as KeyboardInterrupt, by * passing up PY_LOCK_INTR. */ if (Py_MakePendingCalls() < 0) { return NULL; }
/* If we're using a timeout, recompute the timeout after processing * signals, since those can take time. */ if (timeout > 0) { timeout = end - _PyTime_GetMonotonicClock();
/* Check for negative values, since those mean block forever. */ if (timeout < 0) { r = PY_LOCK_FAILURE; } } } } while (r == PY_LOCK_INTR); /* Retry if we were interrupted. */ if (r == PY_LOCK_FAILURE) { Py_RETURN_FALSE; }
// Success! _lockobj_release(lock); lock->done = 1; Py_RETURN_TRUE; }
static PyTypeObject _lockobjtype = { PyVarObject_HEAD_INIT(&PyType_Type, 0) .tp_name = "_xxsubinterpreters.lock", .tp_doc = PyDoc_STR("a basic waitable wrapper around a mutex"), .tp_flags = Py_TPFLAGS_DEFAULT, .tp_basicsize = sizeof(_lockobj), // functionality .tp_new = NULL, // It cannot be instantiated from Python code. .tp_dealloc = (destructor)_lockobj_dealloc, .tp_call = _lockobj_call, };
static _lockobj * _lockobj_new(void) { _lockobj *lock = PyObject_New(_lockobj, &_lockobjtype); if (lock == NULL) { return NULL; } if (_lockobj_init(lock) != 0) { PyMem_Free(lock); return NULL; } return lock; }
static void _lockobj_free(_lockobj *lock) { _lockobj_release(lock); if (lock->owner == NULL || lock->owner == _get_current()) { Py_DECREF(lock); } else { int res = _Py_DECREF_in_interpreter(lock->owner, (PyObject *)lock); assert(res == 0); } }
/* channel-specific code ****************************************************/
typedef struct _channelitem { _PyCrossInterpreterData *data; _lockobj *recvlock; struct _channelitem *next; } _channelitem;
static int _channelqueue_put(_channelqueue *queue, _PyCrossInterpreterData *data) _channelqueue_put(_channelqueue *queue, _PyCrossInterpreterData *data, _lockobj *recvlock) { _channelitem *item = _channelitem_new(); if (item == NULL) { return -1; } item->data = data; item->recvlock = recvlock;
queue->count += 1; if (queue->first == NULL) {
static int _channel_add(_PyChannelState *chan, int64_t interp, _PyCrossInterpreterData *data) _PyCrossInterpreterData *data, _lockobj *recvlock) { int res = -1; PyThread_acquire_lock(chan->mutex, WAIT_LOCK);
if (_channelqueue_put(chan->queue, data) != 0) { if (_channelqueue_put(chan->queue, data, recvlock) != 0) { goto done; }
static int _channel_send(_channels *channels, int64_t id, PyObject *obj) _channel_send(_channels *channels, int64_t id, PyObject *obj, _lockobj *recvlock) { PyInterpreterState *interp = _get_current(); if (interp == NULL) {
// Add the data to the channel. int res = _channel_add(chan, PyInterpreterState_GetID(interp), data); int res = _channel_add(chan, PyInterpreterState_GetID(interp), data, recvlock); PyThread_release_lock(mutex); if (res != 0) { _PyCrossInterpreterData_Release(data);
if (_channel_send(&_globals.channels, cid, obj) != 0) { if (_channel_send(&_globals.channels, cid, obj, NULL) != 0) { return NULL; } Py_RETURN_NONE;
static PyObject * channel_send_wait(PyObject *self, PyObject *args, PyObject *kwds) { static char *kwlist[] = {"cid", "obj", NULL}; int64_t cid; PyObject *obj; if (!PyArg_ParseTupleAndKeywords(args, kwds, "O&O:channel_send", kwlist, channel_id_converter, &cid, &obj)) { return NULL; }
_lockobj *lock = _lockobj_new(); if (lock == NULL) { return NULL; } if (_lockobj_acquire(lock) != 0) { PyErr_SetString(PyExc_RuntimeError, "could not acquire lock"); _lockobj_dealloc(lock); return NULL; } if (_channel_send(&_globals.channels, cid, obj, lock) != 0) { _lockobj_dealloc(lock); return NULL; } Py_INCREF(lock); return (PyObject*)lock; }
PyDoc_STRVAR(channel_send_wait_doc, "channel_send_wait(cid, obj)\n\ \n\ Add the object's data to the channel's queue.\n\ \n\ The returned callable will block until the object is received.\n\ Note that it takes an optional 'timeout' arg like\n\ threading.Lock.acquire() does.");
static PyObject * channel_recv(PyObject *self, PyObject *args, PyObject *kwds) {
/* Create the module */ PyObject *module = PyModule_Create(&interpretersmodule);