Compare commits
No commits in common. "b209990d045c9069b11279916dc529bbe3152b0b" and "eec240a70a6fea5412f970a22c6b91fc3472410c" have entirely different histories.
b209990d04
...
eec240a70a
|
|
@ -38,7 +38,6 @@ async def main():
|
|||
"""
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
# loglevel='runtime',
|
||||
) as n:
|
||||
|
||||
# Spawn both actors, don't bother with collecting results
|
||||
|
|
|
|||
|
|
@ -3,20 +3,17 @@ import tractor
|
|||
|
||||
|
||||
async def breakpoint_forever():
|
||||
'''
|
||||
Indefinitely re-enter debugger in child actor.
|
||||
|
||||
'''
|
||||
"""Indefinitely re-enter debugger in child actor.
|
||||
"""
|
||||
while True:
|
||||
await trio.sleep(0.1)
|
||||
await tractor.pause()
|
||||
await tractor.breakpoint()
|
||||
|
||||
|
||||
async def main():
|
||||
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
loglevel='cancel',
|
||||
) as n:
|
||||
|
||||
portal = await n.run_in_actor(
|
||||
|
|
|
|||
|
|
@ -246,10 +246,10 @@ def test_simple_context(
|
|||
trio.run(main)
|
||||
except error_parent:
|
||||
pass
|
||||
except BaseExceptionGroup as beg:
|
||||
except trio.MultiError as me:
|
||||
# XXX: on windows it seems we may have to expect the group error
|
||||
from tractor._exceptions import is_multi_cancelled
|
||||
assert is_multi_cancelled(beg)
|
||||
assert is_multi_cancelled(me)
|
||||
else:
|
||||
trio.run(main)
|
||||
|
||||
|
|
|
|||
|
|
@ -38,13 +38,10 @@ async def async_gen_stream(sequence):
|
|||
assert cs.cancelled_caught
|
||||
|
||||
|
||||
# TODO: deprecated either remove entirely
|
||||
# or re-impl in terms of `MsgStream` one-sides
|
||||
# wrapper, but at least remove `Portal.open_stream_from()`
|
||||
@tractor.stream
|
||||
async def context_stream(
|
||||
ctx: tractor.Context,
|
||||
sequence: list[int],
|
||||
sequence
|
||||
):
|
||||
for i in sequence:
|
||||
await ctx.send_yield(i)
|
||||
|
|
|
|||
|
|
@ -36,7 +36,6 @@ def parse_ipaddr(arg):
|
|||
|
||||
|
||||
if __name__ == "__main__":
|
||||
__tracebackhide__: bool = True
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--uid", type=parse_uid)
|
||||
|
|
|
|||
|
|
@ -47,7 +47,6 @@ import trio
|
|||
from ._exceptions import (
|
||||
ContextCancelled,
|
||||
InternalError,
|
||||
MsgTypeError,
|
||||
RemoteActorError,
|
||||
StreamOverrun,
|
||||
pack_from_raise,
|
||||
|
|
@ -60,14 +59,12 @@ from .msg import (
|
|||
MsgType,
|
||||
MsgCodec,
|
||||
NamespacePath,
|
||||
PayloadT,
|
||||
Return,
|
||||
Started,
|
||||
Stop,
|
||||
Yield,
|
||||
current_codec,
|
||||
pretty_struct,
|
||||
types as msgtypes,
|
||||
)
|
||||
from ._ipc import Channel
|
||||
from ._streaming import MsgStream
|
||||
|
|
@ -91,10 +88,7 @@ async def _drain_to_final_msg(
|
|||
hide_tb: bool = True,
|
||||
msg_limit: int = 6,
|
||||
|
||||
) -> tuple[
|
||||
Return|None,
|
||||
list[MsgType]
|
||||
]:
|
||||
) -> list[dict]:
|
||||
'''
|
||||
Drain IPC msgs delivered to the underlying rx-mem-chan
|
||||
`Context._recv_chan` from the runtime in search for a final
|
||||
|
|
@ -115,7 +109,6 @@ async def _drain_to_final_msg(
|
|||
# basically ignoring) any bi-dir-stream msgs still in transit
|
||||
# from the far end.
|
||||
pre_result_drained: list[MsgType] = []
|
||||
return_msg: Return|None = None
|
||||
while not (
|
||||
ctx.maybe_error
|
||||
and not ctx._final_result_is_set()
|
||||
|
|
@ -176,6 +169,8 @@ async def _drain_to_final_msg(
|
|||
# pray to the `trio` gawds that we're corrent with this
|
||||
# msg: dict = await ctx._recv_chan.receive()
|
||||
msg: MsgType = await ctx._recv_chan.receive()
|
||||
# always capture unexpected/non-result msgs
|
||||
pre_result_drained.append(msg)
|
||||
|
||||
# NOTE: we get here if the far end was
|
||||
# `ContextCancelled` in 2 cases:
|
||||
|
|
@ -212,13 +207,11 @@ async def _drain_to_final_msg(
|
|||
# if ctx._recv_chan:
|
||||
# await ctx._recv_chan.aclose()
|
||||
# TODO: ^ we don't need it right?
|
||||
return_msg = msg
|
||||
break
|
||||
|
||||
# far end task is still streaming to us so discard
|
||||
# and report depending on local ctx state.
|
||||
case Yield():
|
||||
pre_result_drained.append(msg)
|
||||
if (
|
||||
(ctx._stream.closed
|
||||
and (reason := 'stream was already closed')
|
||||
|
|
@ -243,10 +236,7 @@ async def _drain_to_final_msg(
|
|||
|
||||
f'{pformat(msg)}\n'
|
||||
)
|
||||
return (
|
||||
return_msg,
|
||||
pre_result_drained,
|
||||
)
|
||||
return pre_result_drained
|
||||
|
||||
# drain up to the `msg_limit` hoping to get
|
||||
# a final result or error/ctxc.
|
||||
|
|
@ -270,7 +260,6 @@ async def _drain_to_final_msg(
|
|||
# -[ ] should be a runtime error if a stream is open right?
|
||||
# Stop()
|
||||
case Stop():
|
||||
pre_result_drained.append(msg)
|
||||
log.cancel(
|
||||
'Remote stream terminated due to "stop" msg:\n\n'
|
||||
f'{pformat(msg)}\n'
|
||||
|
|
@ -280,6 +269,7 @@ async def _drain_to_final_msg(
|
|||
# remote error msg, likely already handled inside
|
||||
# `Context._deliver_msg()`
|
||||
case Error():
|
||||
|
||||
# TODO: can we replace this with `ctx.maybe_raise()`?
|
||||
# -[ ] would this be handier for this case maybe?
|
||||
# async with maybe_raise_on_exit() as raises:
|
||||
|
|
@ -346,7 +336,6 @@ async def _drain_to_final_msg(
|
|||
# XXX should pretty much never get here unless someone
|
||||
# overrides the default `MsgType` spec.
|
||||
case _:
|
||||
pre_result_drained.append(msg)
|
||||
# It's definitely an internal error if any other
|
||||
# msg type without a`'cid'` field arrives here!
|
||||
if not msg.cid:
|
||||
|
|
@ -363,10 +352,7 @@ async def _drain_to_final_msg(
|
|||
f'{ctx.outcome}\n'
|
||||
)
|
||||
|
||||
return (
|
||||
return_msg,
|
||||
pre_result_drained,
|
||||
)
|
||||
return pre_result_drained
|
||||
|
||||
|
||||
class Unresolved:
|
||||
|
|
@ -733,36 +719,21 @@ class Context:
|
|||
Return string indicating which task this instance is wrapping.
|
||||
|
||||
'''
|
||||
return 'parent' if self._portal else 'child'
|
||||
return 'caller' if self._portal else 'callee'
|
||||
|
||||
@staticmethod
|
||||
def peer_side(side: str) -> str:
|
||||
match side:
|
||||
case 'child':
|
||||
return 'parent'
|
||||
case 'parent':
|
||||
return 'child'
|
||||
|
||||
# TODO: remove stat!
|
||||
# -[ ] re-implement the `.experiemental._pubsub` stuff
|
||||
# with `MsgStream` and that should be last usage?
|
||||
# -[ ] remove from `tests/legacy_one_way_streaming.py`!
|
||||
async def send_yield(
|
||||
self,
|
||||
data: Any,
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
Deprecated method for what now is implemented in `MsgStream`.
|
||||
|
||||
We need to rework / remove some stuff tho, see above.
|
||||
|
||||
'''
|
||||
warnings.warn(
|
||||
"`Context.send_yield()` is now deprecated. "
|
||||
"Use ``MessageStream.send()``. ",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
# await self.chan.send({'yield': data, 'cid': self.cid})
|
||||
await self.chan.send(
|
||||
Yield(
|
||||
cid=self.cid,
|
||||
|
|
@ -771,11 +742,12 @@ class Context:
|
|||
)
|
||||
|
||||
async def send_stop(self) -> None:
|
||||
'''
|
||||
Terminate a `MsgStream` dialog-phase by sending the IPC
|
||||
equiv of a `StopIteration`.
|
||||
|
||||
'''
|
||||
# await pause()
|
||||
# await self.chan.send({
|
||||
# # Stop(
|
||||
# 'stop': True,
|
||||
# 'cid': self.cid
|
||||
# })
|
||||
await self.chan.send(
|
||||
Stop(cid=self.cid)
|
||||
)
|
||||
|
|
@ -871,7 +843,6 @@ class Context:
|
|||
|
||||
# self-cancel (ack) or,
|
||||
# peer propagated remote cancellation.
|
||||
msgtyperr: bool = False
|
||||
if isinstance(error, ContextCancelled):
|
||||
|
||||
whom: str = (
|
||||
|
|
@ -883,16 +854,6 @@ class Context:
|
|||
f'{error}'
|
||||
)
|
||||
|
||||
elif isinstance(error, MsgTypeError):
|
||||
msgtyperr = True
|
||||
peer_side: str = self.peer_side(self.side)
|
||||
log.error(
|
||||
f'IPC dialog error due to msg-type caused by {peer_side!r} side\n\n'
|
||||
|
||||
f'{error}\n'
|
||||
f'{pformat(self)}\n'
|
||||
)
|
||||
|
||||
else:
|
||||
log.error(
|
||||
f'Remote context error:\n\n'
|
||||
|
|
@ -933,9 +894,9 @@ class Context:
|
|||
# if `._cancel_called` then `.cancel_acked and .cancel_called`
|
||||
# always should be set.
|
||||
and not self._is_self_cancelled()
|
||||
|
||||
and not cs.cancel_called
|
||||
and not cs.cancelled_caught
|
||||
and not msgtyperr
|
||||
):
|
||||
# TODO: it'd sure be handy to inject our own
|
||||
# `trio.Cancelled` subtype here ;)
|
||||
|
|
@ -1040,7 +1001,7 @@ class Context:
|
|||
# when the runtime finally receives it during teardown
|
||||
# (normally in `.result()` called from
|
||||
# `Portal.open_context().__aexit__()`)
|
||||
if side == 'parent':
|
||||
if side == 'caller':
|
||||
if not self._portal:
|
||||
raise InternalError(
|
||||
'No portal found!?\n'
|
||||
|
|
@ -1462,10 +1423,7 @@ class Context:
|
|||
# wait for a final context result/error by "draining"
|
||||
# (by more or less ignoring) any bi-dir-stream "yield"
|
||||
# msgs still in transit from the far end.
|
||||
(
|
||||
return_msg,
|
||||
drained_msgs,
|
||||
) = await _drain_to_final_msg(
|
||||
drained_msgs: list[dict] = await _drain_to_final_msg(
|
||||
ctx=self,
|
||||
hide_tb=hide_tb,
|
||||
)
|
||||
|
|
@ -1483,10 +1441,7 @@ class Context:
|
|||
|
||||
log.cancel(
|
||||
'Ctx drained pre-result msgs:\n'
|
||||
f'{pformat(drained_msgs)}\n\n'
|
||||
|
||||
f'Final return msg:\n'
|
||||
f'{return_msg}\n'
|
||||
f'{pformat(drained_msgs)}'
|
||||
)
|
||||
|
||||
self.maybe_raise(
|
||||
|
|
@ -1653,13 +1608,7 @@ class Context:
|
|||
|
||||
async def started(
|
||||
self,
|
||||
|
||||
# TODO: how to type this so that it's the
|
||||
# same as the payload type? Is this enough?
|
||||
value: PayloadT|None = None,
|
||||
|
||||
strict_parity: bool = False,
|
||||
complain_no_parity: bool = True,
|
||||
value: Any | None = None
|
||||
|
||||
) -> None:
|
||||
'''
|
||||
|
|
@ -1680,7 +1629,7 @@ class Context:
|
|||
f'called `.started()` twice on context with {self.chan.uid}'
|
||||
)
|
||||
|
||||
started_msg = Started(
|
||||
started = Started(
|
||||
cid=self.cid,
|
||||
pld=value,
|
||||
)
|
||||
|
|
@ -1701,54 +1650,28 @@ class Context:
|
|||
# https://zguide.zeromq.org/docs/chapter7/#The-Cheap-or-Nasty-Pattern
|
||||
#
|
||||
codec: MsgCodec = current_codec()
|
||||
msg_bytes: bytes = codec.encode(started_msg)
|
||||
msg_bytes: bytes = codec.encode(started)
|
||||
try:
|
||||
# be a "cheap" dialog (see above!)
|
||||
if (
|
||||
strict_parity
|
||||
or
|
||||
complain_no_parity
|
||||
):
|
||||
rt_started: Started = codec.decode(msg_bytes)
|
||||
rt_started = codec.decode(msg_bytes)
|
||||
if rt_started != started:
|
||||
|
||||
# XXX something is prolly totes cucked with the
|
||||
# codec state!
|
||||
if isinstance(rt_started, dict):
|
||||
rt_started = msgtypes.from_dict_msg(
|
||||
dict_msg=rt_started,
|
||||
)
|
||||
raise RuntimeError(
|
||||
'Failed to roundtrip `Started` msg?\n'
|
||||
f'{pformat(rt_started)}\n'
|
||||
)
|
||||
# TODO: break these methods out from the struct subtype?
|
||||
diff = pretty_struct.Struct.__sub__(rt_started, started)
|
||||
|
||||
if rt_started != started_msg:
|
||||
# TODO: break these methods out from the struct subtype?
|
||||
complaint: str = (
|
||||
'Started value does not match after codec rountrip?\n\n'
|
||||
f'{diff}'
|
||||
)
|
||||
# TODO: rn this will pretty much always fail with
|
||||
# any other sequence type embeded in the
|
||||
# payload...
|
||||
if self._strict_started:
|
||||
raise ValueError(complaint)
|
||||
else:
|
||||
log.warning(complaint)
|
||||
|
||||
diff = pretty_struct.Struct.__sub__(
|
||||
rt_started,
|
||||
started_msg,
|
||||
)
|
||||
complaint: str = (
|
||||
'Started value does not match after codec rountrip?\n\n'
|
||||
f'{diff}'
|
||||
)
|
||||
|
||||
# TODO: rn this will pretty much always fail with
|
||||
# any other sequence type embeded in the
|
||||
# payload...
|
||||
if (
|
||||
self._strict_started
|
||||
or
|
||||
strict_parity
|
||||
):
|
||||
raise ValueError(complaint)
|
||||
else:
|
||||
log.warning(complaint)
|
||||
|
||||
# started_msg = rt_started
|
||||
|
||||
await self.chan.send(started_msg)
|
||||
await self.chan.send(rt_started)
|
||||
|
||||
# raise any msg type error NO MATTER WHAT!
|
||||
except msgspec.ValidationError as verr:
|
||||
|
|
@ -1759,7 +1682,7 @@ class Context:
|
|||
src_validation_error=verr,
|
||||
verb_header='Trying to send payload'
|
||||
# > 'invalid `Started IPC msgs\n'
|
||||
) from verr
|
||||
)
|
||||
|
||||
self._started_called = True
|
||||
|
||||
|
|
@ -1860,17 +1783,13 @@ class Context:
|
|||
else:
|
||||
log_meth = log.runtime
|
||||
|
||||
side: str = self.side
|
||||
|
||||
peer_side: str = self.peer_side(side)
|
||||
|
||||
log_meth(
|
||||
f'Delivering IPC ctx error from {peer_side!r} to {side!r} task\n\n'
|
||||
f'Delivering error-msg to caller\n\n'
|
||||
|
||||
f'<= peer {peer_side!r}: {from_uid}\n'
|
||||
f'<= peer: {from_uid}\n'
|
||||
f' |_ {nsf}()\n\n'
|
||||
|
||||
f'=> {side!r} cid: {cid}\n'
|
||||
f'=> cid: {cid}\n'
|
||||
f' |_{self._task}\n\n'
|
||||
|
||||
f'{pformat(re)}\n'
|
||||
|
|
@ -1885,7 +1804,6 @@ class Context:
|
|||
self._maybe_cancel_and_set_remote_error(re)
|
||||
|
||||
# XXX only case where returning early is fine!
|
||||
structfmt = pretty_struct.Struct.pformat
|
||||
if self._in_overrun:
|
||||
log.warning(
|
||||
f'Queueing OVERRUN msg on caller task:\n'
|
||||
|
|
@ -1895,7 +1813,7 @@ class Context:
|
|||
f'=> cid: {cid}\n'
|
||||
f' |_{self._task}\n\n'
|
||||
|
||||
f'{structfmt(msg)}\n'
|
||||
f'{pformat(msg)}\n'
|
||||
)
|
||||
self._overflow_q.append(msg)
|
||||
return False
|
||||
|
|
@ -1909,7 +1827,7 @@ class Context:
|
|||
f'=> {self._task}\n'
|
||||
f' |_cid={self.cid}\n\n'
|
||||
|
||||
f'{structfmt(msg)}\n'
|
||||
f'{pformat(msg)}\n'
|
||||
)
|
||||
|
||||
# NOTE: if an error is deteced we should always still
|
||||
|
|
@ -2129,9 +2047,6 @@ async def open_context_from_portal(
|
|||
# place..
|
||||
allow_overruns=allow_overruns,
|
||||
)
|
||||
# ASAP, so that `Context.side: str` can be determined for
|
||||
# logging / tracing / debug!
|
||||
ctx._portal: Portal = portal
|
||||
|
||||
assert ctx._remote_func_type == 'context'
|
||||
msg: Started = await ctx._recv_chan.receive()
|
||||
|
|
@ -2150,9 +2065,10 @@ async def open_context_from_portal(
|
|||
msg=msg,
|
||||
src_err=src_error,
|
||||
log=log,
|
||||
expect_msg=Started,
|
||||
expect_key='started',
|
||||
)
|
||||
|
||||
ctx._portal: Portal = portal
|
||||
uid: tuple = portal.channel.uid
|
||||
cid: str = ctx.cid
|
||||
|
||||
|
|
|
|||
|
|
@ -106,7 +106,6 @@ def _trio_main(
|
|||
Entry point for a `trio_run_in_process` subactor.
|
||||
|
||||
'''
|
||||
__tracebackhide__: bool = True
|
||||
_state._current_actor = actor
|
||||
trio_main = partial(
|
||||
async_main,
|
||||
|
|
|
|||
|
|
@ -43,12 +43,9 @@ from tractor.msg import (
|
|||
MsgType,
|
||||
Stop,
|
||||
Yield,
|
||||
pretty_struct,
|
||||
types as msgtypes,
|
||||
)
|
||||
from tractor.msg.pretty_struct import (
|
||||
iter_fields,
|
||||
Struct,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ._context import Context
|
||||
|
|
@ -85,7 +82,7 @@ class InternalError(RuntimeError):
|
|||
_ipcmsg_keys: list[str] = [
|
||||
fi.name
|
||||
for fi, k, v
|
||||
in iter_fields(Error)
|
||||
in pretty_struct.iter_fields(Error)
|
||||
|
||||
]
|
||||
|
||||
|
|
@ -324,7 +321,7 @@ class RemoteActorError(Exception):
|
|||
assert self.boxed_type is boxed_type
|
||||
|
||||
@property
|
||||
def ipc_msg(self) -> Struct:
|
||||
def ipc_msg(self) -> pretty_struct.Struct:
|
||||
'''
|
||||
Re-render the underlying `._ipc_msg: Msg` as
|
||||
a `pretty_struct.Struct` for introspection such that the
|
||||
|
|
@ -337,12 +334,12 @@ class RemoteActorError(Exception):
|
|||
msg_type: MsgType = type(self._ipc_msg)
|
||||
fields: dict[str, Any] = {
|
||||
k: v for _, k, v in
|
||||
iter_fields(self._ipc_msg)
|
||||
pretty_struct.iter_fields(self._ipc_msg)
|
||||
}
|
||||
return defstruct(
|
||||
msg_type.__name__,
|
||||
fields=fields.keys(),
|
||||
bases=(msg_type, Struct),
|
||||
bases=(msg_type, pretty_struct.Struct),
|
||||
)(**fields)
|
||||
|
||||
@property
|
||||
|
|
@ -644,11 +641,11 @@ class MsgTypeError(
|
|||
|
||||
'''
|
||||
reprol_fields: list[str] = [
|
||||
'expected_msg_type',
|
||||
'payload_msg',
|
||||
]
|
||||
extra_body_fields: list[str] = [
|
||||
'cid',
|
||||
'expected_msg',
|
||||
'payload_msg',
|
||||
]
|
||||
|
||||
@property
|
||||
|
|
@ -664,7 +661,9 @@ class MsgTypeError(
|
|||
return self.msgdata.get('_msg_dict')
|
||||
|
||||
@property
|
||||
def expected_msg(self) -> MsgType|None:
|
||||
def payload_msg(
|
||||
self,
|
||||
) -> MsgType|None:
|
||||
'''
|
||||
Attempt to construct what would have been the original
|
||||
`MsgType`-with-payload subtype (i.e. an instance from the set
|
||||
|
|
@ -675,17 +674,9 @@ class MsgTypeError(
|
|||
if msg_dict := self.msg_dict.copy():
|
||||
return msgtypes.from_dict_msg(
|
||||
dict_msg=msg_dict,
|
||||
# use_pretty=True,
|
||||
# ^-TODO-^ would luv to use this BUT then the
|
||||
# `field_prefix` in `pformat_boxed_tb()` cucks it
|
||||
# all up.. XD
|
||||
)
|
||||
return None
|
||||
|
||||
@property
|
||||
def expected_msg_type(self) -> Type[MsgType]|None:
|
||||
return type(self.expected_msg)
|
||||
|
||||
@property
|
||||
def cid(self) -> str:
|
||||
# pre-packed using `.from_decode()` constructor
|
||||
|
|
@ -938,6 +929,7 @@ def _raise_from_no_key_in_msg(
|
|||
src_err: KeyError,
|
||||
log: StackLevelAdapter, # caller specific `log` obj
|
||||
|
||||
expect_key: str = 'yield',
|
||||
expect_msg: str = Yield,
|
||||
stream: MsgStream | None = None,
|
||||
|
||||
|
|
@ -1052,7 +1044,7 @@ def _raise_from_no_key_in_msg(
|
|||
# is activated above.
|
||||
_type: str = 'Stream' if stream else 'Context'
|
||||
raise MessagingError(
|
||||
f"{_type} was expecting a {expect_msg} message"
|
||||
f"{_type} was expecting a '{expect_key.upper()}' message"
|
||||
" BUT received a non-error msg:\n"
|
||||
f'{pformat(msg)}'
|
||||
) from src_err
|
||||
|
|
|
|||
|
|
@ -130,8 +130,6 @@ def _mk_msg_type_err(
|
|||
|
||||
) -> MsgTypeError:
|
||||
|
||||
import textwrap
|
||||
|
||||
# `Channel.send()` case
|
||||
if src_validation_error is None: # send-side
|
||||
|
||||
|
|
@ -211,24 +209,10 @@ def _mk_msg_type_err(
|
|||
msg, _, maybe_field = msgspec_msg.rpartition('$.')
|
||||
obj = object()
|
||||
if (field_val := msg_dict.get(maybe_field, obj)) is not obj:
|
||||
field_name_expr: str = (
|
||||
f' |_{maybe_field}: {codec.pld_spec_str} = '
|
||||
)
|
||||
fmt_val_lines: list[str] = pformat(field_val).splitlines()
|
||||
fmt_val: str = (
|
||||
f'{fmt_val_lines[0]}\n'
|
||||
+
|
||||
textwrap.indent(
|
||||
'\n'.join(fmt_val_lines[1:]),
|
||||
prefix=' '*len(field_name_expr),
|
||||
)
|
||||
)
|
||||
message += (
|
||||
f'{msg.rstrip("`")}\n\n'
|
||||
f'<{msg_type.__qualname__}(\n'
|
||||
# f'{".".join([msg_type.__module__, msg_type.__qualname__])}\n'
|
||||
f'{field_name_expr}{fmt_val}\n'
|
||||
f')>'
|
||||
f'{msg_type}\n'
|
||||
f' |_.{maybe_field}: {codec.pld_spec_str} = {field_val!r}\n'
|
||||
)
|
||||
|
||||
msgtyperr = MsgTypeError.from_decode(
|
||||
|
|
@ -354,7 +338,7 @@ class MsgpackTCPStream(MsgTransport):
|
|||
# self._task = task
|
||||
self._codec = codec
|
||||
log.runtime(
|
||||
f'Using new codec in {self}.recv()\n'
|
||||
'Using new codec in {self}.recv()\n'
|
||||
f'codec: {self._codec}\n\n'
|
||||
f'msg_bytes: {msg_bytes}\n'
|
||||
)
|
||||
|
|
@ -436,7 +420,7 @@ class MsgpackTCPStream(MsgTransport):
|
|||
if self._codec.pld_spec != codec.pld_spec:
|
||||
self._codec = codec
|
||||
log.runtime(
|
||||
f'Using new codec in {self}.send()\n'
|
||||
'Using new codec in {self}.send()\n'
|
||||
f'codec: {self._codec}\n\n'
|
||||
f'msg: {msg}\n'
|
||||
)
|
||||
|
|
|
|||
|
|
@ -79,7 +79,6 @@ async def open_root_actor(
|
|||
|
||||
# enables the multi-process debugger support
|
||||
debug_mode: bool = False,
|
||||
maybe_enable_greenback: bool = False, # `.pause_from_sync()/breakpoint()` support
|
||||
|
||||
# internal logging
|
||||
loglevel: str|None = None,
|
||||
|
|
@ -108,16 +107,14 @@ async def open_root_actor(
|
|||
)
|
||||
if (
|
||||
debug_mode
|
||||
and maybe_enable_greenback
|
||||
and await _debug.maybe_init_greenback(
|
||||
and
|
||||
await _debug.maybe_init_greenback(
|
||||
raise_not_found=False,
|
||||
)
|
||||
):
|
||||
os.environ['PYTHONBREAKPOINT'] = (
|
||||
'tractor.devx._debug.pause_from_sync'
|
||||
)
|
||||
_state._runtime_vars['use_greenback'] = True
|
||||
|
||||
else:
|
||||
# TODO: disable `breakpoint()` by default (without
|
||||
# `greenback`) since it will break any multi-actor
|
||||
|
|
@ -388,20 +385,14 @@ async def open_root_actor(
|
|||
_state._last_actor_terminated = actor
|
||||
|
||||
# restore built-in `breakpoint()` hook state
|
||||
if (
|
||||
debug_mode
|
||||
and
|
||||
maybe_enable_greenback
|
||||
):
|
||||
if debug_mode:
|
||||
if builtin_bp_handler is not None:
|
||||
sys.breakpointhook = builtin_bp_handler
|
||||
|
||||
if orig_bp_path is not None:
|
||||
os.environ['PYTHONBREAKPOINT'] = orig_bp_path
|
||||
|
||||
else:
|
||||
# clear env back to having no entry
|
||||
os.environ.pop('PYTHONBREAKPOINT', None)
|
||||
os.environ.pop('PYTHONBREAKPOINT')
|
||||
|
||||
logger.runtime("Root actor terminated")
|
||||
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ from trio import (
|
|||
TaskStatus,
|
||||
)
|
||||
|
||||
from .msg import NamespacePath
|
||||
from ._ipc import Channel
|
||||
from ._context import (
|
||||
Context,
|
||||
|
|
@ -60,11 +61,6 @@ from .devx import (
|
|||
)
|
||||
from . import _state
|
||||
from .log import get_logger
|
||||
from .msg import (
|
||||
current_codec,
|
||||
MsgCodec,
|
||||
NamespacePath,
|
||||
)
|
||||
from tractor.msg.types import (
|
||||
CancelAck,
|
||||
Error,
|
||||
|
|
@ -102,7 +98,6 @@ async def _invoke_non_context(
|
|||
Context | BaseException
|
||||
] = trio.TASK_STATUS_IGNORED,
|
||||
):
|
||||
__tracebackhide__: bool = True
|
||||
|
||||
# TODO: can we unify this with the `context=True` impl below?
|
||||
if inspect.isasyncgen(coro):
|
||||
|
|
@ -403,11 +398,7 @@ async def _invoke(
|
|||
__tracebackhide__: bool = hide_tb
|
||||
treat_as_gen: bool = False
|
||||
|
||||
if (
|
||||
_state.debug_mode()
|
||||
and
|
||||
_state._runtime_vars['use_greenback']
|
||||
):
|
||||
if _state.debug_mode():
|
||||
# XXX for .pause_from_sync()` usage we need to make sure
|
||||
# `greenback` is boostrapped in the subactor!
|
||||
await _debug.maybe_init_greenback()
|
||||
|
|
@ -521,22 +512,10 @@ async def _invoke(
|
|||
# wrapper that calls `Context.started()` and then does
|
||||
# the `await coro()`?
|
||||
|
||||
# ------ - ------
|
||||
# a "context" endpoint is the most general and
|
||||
# "least sugary" type of RPC with support for
|
||||
# a "context" endpoint type is the most general and
|
||||
# "least sugary" type of RPC ep with support for
|
||||
# bi-dir streaming B)
|
||||
#
|
||||
# the concurrency relation is simlar to a task nursery
|
||||
# wherein a "parent" task (the one that enters
|
||||
# `trio.open_nursery()` in some actor "opens" (via
|
||||
# `Portal.open_context()`) an IPC ctx to another peer
|
||||
# (which is maybe a sub-) actor who then schedules (aka
|
||||
# `trio.Nursery.start()`s) a new "child" task to execute
|
||||
# the `@context` annotated func; that is this func we're
|
||||
# running directly below!
|
||||
# ------ - ------
|
||||
#
|
||||
# StartAck: respond immediately with endpoint info
|
||||
# StartAck
|
||||
await chan.send(
|
||||
StartAck(
|
||||
cid=cid,
|
||||
|
|
@ -545,11 +524,11 @@ async def _invoke(
|
|||
)
|
||||
|
||||
# TODO: should we also use an `.open_context()` equiv
|
||||
# for this child side by factoring the impl from
|
||||
# for this callee side by factoring the impl from
|
||||
# `Portal.open_context()` into a common helper?
|
||||
#
|
||||
# NOTE: there are many different ctx state details
|
||||
# in a child side instance according to current impl:
|
||||
# in a callee side instance according to current impl:
|
||||
# - `.cancelled_caught` can never be `True`.
|
||||
# -> the below scope is never exposed to the
|
||||
# `@context` marked RPC function.
|
||||
|
|
@ -575,7 +554,7 @@ async def _invoke(
|
|||
|
||||
# NOTE: this happens IFF `ctx._scope.cancel()` is
|
||||
# called by any of,
|
||||
# - *this* child task manually calling `ctx.cancel()`.
|
||||
# - *this* callee task manually calling `ctx.cancel()`.
|
||||
# - the runtime calling `ctx._deliver_msg()` which
|
||||
# itself calls `ctx._maybe_cancel_and_set_remote_error()`
|
||||
# which cancels the scope presuming the input error
|
||||
|
|
@ -652,11 +631,10 @@ async def _invoke(
|
|||
# f' |_{ctx}'
|
||||
)
|
||||
|
||||
# task-contex was either cancelled by request
|
||||
# using ``Portal.cancel_actor()`` or
|
||||
# ``Context.cancel()`` on the far end, or it
|
||||
# was cancelled by the local child (or callee)
|
||||
# task, so relay this cancel signal to the
|
||||
# task-contex was either cancelled by request using
|
||||
# ``Portal.cancel_actor()`` or ``Context.cancel()``
|
||||
# on the far end, or it was cancelled by the local
|
||||
# (callee) task, so relay this cancel signal to the
|
||||
# other side.
|
||||
ctxc = ContextCancelled(
|
||||
message=msg,
|
||||
|
|
@ -677,7 +655,7 @@ async def _invoke(
|
|||
|
||||
) as scope_error:
|
||||
|
||||
# always set this (child) side's exception as the
|
||||
# always set this (callee) side's exception as the
|
||||
# local error on the context
|
||||
ctx._local_error: BaseException = scope_error
|
||||
|
||||
|
|
@ -1046,8 +1024,9 @@ async def process_messages(
|
|||
trio.Event(),
|
||||
)
|
||||
|
||||
# runtime-scoped remote error (since no `.cid`)
|
||||
case Error():
|
||||
# XXX remote (runtime scoped) error or uknown
|
||||
# msg (type).
|
||||
case Error() | _:
|
||||
# NOTE: this is the non-rpc error case,
|
||||
# that is, an error **not** raised inside
|
||||
# a call to ``_invoke()`` (i.e. no cid was
|
||||
|
|
@ -1055,6 +1034,10 @@ async def process_messages(
|
|||
# this error to all local channel
|
||||
# consumers (normally portals) by marking
|
||||
# the channel as errored
|
||||
log.exception(
|
||||
f'Unhandled IPC msg:\n\n'
|
||||
f'{msg}\n'
|
||||
)
|
||||
# assert chan.uid
|
||||
chan._exc: Exception = unpack_error(
|
||||
msg,
|
||||
|
|
@ -1062,17 +1045,6 @@ async def process_messages(
|
|||
)
|
||||
raise chan._exc
|
||||
|
||||
# unknown/invalid msg type?
|
||||
case _:
|
||||
codec: MsgCodec = current_codec()
|
||||
message: str = (
|
||||
f'Unhandled IPC msg for codec?\n\n'
|
||||
f'|_{codec}\n\n'
|
||||
f'{msg}\n'
|
||||
)
|
||||
log.exception(message)
|
||||
raise RuntimeError(message)
|
||||
|
||||
log.runtime(
|
||||
'Waiting on next IPC msg from\n'
|
||||
f'peer: {chan.uid}\n'
|
||||
|
|
|
|||
|
|
@ -513,7 +513,7 @@ async def trio_proc(
|
|||
# })
|
||||
|
||||
# track subactor in current nursery
|
||||
curr_actor: Actor = current_actor()
|
||||
curr_actor = current_actor()
|
||||
curr_actor._actoruid2nursery[subactor.uid] = actor_nursery
|
||||
|
||||
# resume caller at next checkpoint now that child is up
|
||||
|
|
|
|||
|
|
@ -44,7 +44,6 @@ from .trionics import (
|
|||
BroadcastReceiver,
|
||||
)
|
||||
from tractor.msg import (
|
||||
Return,
|
||||
Stop,
|
||||
Yield,
|
||||
)
|
||||
|
|
@ -83,7 +82,7 @@ class MsgStream(trio.abc.Channel):
|
|||
self,
|
||||
ctx: Context, # typing: ignore # noqa
|
||||
rx_chan: trio.MemoryReceiveChannel,
|
||||
_broadcaster: BroadcastReceiver|None = None,
|
||||
_broadcaster: BroadcastReceiver | None = None,
|
||||
|
||||
) -> None:
|
||||
self._ctx = ctx
|
||||
|
|
@ -97,26 +96,36 @@ class MsgStream(trio.abc.Channel):
|
|||
# delegate directly to underlying mem channel
|
||||
def receive_nowait(
|
||||
self,
|
||||
allow_msgs: list[str] = Yield,
|
||||
allow_msg_keys: list[str] = ['yield'],
|
||||
):
|
||||
# msg: dict = self._rx_chan.receive_nowait()
|
||||
msg: Yield|Stop = self._rx_chan.receive_nowait()
|
||||
# TODO: replace msg equiv of this or does the `.pld`
|
||||
# interface read already satisfy it? I think so, yes?
|
||||
try:
|
||||
return msg.pld
|
||||
except AttributeError as attrerr:
|
||||
_raise_from_no_key_in_msg(
|
||||
ctx=self._ctx,
|
||||
msg=msg,
|
||||
src_err=attrerr,
|
||||
log=log,
|
||||
stream=self,
|
||||
)
|
||||
for (
|
||||
i,
|
||||
key,
|
||||
) in enumerate(allow_msg_keys):
|
||||
try:
|
||||
# return msg[key]
|
||||
return msg.pld
|
||||
# except KeyError as kerr:
|
||||
except AttributeError as attrerr:
|
||||
if i < (len(allow_msg_keys) - 1):
|
||||
continue
|
||||
|
||||
_raise_from_no_key_in_msg(
|
||||
ctx=self._ctx,
|
||||
msg=msg,
|
||||
# src_err=kerr,
|
||||
src_err=attrerr,
|
||||
log=log,
|
||||
expect_key=key,
|
||||
stream=self,
|
||||
)
|
||||
|
||||
async def receive(
|
||||
self,
|
||||
|
||||
hide_tb: bool = False,
|
||||
hide_tb: bool = True,
|
||||
):
|
||||
'''
|
||||
Receive a single msg from the IPC transport, the next in
|
||||
|
|
@ -148,9 +157,10 @@ class MsgStream(trio.abc.Channel):
|
|||
try:
|
||||
try:
|
||||
msg: Yield = await self._rx_chan.receive()
|
||||
# return msg['yield']
|
||||
return msg.pld
|
||||
|
||||
# TODO: implement with match: instead?
|
||||
# except KeyError as kerr:
|
||||
except AttributeError as attrerr:
|
||||
# src_err = kerr
|
||||
src_err = attrerr
|
||||
|
|
@ -160,8 +170,10 @@ class MsgStream(trio.abc.Channel):
|
|||
_raise_from_no_key_in_msg(
|
||||
ctx=self._ctx,
|
||||
msg=msg,
|
||||
# src_err=kerr,
|
||||
src_err=attrerr,
|
||||
log=log,
|
||||
expect_key='yield',
|
||||
stream=self,
|
||||
)
|
||||
|
||||
|
|
@ -292,7 +304,7 @@ class MsgStream(trio.abc.Channel):
|
|||
while not drained:
|
||||
try:
|
||||
maybe_final_msg = self.receive_nowait(
|
||||
allow_msgs=[Yield, Return],
|
||||
allow_msg_keys=['yield', 'return'],
|
||||
)
|
||||
if maybe_final_msg:
|
||||
log.debug(
|
||||
|
|
|
|||
|
|
@ -119,11 +119,11 @@ class ActorNursery:
|
|||
name: str,
|
||||
*,
|
||||
bind_addrs: list[tuple[str, int]] = [_default_bind_addr],
|
||||
rpc_module_paths: list[str]|None = None,
|
||||
enable_modules: list[str]|None = None,
|
||||
loglevel: str|None = None, # set log level per subactor
|
||||
nursery: trio.Nursery|None = None,
|
||||
debug_mode: bool|None = None,
|
||||
rpc_module_paths: list[str] | None = None,
|
||||
enable_modules: list[str] | None = None,
|
||||
loglevel: str | None = None, # set log level per subactor
|
||||
nursery: trio.Nursery | None = None,
|
||||
debug_mode: bool | None = None,
|
||||
infect_asyncio: bool = False,
|
||||
) -> Portal:
|
||||
'''
|
||||
|
|
|
|||
|
|
@ -420,7 +420,7 @@ def mk_codec(
|
|||
|
||||
# instance of the default `msgspec.msgpack` codec settings, i.e.
|
||||
# no custom structs, hooks or other special types.
|
||||
_def_msgspec_codec: MsgCodec = mk_codec(ipc_pld_spec=Any)
|
||||
_def_msgspec_codec: MsgCodec = mk_codec(ipc_msg_spec=Any)
|
||||
|
||||
# The built-in IPC `Msg` spec.
|
||||
# Our composing "shuttle" protocol which allows `tractor`-app code
|
||||
|
|
|
|||
|
|
@ -451,8 +451,7 @@ def from_dict_msg(
|
|||
dict_msg: dict,
|
||||
|
||||
msgT: MsgType|None = None,
|
||||
tag_field: str = 'msg_type',
|
||||
use_pretty: bool = False,
|
||||
tag_field: str = 'msg_type'
|
||||
|
||||
) -> MsgType:
|
||||
'''
|
||||
|
|
@ -469,19 +468,6 @@ def from_dict_msg(
|
|||
# XXX ensure tag field is removed
|
||||
msgT_name: str = dict_msg.pop(msg_type_tag_field)
|
||||
msgT: MsgType = _msg_table[msgT_name]
|
||||
if use_pretty:
|
||||
msgT = defstruct(
|
||||
name=msgT_name,
|
||||
fields=[
|
||||
(key, fi.type)
|
||||
for fi, key, _
|
||||
in pretty_struct.iter_fields(msgT)
|
||||
],
|
||||
bases=(
|
||||
pretty_struct.Struct,
|
||||
msgT,
|
||||
),
|
||||
)
|
||||
return msgT(**dict_msg)
|
||||
|
||||
# TODO: should be make a msg version of `ContextCancelled?`
|
||||
|
|
|
|||
Loading…
Reference in New Issue