Compare commits
No commits in common. "af013912acbb5aabadae26d168ae5c45599c9f71" and "cf48fdecfeb76cdf67a3f779d98c086e75659dd6" have entirely different histories.
af013912ac
...
cf48fdecfe
|
|
@ -49,21 +49,20 @@ from ._exceptions import (
|
|||
InternalError,
|
||||
RemoteActorError,
|
||||
StreamOverrun,
|
||||
pack_from_raise,
|
||||
pack_error,
|
||||
unpack_error,
|
||||
_raise_from_no_key_in_msg,
|
||||
)
|
||||
from .log import get_logger
|
||||
from .msg import (
|
||||
Error,
|
||||
MsgType,
|
||||
MsgCodec,
|
||||
NamespacePath,
|
||||
Msg,
|
||||
Return,
|
||||
Started,
|
||||
Stop,
|
||||
Yield,
|
||||
current_codec,
|
||||
MsgCodec,
|
||||
pretty_struct,
|
||||
)
|
||||
from ._ipc import Channel
|
||||
|
|
@ -108,7 +107,8 @@ async def _drain_to_final_msg(
|
|||
# wait for a final context result by collecting (but
|
||||
# basically ignoring) any bi-dir-stream msgs still in transit
|
||||
# from the far end.
|
||||
pre_result_drained: list[MsgType] = []
|
||||
# pre_result_drained: list[dict] = []
|
||||
pre_result_drained: list[Msg] = []
|
||||
while not (
|
||||
ctx.maybe_error
|
||||
and not ctx._final_result_is_set()
|
||||
|
|
@ -168,7 +168,7 @@ 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()
|
||||
msg: Msg = await ctx._recv_chan.receive()
|
||||
# always capture unexpected/non-result msgs
|
||||
pre_result_drained.append(msg)
|
||||
|
||||
|
|
@ -191,12 +191,13 @@ async def _drain_to_final_msg(
|
|||
raise
|
||||
|
||||
match msg:
|
||||
|
||||
# final result arrived!
|
||||
case Return(
|
||||
# cid=cid,
|
||||
cid=cid,
|
||||
pld=res,
|
||||
):
|
||||
# try:
|
||||
# ctx._result: Any = msg['return']
|
||||
# ctx._result: Any = msg.pld
|
||||
ctx._result: Any = res
|
||||
log.runtime(
|
||||
'Context delivered final draining msg:\n'
|
||||
|
|
@ -209,9 +210,13 @@ async def _drain_to_final_msg(
|
|||
# TODO: ^ we don't need it right?
|
||||
break
|
||||
|
||||
# far end task is still streaming to us so discard
|
||||
# and report depending on local ctx state.
|
||||
# except KeyError:
|
||||
# except AttributeError:
|
||||
case Yield():
|
||||
# if 'yield' in msg:
|
||||
|
||||
# far end task is still streaming to us so discard
|
||||
# and report per local context state.
|
||||
if (
|
||||
(ctx._stream.closed
|
||||
and (reason := 'stream was already closed')
|
||||
|
|
@ -252,34 +257,45 @@ async def _drain_to_final_msg(
|
|||
)
|
||||
continue
|
||||
|
||||
# stream terminated, but no result yet..
|
||||
#
|
||||
# TODO: work out edge cases here where
|
||||
# a stream is open but the task also calls
|
||||
# this?
|
||||
# -[ ] should be a runtime error if a stream is open right?
|
||||
# Stop()
|
||||
case Stop():
|
||||
# elif 'stop' in msg:
|
||||
log.cancel(
|
||||
'Remote stream terminated due to "stop" msg:\n\n'
|
||||
f'{pformat(msg)}\n'
|
||||
)
|
||||
continue
|
||||
|
||||
# remote error msg, likely already handled inside
|
||||
# `Context._deliver_msg()`
|
||||
case Error():
|
||||
# It's an internal error if any other msg type without
|
||||
# a`'cid'` field arrives here!
|
||||
case _:
|
||||
# if not msg.get('cid'):
|
||||
if not msg.cid:
|
||||
raise InternalError(
|
||||
'Unexpected cid-missing msg?\n\n'
|
||||
f'{msg}\n'
|
||||
)
|
||||
|
||||
# TODO: can we replace this with `ctx.maybe_raise()`?
|
||||
# -[ ] would this be handier for this case maybe?
|
||||
# XXX fallthrough to handle expected error XXX
|
||||
# TODO: replace this with `ctx.maybe_raise()`
|
||||
#
|
||||
# TODO: would this be handier for this case maybe?
|
||||
# async with maybe_raise_on_exit() as raises:
|
||||
# if raises:
|
||||
# log.error('some msg about raising..')
|
||||
#
|
||||
|
||||
re: Exception|None = ctx._remote_error
|
||||
if re:
|
||||
log.critical(
|
||||
'Remote ctx terminated due to "error" msg:\n'
|
||||
f'{re}'
|
||||
)
|
||||
assert msg is ctx._cancel_msg
|
||||
# NOTE: this solved a super duper edge case XD
|
||||
# NOTE: this solved a super dupe edge case XD
|
||||
# this was THE super duper edge case of:
|
||||
# - local task opens a remote task,
|
||||
# - requests remote cancellation of far end
|
||||
|
|
@ -296,10 +312,9 @@ async def _drain_to_final_msg(
|
|||
# does not re-raise any ctxc it receives
|
||||
# IFF **it** was the cancellation
|
||||
# requester..
|
||||
#
|
||||
# XXX will raise if necessary but ow break
|
||||
# from loop presuming any supressed error
|
||||
# (ctxc) should terminate the context!
|
||||
# will raise if necessary, ow break from
|
||||
# loop presuming any error terminates the
|
||||
# context!
|
||||
ctx._maybe_raise_remote_err(
|
||||
re,
|
||||
# NOTE: obvi we don't care if we
|
||||
|
|
@ -323,7 +338,6 @@ async def _drain_to_final_msg(
|
|||
log.critical('SHOULD NEVER GET HERE!?')
|
||||
assert msg is ctx._cancel_msg
|
||||
assert error.msgdata == ctx._remote_error.msgdata
|
||||
assert error.ipc_msg == ctx._remote_error.ipc_msg
|
||||
from .devx._debug import pause
|
||||
await pause()
|
||||
ctx._maybe_cancel_and_set_remote_error(error)
|
||||
|
|
@ -332,20 +346,6 @@ async def _drain_to_final_msg(
|
|||
else:
|
||||
# bubble the original src key error
|
||||
raise
|
||||
|
||||
# XXX should pretty much never get here unless someone
|
||||
# overrides the default `MsgType` spec.
|
||||
case _:
|
||||
# It's definitely an internal error if any other
|
||||
# msg type without a`'cid'` field arrives here!
|
||||
if not msg.cid:
|
||||
raise InternalError(
|
||||
'Unexpected cid-missing msg?\n\n'
|
||||
f'{msg}\n'
|
||||
)
|
||||
|
||||
raise RuntimeError('Unknown msg type: {msg}')
|
||||
|
||||
else:
|
||||
log.cancel(
|
||||
'Skipping `MsgStream` drain since final outcome is set\n\n'
|
||||
|
|
@ -1342,11 +1342,8 @@ class Context:
|
|||
# `._cancel_called == True`.
|
||||
not raise_overrun_from_self
|
||||
and isinstance(remote_error, RemoteActorError)
|
||||
|
||||
and remote_error.boxed_type_str == 'StreamOverrun'
|
||||
|
||||
# and tuple(remote_error.msgdata['sender']) == our_uid
|
||||
and tuple(remote_error.sender) == our_uid
|
||||
and remote_error.msgdata['boxed_type_str'] == 'StreamOverrun'
|
||||
and tuple(remote_error.msgdata['sender']) == our_uid
|
||||
):
|
||||
# NOTE: we set the local scope error to any "self
|
||||
# cancellation" error-response thus "absorbing"
|
||||
|
|
@ -1415,11 +1412,16 @@ class Context:
|
|||
|
||||
assert self._recv_chan
|
||||
raise_overrun: bool = not self._allow_overruns
|
||||
# res_placeholder: int = id(self)
|
||||
if (
|
||||
# self._result == res_placeholder
|
||||
# and not self._remote_error
|
||||
self.maybe_error is None
|
||||
and
|
||||
not self._recv_chan._closed # type: ignore
|
||||
# not self._remote_error
|
||||
# and not self._local_error
|
||||
and not self._recv_chan._closed # type: ignore
|
||||
):
|
||||
|
||||
# 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.
|
||||
|
|
@ -1430,6 +1432,7 @@ class Context:
|
|||
for msg in drained_msgs:
|
||||
|
||||
# TODO: mask this by default..
|
||||
# if 'return' in msg:
|
||||
if isinstance(msg, Return):
|
||||
# from .devx import pause
|
||||
# await pause()
|
||||
|
|
@ -1445,9 +1448,6 @@ class Context:
|
|||
)
|
||||
|
||||
self.maybe_raise(
|
||||
# NOTE: obvi we don't care if we
|
||||
# overran the far end if we're already
|
||||
# waiting on a final result (msg).
|
||||
raise_overrun_from_self=(
|
||||
raise_overrun
|
||||
and
|
||||
|
|
@ -1458,12 +1458,34 @@ class Context:
|
|||
(not self._cancel_called)
|
||||
)
|
||||
)
|
||||
# if (
|
||||
# (re := self._remote_error)
|
||||
# # and self._result == res_placeholder
|
||||
# ):
|
||||
# self._maybe_raise_remote_err(
|
||||
# re,
|
||||
# # NOTE: obvi we don't care if we
|
||||
# # overran the far end if we're already
|
||||
# # waiting on a final result (msg).
|
||||
# # raise_overrun_from_self=False,
|
||||
# raise_overrun_from_self=(
|
||||
# raise_overrun
|
||||
# and
|
||||
# # only when we ARE NOT the canceller
|
||||
# # should we raise overruns, bc ow we're
|
||||
# # raising something we know might happen
|
||||
# # during cancellation ;)
|
||||
# (not self._cancel_called)
|
||||
# ),
|
||||
# )
|
||||
# if maybe_err:
|
||||
# self._result = maybe_err
|
||||
|
||||
return self.outcome
|
||||
|
||||
# TODO: switch this with above!
|
||||
# -[ ] should be named `.wait_for_outcome()` and instead do
|
||||
# TODO: switch this with above which should be named
|
||||
# `.wait_for_outcome()` and instead do
|
||||
# a `.outcome.Outcome.unwrap()` ?
|
||||
#
|
||||
# @property
|
||||
# def result(self) -> Any|None:
|
||||
# if self._final_result_is_set():
|
||||
|
|
@ -1522,6 +1544,7 @@ class Context:
|
|||
return None
|
||||
|
||||
def _final_result_is_set(self) -> bool:
|
||||
# return not (self._result == id(self))
|
||||
return self._result is not Unresolved
|
||||
|
||||
# def get_result_nowait(self) -> Any|None:
|
||||
|
|
@ -1738,7 +1761,8 @@ class Context:
|
|||
|
||||
async def _deliver_msg(
|
||||
self,
|
||||
msg: MsgType,
|
||||
# msg: dict,
|
||||
msg: Msg,
|
||||
|
||||
) -> bool:
|
||||
'''
|
||||
|
|
@ -1752,20 +1776,6 @@ class Context:
|
|||
`._scope_nursery: trio.Nursery`) which ensures that such
|
||||
messages are queued up and eventually sent if possible.
|
||||
|
||||
XXX RULES XXX
|
||||
------ - ------
|
||||
- NEVER raise remote errors from this method; a runtime task caller.
|
||||
An error "delivered" to a ctx should always be raised by
|
||||
the corresponding local task operating on the
|
||||
`Portal`/`Context` APIs.
|
||||
|
||||
- NEVER `return` early before delivering the msg!
|
||||
bc if the error is a ctxc and there is a task waiting on
|
||||
`.result()` we need the msg to be
|
||||
`send_chan.send_nowait()`-ed over the `._recv_chan` so
|
||||
that the error is relayed to that waiter task and thus
|
||||
raised in user code!
|
||||
|
||||
'''
|
||||
cid: str = self.cid
|
||||
chan: Channel = self.chan
|
||||
|
|
@ -1796,14 +1806,28 @@ class Context:
|
|||
)
|
||||
self._cancel_msg: dict = msg
|
||||
|
||||
# XXX NOTE: this will not raise an error, merely set
|
||||
# NOTE: this will not raise an error, merely set
|
||||
# `._remote_error` and maybe cancel any task currently
|
||||
# entered in `Portal.open_context()` presuming the
|
||||
# error is "cancel causing" (i.e. a `ContextCancelled`
|
||||
# or `RemoteActorError`).
|
||||
self._maybe_cancel_and_set_remote_error(re)
|
||||
|
||||
# XXX only case where returning early is fine!
|
||||
# XXX NEVER do this XXX..!!
|
||||
# bc if the error is a ctxc and there is a task
|
||||
# waiting on `.result()` we need the msg to be sent
|
||||
# over the `send_chan`/`._recv_chan` so that the error
|
||||
# is relayed to that waiter task..
|
||||
# return True
|
||||
#
|
||||
# XXX ALSO NO!! XXX
|
||||
# => NEVER raise remote errors from the calling
|
||||
# runtime task, they should always be raised by
|
||||
# consumer side tasks operating on the
|
||||
# `Portal`/`Context` APIs.
|
||||
# if self._remote_error:
|
||||
# self._maybe_raise_remote_err(error)
|
||||
|
||||
if self._in_overrun:
|
||||
log.warning(
|
||||
f'Queueing OVERRUN msg on caller task:\n'
|
||||
|
|
@ -1922,13 +1946,17 @@ class Context:
|
|||
# anything different.
|
||||
return False
|
||||
else:
|
||||
# txt += f'\n{msg}\n'
|
||||
# raise local overrun and immediately pack as IPC
|
||||
# msg for far end.
|
||||
err_msg: Error = pack_from_raise(
|
||||
local_err=StreamOverrun(
|
||||
try:
|
||||
raise StreamOverrun(
|
||||
txt,
|
||||
sender=from_uid,
|
||||
),
|
||||
)
|
||||
except StreamOverrun as err:
|
||||
err_msg: dict[str, dict] = pack_error(
|
||||
err,
|
||||
cid=cid,
|
||||
)
|
||||
try:
|
||||
|
|
@ -1936,9 +1964,9 @@ class Context:
|
|||
await chan.send(err_msg)
|
||||
return True
|
||||
|
||||
# XXX: local consumer has closed their side of
|
||||
# the IPC so cancel the far end streaming task
|
||||
except trio.BrokenResourceError:
|
||||
# XXX: local consumer has closed their side
|
||||
# so cancel the far end streaming task
|
||||
log.warning(
|
||||
'Channel for ctx is already closed?\n'
|
||||
f'|_{chan}\n'
|
||||
|
|
@ -2351,17 +2379,28 @@ async def open_context_from_portal(
|
|||
# an exception type boxed in a `RemoteActorError`
|
||||
# is returned (meaning it was obvi not raised)
|
||||
# that we want to log-report on.
|
||||
match result_or_err:
|
||||
case ContextCancelled() as ctxc:
|
||||
log.cancel(ctxc.tb_str)
|
||||
msgdata: str|None = getattr(
|
||||
result_or_err,
|
||||
'msgdata',
|
||||
None
|
||||
)
|
||||
match (msgdata, result_or_err):
|
||||
case (
|
||||
{'tb_str': tbstr},
|
||||
ContextCancelled(),
|
||||
):
|
||||
log.cancel(tbstr)
|
||||
|
||||
case RemoteActorError() as rae:
|
||||
case (
|
||||
{'tb_str': tbstr},
|
||||
RemoteActorError(),
|
||||
):
|
||||
log.exception(
|
||||
'Context remotely errored!\n'
|
||||
f'<= peer: {uid}\n'
|
||||
f' |_ {nsf}()\n\n'
|
||||
|
||||
f'{rae.tb_str}'
|
||||
f'{tbstr}'
|
||||
)
|
||||
case (None, _):
|
||||
log.runtime(
|
||||
|
|
@ -2371,6 +2410,7 @@ async def open_context_from_portal(
|
|||
|
||||
f'`{result_or_err}`\n'
|
||||
)
|
||||
|
||||
finally:
|
||||
# XXX: (MEGA IMPORTANT) if this is a root opened process we
|
||||
# wait for any immediate child in debug before popping the
|
||||
|
|
|
|||
123
tractor/_ipc.py
123
tractor/_ipc.py
|
|
@ -38,6 +38,7 @@ from typing import (
|
|||
Protocol,
|
||||
Type,
|
||||
TypeVar,
|
||||
Union,
|
||||
)
|
||||
|
||||
import msgspec
|
||||
|
|
@ -46,9 +47,8 @@ import trio
|
|||
|
||||
from tractor.log import get_logger
|
||||
from tractor._exceptions import (
|
||||
MsgTypeError,
|
||||
pack_from_raise,
|
||||
TransportClosed,
|
||||
MsgTypeError,
|
||||
)
|
||||
from tractor.msg import (
|
||||
_ctxvar_MsgCodec,
|
||||
|
|
@ -118,24 +118,17 @@ class MsgTransport(Protocol[MsgType]):
|
|||
...
|
||||
|
||||
|
||||
def _mk_msg_type_err(
|
||||
def _raise_msg_type_err(
|
||||
msg: Any|bytes,
|
||||
codec: MsgCodec,
|
||||
|
||||
message: str|None = None,
|
||||
validation_err: msgspec.ValidationError|None = None,
|
||||
verb_header: str = '',
|
||||
|
||||
src_validation_error: msgspec.ValidationError|None = None,
|
||||
src_type_error: TypeError|None = None,
|
||||
) -> None:
|
||||
|
||||
) -> MsgTypeError:
|
||||
# if side == 'send':
|
||||
if validation_err is None: # send-side
|
||||
|
||||
# `Channel.send()` case
|
||||
if src_validation_error is None: # send-side
|
||||
|
||||
# no src error from `msgspec.msgpack.Decoder.decode()` so
|
||||
# prolly a manual type-check on our part.
|
||||
if message is None:
|
||||
import traceback
|
||||
from tractor._exceptions import pformat_boxed_tb
|
||||
|
||||
|
|
@ -151,42 +144,14 @@ def _mk_msg_type_err(
|
|||
field_prefix=' ',
|
||||
indent='',
|
||||
)
|
||||
message: str = (
|
||||
raise MsgTypeError(
|
||||
f'invalid msg -> {msg}: {type(msg)}\n\n'
|
||||
f'{tb_fmt}\n'
|
||||
f'Valid IPC msgs are:\n\n'
|
||||
# f' ------ - ------\n'
|
||||
f'{fmt_spec}\n',
|
||||
)
|
||||
elif src_type_error:
|
||||
src_message: str = str(src_type_error)
|
||||
patt: str = 'type '
|
||||
type_idx: int = src_message.find('type ')
|
||||
invalid_type: str = src_message[type_idx + len(patt):].split()[0]
|
||||
|
||||
enc_hook: Callable|None = codec.enc.enc_hook
|
||||
if enc_hook is None:
|
||||
message += (
|
||||
'\n\n'
|
||||
|
||||
f"The current IPC-msg codec can't encode type `{invalid_type}` !\n"
|
||||
f'Maybe a `msgpack.Encoder.enc_hook()` extension is needed?\n\n'
|
||||
|
||||
f'Check the `msgspec` docs for ad-hoc type extending:\n'
|
||||
'|_ https://jcristharif.com/msgspec/extending.html\n'
|
||||
'|_ https://jcristharif.com/msgspec/extending.html#defining-a-custom-extension-messagepack-only\n'
|
||||
f'{fmt_spec}\n'
|
||||
)
|
||||
|
||||
|
||||
msgtyperr = MsgTypeError(
|
||||
message=message,
|
||||
ipc_msg=msg,
|
||||
)
|
||||
# ya, might be `None`
|
||||
msgtyperr.__cause__ = src_type_error
|
||||
return msgtyperr
|
||||
|
||||
# `Channel.recv()` case
|
||||
else:
|
||||
# decode the msg-bytes using the std msgpack
|
||||
# interchange-prot (i.e. without any
|
||||
|
|
@ -196,31 +161,29 @@ def _mk_msg_type_err(
|
|||
msg_dict: dict = msgspec.msgpack.decode(msg)
|
||||
msg_type_name: str = msg_dict['msg_type']
|
||||
msg_type = getattr(msgtypes, msg_type_name)
|
||||
message: str = (
|
||||
errmsg: str = (
|
||||
f'invalid `{msg_type_name}` IPC msg\n\n'
|
||||
)
|
||||
if verb_header:
|
||||
message = f'{verb_header} ' + message
|
||||
errmsg = f'{verb_header} ' + errmsg
|
||||
|
||||
# XXX see if we can determine the exact invalid field
|
||||
# such that we can comprehensively report the
|
||||
# specific field's type problem
|
||||
msgspec_msg: str = src_validation_error.args[0].rstrip('`')
|
||||
msgspec_msg: str = validation_err.args[0].rstrip('`')
|
||||
msg, _, maybe_field = msgspec_msg.rpartition('$.')
|
||||
obj = object()
|
||||
if (field_val := msg_dict.get(maybe_field, obj)) is not obj:
|
||||
message += (
|
||||
field_type: Union[Type] = msg_type.__signature__.parameters[
|
||||
maybe_field
|
||||
].annotation
|
||||
errmsg += (
|
||||
f'{msg.rstrip("`")}\n\n'
|
||||
f'{msg_type}\n'
|
||||
f' |_.{maybe_field}: {codec.pld_spec_str} = {field_val!r}\n'
|
||||
f' |_.{maybe_field}: {field_type} = {field_val!r}\n'
|
||||
)
|
||||
|
||||
msgtyperr = MsgTypeError.from_decode(
|
||||
message=message,
|
||||
msgdict=msg_dict,
|
||||
)
|
||||
msgtyperr.__cause__ = src_validation_error
|
||||
return msgtyperr
|
||||
raise MsgTypeError(errmsg) from validation_err
|
||||
|
||||
|
||||
# TODO: not sure why we have to inherit here, but it seems to be an
|
||||
|
|
@ -362,15 +325,12 @@ class MsgpackTCPStream(MsgTransport):
|
|||
# and always raise such that spec violations
|
||||
# are never allowed to be caught silently!
|
||||
except msgspec.ValidationError as verr:
|
||||
msgtyperr: MsgTypeError = _mk_msg_type_err(
|
||||
# re-raise as type error
|
||||
_raise_msg_type_err(
|
||||
msg=msg_bytes,
|
||||
codec=codec,
|
||||
src_validation_error=verr,
|
||||
validation_err=verr,
|
||||
)
|
||||
# XXX deliver up to `Channel.recv()` where
|
||||
# a re-raise and `Error`-pack can inject the far
|
||||
# end actor `.uid`.
|
||||
yield msgtyperr
|
||||
|
||||
except (
|
||||
msgspec.DecodeError,
|
||||
|
|
@ -427,7 +387,7 @@ class MsgpackTCPStream(MsgTransport):
|
|||
|
||||
if type(msg) not in msgtypes.__msg_types__:
|
||||
if strict_types:
|
||||
raise _mk_msg_type_err(
|
||||
_raise_msg_type_err(
|
||||
msg,
|
||||
codec=codec,
|
||||
)
|
||||
|
|
@ -440,16 +400,11 @@ class MsgpackTCPStream(MsgTransport):
|
|||
try:
|
||||
bytes_data: bytes = codec.encode(msg)
|
||||
except TypeError as typerr:
|
||||
msgtyperr: MsgTypeError = _mk_msg_type_err(
|
||||
msg,
|
||||
codec=codec,
|
||||
message=(
|
||||
f'IPC-msg-spec violation in\n\n'
|
||||
raise MsgTypeError(
|
||||
'A msg field violates the current spec\n'
|
||||
f'{codec.pld_spec}\n\n'
|
||||
f'{pretty_struct.Struct.pformat(msg)}'
|
||||
),
|
||||
src_type_error=typerr,
|
||||
)
|
||||
raise msgtyperr from typerr
|
||||
) from typerr
|
||||
|
||||
# supposedly the fastest says,
|
||||
# https://stackoverflow.com/a/54027962
|
||||
|
|
@ -764,35 +719,13 @@ class Channel:
|
|||
assert self._transport
|
||||
while True:
|
||||
try:
|
||||
async for msg in self._transport:
|
||||
match msg:
|
||||
# NOTE: if transport/interchange delivers
|
||||
# a type error, we pack it with the far
|
||||
# end peer `Actor.uid` and relay the
|
||||
# `Error`-msg upward to the `._rpc` stack
|
||||
# for normal RAE handling.
|
||||
case MsgTypeError():
|
||||
yield pack_from_raise(
|
||||
local_err=msg,
|
||||
cid=msg.cid,
|
||||
|
||||
# XXX we pack it here bc lower
|
||||
# layers have no notion of an
|
||||
# actor-id ;)
|
||||
src_uid=self.uid,
|
||||
)
|
||||
case _:
|
||||
yield msg
|
||||
|
||||
# TODO: if we were gonna do this it should be
|
||||
# done up at the `MsgStream` layer!
|
||||
#
|
||||
async for item in self._transport:
|
||||
yield item
|
||||
# sent = yield item
|
||||
# if sent is not None:
|
||||
# # optimization, passing None through all the
|
||||
# # time is pointless
|
||||
# await self._transport.send(sent)
|
||||
|
||||
except trio.BrokenResourceError:
|
||||
|
||||
# if not self._autorecon:
|
||||
|
|
|
|||
|
|
@ -46,7 +46,6 @@ from ._state import (
|
|||
from ._ipc import Channel
|
||||
from .log import get_logger
|
||||
from .msg import (
|
||||
Error,
|
||||
NamespacePath,
|
||||
Return,
|
||||
)
|
||||
|
|
@ -70,7 +69,8 @@ log = get_logger(__name__)
|
|||
# `._raise_from_no_key_in_msg()` (after tweak to
|
||||
# accept a `chan: Channel` arg) in key block!
|
||||
def _unwrap_msg(
|
||||
msg: Return|Error,
|
||||
# msg: dict[str, Any],
|
||||
msg: Return,
|
||||
channel: Channel,
|
||||
|
||||
hide_tb: bool = True,
|
||||
|
|
|
|||
|
|
@ -47,13 +47,12 @@ from ._context import (
|
|||
Context,
|
||||
)
|
||||
from ._exceptions import (
|
||||
ContextCancelled,
|
||||
ModuleNotExposed,
|
||||
MsgTypeError,
|
||||
TransportClosed,
|
||||
is_multi_cancelled,
|
||||
ContextCancelled,
|
||||
pack_error,
|
||||
unpack_error,
|
||||
TransportClosed,
|
||||
)
|
||||
from .devx import (
|
||||
maybe_wait_for_debugger,
|
||||
|
|
@ -637,7 +636,7 @@ async def _invoke(
|
|||
# (callee) task, so relay this cancel signal to the
|
||||
# other side.
|
||||
ctxc = ContextCancelled(
|
||||
message=msg,
|
||||
msg,
|
||||
boxed_type=trio.Cancelled,
|
||||
canceller=canceller,
|
||||
)
|
||||
|
|
@ -827,12 +826,7 @@ async def process_messages(
|
|||
| Stop(cid=cid)
|
||||
| Return(cid=cid)
|
||||
| CancelAck(cid=cid)
|
||||
|
||||
# `.cid` means RPC-ctx-task specific
|
||||
| Error(cid=cid)
|
||||
|
||||
# recv-side `MsgType` decode violation
|
||||
| MsgTypeError(cid=cid)
|
||||
| Error(cid=cid) # RPC-task ctx specific
|
||||
):
|
||||
# deliver response to local caller/waiter
|
||||
# via its per-remote-context memory channel.
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ from pprint import pformat
|
|||
import signal
|
||||
import sys
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
TYPE_CHECKING,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ Built-in messaging patterns, types, APIs and helpers.
|
|||
|
||||
'''
|
||||
from typing import (
|
||||
Union,
|
||||
TypeAlias,
|
||||
)
|
||||
from .ptr import (
|
||||
|
|
@ -55,9 +56,8 @@ from .types import (
|
|||
|
||||
# full msg class set from above as list
|
||||
__msg_types__ as __msg_types__,
|
||||
|
||||
# type-alias for union of all msgs
|
||||
MsgType as MsgType,
|
||||
)
|
||||
|
||||
__msg_spec__: TypeAlias = MsgType
|
||||
# TODO: use new type declaration syntax for msg-type-spec
|
||||
# https://docs.python.org/3/library/typing.html#type-aliases
|
||||
# https://docs.python.org/3/reference/simple_stmts.html#type
|
||||
__msg_spec__: TypeAlias = Union[*__msg_types__]
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ from trio.lowlevel import (
|
|||
from tractor.msg.pretty_struct import Struct
|
||||
from tractor.msg.types import (
|
||||
mk_msg_spec,
|
||||
MsgType,
|
||||
Msg,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -87,50 +87,12 @@ class MsgCodec(Struct):
|
|||
|
||||
pld_spec: Union[Type[Struct]]|None
|
||||
|
||||
@property
|
||||
def pld_spec_str(self) -> str:
|
||||
spec: Union[Type]|Type = self.pld_spec
|
||||
|
||||
# TODO: could also use match: instead?
|
||||
if getattr(spec, '__args__', False):
|
||||
# `typing.Union` case
|
||||
return str(spec)
|
||||
else:
|
||||
return spec.__name__
|
||||
|
||||
# struct type unions
|
||||
# https://jcristharif.com/msgspec/structs.html#tagged-unions
|
||||
@property
|
||||
def msg_spec(self) -> Union[Type[Struct]]:
|
||||
return self._dec.type
|
||||
|
||||
def msg_spec_items(
|
||||
self,
|
||||
msg: MsgType|None = None,
|
||||
|
||||
) -> dict[str, MsgType]|str:
|
||||
|
||||
msgt_table: dict[str, MsgType] = {
|
||||
msgt: str(msgt)
|
||||
for msgt in self.msg_spec.__args__
|
||||
}
|
||||
if msg:
|
||||
msgt: MsgType = type(msg)
|
||||
str_repr: str = msgt_table[msgt]
|
||||
return {msgt: str_repr}
|
||||
|
||||
return msgt_table
|
||||
|
||||
# TODO: some way to make `pretty_struct.Struct` use this
|
||||
# wrapped field over the `.msg_spec` one?
|
||||
def pformat_msg_spec(
|
||||
self,
|
||||
msg: MsgType|None = None,
|
||||
) -> str:
|
||||
return '\n'.join(
|
||||
self.msg_spec_items(msg=msg).values()
|
||||
)
|
||||
|
||||
lib: ModuleType = msgspec
|
||||
|
||||
# TODO: a sub-decoder system as well?
|
||||
|
|
@ -146,7 +108,7 @@ class MsgCodec(Struct):
|
|||
# OR
|
||||
# ) = {
|
||||
# # pre-seed decoders for std-py-type-set for use when
|
||||
# # `MsgType.pld == None|Any`.
|
||||
# # `Msg.pld == None|Any`.
|
||||
# None: msgpack.Decoder(Any),
|
||||
# Any: msgpack.Decoder(Any),
|
||||
# }
|
||||
|
|
@ -341,7 +303,7 @@ def mk_codec(
|
|||
# by `tag_field: str` value key?
|
||||
# payload_msg_specs: dict[
|
||||
# str, # tag_field value as sub-decoder key
|
||||
# Union[Type[Struct]] # `MsgType.pld` type spec
|
||||
# Union[Type[Struct]] # `Msg.pld` type spec
|
||||
# ]|None = None,
|
||||
|
||||
libname: str = 'msgspec',
|
||||
|
|
@ -374,7 +336,7 @@ def mk_codec(
|
|||
raise RuntimeError(
|
||||
f'If a payload spec is provided,\n'
|
||||
"the builtin SC-shuttle-protocol's msg set\n"
|
||||
f'(i.e. a `{MsgType}`) MUST be used!\n\n'
|
||||
f'(i.e. `{Msg}`) MUST be used!\n\n'
|
||||
f'However both values were passed as => mk_codec(\n'
|
||||
f' ipc_msg_spec={ipc_msg_spec}`\n'
|
||||
f' ipc_pld_spec={ipc_pld_spec}`\n)\n'
|
||||
|
|
|
|||
|
|
@ -31,7 +31,6 @@ from typing import (
|
|||
Literal,
|
||||
Type,
|
||||
TypeVar,
|
||||
TypeAlias,
|
||||
Union,
|
||||
)
|
||||
|
||||
|
|
@ -401,29 +400,16 @@ class CancelAck(
|
|||
pld: bool
|
||||
|
||||
|
||||
# TODO: unify this with `._exceptions.RemoteActorError`
|
||||
# such that we can have a msg which is both raisable and
|
||||
# IPC-wire ready?
|
||||
# B~o
|
||||
class Error(
|
||||
Struct,
|
||||
tag=True,
|
||||
tag_field='msg_type',
|
||||
|
||||
# TODO may omit defaults?
|
||||
# https://jcristharif.com/msgspec/structs.html#omitting-default-values
|
||||
# omit_defaults=True,
|
||||
):
|
||||
'''
|
||||
A pkt that wraps `RemoteActorError`s for relay and raising.
|
||||
|
||||
Fields are 1-to-1 meta-data as needed originally by
|
||||
`RemoteActorError.msgdata: dict` but now are defined here.
|
||||
|
||||
Note: this msg shuttles `ContextCancelled` and `StreamOverrun`
|
||||
as well is used to rewrap any `MsgTypeError` for relay-reponse
|
||||
to bad `Yield.pld` senders during an IPC ctx's streaming dialog
|
||||
phase.
|
||||
`RemoteActorError.msgdata: dict`.
|
||||
|
||||
'''
|
||||
src_uid: tuple[str, str]
|
||||
|
|
@ -442,10 +428,6 @@ class Error(
|
|||
# `StreamOverrun`
|
||||
sender: tuple[str, str]|None = None
|
||||
|
||||
# for the `MsgTypeError` case where the receiver side
|
||||
# decodes the underlying original `Msg`-subtype
|
||||
_msg_dict: dict|None = None
|
||||
|
||||
|
||||
# TODO: should be make a msg version of `ContextCancelled?`
|
||||
# and/or with a scope field or a full `ActorCancelled`?
|
||||
|
|
@ -504,11 +486,6 @@ __msg_types__: list[Msg] = (
|
|||
_payload_msgs
|
||||
)
|
||||
|
||||
# TODO: use new type declaration syntax for msg-type-spec
|
||||
# https://docs.python.org/3/library/typing.html#type-aliases
|
||||
# https://docs.python.org/3/reference/simple_stmts.html#type
|
||||
MsgType: TypeAlias = Union[*__msg_types__]
|
||||
|
||||
|
||||
def mk_msg_spec(
|
||||
payload_type_union: Union[Type] = Any,
|
||||
|
|
|
|||
Loading…
Reference in New Issue