cpython/Modules/_interpqueuesmodule.c

/* interpreters module */
/* low-level access to interpreter primitives */

#ifndef Py_BUILD_CORE_BUILTIN
#define Py_BUILD_CORE_MODULE …
#endif

#include "Python.h"
#include "pycore_crossinterp.h"   // struct _xid

#define REGISTERS_HEAP_TYPES
#define HAS_UNBOUND_ITEMS
#include "_interpreters_common.h"
#undef HAS_UNBOUND_ITEMS
#undef REGISTERS_HEAP_TYPES


#define MODULE_NAME …
#define MODULE_NAME_STR …
#define MODINIT_FUNC_NAME …


#define GLOBAL_MALLOC(TYPE) …
#define GLOBAL_FREE(VAR) …


#define XID_IGNORE_EXC …
#define XID_FREE …

static int
_release_xid_data(_PyCrossInterpreterData *data, int flags)
{ … }

static PyInterpreterState *
_get_current_interp(void)
{ … }

static PyObject *
_get_current_module(void)
{ … }


struct idarg_int64_converter_data { … };

static int
idarg_int64_converter(PyObject *arg, void *ptr)
{ … }


static int
ensure_highlevel_module_loaded(void)
{ … }


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

module_state;

static inline module_state *
get_module_state(PyObject *mod)
{ … }

static int
traverse_module_state(module_state *state, visitproc visit, void *arg)
{ … }

static int
clear_module_state(module_state *state)
{ … }


/* error codes **************************************************************/

#define ERR_EXCEPTION_RAISED …
// multi-queue errors
#define ERR_QUEUES_ALLOC …
#define ERR_QUEUE_ALLOC …
#define ERR_NO_NEXT_QUEUE_ID …
#define ERR_QUEUE_NOT_FOUND …
// single-queue errors
#define ERR_QUEUE_EMPTY …
#define ERR_QUEUE_FULL …
#define ERR_QUEUE_NEVER_BOUND …

static int ensure_external_exc_types(module_state *);

static int
resolve_module_errcode(module_state *state, int errcode, int64_t qid,
                       PyObject **p_exctype, PyObject **p_msgobj)
{ … }


/* QueueError ***************************************************************/

static int
add_exctype(PyObject *mod, PyObject **p_state_field,
            const char *qualname, const char *doc, PyObject *base)
{ … }

static int
add_QueueError(PyObject *mod)
{ … }

static int
set_external_exc_types(module_state *state,
                       PyObject *emptyerror, PyObject *fullerror)
{ … }

static int
ensure_external_exc_types(module_state *state)
{ … }

static int
handle_queue_error(int err, PyObject *mod, int64_t qid)
{ … }


/* the basic queue **********************************************************/

struct _queueitem;

_queueitem;

static void
_queueitem_init(_queueitem *item,
                int64_t interpid, _PyCrossInterpreterData *data,
                int fmt, int unboundop)
{ … }

static void
_queueitem_clear_data(_queueitem *item)
{ … }

static void
_queueitem_clear(_queueitem *item)
{ … }

static _queueitem *
_queueitem_new(int64_t interpid, _PyCrossInterpreterData *data,
               int fmt, int unboundop)
{ … }

static void
_queueitem_free(_queueitem *item)
{ … }

static void
_queueitem_free_all(_queueitem *item)
{ … }

static void
_queueitem_popped(_queueitem *item,
                  _PyCrossInterpreterData **p_data, int *p_fmt, int *p_unboundop)
{ … }

static int
_queueitem_clear_interpreter(_queueitem *item)
{ … }


/* the queue */

_queue;

static int
_queue_init(_queue *queue, Py_ssize_t maxsize, int fmt, int unboundop)
{ … }

static void
_queue_clear(_queue *queue)
{ … }

static void _queue_free(_queue *);

static void
_queue_kill_and_wait(_queue *queue)
{ … }

static void
_queue_mark_waiter(_queue *queue, PyThread_type_lock parent_mutex)
{ … }

static void
_queue_unmark_waiter(_queue *queue, PyThread_type_lock parent_mutex)
{ … }

static int
_queue_lock(_queue *queue)
{ … }

static void
_queue_unlock(_queue *queue)
{ … }

static int
_queue_add(_queue *queue, int64_t interpid, _PyCrossInterpreterData *data,
           int fmt, int unboundop)
{ … }

static int
_queue_next(_queue *queue,
            _PyCrossInterpreterData **p_data, int *p_fmt, int *p_unboundop)
{ … }

static int
_queue_get_maxsize(_queue *queue, Py_ssize_t *p_maxsize)
{ … }

static int
_queue_is_full(_queue *queue, int *p_is_full)
{ … }

static int
_queue_get_count(_queue *queue, Py_ssize_t *p_count)
{ … }

static void
_queue_clear_interpreter(_queue *queue, int64_t interpid)
{ … }


/* external queue references ************************************************/

struct _queueref;

_queueref;

static _queueref *
_queuerefs_find(_queueref *first, int64_t qid, _queueref **pprev)
{ … }

static void
_queuerefs_clear(_queueref *head)
{ … }


/* a collection of queues ***************************************************/

_queues;

static void
_queues_init(_queues *queues, PyThread_type_lock mutex)
{ … }

static void
_queues_fini(_queues *queues)
{ … }

static int64_t
_queues_next_id(_queues *queues)  // needs lock
{ … }

static int
_queues_lookup(_queues *queues, int64_t qid, _queue **res)
{ … }

static int64_t
_queues_add(_queues *queues, _queue *queue)
{ … }

static void
_queues_remove_ref(_queues *queues, _queueref *ref, _queueref *prev,
                   _queue **p_queue)
{ … }

static int
_queues_remove(_queues *queues, int64_t qid, _queue **p_queue)
{ … }

static int
_queues_incref(_queues *queues, int64_t qid)
{ … }

static int
_queues_decref(_queues *queues, int64_t qid)
{ … }

struct queue_id_and_info { … };

static struct queue_id_and_info *
_queues_list_all(_queues *queues, int64_t *p_count)
{ … }

static void
_queues_clear_interpreter(_queues *queues, int64_t interpid)
{ … }


/* "high"-level queue-related functions *************************************/

static void
_queue_free(_queue *queue)
{ … }

// Create a new queue.
static int64_t
queue_create(_queues *queues, Py_ssize_t maxsize, int fmt, int unboundop)
{ … }

// Completely destroy the queue.
static int
queue_destroy(_queues *queues, int64_t qid)
{ … }

// Push an object onto the queue.
static int
queue_put(_queues *queues, int64_t qid, PyObject *obj, int fmt, int unboundop)
{ … }

// Pop the next object off the queue.  Fail if empty.
// XXX Support a "wait" mutex?
static int
queue_get(_queues *queues, int64_t qid,
          PyObject **res, int *p_fmt, int *p_unboundop)
{ … }

static int
queue_get_maxsize(_queues *queues, int64_t qid, Py_ssize_t *p_maxsize)
{ … }

static int
queue_is_full(_queues *queues, int64_t qid, int *p_is_full)
{ … }

static int
queue_get_count(_queues *queues, int64_t qid, Py_ssize_t *p_count)
{ … }


/* external Queue objects ***************************************************/

static int _queueobj_shared(PyThreadState *,
                            PyObject *, _PyCrossInterpreterData *);

static int
set_external_queue_type(module_state *state, PyTypeObject *queue_type)
{ … }

static PyTypeObject *
get_external_queue_type(PyObject *module)
{ … }


// XXX Use a new __xid__ protocol instead?

struct _queueid_xid { … };

static _queues * _get_global_queues(void);

static void *
_queueid_xid_new(int64_t qid)
{ … }

static void
_queueid_xid_free(void *data)
{ … }

static PyObject *
_queueobj_from_xid(_PyCrossInterpreterData *data)
{ … }

static int
_queueobj_shared(PyThreadState *tstate, PyObject *queueobj,
                 _PyCrossInterpreterData *data)
{ … }


/* module level code ********************************************************/

/* globals is the process-global state for the module.  It holds all
   the data that we need to share between interpreters, so it cannot
   hold PyObject values. */
static struct globals { … } _globals = …;

static int
_globals_init(void)
{ … }

static void
_globals_fini(void)
{ … }

static _queues *
_get_global_queues(void)
{ … }


static void
clear_interpreter(void *data)
{ … }


qidarg_converter_data;

static int
qidarg_converter(PyObject *arg, void *ptr)
{ … }


static PyObject *
queuesmod_create(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_create_doc,
"create(maxsize, fmt, unboundop) -> qid\n\
\n\
Create a new cross-interpreter queue and return its unique generated ID.\n\
It is a new reference as though bind() had been called on the queue.\n\
\n\
The caller is responsible for calling destroy() for the new queue\n\
before the runtime is finalized.");

static PyObject *
queuesmod_destroy(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_destroy_doc,
"destroy(qid)\n\
\n\
Clear and destroy the queue.  Afterward attempts to use the queue\n\
will behave as though it never existed.");

static PyObject *
queuesmod_list_all(PyObject *self, PyObject *Py_UNUSED(ignored))
{ … }

PyDoc_STRVAR(queuesmod_list_all_doc,
"list_all() -> [(qid, fmt)]\n\
\n\
Return the list of IDs for all queues.\n\
Each corresponding default format is also included.");

static PyObject *
queuesmod_put(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_put_doc,
"put(qid, obj, fmt)\n\
\n\
Add the object's data to the queue.");

static PyObject *
queuesmod_get(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_get_doc,
"get(qid) -> (obj, fmt)\n\
\n\
Return a new object from the data at the front of the queue.\n\
The object's format is also returned.\n\
\n\
If there is nothing to receive then raise QueueEmpty.");

static PyObject *
queuesmod_bind(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_bind_doc,
"bind(qid)\n\
\n\
Take a reference to the identified queue.\n\
The queue is not destroyed until there are no references left.");

static PyObject *
queuesmod_release(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_release_doc,
"release(qid)\n\
\n\
Release a reference to the queue.\n\
The queue is destroyed once there are no references left.");

static PyObject *
queuesmod_get_maxsize(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_get_maxsize_doc,
"get_maxsize(qid)\n\
\n\
Return the maximum number of items in the queue.");

static PyObject *
queuesmod_get_queue_defaults(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_get_queue_defaults_doc,
"get_queue_defaults(qid)\n\
\n\
Return the queue's default values, set when it was created.");

static PyObject *
queuesmod_is_full(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_is_full_doc,
"is_full(qid)\n\
\n\
Return true if the queue has a maxsize and has reached it.");

static PyObject *
queuesmod_get_count(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

PyDoc_STRVAR(queuesmod_get_count_doc,
"get_count(qid)\n\
\n\
Return the number of items in the queue.");

static PyObject *
queuesmod__register_heap_types(PyObject *self, PyObject *args, PyObject *kwds)
{ … }

static PyMethodDef module_functions[] = …;


/* initialization function */

PyDoc_STRVAR(module_doc,
"This module provides primitive operations to manage Python interpreters.\n\
The 'interpreters' module provides a more convenient interface.");

static int
module_exec(PyObject *mod)
{ … }

static struct PyModuleDef_Slot module_slots[] = …;

static int
module_traverse(PyObject *mod, visitproc visit, void *arg)
{ … }

static int
module_clear(PyObject *mod)
{ … }

static void
module_free(void *mod)
{ … }

static struct PyModuleDef moduledef = …;

PyMODINIT_FUNC
MODINIT_FUNC_NAME(void)
{ … }