From 4b3a595bca32b954c2c89320f8bf362985c4017b Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Fri, 12 Jul 2024 12:29:39 -0600 Subject: [PATCH 1/7] Factor out interpreters._crossinterp. --- Lib/test/support/interpreters/_crossinterp.py | 102 ++++++++++++++++++ Lib/test/support/interpreters/queues.py | 60 +++-------- Lib/test/test_interpreters/test_queues.py | 4 +- Modules/_interpqueuesmodule.c | 28 +---- Modules/_interpreters_common.h | 30 ++++++ 5 files changed, 150 insertions(+), 74 deletions(-) create mode 100644 Lib/test/support/interpreters/_crossinterp.py diff --git a/Lib/test/support/interpreters/_crossinterp.py b/Lib/test/support/interpreters/_crossinterp.py new file mode 100644 index 00000000000000..544e197ba4c028 --- /dev/null +++ b/Lib/test/support/interpreters/_crossinterp.py @@ -0,0 +1,102 @@ +"""Common code between queues and channels.""" + + +class ItemInterpreterDestroyed(Exception): + """Raised when trying to get an item whose interpreter was destroyed.""" + + +class classonly: + """A non-data descriptor that makes a value only visible on the class. + + This is like the "classmethod" builtin, but does not show up on + instances of the class. It may be used as a decorator. + """ + + def __init__(self, value): + self.value = value + self.getter = classmethod(value).__get__ + self.name = None + + def __set_name__(self, cls, name): + if self.name is not None: + raise TypeError('already used') + self.name = name + + def __get__(self, obj, cls): + if obj is not None: + raise AttributeError(self.name) + # called on the class + return self.getter(None, cls) + + +class UnboundItem: + """Represents a cross-interpreter item no longer bound to an interpreter. + + An item is unbound when the interpreter that added it to the + cross-interpreter container is destroyed. + """ + + __slots__ = () + + @classonly + def singleton(cls, kind, module, name='UNBOUND'): + doc = cls.__doc__.replace('cross-interpreter container', kind) + doc = doc.replace('cross-interpreter', kind) + subclass = type( + f'Unbound{kind.capitalize()}Item', + (cls,), + dict( + _MODULE=module, + _NAME=name, + __doc__=doc, + ), + ) + return object.__new__(subclass) + + _MODULE = __name__ + _NAME = 'UNBOUND' + + def __new__(cls): + raise Exception(f'use {cls._MODULE}.{cls._NAME}') + + def __repr__(self): + return f'{self._MODULE}.{self._NAME}' +# return f'interpreters.queues.UNBOUND' + + +UNBOUND = object.__new__(UnboundItem) +UNBOUND_ERROR = object() +UNBOUND_REMOVE = object() + +_UNBOUND_CONSTANT_TO_FLAG = { + UNBOUND_REMOVE: 1, + UNBOUND_ERROR: 2, + UNBOUND: 3, +} +_UNBOUND_FLAG_TO_CONSTANT = {v: k + for k, v in _UNBOUND_CONSTANT_TO_FLAG.items()} + + +def serialize_unbound(unbound): + op = unbound + try: + flag = _UNBOUND_CONSTANT_TO_FLAG[op] + except KeyError: + raise NotImplementedError(f'unsupported unbound replacement op {op!r}') + return flag, + + +def resolve_unbound(flag, exctype_destroyed): + try: + op = _UNBOUND_FLAG_TO_CONSTANT[flag] + except KeyError: + raise NotImplementedError(f'unsupported unbound replacement op {flag!r}') + if op is UNBOUND_REMOVE: + # "remove" not possible here + raise NotImplementedError + elif op is UNBOUND_ERROR: + raise exctype_destroyed("item's original interpreter destroyed") + elif op is UNBOUND: + return UNBOUND + else: + raise NotImplementedError(repr(op)) diff --git a/Lib/test/support/interpreters/queues.py b/Lib/test/support/interpreters/queues.py index 402ceffd1bb21c..deb8e8613af731 100644 --- a/Lib/test/support/interpreters/queues.py +++ b/Lib/test/support/interpreters/queues.py @@ -5,11 +5,15 @@ import time import weakref import _interpqueues as _queues +from . import _crossinterp # aliases: from _interpqueues import ( QueueError, QueueNotFoundError, ) +from ._crossinterp import ( + UNBOUND_ERROR, UNBOUND_REMOVE, +) __all__ = [ 'UNBOUND', 'UNBOUND_ERROR', 'UNBOUND_REMOVE', @@ -34,7 +38,8 @@ class QueueFull(QueueError, queue.Full): """ -class ItemInterpreterDestroyed(QueueError): +class ItemInterpreterDestroyed(QueueError, + _crossinterp.ItemInterpreterDestroyed): """Raised from get() and get_nowait().""" @@ -42,57 +47,20 @@ class ItemInterpreterDestroyed(QueueError): _PICKLED = 1 -class UnboundItem: - """Represents a Queue item no longer bound to an interpreter. - - An item is unbound when the interpreter that added it to the queue - is destroyed. - """ - - __slots__ = () - - def __new__(cls): - return UNBOUND - - def __repr__(self): - return f'interpreters.queues.UNBOUND' - - -UNBOUND = object.__new__(UnboundItem) -UNBOUND_ERROR = object() -UNBOUND_REMOVE = object() +UNBOUND = _crossinterp.UnboundItem.singleton('queue', __name__) -_UNBOUND_CONSTANT_TO_FLAG = { - UNBOUND_REMOVE: 1, - UNBOUND_ERROR: 2, - UNBOUND: 3, -} -_UNBOUND_FLAG_TO_CONSTANT = {v: k - for k, v in _UNBOUND_CONSTANT_TO_FLAG.items()} def _serialize_unbound(unbound): - op = unbound - try: - flag = _UNBOUND_CONSTANT_TO_FLAG[op] - except KeyError: - raise NotImplementedError(f'unsupported unbound replacement op {op!r}') - return flag, + if unbound is UNBOUND: + unbound = _crossinterp.UNBOUND + return _crossinterp.serialize_unbound(unbound) def _resolve_unbound(flag): - try: - op = _UNBOUND_FLAG_TO_CONSTANT[flag] - except KeyError: - raise NotImplementedError(f'unsupported unbound replacement op {flag!r}') - if op is UNBOUND_REMOVE: - # "remove" not possible here - raise NotImplementedError - elif op is UNBOUND_ERROR: - raise ItemInterpreterDestroyed("item's original interpreter destroyed") - elif op is UNBOUND: - return UNBOUND - else: - raise NotImplementedError(repr(op)) + resolved = _crossinterp.resolve_unbound(flag, ItemInterpreterDestroyed) + if resolved is _crossinterp.UNBOUND: + resolved = UNBOUND + return resolved def create(maxsize=0, *, syncobj=False, unbounditems=UNBOUND): diff --git a/Lib/test/test_interpreters/test_queues.py b/Lib/test/test_interpreters/test_queues.py index 30d58a5b29107f..18f83d097eb360 100644 --- a/Lib/test/test_interpreters/test_queues.py +++ b/Lib/test/test_interpreters/test_queues.py @@ -8,11 +8,11 @@ # Raise SkipTest if subinterpreters not supported. _queues = import_helper.import_module('_interpqueues') from test.support import interpreters -from test.support.interpreters import queues +from test.support.interpreters import queues, _crossinterp from .utils import _run_output, TestBase as _TestBase -REPLACE = queues._UNBOUND_CONSTANT_TO_FLAG[queues.UNBOUND] +REPLACE = _crossinterp._UNBOUND_CONSTANT_TO_FLAG[_crossinterp.UNBOUND] def get_num_queues(): diff --git a/Modules/_interpqueuesmodule.c b/Modules/_interpqueuesmodule.c index 8e82789198792f..8534be4ee2f5bc 100644 --- a/Modules/_interpqueuesmodule.c +++ b/Modules/_interpqueuesmodule.c @@ -9,7 +9,9 @@ #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 @@ -402,32 +404,6 @@ handle_queue_error(int err, PyObject *mod, int64_t qid) } -/* unbound items ************************************************************/ - -#define UNBOUND_REMOVE 1 -#define UNBOUND_ERROR 2 -#define UNBOUND_REPLACE 3 - -// It would also be possible to add UNBOUND_REPLACE where the replacement -// value is user-provided. There would be some limitations there, though. -// Another possibility would be something like UNBOUND_COPY, where the -// object is released but the underlying data is copied (with the "raw" -// allocator) and used when the item is popped off the queue. - -static int -check_unbound(int unboundop) -{ - switch (unboundop) { - case UNBOUND_REMOVE: - case UNBOUND_ERROR: - case UNBOUND_REPLACE: - return 1; - default: - return 0; - } -} - - /* the basic queue **********************************************************/ struct _queueitem; diff --git a/Modules/_interpreters_common.h b/Modules/_interpreters_common.h index 07120f6ccc7207..feb925063a66eb 100644 --- a/Modules/_interpreters_common.h +++ b/Modules/_interpreters_common.h @@ -19,3 +19,33 @@ clear_xid_class(PyTypeObject *cls) return _PyCrossInterpreterData_UnregisterClass(cls); } #endif + + +/* unbound items ************************************************************/ + +#ifdef HAS_UNBOUND_ITEMS + +#define UNBOUND_REMOVE 1 +#define UNBOUND_ERROR 2 +#define UNBOUND_REPLACE 3 + +// It would also be possible to add UNBOUND_REPLACE where the replacement +// value is user-provided. There would be some limitations there, though. +// Another possibility would be something like UNBOUND_COPY, where the +// object is released but the underlying data is copied (with the "raw" +// allocator) and used when the item is popped off the queue. + +static int +check_unbound(int unboundop) +{ + switch (unboundop) { + case UNBOUND_REMOVE: + case UNBOUND_ERROR: + case UNBOUND_REPLACE: + return 1; + default: + return 0; + } +} + +#endif From 2835ec7e72c311b7bb3d327d464779eadaad8e6b Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Fri, 12 Jul 2024 14:24:25 -0600 Subject: [PATCH 2/7] Improve unbound item support for channels. --- Lib/test/support/interpreters/channels.py | 108 ++++++++++++++++--- Lib/test/test__interpchannels.py | 120 ++++++++++++---------- Modules/_interpchannelsmodule.c | 112 +++++++++++++++++--- 3 files changed, 255 insertions(+), 85 deletions(-) diff --git a/Lib/test/support/interpreters/channels.py b/Lib/test/support/interpreters/channels.py index fbae7e634cf34d..008e0cef163595 100644 --- a/Lib/test/support/interpreters/channels.py +++ b/Lib/test/support/interpreters/channels.py @@ -2,35 +2,69 @@ import time import _interpchannels as _channels +from . import _crossinterp # aliases: from _interpchannels import ( ChannelError, ChannelNotFoundError, ChannelClosedError, ChannelEmptyError, ChannelNotEmptyError, ) +from ._crossinterp import ( + UNBOUND_ERROR, UNBOUND_REMOVE, +) __all__ = [ + 'UNBOUND', 'UNBOUND_ERROR', 'UNBOUND_REMOVE', 'create', 'list_all', 'SendChannel', 'RecvChannel', 'ChannelError', 'ChannelNotFoundError', 'ChannelEmptyError', + 'ItemInterpreterDestroyed', ] -def create(): +class ItemInterpreterDestroyed(ChannelError, + _crossinterp.ItemInterpreterDestroyed): + """Raised from get() and get_nowait().""" + + +UNBOUND = _crossinterp.UnboundItem.singleton('queue', __name__) + + +def _serialize_unbound(unbound): + if unbound is UNBOUND: + unbound = _crossinterp.UNBOUND + return _crossinterp.serialize_unbound(unbound) + + +def _resolve_unbound(flag): + resolved = _crossinterp.resolve_unbound(flag, ItemInterpreterDestroyed) + if resolved is _crossinterp.UNBOUND: + resolved = UNBOUND + return resolved + + +# XXX Change the default to UNBOUND! +def create(*, unbounditems=UNBOUND_REMOVE): """Return (recv, send) for a new cross-interpreter channel. The channel may be used to pass data safely between interpreters. + + "unbounditems" sets the default for the send end of the channel. + See SendChannel.send() for supported values. The default value + is UNBOUND, which replaces the unbound item when received. """ - cid = _channels.create() - recv, send = RecvChannel(cid), SendChannel(cid) + unbound = _serialize_unbound(unbounditems) + unboundop, = unbound + cid = _channels.create(unboundop) + recv, send = RecvChannel(cid), SendChannel(cid, _unbound=unbound) return recv, send def list_all(): """Return a list of (recv, send) for all open channels.""" - return [(RecvChannel(cid), SendChannel(cid)) - for cid in _channels.list_all()] + return [(RecvChannel(cid), SendChannel(cid, _unbound=unbound)) + for cid, unbound in _channels.list_all()] class _ChannelEnd: @@ -106,12 +140,15 @@ def recv(self, timeout=None, *, if timeout < 0: raise ValueError(f'timeout value must be non-negative') end = time.time() + timeout - obj = _channels.recv(self._id, _sentinel) + obj, unboundop = _channels.recv(self._id, _sentinel) while obj is _sentinel: time.sleep(_delay) if timeout is not None and time.time() >= end: raise TimeoutError - obj = _channels.recv(self._id, _sentinel) + obj, unboundop = _channels.recv(self._id, _sentinel) + if unboundop is not None: + assert obj is None, repr(obj) + return _resolve_unbound(unboundop) return obj def recv_nowait(self, default=_NOT_SET): @@ -122,9 +159,13 @@ def recv_nowait(self, default=_NOT_SET): is the same as recv(). """ if default is _NOT_SET: - return _channels.recv(self._id) + obj, unboundop = _channels.recv(self._id) else: - return _channels.recv(self._id, default) + obj, unboundop = _channels.recv(self._id, default) + if unboundop is not None: + assert obj is None, repr(obj) + return _resolve_unbound(unboundop) + return obj def close(self): _channels.close(self._id, recv=True) @@ -135,43 +176,76 @@ class SendChannel(_ChannelEnd): _end = 'send' + def __new__(cls, cid, *, _unbound=None): + if _unbound is None: + op = _channels.get_channel_defaults(cid) + _unbound = (op,) + self = super().__new__(cls, cid) + self._unbound = _unbound + return self + @property def is_closed(self): info = self._info return info.closed or info.closing - def send(self, obj, timeout=None): + def send(self, obj, timeout=None, *, + unbound=None, + ): """Send the object (i.e. its data) to the channel's receiving end. This blocks until the object is received. """ - _channels.send(self._id, obj, timeout=timeout, blocking=True) + if unbound is None: + unboundop, = self._unbound + else: + unboundop, = _serialize_unbound(unbound) + _channels.send(self._id, obj, unboundop, timeout=timeout, blocking=True) - def send_nowait(self, obj): + def send_nowait(self, obj, *, + unbound=None, + ): """Send the object to the channel's receiving end. If the object is immediately received then return True (else False). Otherwise this is the same as send(). """ + if unbound is None: + unboundop, = self._unbound + else: + unboundop, = _serialize_unbound(unbound) # XXX Note that at the moment channel_send() only ever returns # None. This should be fixed when channel_send_wait() is added. # See bpo-32604 and gh-19829. - return _channels.send(self._id, obj, blocking=False) + return _channels.send(self._id, obj, unboundop, blocking=False) - def send_buffer(self, obj, timeout=None): + def send_buffer(self, obj, timeout=None, *, + unbound=None, + ): """Send the object's buffer to the channel's receiving end. This blocks until the object is received. """ - _channels.send_buffer(self._id, obj, timeout=timeout, blocking=True) + if unbound is None: + unboundop, = self._unbound + else: + unboundop, = _serialize_unbound(unbound) + _channels.send_buffer(self._id, obj, unboundop, + timeout=timeout, blocking=True) - def send_buffer_nowait(self, obj): + def send_buffer_nowait(self, obj, *, + unbound=None, + ): """Send the object's buffer to the channel's receiving end. If the object is immediately received then return True (else False). Otherwise this is the same as send(). """ - return _channels.send_buffer(self._id, obj, blocking=False) + if unbound is None: + unboundop, = self._unbound + else: + unboundop, = _serialize_unbound(unbound) + return _channels.send_buffer(self._id, obj, unboundop, blocking=False) def close(self): _channels.close(self._id, send=True) diff --git a/Lib/test/test__interpchannels.py b/Lib/test/test__interpchannels.py index b76c58917c0b9c..f1430ce624e5a3 100644 --- a/Lib/test/test__interpchannels.py +++ b/Lib/test/test__interpchannels.py @@ -29,9 +29,19 @@ def recv_wait(cid): while True: try: - return _channels.recv(cid) + obj, unboundop = _channels.recv(cid) except _channels.ChannelEmptyError: time.sleep(0.1) + else: + assert unboundop is None, repr(unboundop) + return obj + + +def recv_nowait(cid, *args): + obj, unboundop = _channels.recv(cid, *args) + assert unboundop is None, repr(unboundop) + return obj + #@contextmanager #def run_threaded(id, source, **shared): @@ -212,7 +222,7 @@ def _run_action(cid, action, end, state): else: raise Exception('expected ChannelEmptyError') else: - _channels.recv(cid) + recv_nowait(cid) return state.decr() else: raise ValueError(end) @@ -235,7 +245,7 @@ def _run_action(cid, action, end, state): def clean_up_channels(): - for cid in _channels.list_all(): + for cid, _ in _channels.list_all(): try: _channels.destroy(cid) except _channels.ChannelNotFoundError: @@ -345,7 +355,7 @@ def test_shareable(self): obj = _channels.create() _channels.send(chan, obj, blocking=False) - got = _channels.recv(chan) + got = recv_nowait(chan) self.assertEqual(got, obj) self.assertIs(type(got), type(obj)) @@ -360,11 +370,11 @@ def test_create_cid(self): self.assertIsInstance(cid, _channels.ChannelID) def test_sequential_ids(self): - before = _channels.list_all() + before = [cid for cid, _ in _channels.list_all()] id1 = _channels.create() id2 = _channels.create() id3 = _channels.create() - after = _channels.list_all() + after = [cid for cid, _ in _channels.list_all()] self.assertEqual(id2, int(id1) + 1) self.assertEqual(id3, int(id2) + 1) @@ -412,7 +422,7 @@ def test_channel_list_interpreters_basic(self): interp1 = _interpreters.create() _run_output(interp1, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) # Test for channel that has both ends associated to an interpreter. send_interps = _channels.list_interpreters(cid, send=True) @@ -435,11 +445,11 @@ def test_channel_list_interpreters_multiple(self): """)) _run_output(interp2, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) _run_output(interp3, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) @@ -454,7 +464,7 @@ def test_channel_list_interpreters_destroyed(self): _channels.send(cid, "send", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) # Should be one interpreter associated with each end. send_interps = _channels.list_interpreters(cid, send=True) @@ -480,12 +490,12 @@ def test_channel_list_interpreters_released(self): _channels.send(cid, "data", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) _channels.send(cid, "data", blocking=False) _run_output(interp2, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + _channels.recv({cid}) """)) # Check the setup. send_interps = _channels.list_interpreters(cid, send=True) @@ -580,7 +590,7 @@ def test_allowed_types(self): for obj in objects: with self.subTest(obj): _channels.send(cid, obj, blocking=False) - got = _channels.recv(cid) + got = recv_nowait(cid) self.assertEqual(got, obj) self.assertIs(type(got), type(obj)) @@ -598,7 +608,7 @@ def test_run_string_arg_unresolved(self): print(cid.end) _channels.send(cid, b'spam', blocking=False) """)) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertEqual(obj, b'spam') self.assertEqual(out.strip(), 'send') @@ -618,7 +628,7 @@ def test_run_string_arg_resolved(self): _channels.send(chan.id, b'spam', blocking=False) """), dict(chan=cid.send)) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertEqual(obj, b'spam') self.assertEqual(out.strip(), 'send') @@ -630,7 +640,7 @@ def test_send_recv_main(self): cid = _channels.create() orig = b'spam' _channels.send(cid, orig, blocking=False) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertEqual(obj, orig) self.assertIsNot(obj, orig) @@ -642,7 +652,7 @@ def test_send_recv_same_interpreter(self): cid = _channels.create() orig = b'spam' _channels.send(cid, orig, blocking=False) - obj = _channels.recv(cid) + obj, _ = _channels.recv(cid) assert obj is not orig assert obj == orig """)) @@ -654,7 +664,7 @@ def test_send_recv_different_interpreters(self): import _interpchannels as _channels _channels.send({cid}, b'spam', blocking=False) """)) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertEqual(obj, b'spam') @@ -685,7 +695,7 @@ def f(): import _interpchannels as _channels while True: try: - obj = _channels.recv({cid}) + obj, _ = _channels.recv({cid}) break except _channels.ChannelEmptyError: time.sleep(0.1) @@ -717,16 +727,16 @@ def test_recv_empty(self): def test_recv_default(self): default = object() cid = _channels.create() - obj1 = _channels.recv(cid, default) + obj1 = recv_nowait(cid, default) _channels.send(cid, None, blocking=False) _channels.send(cid, 1, blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'eggs', blocking=False) - obj2 = _channels.recv(cid, default) - obj3 = _channels.recv(cid, default) - obj4 = _channels.recv(cid) - obj5 = _channels.recv(cid, default) - obj6 = _channels.recv(cid, default) + obj2 = recv_nowait(cid, default) + obj3 = recv_nowait(cid, default) + obj4 = recv_nowait(cid) + obj5 = recv_nowait(cid, default) + obj6 = recv_nowait(cid, default) self.assertIs(obj1, default) self.assertIs(obj2, None) @@ -759,7 +769,7 @@ def test_recv_sending_interp_destroyed(self): _channels.send(cid2, b'eggs', blocking=False) _interpreters.destroy(interp) - _channels.recv(cid2) + recv_nowait(cid2) with self.assertRaisesRegex(RuntimeError, f'channel {cid2} is empty'): _channels.recv(cid2) @@ -772,7 +782,7 @@ def test_send_buffer(self): buf = bytearray(b'spamspamspam') cid = _channels.create() _channels.send_buffer(cid, buf, blocking=False) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertIsNot(obj, buf) self.assertIsInstance(obj, memoryview) @@ -799,7 +809,7 @@ def build_send_waiter(self, obj, *, buffer=False): started = time.monotonic() send(cid, obj, blocking=False) stopped = time.monotonic() - _channels.recv(cid) + recv_nowait(cid) finally: _channels.destroy(cid) delay = stopped - started # seconds @@ -882,7 +892,7 @@ def test_send_timeout(self): with self.assertRaises(TimeoutError): _channels.send(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): - received = _channels.recv(cid) + received = recv_nowait(cid) print(repr(received)) with self.subTest('timeout not hit'): @@ -919,7 +929,7 @@ def test_send_buffer_timeout(self): with self.assertRaises(TimeoutError): _channels.send_buffer(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): - received = _channels.recv(cid) + received = recv_nowait(cid) print(repr(received)) with self.subTest('timeout not hit'): @@ -1001,7 +1011,7 @@ def f(): def test_close_single_user(self): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1036,7 +1046,7 @@ def test_close_multiple_users(self): def test_close_multiple_times(self): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1053,7 +1063,7 @@ def test_close_empty(self): with self.subTest((send, recv)): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.close(cid, send=send, recv=recv) with self.assertRaises(_channels.ChannelClosedError): @@ -1068,7 +1078,7 @@ def test_close_defaults_with_unused_items(self): with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid) - _channels.recv(cid) + recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) def test_close_recv_with_unused_items_unforced(self): @@ -1078,10 +1088,10 @@ def test_close_recv_with_unused_items_unforced(self): with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid, recv=True) - _channels.recv(cid) + recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) - _channels.recv(cid) - _channels.recv(cid) + recv_nowait(cid) + recv_nowait(cid) _channels.close(cid, recv=True) def test_close_send_with_unused_items_unforced(self): @@ -1092,8 +1102,8 @@ def test_close_send_with_unused_items_unforced(self): with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') - _channels.recv(cid) - _channels.recv(cid) + recv_nowait(cid) + recv_nowait(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) @@ -1104,10 +1114,10 @@ def test_close_both_with_unused_items_unforced(self): with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid, recv=True, send=True) - _channels.recv(cid) + recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) - _channels.recv(cid) - _channels.recv(cid) + recv_nowait(cid) + recv_nowait(cid) _channels.close(cid, recv=True) def test_close_recv_with_unused_items_forced(self): @@ -1170,7 +1180,7 @@ def test_close_used_multiple_times_by_single_user(self): _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.close(cid, force=True) with self.assertRaises(_channels.ChannelClosedError): @@ -1242,7 +1252,7 @@ class ChannelReleaseTests(TestBase): def test_single_user(self): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): @@ -1260,7 +1270,7 @@ def test_multiple_users(self): """)) out = _run_output(id2, dedent(f""" import _interpchannels as _channels - obj = _channels.recv({cid}) + obj, _ = _channels.recv({cid}) _channels.release({cid}) print(repr(obj)) """)) @@ -1273,7 +1283,7 @@ def test_multiple_users(self): def test_no_kwargs(self): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1284,7 +1294,7 @@ def test_no_kwargs(self): def test_multiple_times(self): cid = _channels.create() _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): @@ -1316,7 +1326,7 @@ def test_by_unassociated_interp(self): import _interpchannels as _channels _channels.release({cid}) """)) - obj = _channels.recv(cid) + obj = recv_nowait(cid) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1340,10 +1350,10 @@ def test_partially(self): # XXX Is partial close too weird/confusing? cid = _channels.create() _channels.send(cid, None, blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.send(cid, b'spam', blocking=False) _channels.release(cid, send=True) - obj = _channels.recv(cid) + obj = recv_nowait(cid) self.assertEqual(obj, b'spam') @@ -1352,7 +1362,7 @@ def test_used_multiple_times_by_single_user(self): _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) - _channels.recv(cid) + recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): @@ -1439,7 +1449,7 @@ def _new_channel(self, creator): _xxsubchannels.send({ch}, int(cid), blocking=False) del _interpreters """) - self._cid = _channels.recv(ch) + self._cid = recv_nowait(ch) return self._cid def _get_interpreter(self, interp): @@ -1670,8 +1680,8 @@ def run_action(self, fix, action, *, hideclosed=True): _channels.send({_cid}, b'X' if result.closed else b'', blocking=False) """) result = ChannelState( - pending=int.from_bytes(_channels.recv(_cid), 'little'), - closed=bool(_channels.recv(_cid)), + pending=int.from_bytes(recv_nowait(_cid), 'little'), + closed=bool(recv_nowait(_cid)), ) fix.record_action(action, result) @@ -1729,7 +1739,7 @@ def _assert_closed(self, fix): self.assertTrue(fix.state.closed) for _ in range(fix.state.pending): - _channels.recv(fix.cid) + recv_nowait(fix.cid) self._assert_closed_in_interp(fix) for interp in ('same', 'other'): diff --git a/Modules/_interpchannelsmodule.c b/Modules/_interpchannelsmodule.c index f0447475c49116..18737186dd9074 100644 --- a/Modules/_interpchannelsmodule.c +++ b/Modules/_interpchannelsmodule.c @@ -18,7 +18,9 @@ #endif #define REGISTERS_HEAP_TYPES +#define HAS_UNBOUND_ITEMS #include "_interpreters_common.h" +#undef HAS_UNBOUND_ITEMS #undef REGISTERS_HEAP_TYPES @@ -2767,8 +2769,23 @@ clear_interpreter(void *data) static PyObject * -channelsmod_create(PyObject *self, PyObject *Py_UNUSED(ignored)) +channelsmod_create(PyObject *self, PyObject *args, PyObject *kwds) { + static char *kwlist[] = {"unboundop", NULL}; + // XXX Make it required. + int unboundop = UNBOUND_REMOVE; + if (!PyArg_ParseTupleAndKeywords(args, kwds, "|i:create", kwlist, + &unboundop)) + { + return NULL; + } + if (!check_unbound(unboundop)) { + PyErr_Format(PyExc_ValueError, + "unsupported unboundop %d", unboundop); + return NULL; + } + + // XXX Save unboundop. int64_t cid = channel_create(&_globals.channels); if (cid < 0) { (void)handle_channel_error(-1, self, cid); @@ -2796,7 +2813,7 @@ channelsmod_create(PyObject *self, PyObject *Py_UNUSED(ignored)) } PyDoc_STRVAR(channelsmod_create_doc, -"channel_create() -> cid\n\ +"channel_create(unboundop) -> cid\n\ \n\ Create a new cross-interpreter channel and return a unique generated ID."); @@ -2860,7 +2877,16 @@ channelsmod_list_all(PyObject *self, PyObject *Py_UNUSED(ignored)) break; } assert(cidobj != NULL); - PyList_SET_ITEM(ids, (Py_ssize_t)i, cidobj); + + // XXX get unboundop + int unboundop = UNBOUND_REPLACE; + PyObject *item = Py_BuildValue("Oi", cidobj, unboundop); + Py_DECREF(cidobj); + if (item == NULL) { + Py_SETREF(ids, NULL); + break; + } + PyList_SET_ITEM(ids, (Py_ssize_t)i, item); } finally: @@ -2942,16 +2968,25 @@ receive end."); static PyObject * channelsmod_send(PyObject *self, PyObject *args, PyObject *kwds) { - static char *kwlist[] = {"cid", "obj", "blocking", "timeout", NULL}; + static char *kwlist[] = {"cid", "obj", "unboundop", "blocking", "timeout", + NULL}; struct channel_id_converter_data cid_data = { .module = self, }; PyObject *obj; + // XXX Make unboundop required. + int unboundop = UNBOUND_REMOVE; int blocking = 1; PyObject *timeout_obj = NULL; - if (!PyArg_ParseTupleAndKeywords(args, kwds, "O&O|$pO:channel_send", kwlist, + if (!PyArg_ParseTupleAndKeywords(args, kwds, "O&O|i$pO:channel_send", kwlist, channel_id_converter, &cid_data, &obj, - &blocking, &timeout_obj)) { + &unboundop, &blocking, &timeout_obj)) + { + return NULL; + } + if (!check_unbound(unboundop)) { + PyErr_Format(PyExc_ValueError, + "unsupported unboundop %d", unboundop); return NULL; } @@ -2963,6 +2998,7 @@ channelsmod_send(PyObject *self, PyObject *args, PyObject *kwds) /* Queue up the object. */ int err = 0; + // XXX Store unboundop if (blocking) { err = channel_send_wait(&_globals.channels, cid, obj, timeout); } @@ -2985,17 +3021,25 @@ By default this waits for the object to be received."); static PyObject * channelsmod_send_buffer(PyObject *self, PyObject *args, PyObject *kwds) { - static char *kwlist[] = {"cid", "obj", "blocking", "timeout", NULL}; + static char *kwlist[] = {"cid", "obj", "unboundop", "blocking", "timeout", + NULL}; struct channel_id_converter_data cid_data = { .module = self, }; PyObject *obj; + // XXX Make unboundop required. + int unboundop = UNBOUND_REMOVE; int blocking = 1; PyObject *timeout_obj = NULL; if (!PyArg_ParseTupleAndKeywords(args, kwds, - "O&O|$pO:channel_send_buffer", kwlist, + "O&O|i$pO:channel_send_buffer", kwlist, channel_id_converter, &cid_data, &obj, - &blocking, &timeout_obj)) { + &unboundop, &blocking, &timeout_obj)) { + return NULL; + } + if (!check_unbound(unboundop)) { + PyErr_Format(PyExc_ValueError, + "unsupported unboundop %d", unboundop); return NULL; } @@ -3012,6 +3056,7 @@ channelsmod_send_buffer(PyObject *self, PyObject *args, PyObject *kwds) /* Queue up the object. */ int err = 0; + // XXX Store unboundop if (blocking) { err = channel_send_wait(&_globals.channels, cid, tempobj, timeout); } @@ -3062,11 +3107,15 @@ channelsmod_recv(PyObject *self, PyObject *args, PyObject *kwds) obj = Py_NewRef(dflt); } Py_XDECREF(dflt); - return obj; + + // XXX return unboundop if unbound + PyObject *res = Py_BuildValue("OO", obj, Py_None); + Py_DECREF(obj); + return res; } PyDoc_STRVAR(channelsmod_recv_doc, -"channel_recv(cid, [default]) -> obj\n\ +"channel_recv(cid, [default]) -> (obj, unboundop)\n\ \n\ Return a new object from the data at the front of the channel's queue.\n\ \n\ @@ -3194,6 +3243,41 @@ PyDoc_STRVAR(channelsmod_get_info_doc, \n\ Return details about the channel."); +static PyObject * +channelsmod_get_channel_defaults(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"cid", NULL}; + struct channel_id_converter_data cid_data = { + .module = self, + }; + if (!PyArg_ParseTupleAndKeywords(args, kwds, + "O&:get_channel_defaults", kwlist, + channel_id_converter, &cid_data)) { + return NULL; + } + int64_t cid = cid_data.cid; + + // XXX get stored defaults. +// _channel *channel = NULL; +// int err = _channels_lookup(&_globals.channels, cid, &channel); +// if (handle_channel_error(err, self, cid)) { +// return NULL; +// } +// int unboundop = channel->defaults.unboundop; +// _channel_unmark_waiter(channel, _globals.channels.mutex); + + (void)cid; + int unboundop = UNBOUND_REPLACE; + + PyObject *defaults = Py_BuildValue("i", unboundop); + return defaults; +} + +PyDoc_STRVAR(channelsmod_get_channel_defaults_doc, +"get_channel_defaults(cid)\n\ +\n\ +Return the channel's default values, set when it was created."); + static PyObject * channelsmod__channel_id(PyObject *self, PyObject *args, PyObject *kwds) { @@ -3240,8 +3324,8 @@ channelsmod__register_end_types(PyObject *self, PyObject *args, PyObject *kwds) } static PyMethodDef module_functions[] = { - {"create", channelsmod_create, - METH_NOARGS, channelsmod_create_doc}, + {"create", _PyCFunction_CAST(channelsmod_create), + METH_VARARGS | METH_KEYWORDS, channelsmod_create_doc}, {"destroy", _PyCFunction_CAST(channelsmod_destroy), METH_VARARGS | METH_KEYWORDS, channelsmod_destroy_doc}, {"list_all", channelsmod_list_all, @@ -3260,6 +3344,8 @@ static PyMethodDef module_functions[] = { METH_VARARGS | METH_KEYWORDS, channelsmod_release_doc}, {"get_info", _PyCFunction_CAST(channelsmod_get_info), METH_VARARGS | METH_KEYWORDS, channelsmod_get_info_doc}, + {"get_channel_defaults", _PyCFunction_CAST(channelsmod_get_channel_defaults), + METH_VARARGS | METH_KEYWORDS, channelsmod_get_channel_defaults_doc}, {"_channel_id", _PyCFunction_CAST(channelsmod__channel_id), METH_VARARGS | METH_KEYWORDS, NULL}, {"_register_end_types", _PyCFunction_CAST(channelsmod__register_end_types), From c6b3a0e64381c102c1c43d8708b7dd032fe022ca Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Fri, 12 Jul 2024 14:59:02 -0600 Subject: [PATCH 3/7] Make UNBOUND_REPLACE the default. --- Lib/test/support/interpreters/channels.py | 3 +- Lib/test/test__interpchannels.py | 155 +++++++++++----------- Modules/_interpchannelsmodule.c | 5 +- 3 files changed, 82 insertions(+), 81 deletions(-) diff --git a/Lib/test/support/interpreters/channels.py b/Lib/test/support/interpreters/channels.py index 008e0cef163595..237b82def9fa6a 100644 --- a/Lib/test/support/interpreters/channels.py +++ b/Lib/test/support/interpreters/channels.py @@ -44,8 +44,7 @@ def _resolve_unbound(flag): return resolved -# XXX Change the default to UNBOUND! -def create(*, unbounditems=UNBOUND_REMOVE): +def create(*, unbounditems=UNBOUND): """Return (recv, send) for a new cross-interpreter channel. The channel may be used to pass data safely between interpreters. diff --git a/Lib/test/test__interpchannels.py b/Lib/test/test__interpchannels.py index f1430ce624e5a3..6324222a1dc90a 100644 --- a/Lib/test/test__interpchannels.py +++ b/Lib/test/test__interpchannels.py @@ -8,16 +8,19 @@ from test.support import import_helper +from test.support.interpreters import _crossinterp from test.test__interpreters import ( _interpreters, _run_output, clean_up_interpreters, ) - _channels = import_helper.import_module('_interpchannels') +REPLACE = _crossinterp._UNBOUND_CONSTANT_TO_FLAG[_crossinterp.UNBOUND] + + # Additional tests are found in Lib/test/test_interpreters/test_channels.py. # New tests should be added there. # XXX The tests here should be moved there. See the note under LowLevelTests. @@ -307,7 +310,7 @@ def test_bad_kwargs(self): _channels._channel_id(10, send=False, recv=False) def test_does_not_exist(self): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(_channels.ChannelNotFoundError): _channels._channel_id(int(cid) + 1) # unforced @@ -329,9 +332,9 @@ def test_repr(self): self.assertEqual(repr(cid), 'ChannelID(10)') def test_equality(self): - cid1 = _channels.create() + cid1 = _channels.create(REPLACE) cid2 = _channels._channel_id(int(cid1)) - cid3 = _channels.create() + cid3 = _channels.create(REPLACE) self.assertTrue(cid1 == cid1) self.assertTrue(cid1 == cid2) @@ -351,9 +354,9 @@ def test_equality(self): self.assertTrue(cid1 != cid3) def test_shareable(self): - chan = _channels.create() + chan = _channels.create(REPLACE) - obj = _channels.create() + obj = _channels.create(REPLACE) _channels.send(chan, obj, blocking=False) got = recv_nowait(chan) @@ -366,14 +369,14 @@ def test_shareable(self): class ChannelTests(TestBase): def test_create_cid(self): - cid = _channels.create() + cid = _channels.create(REPLACE) self.assertIsInstance(cid, _channels.ChannelID) def test_sequential_ids(self): before = [cid for cid, _ in _channels.list_all()] - id1 = _channels.create() - id2 = _channels.create() - id3 = _channels.create() + id1 = _channels.create(REPLACE) + id2 = _channels.create(REPLACE) + id3 = _channels.create(REPLACE) after = [cid for cid, _ in _channels.list_all()] self.assertEqual(id2, int(id1) + 1) @@ -384,7 +387,7 @@ def test_ids_global(self): id1 = _interpreters.create() out = _run_output(id1, dedent(""" import _interpchannels as _channels - cid = _channels.create() + cid = _channels.create(3) print(cid) """)) cid1 = int(out.strip()) @@ -392,7 +395,7 @@ def test_ids_global(self): id2 = _interpreters.create() out = _run_output(id2, dedent(""" import _interpchannels as _channels - cid = _channels.create() + cid = _channels.create(3) print(cid) """)) cid2 = int(out.strip()) @@ -402,7 +405,7 @@ def test_ids_global(self): def test_channel_list_interpreters_none(self): """Test listing interpreters for a channel with no associations.""" # Test for channel with no associated _interpreters. - cid = _channels.create() + cid = _channels.create(REPLACE) send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, []) @@ -411,7 +414,7 @@ def test_channel_list_interpreters_none(self): def test_channel_list_interpreters_basic(self): """Test basic listing channel _interpreters.""" interp0, *_ = _interpreters.get_main() - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) # Test for a channel that has one end associated to an interpreter. send_interps = _channels.list_interpreters(cid, send=True) @@ -436,7 +439,7 @@ def test_channel_list_interpreters_multiple(self): interp1 = _interpreters.create() interp2 = _interpreters.create() interp3 = _interpreters.create() - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) _run_output(interp1, dedent(f""" @@ -460,7 +463,7 @@ def test_channel_list_interpreters_destroyed(self): """Test listing channel interpreters with a destroyed interpreter.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels @@ -486,7 +489,7 @@ def test_channel_list_interpreters_released(self): interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() interp2 = _interpreters.create() - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, "data", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels @@ -526,7 +529,7 @@ def test_channel_list_interpreters_closed(self): """Test listing channel interpreters with a closed channel.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() - cid = _channels.create() + cid = _channels.create(REPLACE) # Put something in the channel so that it's not empty. _channels.send(cid, "send", blocking=False) @@ -548,7 +551,7 @@ def test_channel_list_interpreters_closed_send_end(self): """Test listing channel interpreters with a channel's send end closed.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() - cid = _channels.create() + cid = _channels.create(REPLACE) # Put something in the channel so that it's not empty. _channels.send(cid, "send", blocking=False) @@ -580,7 +583,7 @@ def test_channel_list_interpreters_closed_send_end(self): _channels.list_interpreters(cid, send=False) def test_allowed_types(self): - cid = _channels.create() + cid = _channels.create(REPLACE) objects = [ None, 'spam', @@ -599,7 +602,7 @@ def test_allowed_types(self): # XXX What about between interpreters? def test_run_string_arg_unresolved(self): - cid = _channels.create() + cid = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.set___main___attrs(interp, dict(cid=cid.send)) @@ -618,7 +621,7 @@ def test_run_string_arg_unresolved(self): # Note: this test caused crashes on some buildbots (bpo-33615). @unittest.skip('disabled until high-level channels exist') def test_run_string_arg_resolved(self): - cid = _channels.create() + cid = _channels.create(REPLACE) cid = _channels._channel_id(cid, _resolve=True) interp = _interpreters.create() @@ -637,7 +640,7 @@ def test_run_string_arg_resolved(self): # send/recv def test_send_recv_main(self): - cid = _channels.create() + cid = _channels.create(REPLACE) orig = b'spam' _channels.send(cid, orig, blocking=False) obj = recv_nowait(cid) @@ -649,7 +652,7 @@ def test_send_recv_same_interpreter(self): id1 = _interpreters.create() out = _run_output(id1, dedent(""" import _interpchannels as _channels - cid = _channels.create() + cid = _channels.create(REPLACE) orig = b'spam' _channels.send(cid, orig, blocking=False) obj, _ = _channels.recv(cid) @@ -658,7 +661,7 @@ def test_send_recv_same_interpreter(self): """)) def test_send_recv_different_interpreters(self): - cid = _channels.create() + cid = _channels.create(REPLACE) id1 = _interpreters.create() out = _run_output(id1, dedent(f""" import _interpchannels as _channels @@ -669,7 +672,7 @@ def test_send_recv_different_interpreters(self): self.assertEqual(obj, b'spam') def test_send_recv_different_threads(self): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): obj = recv_wait(cid) @@ -684,7 +687,7 @@ def f(): self.assertEqual(obj, b'spam') def test_send_recv_different_interpreters_and_threads(self): - cid = _channels.create() + cid = _channels.create(REPLACE) id1 = _interpreters.create() out = None @@ -720,13 +723,13 @@ def test_recv_not_found(self): _channels.recv(10) def test_recv_empty(self): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(_channels.ChannelEmptyError): _channels.recv(cid) def test_recv_default(self): default = object() - cid = _channels.create() + cid = _channels.create(REPLACE) obj1 = recv_nowait(cid, default) _channels.send(cid, None, blocking=False) _channels.send(cid, 1, blocking=False) @@ -747,7 +750,7 @@ def test_recv_default(self): def test_recv_sending_interp_destroyed(self): with self.subTest('closed'): - cid1 = _channels.create() + cid1 = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels @@ -760,7 +763,7 @@ def test_recv_sending_interp_destroyed(self): _channels.recv(cid1) del cid1 with self.subTest('still open'): - cid2 = _channels.create() + cid2 = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels @@ -780,7 +783,7 @@ def test_recv_sending_interp_destroyed(self): def test_send_buffer(self): buf = bytearray(b'spamspamspam') - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send_buffer(cid, buf, blocking=False) obj = recv_nowait(cid) @@ -804,7 +807,7 @@ def build_send_waiter(self, obj, *, buffer=False): else: send = _channels.send - cid = _channels.create() + cid = _channels.create(REPLACE) try: started = time.monotonic() send(cid, obj, blocking=False) @@ -823,7 +826,7 @@ def test_send_blocking_waiting(self): received = None obj = b'spam' wait = self.build_send_waiter(obj) - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): nonlocal received wait() @@ -839,7 +842,7 @@ def test_send_buffer_blocking_waiting(self): received = None obj = bytearray(b'spam') wait = self.build_send_waiter(obj, buffer=True) - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): nonlocal received wait() @@ -854,7 +857,7 @@ def f(): def test_send_blocking_no_wait(self): received = None obj = b'spam' - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): nonlocal received received = recv_wait(cid) @@ -868,7 +871,7 @@ def f(): def test_send_buffer_blocking_no_wait(self): received = None obj = bytearray(b'spam') - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): nonlocal received received = recv_wait(cid) @@ -883,12 +886,12 @@ def test_send_timeout(self): obj = b'spam' with self.subTest('non-blocking with timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(ValueError): _channels.send(cid, obj, blocking=False, timeout=0.1) with self.subTest('timeout hit'): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(TimeoutError): _channels.send(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): @@ -896,7 +899,7 @@ def test_send_timeout(self): print(repr(received)) with self.subTest('timeout not hit'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): recv_wait(cid) t = threading.Thread(target=f) @@ -920,12 +923,12 @@ def test_send_buffer_timeout(self): obj = bytearray(b'spam') with self.subTest('non-blocking with timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(ValueError): _channels.send_buffer(cid, obj, blocking=False, timeout=0.1) with self.subTest('timeout hit'): - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(TimeoutError): _channels.send_buffer(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): @@ -933,7 +936,7 @@ def test_send_buffer_timeout(self): print(repr(received)) with self.subTest('timeout not hit'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): recv_wait(cid) t = threading.Thread(target=f) @@ -946,7 +949,7 @@ def test_send_closed_while_waiting(self): wait = self.build_send_waiter(obj) with self.subTest('without timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) @@ -957,7 +960,7 @@ def f(): t.join() with self.subTest('with timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) @@ -984,7 +987,7 @@ def test_send_buffer_closed_while_waiting(self): wait = self.build_send_waiter(obj, buffer=True) with self.subTest('without timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) @@ -995,7 +998,7 @@ def f(): t.join() with self.subTest('with timeout'): - cid = _channels.create() + cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) @@ -1009,7 +1012,7 @@ def f(): # close def test_close_single_user(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid) @@ -1020,7 +1023,7 @@ def test_close_single_user(self): _channels.recv(cid) def test_close_multiple_users(self): - cid = _channels.create() + cid = _channels.create(REPLACE) id1 = _interpreters.create() id2 = _interpreters.create() _interpreters.run_string(id1, dedent(f""" @@ -1044,7 +1047,7 @@ def test_close_multiple_users(self): self.assertEqual(excsnap.type.__name__, 'ChannelClosedError') def test_close_multiple_times(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid) @@ -1061,7 +1064,7 @@ def test_close_empty(self): ] for send, recv in tests: with self.subTest((send, recv)): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid, send=send, recv=recv) @@ -1072,7 +1075,7 @@ def test_close_empty(self): _channels.recv(cid) def test_close_defaults_with_unused_items(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) @@ -1082,7 +1085,7 @@ def test_close_defaults_with_unused_items(self): _channels.send(cid, b'eggs', blocking=False) def test_close_recv_with_unused_items_unforced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) @@ -1095,7 +1098,7 @@ def test_close_recv_with_unused_items_unforced(self): _channels.close(cid, recv=True) def test_close_send_with_unused_items_unforced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True) @@ -1108,7 +1111,7 @@ def test_close_send_with_unused_items_unforced(self): _channels.recv(cid) def test_close_both_with_unused_items_unforced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) @@ -1121,7 +1124,7 @@ def test_close_both_with_unused_items_unforced(self): _channels.close(cid, recv=True) def test_close_recv_with_unused_items_forced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, recv=True, force=True) @@ -1132,7 +1135,7 @@ def test_close_recv_with_unused_items_forced(self): _channels.recv(cid) def test_close_send_with_unused_items_forced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True, force=True) @@ -1143,7 +1146,7 @@ def test_close_send_with_unused_items_forced(self): _channels.recv(cid) def test_close_both_with_unused_items_forced(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True, recv=True, force=True) @@ -1154,7 +1157,7 @@ def test_close_both_with_unused_items_forced(self): _channels.recv(cid) def test_close_never_used(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1163,7 +1166,7 @@ def test_close_never_used(self): _channels.recv(cid) def test_close_by_unassociated_interp(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" @@ -1176,7 +1179,7 @@ def test_close_by_unassociated_interp(self): _channels.close(cid) def test_close_used_multiple_times_by_single_user(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) @@ -1189,7 +1192,7 @@ def test_close_used_multiple_times_by_single_user(self): _channels.recv(cid) def test_channel_list_interpreters_invalid_channel(self): - cid = _channels.create() + cid = _channels.create(REPLACE) # Test for invalid channel ID. with self.assertRaises(_channels.ChannelNotFoundError): _channels.list_interpreters(1000, send=True) @@ -1201,7 +1204,7 @@ def test_channel_list_interpreters_invalid_channel(self): def test_channel_list_interpreters_invalid_args(self): # Tests for invalid arguments passed to the API. - cid = _channels.create() + cid = _channels.create(REPLACE) with self.assertRaises(TypeError): _channels.list_interpreters(cid) @@ -1250,7 +1253,7 @@ class ChannelReleaseTests(TestBase): """ def test_single_user(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid, send=True, recv=True) @@ -1261,7 +1264,7 @@ def test_single_user(self): _channels.recv(cid) def test_multiple_users(self): - cid = _channels.create() + cid = _channels.create(REPLACE) id1 = _interpreters.create() id2 = _interpreters.create() _interpreters.run_string(id1, dedent(f""" @@ -1281,7 +1284,7 @@ def test_multiple_users(self): self.assertEqual(out.strip(), "b'spam'") def test_no_kwargs(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid) @@ -1292,7 +1295,7 @@ def test_no_kwargs(self): _channels.recv(cid) def test_multiple_times(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid, send=True, recv=True) @@ -1301,7 +1304,7 @@ def test_multiple_times(self): _channels.release(cid, send=True, recv=True) def test_with_unused_items(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.release(cid, send=True, recv=True) @@ -1310,7 +1313,7 @@ def test_with_unused_items(self): _channels.recv(cid) def test_never_used(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): @@ -1319,7 +1322,7 @@ def test_never_used(self): _channels.recv(cid) def test_by_unassociated_interp(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" @@ -1335,7 +1338,7 @@ def test_by_unassociated_interp(self): def test_close_if_unassociated(self): # XXX Something's not right with this test... - cid = _channels.create() + cid = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels @@ -1348,7 +1351,7 @@ def test_close_if_unassociated(self): def test_partially(self): # XXX Is partial close too weird/confusing? - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, None, blocking=False) recv_nowait(cid) _channels.send(cid, b'spam', blocking=False) @@ -1358,7 +1361,7 @@ def test_partially(self): self.assertEqual(obj, b'spam') def test_used_multiple_times_by_single_user(self): - cid = _channels.create() + cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) @@ -1438,9 +1441,9 @@ def clean_up(self): def _new_channel(self, creator): if creator.name == 'main': - return _channels.create() + return _channels.create(REPLACE) else: - ch = _channels.create() + ch = _channels.create(REPLACE) run_interp(creator.id, f""" import _interpreters cid = _xxsubchannels.create() @@ -1667,7 +1670,7 @@ def run_action(self, fix, action, *, hideclosed=True): ) fix.record_action(action, result) else: - _cid = _channels.create() + _cid = _channels.create(REPLACE) run_interp(interp.id, f""" result = helpers.run_action( {fix.cid}, diff --git a/Modules/_interpchannelsmodule.c b/Modules/_interpchannelsmodule.c index 18737186dd9074..5da30825a533f8 100644 --- a/Modules/_interpchannelsmodule.c +++ b/Modules/_interpchannelsmodule.c @@ -2772,9 +2772,8 @@ static PyObject * channelsmod_create(PyObject *self, PyObject *args, PyObject *kwds) { static char *kwlist[] = {"unboundop", NULL}; - // XXX Make it required. - int unboundop = UNBOUND_REMOVE; - if (!PyArg_ParseTupleAndKeywords(args, kwds, "|i:create", kwlist, + int unboundop; + if (!PyArg_ParseTupleAndKeywords(args, kwds, "i:create", kwlist, &unboundop)) { return NULL; From f7125abec93590ad13baff493165bfecc746a6ae Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Fri, 12 Jul 2024 15:30:23 -0600 Subject: [PATCH 4/7] Add _interpchannels.get_count(). --- Modules/_interpchannelsmodule.c | 47 +++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/Modules/_interpchannelsmodule.c b/Modules/_interpchannelsmodule.c index 5da30825a533f8..b519cb5b4061fc 100644 --- a/Modules/_interpchannelsmodule.c +++ b/Modules/_interpchannelsmodule.c @@ -1917,6 +1917,23 @@ channel_is_associated(_channels *channels, int64_t cid, int64_t interpid, return (end != NULL && end->open); } +static int +_channel_get_count(_channels *channels, int64_t cid, Py_ssize_t *p_count) +{ + PyThread_type_lock mutex = NULL; + _channel_state *chan = NULL; + int err = _channels_lookup(channels, cid, &mutex, &chan); + if (err != 0) { + return err; + } + assert(chan != NULL); + int64_t count = chan->queue->count; + PyThread_release_lock(mutex); + + *p_count = (Py_ssize_t)count; + return 0; +} + /* channel info */ @@ -3215,6 +3232,34 @@ Close the channel for the current interpreter. 'send' and 'recv'\n\ (bool) may be used to indicate the ends to close. By default both\n\ ends are closed. Closing an already closed end is a noop."); +static PyObject * +channelsmod_get_count(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"cid", NULL}; + struct channel_id_converter_data cid_data = { + .module = self, + }; + if (!PyArg_ParseTupleAndKeywords(args, kwds, + "O&:get_count", kwlist, + channel_id_converter, &cid_data)) { + return NULL; + } + int64_t cid = cid_data.cid; + + Py_ssize_t count = -1; + int err = _channel_get_count(&_globals.channels, cid, &count); + if (handle_channel_error(err, self, cid)) { + return NULL; + } + assert(count >= 0); + return PyLong_FromSsize_t(count); +} + +PyDoc_STRVAR(channelsmod_get_count_doc, +"get_count(cid)\n\ +\n\ +Return the number of items in the channel."); + static PyObject * channelsmod_get_info(PyObject *self, PyObject *args, PyObject *kwds) { @@ -3341,6 +3386,8 @@ static PyMethodDef module_functions[] = { METH_VARARGS | METH_KEYWORDS, channelsmod_close_doc}, {"release", _PyCFunction_CAST(channelsmod_release), METH_VARARGS | METH_KEYWORDS, channelsmod_release_doc}, + {"get_count", _PyCFunction_CAST(channelsmod_get_count), + METH_VARARGS | METH_KEYWORDS, channelsmod_get_count_doc}, {"get_info", _PyCFunction_CAST(channelsmod_get_info), METH_VARARGS | METH_KEYWORDS, channelsmod_get_info_doc}, {"get_channel_defaults", _PyCFunction_CAST(channelsmod_get_channel_defaults), From 51c6f5ddb1e6a598285013f80ee1c24d090ed43c Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Mon, 15 Jul 2024 12:18:16 -0600 Subject: [PATCH 5/7] Implement all the unbound logic. --- Lib/test/support/interpreters/channels.py | 7 +- Lib/test/test__interpchannels.py | 7 +- Lib/test/test_interpreters/test_channels.py | 222 +++++++++++++++++++ Modules/_interpchannelsmodule.c | 228 +++++++++++++------- Modules/_interpqueuesmodule.c | 14 -- Modules/_interpreters_common.h | 15 ++ 6 files changed, 397 insertions(+), 96 deletions(-) diff --git a/Lib/test/support/interpreters/channels.py b/Lib/test/support/interpreters/channels.py index 237b82def9fa6a..d2bd93d77f7169 100644 --- a/Lib/test/support/interpreters/channels.py +++ b/Lib/test/support/interpreters/channels.py @@ -177,8 +177,11 @@ class SendChannel(_ChannelEnd): def __new__(cls, cid, *, _unbound=None): if _unbound is None: - op = _channels.get_channel_defaults(cid) - _unbound = (op,) + try: + op = _channels.get_channel_defaults(cid) + _unbound = (op,) + except ChannelNotFoundError: + _unbound = _serialize_unbound(UNBOUND) self = super().__new__(cls, cid) self._unbound = _unbound return self diff --git a/Lib/test/test__interpchannels.py b/Lib/test/test__interpchannels.py index 6324222a1dc90a..de9f62a98c2a24 100644 --- a/Lib/test/test__interpchannels.py +++ b/Lib/test/test__interpchannels.py @@ -40,9 +40,9 @@ def recv_wait(cid): return obj -def recv_nowait(cid, *args): +def recv_nowait(cid, *args, unbound=False): obj, unboundop = _channels.recv(cid, *args) - assert unboundop is None, repr(unboundop) + assert (unboundop is None) != unbound, repr(unboundop) return obj @@ -772,7 +772,8 @@ def test_recv_sending_interp_destroyed(self): _channels.send(cid2, b'eggs', blocking=False) _interpreters.destroy(interp) - recv_nowait(cid2) + recv_nowait(cid2, unbound=True) + recv_nowait(cid2, unbound=False) with self.assertRaisesRegex(RuntimeError, f'channel {cid2} is empty'): _channels.recv(cid2) diff --git a/Lib/test/test_interpreters/test_channels.py b/Lib/test/test_interpreters/test_channels.py index 6c37754142e361..eada18f99d04db 100644 --- a/Lib/test/test_interpreters/test_channels.py +++ b/Lib/test/test_interpreters/test_channels.py @@ -372,6 +372,228 @@ def test_send_buffer_nowait(self): obj[4:8] = b'ham.' self.assertEqual(obj, buf) + def test_send_cleared_with_subinterpreter(self): + def common(rch, sch, unbound=None, presize=0): + if not unbound: + extraargs = '' + elif unbound is channels.UNBOUND: + extraargs = ', unbound=channels.UNBOUND' + elif unbound is channels.UNBOUND_ERROR: + extraargs = ', unbound=channels.UNBOUND_ERROR' + elif unbound is channels.UNBOUND_REMOVE: + extraargs = ', unbound=channels.UNBOUND_REMOVE' + else: + raise NotImplementedError(repr(unbound)) + interp = interpreters.create() + + _run_output(interp, dedent(f""" + from test.support.interpreters import channels + sch = channels.SendChannel({sch.id}) + obj1 = b'spam' + obj2 = b'eggs' + sch.send_nowait(obj1{extraargs}) + sch.send_nowait(obj2{extraargs}) + """)) + self.assertEqual( + _channels.get_count(rch.id), + presize + 2, + ) + + if presize == 0: + obj1 = rch.recv() + self.assertEqual(obj1, b'spam') + self.assertEqual( + _channels.get_count(rch.id), + presize + 1, + ) + + return interp + + with self.subTest('default'): # UNBOUND + rch, sch = channels.create() + interp = common(rch, sch) + del interp + self.assertEqual(_channels.get_count(rch.id), 1) + obj1 = rch.recv() + self.assertEqual(_channels.get_count(rch.id), 0) + self.assertIs(obj1, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 0) + with self.assertRaises(channels.ChannelEmptyError): + rch.recv_nowait() + + with self.subTest('UNBOUND'): + rch, sch = channels.create() + interp = common(rch, sch, channels.UNBOUND) + del interp + self.assertEqual(_channels.get_count(rch.id), 1) + obj1 = rch.recv() + self.assertIs(obj1, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 0) + with self.assertRaises(channels.ChannelEmptyError): + rch.recv_nowait() + + with self.subTest('UNBOUND_ERROR'): + rch, sch = channels.create() + interp = common(rch, sch, channels.UNBOUND_ERROR) + + del interp + self.assertEqual(_channels.get_count(rch.id), 1) + with self.assertRaises(channels.ItemInterpreterDestroyed): + rch.recv() + + self.assertEqual(_channels.get_count(rch.id), 0) + with self.assertRaises(channels.ChannelEmptyError): + rch.recv_nowait() + + with self.subTest('UNBOUND_REMOVE'): + rch, sch = channels.create() + + interp = common(rch, sch, channels.UNBOUND_REMOVE) + del interp + self.assertEqual(_channels.get_count(rch.id), 0) + with self.assertRaises(channels.ChannelEmptyError): + rch.recv_nowait() + + sch.send_nowait(b'ham', unbound=channels.UNBOUND_REMOVE) + self.assertEqual(_channels.get_count(rch.id), 1) + interp = common(rch, sch, channels.UNBOUND_REMOVE, 1) + self.assertEqual(_channels.get_count(rch.id), 3) + sch.send_nowait(42, unbound=channels.UNBOUND_REMOVE) + self.assertEqual(_channels.get_count(rch.id), 4) + del interp + self.assertEqual(_channels.get_count(rch.id), 2) + obj1 = rch.recv() + obj2 = rch.recv() + self.assertEqual(obj1, b'ham') + self.assertEqual(obj2, 42) + self.assertEqual(_channels.get_count(rch.id), 0) + with self.assertRaises(channels.ChannelEmptyError): + rch.recv_nowait() + + def test_send_cleared_with_subinterpreter_mixed(self): + rch, sch = channels.create() + interp = interpreters.create() + + # If we don't associate the main interpreter with the channel + # then the channel will be automatically closed when interp + # is destroyed. + sch.send_nowait(None) + rch.recv() + self.assertEqual(_channels.get_count(rch.id), 0) + + _run_output(interp, dedent(f""" + from test.support.interpreters import channels + sch = channels.SendChannel({sch.id}) + sch.send_nowait(1, unbound=channels.UNBOUND) + sch.send_nowait(2, unbound=channels.UNBOUND_ERROR) + sch.send_nowait(3) + sch.send_nowait(4, unbound=channels.UNBOUND_REMOVE) + sch.send_nowait(5, unbound=channels.UNBOUND) + """)) + self.assertEqual(_channels.get_count(rch.id), 5) + + del interp + self.assertEqual(_channels.get_count(rch.id), 4) + + obj1 = rch.recv() + self.assertIs(obj1, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 3) + + with self.assertRaises(channels.ItemInterpreterDestroyed): + rch.recv() + self.assertEqual(_channels.get_count(rch.id), 2) + + obj2 = rch.recv() + self.assertIs(obj2, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 1) + + obj3 = rch.recv() + self.assertIs(obj3, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 0) + + def test_send_cleared_with_subinterpreter_multiple(self): + rch, sch = channels.create() + interp1 = interpreters.create() + interp2 = interpreters.create() + + sch.send_nowait(1) + _run_output(interp1, dedent(f""" + from test.support.interpreters import channels + rch = channels.RecvChannel({rch.id}) + sch = channels.SendChannel({sch.id}) + obj1 = rch.recv() + sch.send_nowait(2, unbound=channels.UNBOUND) + sch.send_nowait(obj1, unbound=channels.UNBOUND_REMOVE) + """)) + _run_output(interp2, dedent(f""" + from test.support.interpreters import channels + rch = channels.RecvChannel({rch.id}) + sch = channels.SendChannel({sch.id}) + obj2 = rch.recv() + obj1 = rch.recv() + """)) + self.assertEqual(_channels.get_count(rch.id), 0) + sch.send_nowait(3) + _run_output(interp1, dedent(""" + sch.send_nowait(4, unbound=channels.UNBOUND) + # interp closed here + sch.send_nowait(5, unbound=channels.UNBOUND_REMOVE) + sch.send_nowait(6, unbound=channels.UNBOUND) + """)) + _run_output(interp2, dedent(""" + sch.send_nowait(7, unbound=channels.UNBOUND_ERROR) + # interp closed here + sch.send_nowait(obj1, unbound=channels.UNBOUND_ERROR) + sch.send_nowait(obj2, unbound=channels.UNBOUND_REMOVE) + sch.send_nowait(8, unbound=channels.UNBOUND) + """)) + _run_output(interp1, dedent(""" + sch.send_nowait(9, unbound=channels.UNBOUND_REMOVE) + sch.send_nowait(10, unbound=channels.UNBOUND) + """)) + self.assertEqual(_channels.get_count(rch.id), 10) + + obj3 = rch.recv() + self.assertEqual(obj3, 3) + self.assertEqual(_channels.get_count(rch.id), 9) + + obj4 = rch.recv() + self.assertEqual(obj4, 4) + self.assertEqual(_channels.get_count(rch.id), 8) + + del interp1 + self.assertEqual(_channels.get_count(rch.id), 6) + + # obj5 was removed + + obj6 = rch.recv() + self.assertIs(obj6, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 5) + + obj7 = rch.recv() + self.assertEqual(obj7, 7) + self.assertEqual(_channels.get_count(rch.id), 4) + + del interp2 + self.assertEqual(_channels.get_count(rch.id), 3) + + # obj1 + with self.assertRaises(channels.ItemInterpreterDestroyed): + rch.recv() + self.assertEqual(_channels.get_count(rch.id), 2) + + # obj2 was removed + + obj8 = rch.recv() + self.assertIs(obj8, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 1) + + # obj9 was removed + + obj10 = rch.recv() + self.assertIs(obj10, channels.UNBOUND) + self.assertEqual(_channels.get_count(rch.id), 0) + if __name__ == '__main__': # Test needs to be a package, so we can do relative imports. diff --git a/Modules/_interpchannelsmodule.c b/Modules/_interpchannelsmodule.c index b519cb5b4061fc..b4e068a3498cb0 100644 --- a/Modules/_interpchannelsmodule.c +++ b/Modules/_interpchannelsmodule.c @@ -513,8 +513,14 @@ _waiting_finish_releasing(_waiting_t *waiting) 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; _PyCrossInterpreterData *data; _waiting_t *waiting; + int unboundop; struct _channelitem *next; } _channelitem; @@ -526,11 +532,22 @@ _channelitem_ID(_channelitem *item) static void _channelitem_init(_channelitem *item, - _PyCrossInterpreterData *data, _waiting_t *waiting) + int64_t interpid, _PyCrossInterpreterData *data, + _waiting_t *waiting, int unboundop) { + if (interpid < 0) { + interpid = _get_interpid(data); + } + else { + assert(data == NULL + || _PyCrossInterpreterData_INTERPID(data) < 0 + || interpid == _PyCrossInterpreterData_INTERPID(data)); + } *item = (_channelitem){ + .interpid = interpid, .data = data, .waiting = waiting, + .unboundop = unboundop, }; if (waiting != NULL) { waiting->itemid = _channelitem_ID(item); @@ -538,17 +555,15 @@ _channelitem_init(_channelitem *item, } static void -_channelitem_clear(_channelitem *item) +_channelitem_clear_data(_channelitem *item, int removed) { - 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 != NULL && removed) { if (item->waiting->status == WAITING_ACQUIRED) { _waiting_release(item->waiting, 0); } @@ -556,15 +571,23 @@ _channelitem_clear(_channelitem *item) } } +static void +_channelitem_clear(_channelitem *item) +{ + item->next = NULL; + _channelitem_clear_data(item, 1); +} + static _channelitem * -_channelitem_new(_PyCrossInterpreterData *data, _waiting_t *waiting) +_channelitem_new(int64_t interpid, _PyCrossInterpreterData *data, + _waiting_t *waiting, int unboundop) { _channelitem *item = GLOBAL_MALLOC(_channelitem); if (item == NULL) { PyErr_NoMemory(); return NULL; } - _channelitem_init(item, data, waiting); + _channelitem_init(item, interpid, data, waiting, unboundop); return item; } @@ -587,17 +610,48 @@ _channelitem_free_all(_channelitem *item) static void _channelitem_popped(_channelitem *item, - _PyCrossInterpreterData **p_data, _waiting_t **p_waiting) + _PyCrossInterpreterData **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(_PyCrossInterpreterData_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; @@ -636,9 +690,10 @@ _channelqueue_free(_channelqueue *queue) static int _channelqueue_put(_channelqueue *queue, - _PyCrossInterpreterData *data, _waiting_t *waiting) + int64_t interpid, _PyCrossInterpreterData *data, + _waiting_t *waiting, int unboundop) { - _channelitem *item = _channelitem_new(data, waiting); + _channelitem *item = _channelitem_new(interpid, data, waiting, unboundop); if (item == NULL) { return -1; } @@ -661,7 +716,8 @@ _channelqueue_put(_channelqueue *queue, static int _channelqueue_get(_channelqueue *queue, - _PyCrossInterpreterData **p_data, _waiting_t **p_waiting) + _PyCrossInterpreterData **p_data, _waiting_t **p_waiting, + int *p_unboundop) { _channelitem *item = queue->first; if (item == NULL) { @@ -673,7 +729,7 @@ _channelqueue_get(_channelqueue *queue, } queue->count -= 1; - _channelitem_popped(item, p_data, p_waiting); + _channelitem_popped(item, p_data, p_waiting, p_unboundop); return 0; } @@ -739,7 +795,8 @@ _channelqueue_remove(_channelqueue *queue, _channelitem_id_t itemid, } queue->count -= 1; - _channelitem_popped(item, p_data, p_waiting); + int unboundop; + _channelitem_popped(item, p_data, p_waiting, &unboundop); } static void @@ -750,14 +807,17 @@ _channelqueue_clear_interpreter(_channelqueue *queue, int64_t interpid) while (next != NULL) { _channelitem *item = next; next = item->next; - if (_PyCrossInterpreterData_INTERPID(item->data) == interpid) { + int remove = (item->interpid == interpid) + ? _channelitem_clear_interpreter(item) + : 0; + if (remove) { + _channelitem_free(item); if (prev == NULL) { - queue->first = item->next; + queue->first = next; } else { - prev->next = item->next; + prev->next = next; } - _channelitem_free(item); queue->count -= 1; } else { @@ -1020,12 +1080,15 @@ typedef struct _channel { PyThread_type_lock mutex; _channelqueue *queue; _channelends *ends; + struct { + int unboundop; + } defaults; int open; struct _channel_closing *closing; } _channel_state; static _channel_state * -_channel_new(PyThread_type_lock mutex) +_channel_new(PyThread_type_lock mutex, int unboundop) { _channel_state *chan = GLOBAL_MALLOC(_channel_state); if (chan == NULL) { @@ -1043,6 +1106,7 @@ _channel_new(PyThread_type_lock mutex) GLOBAL_FREE(chan); return NULL; } + chan->defaults.unboundop = unboundop; chan->open = 1; chan->closing = NULL; return chan; @@ -1063,7 +1127,8 @@ _channel_free(_channel_state *chan) static int _channel_add(_channel_state *chan, int64_t interpid, - _PyCrossInterpreterData *data, _waiting_t *waiting) + _PyCrossInterpreterData *data, _waiting_t *waiting, + int unboundop) { int res = -1; PyThread_acquire_lock(chan->mutex, WAIT_LOCK); @@ -1077,7 +1142,7 @@ _channel_add(_channel_state *chan, int64_t interpid, goto done; } - if (_channelqueue_put(chan->queue, data, waiting) != 0) { + if (_channelqueue_put(chan->queue, interpid, data, waiting, unboundop) != 0) { goto done; } // Any errors past this point must cause a _waiting_release() call. @@ -1090,7 +1155,8 @@ _channel_add(_channel_state *chan, int64_t interpid, static int _channel_next(_channel_state *chan, int64_t interpid, - _PyCrossInterpreterData **p_data, _waiting_t **p_waiting) + _PyCrossInterpreterData **p_data, _waiting_t **p_waiting, + int *p_unboundop) { int err = 0; PyThread_acquire_lock(chan->mutex, WAIT_LOCK); @@ -1104,11 +1170,15 @@ _channel_next(_channel_state *chan, int64_t interpid, goto done; } - int empty = _channelqueue_get(chan->queue, p_data, p_waiting); - assert(empty == 0 || empty == ERR_CHANNEL_EMPTY); + int empty = _channelqueue_get(chan->queue, p_data, p_waiting, p_unboundop); assert(!PyErr_Occurred()); - if (empty && chan->closing != NULL) { - chan->open = 0; + if (empty) { + assert(empty == ERR_CHANNEL_EMPTY); + if (chan->closing != NULL) { + chan->open = 0; + } + err = ERR_CHANNEL_EMPTY; + goto done; } done: @@ -1530,18 +1600,27 @@ _channels_release_cid_object(_channels *channels, int64_t cid) PyThread_release_lock(channels->mutex); } -static int64_t * +struct channel_id_and_info { + int64_t id; + int unboundop; +}; + +static struct channel_id_and_info * _channels_list_all(_channels *channels, int64_t *count) { - int64_t *cids = NULL; + struct channel_id_and_info *cids = NULL; PyThread_acquire_lock(channels->mutex, WAIT_LOCK); - int64_t *ids = PyMem_NEW(int64_t, (Py_ssize_t)(channels->numopen)); + struct channel_id_and_info *ids = + PyMem_NEW(struct channel_id_and_info, (Py_ssize_t)(channels->numopen)); if (ids == NULL) { goto done; } _channelref *ref = channels->head; for (int64_t i=0; ref != NULL; ref = ref->next, i++) { - ids[i] = ref->cid; + ids[i] = (struct channel_id_and_info){ + .id = ref->cid, + .unboundop = ref->chan->defaults.unboundop, + }; } *count = channels->numopen; @@ -1626,13 +1705,13 @@ _channel_finish_closing(_channel_state *chan) { // Create a new channel. static int64_t -channel_create(_channels *channels) +channel_create(_channels *channels, int unboundop) { PyThread_type_lock mutex = PyThread_allocate_lock(); if (mutex == NULL) { return ERR_CHANNEL_MUTEX_INIT; } - _channel_state *chan = _channel_new(mutex); + _channel_state *chan = _channel_new(mutex, unboundop); if (chan == NULL) { PyThread_free_lock(mutex); return -1; @@ -1664,7 +1743,7 @@ channel_destroy(_channels *channels, int64_t cid) // Optionally request to be notified when it is received. static int channel_send(_channels *channels, int64_t cid, PyObject *obj, - _waiting_t *waiting) + _waiting_t *waiting, int unboundop) { PyInterpreterState *interp = _get_current_interp(); if (interp == NULL) { @@ -1700,7 +1779,7 @@ channel_send(_channels *channels, int64_t cid, PyObject *obj, } // Add the data to the channel. - int res = _channel_add(chan, interpid, data, waiting); + int res = _channel_add(chan, interpid, data, waiting, unboundop); PyThread_release_lock(mutex); if (res != 0) { // We may chain an exception here: @@ -1737,7 +1816,7 @@ channel_clear_sent(_channels *channels, int64_t cid, _waiting_t *waiting) // Like channel_send(), but strictly wait for the object to be received. static int channel_send_wait(_channels *channels, int64_t cid, PyObject *obj, - PY_TIMEOUT_T timeout) + int unboundop, PY_TIMEOUT_T timeout) { // We use a stack variable here, so we must ensure that &waiting // is not held by any channel item at the point this function exits. @@ -1748,7 +1827,7 @@ channel_send_wait(_channels *channels, int64_t cid, PyObject *obj, } /* Queue up the object. */ - int res = channel_send(channels, cid, obj, &waiting); + int res = channel_send(channels, cid, obj, &waiting, unboundop); if (res < 0) { assert(waiting.status == WAITING_NO_STATUS); goto finally; @@ -1790,7 +1869,7 @@ channel_send_wait(_channels *channels, int64_t cid, PyObject *obj, // The current interpreter gets associated with the recv end of the channel. // XXX Support a "wait" mutex? static int -channel_recv(_channels *channels, int64_t cid, PyObject **res) +channel_recv(_channels *channels, int64_t cid, PyObject **res, int *p_unboundop) { int err; *res = NULL; @@ -1818,13 +1897,15 @@ channel_recv(_channels *channels, int64_t cid, PyObject **res) // Pop off the next item from the channel. _PyCrossInterpreterData *data = NULL; _waiting_t *waiting = NULL; - err = _channel_next(chan, interpid, &data, &waiting); + err = _channel_next(chan, interpid, &data, &waiting, p_unboundop); PyThread_release_lock(mutex); if (err != 0) { return err; } else if (data == NULL) { + // The item was unbound. assert(!PyErr_Occurred()); + *res = NULL; return 0; } @@ -2801,8 +2882,7 @@ channelsmod_create(PyObject *self, PyObject *args, PyObject *kwds) return NULL; } - // XXX Save unboundop. - int64_t cid = channel_create(&_globals.channels); + int64_t cid = channel_create(&_globals.channels, unboundop); if (cid < 0) { (void)handle_channel_error(-1, self, cid); return NULL; @@ -2864,7 +2944,8 @@ static PyObject * channelsmod_list_all(PyObject *self, PyObject *Py_UNUSED(ignored)) { int64_t count = 0; - int64_t *cids = _channels_list_all(&_globals.channels, &count); + struct channel_id_and_info *cids = + _channels_list_all(&_globals.channels, &count); if (cids == NULL) { if (count == 0) { return PyList_New(0); @@ -2881,22 +2962,20 @@ channelsmod_list_all(PyObject *self, PyObject *Py_UNUSED(ignored)) ids = NULL; goto finally; } - int64_t *cur = cids; + struct channel_id_and_info *cur = cids; for (int64_t i=0; i < count; cur++, i++) { PyObject *cidobj = NULL; - int err = newchannelid(state->ChannelIDType, *cur, 0, + int err = newchannelid(state->ChannelIDType, cur->id, 0, &_globals.channels, 0, 0, (channelid **)&cidobj); - if (handle_channel_error(err, self, *cur)) { + if (handle_channel_error(err, self, cur->id)) { assert(cidobj == NULL); Py_SETREF(ids, NULL); break; } assert(cidobj != NULL); - // XXX get unboundop - int unboundop = UNBOUND_REPLACE; - PyObject *item = Py_BuildValue("Oi", cidobj, unboundop); + PyObject *item = Py_BuildValue("Oi", cidobj, cur->unboundop); Py_DECREF(cidobj); if (item == NULL) { Py_SETREF(ids, NULL); @@ -2990,8 +3069,7 @@ channelsmod_send(PyObject *self, PyObject *args, PyObject *kwds) .module = self, }; PyObject *obj; - // XXX Make unboundop required. - int unboundop = UNBOUND_REMOVE; + int unboundop = UNBOUND_REPLACE; int blocking = 1; PyObject *timeout_obj = NULL; if (!PyArg_ParseTupleAndKeywords(args, kwds, "O&O|i$pO:channel_send", kwlist, @@ -3016,10 +3094,10 @@ channelsmod_send(PyObject *self, PyObject *args, PyObject *kwds) int err = 0; // XXX Store unboundop if (blocking) { - err = channel_send_wait(&_globals.channels, cid, obj, timeout); + err = channel_send_wait(&_globals.channels, cid, obj, unboundop, timeout); } else { - err = channel_send(&_globals.channels, cid, obj, NULL); + err = channel_send(&_globals.channels, cid, obj, NULL, unboundop); } if (handle_channel_error(err, self, cid)) { return NULL; @@ -3043,8 +3121,7 @@ channelsmod_send_buffer(PyObject *self, PyObject *args, PyObject *kwds) .module = self, }; PyObject *obj; - // XXX Make unboundop required. - int unboundop = UNBOUND_REMOVE; + int unboundop = UNBOUND_REPLACE; int blocking = 1; PyObject *timeout_obj = NULL; if (!PyArg_ParseTupleAndKeywords(args, kwds, @@ -3074,10 +3151,11 @@ channelsmod_send_buffer(PyObject *self, PyObject *args, PyObject *kwds) int err = 0; // XXX Store unboundop if (blocking) { - err = channel_send_wait(&_globals.channels, cid, tempobj, timeout); + err = channel_send_wait( + &_globals.channels, cid, tempobj, unboundop, timeout); } else { - err = channel_send(&_globals.channels, cid, tempobj, NULL); + err = channel_send(&_globals.channels, cid, tempobj, NULL, unboundop); } Py_DECREF(tempobj); if (handle_channel_error(err, self, cid)) { @@ -3109,22 +3187,21 @@ channelsmod_recv(PyObject *self, PyObject *args, PyObject *kwds) cid = cid_data.cid; PyObject *obj = NULL; - int err = channel_recv(&_globals.channels, cid, &obj); - if (handle_channel_error(err, self, cid)) { - return NULL; - } - Py_XINCREF(dflt); - if (obj == NULL) { + int unboundop = 0; + int err = channel_recv(&_globals.channels, cid, &obj, &unboundop); + if (err == ERR_CHANNEL_EMPTY && dflt != NULL) { // Use the default. - if (dflt == NULL) { - (void)handle_channel_error(ERR_CHANNEL_EMPTY, self, cid); - return NULL; - } obj = Py_NewRef(dflt); + err = 0; + } + else if (handle_channel_error(err, self, cid)) { + return NULL; + } + else if (obj == NULL) { + // The item was unbound. + return Py_BuildValue("Oi", Py_None, unboundop); } - Py_XDECREF(dflt); - // XXX return unboundop if unbound PyObject *res = Py_BuildValue("OO", obj, Py_None); Py_DECREF(obj); return res; @@ -3301,17 +3378,14 @@ channelsmod_get_channel_defaults(PyObject *self, PyObject *args, PyObject *kwds) } int64_t cid = cid_data.cid; - // XXX get stored defaults. -// _channel *channel = NULL; -// int err = _channels_lookup(&_globals.channels, cid, &channel); -// if (handle_channel_error(err, self, cid)) { -// return NULL; -// } -// int unboundop = channel->defaults.unboundop; -// _channel_unmark_waiter(channel, _globals.channels.mutex); - - (void)cid; - int unboundop = UNBOUND_REPLACE; + PyThread_type_lock mutex = NULL; + _channel_state *channel = NULL; + int err = _channels_lookup(&_globals.channels, cid, &mutex, &channel); + if (handle_channel_error(err, self, cid)) { + return NULL; + } + int unboundop = channel->defaults.unboundop; + PyThread_release_lock(mutex); PyObject *defaults = Py_BuildValue("i", unboundop); return defaults; diff --git a/Modules/_interpqueuesmodule.c b/Modules/_interpqueuesmodule.c index 8534be4ee2f5bc..5dec240f02c4db 100644 --- a/Modules/_interpqueuesmodule.c +++ b/Modules/_interpqueuesmodule.c @@ -60,20 +60,6 @@ _release_xid_data(_PyCrossInterpreterData *data, int flags) return res; } -static inline int64_t -_get_interpid(_PyCrossInterpreterData *data) -{ - int64_t interpid; - if (data != NULL) { - interpid = _PyCrossInterpreterData_INTERPID(data); - assert(!PyErr_Occurred()); - } - else { - interpid = PyInterpreterState_GetID(PyInterpreterState_Get()); - } - return interpid; -} - static PyInterpreterState * _get_current_interp(void) { diff --git a/Modules/_interpreters_common.h b/Modules/_interpreters_common.h index feb925063a66eb..0d2e0c9efd3837 100644 --- a/Modules/_interpreters_common.h +++ b/Modules/_interpreters_common.h @@ -21,6 +21,21 @@ clear_xid_class(PyTypeObject *cls) #endif +static inline int64_t +_get_interpid(_PyCrossInterpreterData *data) +{ + int64_t interpid; + if (data != NULL) { + interpid = _PyCrossInterpreterData_INTERPID(data); + assert(!PyErr_Occurred()); + } + else { + interpid = PyInterpreterState_GetID(PyInterpreterState_Get()); + } + return interpid; +} + + /* unbound items ************************************************************/ #ifdef HAS_UNBOUND_ITEMS From 7fe548d79d49a37ca22fc35f33089c9441e20413 Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Mon, 15 Jul 2024 13:01:39 -0600 Subject: [PATCH 6/7] Drop some outdated TODO comments. --- Modules/_interpchannelsmodule.c | 2 -- 1 file changed, 2 deletions(-) diff --git a/Modules/_interpchannelsmodule.c b/Modules/_interpchannelsmodule.c index b4e068a3498cb0..a8b4a8d76b0eaa 100644 --- a/Modules/_interpchannelsmodule.c +++ b/Modules/_interpchannelsmodule.c @@ -3092,7 +3092,6 @@ channelsmod_send(PyObject *self, PyObject *args, PyObject *kwds) /* Queue up the object. */ int err = 0; - // XXX Store unboundop if (blocking) { err = channel_send_wait(&_globals.channels, cid, obj, unboundop, timeout); } @@ -3149,7 +3148,6 @@ channelsmod_send_buffer(PyObject *self, PyObject *args, PyObject *kwds) /* Queue up the object. */ int err = 0; - // XXX Store unboundop if (blocking) { err = channel_send_wait( &_globals.channels, cid, tempobj, unboundop, timeout); From 971011b8c00397b3cc88d8d8b7d05d5fc0d36c8a Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Mon, 15 Jul 2024 13:16:56 -0600 Subject: [PATCH 7/7] Fix WASI. --- Lib/test/test__interpchannels.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Lib/test/test__interpchannels.py b/Lib/test/test__interpchannels.py index de9f62a98c2a24..4a7f04b9df9843 100644 --- a/Lib/test/test__interpchannels.py +++ b/Lib/test/test__interpchannels.py @@ -8,6 +8,7 @@ from test.support import import_helper +_channels = import_helper.import_module('_interpchannels') from test.support.interpreters import _crossinterp from test.test__interpreters import ( _interpreters, @@ -15,8 +16,6 @@ clean_up_interpreters, ) -_channels = import_helper.import_module('_interpchannels') - REPLACE = _crossinterp._UNBOUND_CONSTANT_TO_FLAG[_crossinterp.UNBOUND]