support for Python 3.11

[hroncok]

No description provided.

[hroncok]

Unfortunately, I still get errors/failures like:

______ ERROR at teardown of AsyncResultTest.test_error_engine_info_apply _______
ipyparallel/tests/clienttest.py:202: in tearDown
self.client[:].use_pickle()
ipyparallel/client/client.py:1350: in __getitem__
return self.direct_view(key)
ipyparallel/client/client.py:2080: in direct_view
targets = self._build_targets(targets)[1]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <ipyparallel.client.client.Client object at 0x7f3740211ed0>
targets = slice(None, None, None)
def _build_targets(self, targets):
"""Turn valid target IDs or 'all' into two lists:
(int_ids, uuids).
"""
if not self._ids:
# flush notification socket if no engines yet, just in case
if not self.ids:
> raise error.NoEnginesRegistered(
"Can't build targets without any engines"
)
E ipyparallel.error.NoEnginesRegistered: This operation requires engines. Try client.wait_for_engines(n) to wait for engines to register.
ipyparallel/client/client.py:706: NoEnginesRegistered
---------------------------- Captured log teardown -----------------------------
ERROR asyncio:base_events.py:1757 Exception in callback BaseAsyncIOLoop._handle_events(11, 1)
handle: <Handle BaseAsyncIOLoop._handle_events(11, 1)>
Traceback (most recent call last):
File "/usr/lib64/python3.11/asyncio/events.py", line 80, in _run
self._context.run(self._callback, *self._args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/tornado/platform/asyncio.py", line 189, in _handle_events
handler_func(fileobj, events)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/eventloop/zmqstream.py", line 445, in _handle_events
zmq_events = self.socket.EVENTS
^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 51, in __getattr__
return self._get_attr_opt(upper_key, opt)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 63, in _get_attr_opt
return self.get(opt)
^^^^^^^^^^^^^
File "zmq/backend/cython/socket.pyx", line 464, in zmq.backend.cython.socket.Socket.get
File "zmq/backend/cython/socket.pyx", line 135, in zmq.backend.cython.socket._check_closed
zmq.error.ZMQError: Socket operation on non-socket
ERROR asyncio:base_events.py:1757 Exception in callback BaseAsyncIOLoop._handle_events(13, 1)
handle: <Handle BaseAsyncIOLoop._handle_events(13, 1)>
Traceback (most recent call last):
File "/usr/lib64/python3.11/asyncio/events.py", line 80, in _run
self._context.run(self._callback, *self._args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/tornado/platform/asyncio.py", line 189, in _handle_events
handler_func(fileobj, events)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/eventloop/zmqstream.py", line 445, in _handle_events
zmq_events = self.socket.EVENTS
^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 51, in __getattr__
return self._get_attr_opt(upper_key, opt)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 63, in _get_attr_opt
return self.get(opt)
^^^^^^^^^^^^^
File "zmq/backend/cython/socket.pyx", line 464, in zmq.backend.cython.socket.Socket.get
File "zmq/backend/cython/socket.pyx", line 135, in zmq.backend.cython.socket._check_closed
zmq.error.ZMQError: Socket operation on non-socket
ERROR asyncio:base_events.py:1757 Exception in callback BaseAsyncIOLoop._handle_events(14, 1)
handle: <Handle BaseAsyncIOLoop._handle_events(14, 1)>
Traceback (most recent call last):
File "/usr/lib64/python3.11/asyncio/events.py", line 80, in _run
self._context.run(self._callback, *self._args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/tornado/platform/asyncio.py", line 189, in _handle_events
handler_func(fileobj, events)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/eventloop/zmqstream.py", line 445, in _handle_events
zmq_events = self.socket.EVENTS
^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 51, in __getattr__
return self._get_attr_opt(upper_key, opt)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/site-packages/zmq/sugar/attrsettr.py", line 63, in _get_attr_opt
return self.get(opt)
^^^^^^^^^^^^^
File "zmq/backend/cython/socket.pyx", line 464, in zmq.backend.cython.socket.Socket.get
File "zmq/backend/cython/socket.pyx", line 135, in zmq.backend.cython.socket._check_closed
zmq.error.ZMQError: Socket operation on non-socket
=================================== FAILURES ===================================
_________________ AsyncResultTest.test_error_engine_info_apply _________________
self = <ipyparallel.tests.test_asyncresult.AsyncResultTest testMethod=test_error_engine_info_apply>
def test_error_engine_info_apply(self):
dv = self.client[:]
targets = self.client.ids
ar = dv.apply_async(lambda: 1 / 0)
try:
ar.get()
except Exception as e:
exc = e
else:
pytest.fail("Should have raised remote ZeroDivisionError")
assert isinstance(exc, ipp.error.CompositeError)
expected_engine_info = [
{
"engine_id": engine_id,
"engine_uuid": self.client._engines[engine_id],
"method": "apply",
}
for engine_id in self.client.ids
]
engine_infos = [e[-1] for e in exc.elist]
> assert engine_infos == expected_engine_info
E assert [{}, {}] == []
E Left contains 2 more items, first extra item: {}
E Full diff:
E - []
E + [{}, {}]
ipyparallel/tests/test_asyncresult.py:527: AssertionError

Or

______________ AsyncResultTest.test_return_exceptions_postmortem _______________
etype = <class 'ValueError'>
@contextmanager
def raises_remote(etype):
if isinstance(etype, str):
# allow Exception or 'Exception'
expected_ename = etype
else:
expected_ename = etype.__name__
try:
try:
> yield
ipyparallel/tests/clienttest.py:105: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <ipyparallel.tests.test_asyncresult.AsyncResultTest testMethod=test_return_exceptions_postmortem>
def test_return_exceptions_postmortem(self):
self.minimum_engines(3)
dv = self.client[:]
bad_id = dv.targets[1]
dv.scatter("rank", dv.targets, flatten=True)
def fail_on_bad_id(rank, bad_id):
if rank == bad_id:
raise ValueError(f"{rank} is bad!")
return rank
ar = dv.apply_async(fail_on_bad_id, ipp.Reference('rank'), bad_id)
with raises_remote(ValueError):
> ar.get()
ipyparallel/tests/test_asyncresult.py:419: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <AsyncResult(fail_on_bad_id): failed>, timeout = None
return_exceptions = False, return_when = None
def get(self, timeout=None, return_exceptions=None, return_when=None):
"""Return the result when it arrives.
Arguments:
timeout : int [default None]
If `timeout` is not ``None`` and the result does not arrive within
`timeout` seconds then ``TimeoutError`` is raised. If the
remote call raised an exception then that exception will be reraised
by get() inside a `RemoteError`.
return_exceptions : bool [default False]
If True, return Exceptions instead of raising them.
return_when : None, ALL_COMPLETED, or FIRST_EXCEPTION
FIRST_COMPLETED is not supported, and treated the same as ALL_COMPLETED.
See :py:func:`concurrent.futures.wait` for documentation.
When return_when=FIRST_EXCEPTION, will raise immediately on the first exception,
rather than waiting for all results to finish before reporting errors.
.. versionchanged:: 8.0
Added `return_when` argument.
"""
if return_when == FIRST_COMPLETED:
# FIRST_COMPLETED unsupported, same as ALL_COMPLETED
warnings.warn(
"Ignoring unsupported AsyncResult.get(return_when=FIRST_COMPLETED)",
UserWarning,
stacklevel=2,
)
return_when = None
elif return_when == ALL_COMPLETED:
# None avoids call to .split() and is a tiny bit more efficient
return_when = None
if not self.ready():
wait_result = self.wait(timeout, return_when=return_when)
if return_exceptions is None:
# default to attribute, if AsyncResult was created with return_exceptions=True
return_exceptions = self._return_exceptions
if self._ready:
if self._success:
return self.result()
else:
e = self.exception()
if return_exceptions:
return self._reconstruct_result(self._raw_results)
else:
> raise e
ipyparallel/client/asyncresult.py:367: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <AsyncResult(fail_on_bad_id): failed>
f = <Future at 0x7f2152fe4250 state=finished returned list>
def _resolve_result(self, f=None):
if self.done():
return
if f:
results = f.result()
else:
results = list(map(self._client.results.get, self.msg_ids))
# store raw results
self._raw_results = results
try:
if self._single_result:
r = results[0]
if isinstance(r, Exception):
raise r
else:
> results = self._collect_exceptions(results)
ipyparallel/client/asyncresult.py:540: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <AsyncResult(fail_on_bad_id): failed>
results = [0, <RemoteErrorWithTB[ ]:EngineError(Engine 1 died while running task '198305e2-ae57d6bd4246c55f7d4bc203_3608_7')>, 2]
def _collect_exceptions(self, results):
"""Wrap Exceptions in a CompositeError
if self._return_exceptions is True, this is a no-op
"""
if self._return_exceptions:
return results
else:
> return error.collect_exceptions(results, self._fname)
ipyparallel/client/asyncresult.py:556: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ rdict_or_list = [0, <RemoteErrorWithTB[ ]:EngineError(Engine 1 died while running task '198305e2-ae57d6bd4246c55f7d4bc203_3608_7')>, 2]
method = 'fail_on_bad_id'
def collect_exceptions(rdict_or_list, method='unspecified'):
"""check a result dict for errors, and raise CompositeError if any exist.
Passthrough otherwise."""
elist = []
if isinstance(rdict_or_list, dict):
rlist = rdict_or_list.values()
else:
rlist = rdict_or_list
for r in rlist:
if isinstance(r, RemoteError):
en, ev, etb, ei = r.ename, r.evalue, r.traceback, r.engine_info
# Sometimes we could have CompositeError in our list. Just take
# the errors out of them and put them in our new list. This
# has the effect of flattening lists of CompositeErrors into one
# CompositeError
if en == 'CompositeError':
for e in ev.elist:
elist.append(e)
else:
elist.append((en, ev, etb, ei))
if len(elist) == 0:
return rdict_or_list
else:
msg = "one or more exceptions raised in: %s" % (method)
err = CompositeError(msg, elist)
> raise err
E ipyparallel.error.CompositeError: one or more exceptions raised in: fail_on_bad_id
E [Engine Exception]EngineError: Engine 1 died while running task '198305e2-ae57d6bd4246c55f7d4bc203_3608_7'
ipyparallel/error.py:248: CompositeError
During handling of the above exception, another exception occurred:
etype = <class 'ValueError'>
@contextmanager
def raises_remote(etype):
if isinstance(etype, str):
# allow Exception or 'Exception'
expected_ename = etype
else:
expected_ename = etype.__name__
try:
try:
yield
except error.AlreadyDisplayedError as e:
e.original_error.raise_exception()
except error.CompositeError as e:
> e.raise_exception()
ipyparallel/tests/clienttest.py:109: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = CompositeError(1), excid = 0
def raise_exception(self, excid=0):
try:
en, ev, etb, ei = self.elist[excid]
except:
raise IndexError("an exception with index %i does not exist" % excid)
else:
> raise RemoteError(en, ev, etb, ei)
E ipyparallel.tests.RemoteErrorWithTB: [Engine Exception] EngineError: Engine 1 died while running task '198305e2-ae57d6bd4246c55f7d4bc203_3608_7'
E Traceback (most recent call last):
E E File "/builddir/build/BUILD/ipyparallel-8.2.1/ipyparallel/client/client.py", line 894, in _handle_stranded_msgs
E raise error.EngineError(
E ^^^^^^^^^^^^^^^^^^^^^^^^
E E ipyparallel.error.EngineError: Engine 1 died while running task '198305e2-ae57d6bd4246c55f7d4bc203_3608_7'
ipyparallel/error.py:201: RemoteErrorWithTB
During handling of the above exception, another exception occurred:
self = <ipyparallel.tests.test_asyncresult.AsyncResultTest testMethod=test_return_exceptions_postmortem>
def test_return_exceptions_postmortem(self):
self.minimum_engines(3)
dv = self.client[:]
bad_id = dv.targets[1]
dv.scatter("rank", dv.targets, flatten=True)
def fail_on_bad_id(rank, bad_id):
if rank == bad_id:
raise ValueError(f"{rank} is bad!")
return rank
ar = dv.apply_async(fail_on_bad_id, ipp.Reference('rank'), bad_id)
> with raises_remote(ValueError):
ipyparallel/tests/test_asyncresult.py:418: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ /usr/lib64/python3.11/contextlib.py:155: in __exit__
self.gen.throw(typ, value, traceback)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ etype = <class 'ValueError'>
@contextmanager
def raises_remote(etype):
if isinstance(etype, str):
# allow Exception or 'Exception'
expected_ename = etype
else:
expected_ename = etype.__name__
try:
try:
yield
except error.AlreadyDisplayedError as e:
e.original_error.raise_exception()
except error.CompositeError as e:
e.raise_exception()
except error.RemoteError as e:
> assert (
expected_ename == e.ename
), f"Should have raised {expected_ename}, but raised {e.ename}"
E AssertionError: Should have raised ValueError, but raised EngineError
ipyparallel/tests/clienttest.py:111: AssertionError

A lot of stuff like:

2022-06-17 15:24:28.046 [KernelNanny.0] Pipe closed, parent 3645 has status: zombie

I ama afraid I don'T know how to debug this :(

[hroncok]

Thanks

[minrk]

Managed to track it down to rendering tracebacks of deserialized functions:

importtracebackfromipyparallel.serializeimportdeserialize_object, serialize_objectdeffail():
1/0pfail=serialize_object(fail)
fail2, _=deserialize_object(pfail)
try:
fail2()
exceptExceptionase:
traceback.print_exc()

That print_exc() fails with:

Traceback (most recent call last):
File "/Users/minrk/dev/ip/parallel/test.py", line 15, in <module>
fail2()
^^^^^^^
File "/Users/minrk/dev/ip/parallel/test.py", line -1, in fail
ZeroDivisionError: division by zero
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 353, in _walk_tb_with_full_positions
positions = _get_code_position(tb.tb_frame.f_code, tb.tb_lasti)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 367, in _get_code_position
return next(itertools.islice(positions_gen, instruction_index // 2, None))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
StopIteration
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/Users/minrk/dev/ip/parallel/test.py", line 17, in <module>
traceback.print_exc()
^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 183, in print_exc
print_exception(*sys.exc_info(), limit=limit, file=file, chain=chain)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 124, in print_exception
te = TracebackException(type(value), value, tb, limit=limit, compact=True)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 684, in __init__
self.stack = StackSummary._extract_from_extended_frame_gen(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/traceback.py", line 416, in _extract_from_extended_frame_gen
for f, (lineno, end_lineno, colno, end_colno) in frame_gen:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: generator raised StopIteration

Ultimately, because similar code is called in log.exception() for unhandled errors, the whole engine crashes when any error is raised.

Maybe you know what the right fix is for that? Seems like there's something to add in the deserialize code.

[hroncok]

Maybe you know what the right fix is for that?

Sorry, no idea yet.

[minrk]

Oops, committed some debug prints. Will clean those out later

[hroncok]

The code that fails was added in python/cpython#26958 (Print columns in tracebacks (PEP 657)).

I suppose @ammaraskar might be able to help here. It seems the position information is missing when ipyparallel serializes and deserializes the function, but I am not sure how exactly to access (and store) this information.

[hroncok]

The current version seems to work for us in Fedora, thanks!

[minrk]

Thanks!