Compare commits

..

1 Commits

Author SHA1 Message Date
Tyler Goodlet 9b4f580470 Go back to `ContextVar` for codec mgmt
Turns out we do want per-task inheritance particularly if there's to be
per `Context` dynamic mutation of the spec; we don't want mutation in
some task to affect any parent/global setting.

Turns out since we use a common "feeder task" in the rpc loop, we need to
offer a per `Context` payload decoder sys anyway in order to enable
per-task controls for inter-actor multi-task-ctx scenarios.
2024-04-18 16:24:59 -04:00
5 changed files with 187 additions and 287 deletions

View File

@ -24,7 +24,6 @@ import importlib
from pprint import pformat
from typing import (
Any,
Callable,
Type,
TYPE_CHECKING,
)
@ -33,11 +32,8 @@ import traceback
import trio
from msgspec import (
defstruct,
msgpack,
Raw,
structs,
ValidationError,
defstruct,
)
from tractor._state import current_actor
@ -48,8 +44,6 @@ from tractor.msg import (
Stop,
Yield,
types as msgtypes,
MsgCodec,
MsgDec,
)
from tractor.msg.pretty_struct import (
iter_fields,
@ -176,7 +170,7 @@ def pformat_boxed_tb(
f' ------ - ------\n\n'
# f'{tb_str}\n'
f'{tb_body}'
f' ------ - ------\n'
f' ------ - ------\n'
f'_|\n'
)
tb_box_indent: str = (
@ -938,7 +932,7 @@ def is_multi_cancelled(exc: BaseException) -> bool:
return False
def _raise_from_unexpected_msg(
def _raise_from_no_key_in_msg(
ctx: Context,
msg: MsgType,
src_err: AttributeError,
@ -978,6 +972,8 @@ def _raise_from_unexpected_msg(
# an internal error should never get here
try:
cid: str = msg.cid
# cid: str = msg['cid']
# except KeyError as src_err:
except AttributeError as src_err:
raise MessagingError(
f'IPC `Context` rx-ed msg without a ctx-id (cid)!?\n'
@ -989,6 +985,7 @@ def _raise_from_unexpected_msg(
# TODO: test that shows stream raising an expected error!!!
# raise the error message in a boxed exception type!
# if msg.get('error'):
if isinstance(msg, Error):
# match msg:
# case Error():
@ -1004,6 +1001,7 @@ def _raise_from_unexpected_msg(
# the stream._eoc outside this in the calleer always?
# case Stop():
elif (
# msg.get('stop')
isinstance(msg, Stop)
or (
stream
@ -1038,6 +1036,7 @@ def _raise_from_unexpected_msg(
# that arrived which is probably the source of this stream
# closure
ctx.maybe_raise()
raise eoc from src_err
if (
@ -1057,128 +1056,3 @@ def _raise_from_unexpected_msg(
" BUT received a non-error msg:\n"
f'{pformat(msg)}'
) from src_err
_raise_from_no_key_in_msg = _raise_from_unexpected_msg
def _mk_msg_type_err(
msg: Any|bytes|Raw,
codec: MsgCodec|MsgDec,
message: str|None = None,
verb_header: str = '',
src_validation_error: ValidationError|None = None,
src_type_error: TypeError|None = None,
) -> MsgTypeError:
'''
Compose a `MsgTypeError` from an input runtime context.
'''
# `Channel.send()` case
if src_validation_error is None:
if isinstance(codec, MsgDec):
raise RuntimeError(
'`codec` must be a `MsgCodec` for send-side errors?'
)
# no src error from `msgspec.msgpack.Decoder.decode()` so
# prolly a manual type-check on our part.
if message is None:
fmt_spec: str = codec.pformat_msg_spec()
fmt_stack: str = (
'\n'.join(traceback.format_stack(limit=3))
)
tb_fmt: str = pformat_boxed_tb(
tb_str=fmt_stack,
# fields_str=header,
field_prefix=' ',
indent='',
)
message: str = (
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'
)
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
# `msgspec.Struct` handling) so that we can
# determine what `.msg.types.Msg` is the culprit
# by reporting the received value.
msg_dict: dict = msgpack.decode(msg)
msg_type_name: str = msg_dict['msg_type']
msg_type = getattr(msgtypes, msg_type_name)
message: str = (
f'invalid `{msg_type_name}` IPC msg\n\n'
)
if verb_header:
message = f'{verb_header} ' + message
# 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('`')
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')>'
)
msgtyperr = MsgTypeError.from_decode(
message=message,
msgdict=msg_dict,
)
msgtyperr.__cause__ = src_validation_error
return msgtyperr

View File

@ -49,7 +49,6 @@ from tractor._exceptions import (
MsgTypeError,
pack_from_raise,
TransportClosed,
_mk_msg_type_err,
)
from tractor.msg import (
_ctxvar_MsgCodec,
@ -119,6 +118,127 @@ class MsgTransport(Protocol[MsgType]):
...
def _mk_msg_type_err(
msg: Any|bytes,
codec: MsgCodec,
message: str|None = None,
verb_header: str = '',
src_validation_error: msgspec.ValidationError|None = None,
src_type_error: TypeError|None = None,
) -> MsgTypeError:
import textwrap
# `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
fmt_spec: str = '\n'.join(
map(str, codec.msg_spec.__args__)
)
fmt_stack: str = (
'\n'.join(traceback.format_stack(limit=3))
)
tb_fmt: str = pformat_boxed_tb(
tb_str=fmt_stack,
# fields_str=header,
field_prefix=' ',
indent='',
)
message: str = (
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'
)
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
# `msgspec.Struct` handling) so that we can
# determine what `.msg.types.Msg` is the culprit
# by reporting the received value.
msg_dict: dict = msgspec.msgpack.decode(msg)
msg_type_name: str = msg_dict['msg_type']
msg_type = getattr(msgtypes, msg_type_name)
message: str = (
f'invalid `{msg_type_name}` IPC msg\n\n'
)
if verb_header:
message = f'{verb_header} ' + message
# 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('`')
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')>'
)
msgtyperr = MsgTypeError.from_decode(
message=message,
msgdict=msg_dict,
)
msgtyperr.__cause__ = src_validation_error
return msgtyperr
# TODO: not sure why we have to inherit here, but it seems to be an
# issue with ``get_msg_transport()`` returning a ``Type[Protocol]``;
# probably should make a `mypy` issue?
@ -179,10 +299,10 @@ class MsgpackTCPStream(MsgTransport):
_codec._ctxvar_MsgCodec.get()
)
# TODO: mask out before release?
# log.runtime(
# f'New {self} created with codec\n'
# f'codec: {self._codec}\n'
# )
log.runtime(
f'New {self} created with codec\n'
f'codec: {self._codec}\n'
)
async def _iter_packets(self) -> AsyncGenerator[dict, None]:
'''

View File

@ -34,7 +34,6 @@ from ._codec import (
apply_codec as apply_codec,
mk_codec as mk_codec,
MsgCodec as MsgCodec,
MsgDec as MsgDec,
current_codec as current_codec,
)
@ -51,7 +50,6 @@ from .types import (
Yield as Yield,
Stop as Stop,
Return as Return,
CancelAck as CancelAck,
Error as Error,

View File

@ -71,108 +71,6 @@ if TYPE_CHECKING:
log = get_logger(__name__)
# TODO: unify with `MsgCodec` by making `._dec` part this?
class MsgDec(Struct):
'''
An IPC msg decoder.
Normally used to decode only a payload: `MsgType.pld:
PayloadT` field before delivery to IPC consumer code.
'''
_dec: msgpack.Decoder
@property
def dec(self) -> msgpack.Decoder:
return self._dec
# struct type unions
# https://jcristharif.com/msgspec/structs.html#tagged-unions
#
# ^-TODO-^: make a wrapper type for this such that alt
# backends can be represented easily without a `Union` needed,
# AND so that we have better support for wire transport.
#
# -[ ] maybe `FieldSpec` is a good name since msg-spec
# better applies to a `MsgType[FieldSpec]`?
#
# -[ ] both as part of the `.open_context()` call AND as part of the
# immediate ack-reponse (see similar below)
# we should do spec matching and fail if anything is awry?
#
# -[ ] eventually spec should be generated/parsed from the
# type-annots as # desired in GH issue:
# https://github.com/goodboy/tractor/issues/365
#
# -[ ] semantics of the mismatch case
# - when caller-callee specs we should raise
# a `MsgTypeError` or `MsgSpecError` or similar?
#
# -[ ] wrapper types for both spec types such that we can easily
# IPC transport them?
# - `TypeSpec: Union[Type]`
# * also a `.__contains__()` for doing `None in
# TypeSpec[None|int]` since rn you need to do it on
# `.__args__` for unions..
# - `MsgSpec: Union[Type[Msg]]
#
# -[ ] auto-genning this from new (in 3.12) type parameter lists Bo
# |_ https://docs.python.org/3/reference/compound_stmts.html#type-params
# |_ historical pep 695: https://peps.python.org/pep-0695/
# |_ full lang spec: https://typing.readthedocs.io/en/latest/spec/
# |_ on annotation scopes:
# https://docs.python.org/3/reference/executionmodel.html#annotation-scopes
# |_ 3.13 will have subscriptable funcs Bo
# https://peps.python.org/pep-0718/
@property
def spec(self) -> Union[Type[Struct]]:
# NOTE: defined and applied inside `mk_codec()`
return self._dec.type
# no difference, as compared to a `MsgCodec` which defines the
# `MsgType.pld: PayloadT` part of its spec separately
pld_spec = spec
# TODO: would get moved into `FieldSpec.__str__()` right?
@property
def spec_str(self) -> str:
# TODO: could also use match: instead?
spec: Union[Type]|Type = self.spec
# `typing.Union` case
if getattr(spec, '__args__', False):
return str(spec)
# just a single type
else:
return spec.__name__
pld_spec_str = spec_str
def decode(
self,
raw: Raw|bytes,
) -> Any:
return self._dec.decode(raw)
@property
def hook(self) -> Callable|None:
return self._dec.dec_hook
def mk_dec(
spec: Union[Type[Struct]]|Any = Any,
dec_hook: Callable|None = None,
) -> MsgDec:
return msgpack.Decoder(
type=spec, # like `Msg[Any]`
dec_hook=dec_hook,
)
# TODO: overall IPC msg-spec features (i.e. in this mod)!
#
# -[ ] API changes towards being interchange lib agnostic!
@ -196,7 +94,8 @@ class MsgCodec(Struct):
'''
_enc: msgpack.Encoder
_dec: msgpack.Decoder
_pld_spec: Type[Struct]|Raw|Any
pld_spec: Union[Type[Struct]]|None
def __repr__(self) -> str:
speclines: str = textwrap.indent(
@ -219,21 +118,14 @@ class MsgCodec(Struct):
')>'
)
@property
def pld_spec(self) -> Type[Struct]|Raw|Any:
return self._pld_spec
@property
def pld_spec_str(self) -> str:
# TODO: could also use match: instead?
spec: Union[Type]|Type = self.pld_spec
# `typing.Union` case
# TODO: could also use match: instead?
if getattr(spec, '__args__', False):
# `typing.Union` case
return str(spec)
# just a single type
else:
return spec.__name__
@ -241,7 +133,6 @@ class MsgCodec(Struct):
# https://jcristharif.com/msgspec/structs.html#tagged-unions
@property
def msg_spec(self) -> Union[Type[Struct]]:
# NOTE: defined and applied inside `mk_codec()`
return self._dec.type
def msg_spec_items(
@ -266,9 +157,8 @@ class MsgCodec(Struct):
def pformat_msg_spec(
self,
msg: MsgType|None = None,
join_char: str = '\n',
) -> str:
return join_char.join(
return '\n'.join(
self.msg_spec_items(msg=msg).values()
)
@ -515,25 +405,18 @@ def mk_codec(
assert len(ipc_msg_spec.__args__) == len(msg_types)
assert ipc_msg_spec
# TODO: use this shim instead?
# bc.. unification, err somethin?
# dec: MsgDec = mk_dec(
# spec=ipc_msg_spec,
# dec_hook=dec_hook,
# )
dec = msgpack.Decoder(
type=ipc_msg_spec,
dec_hook=dec_hook,
)
enc = msgpack.Encoder(
enc_hook=enc_hook,
)
dec = msgpack.Decoder(
type=ipc_msg_spec, # like `Msg[Any]`
dec_hook=dec_hook,
)
codec = MsgCodec(
_enc=enc,
_dec=dec,
_pld_spec=ipc_pld_spec,
pld_spec=ipc_pld_spec,
)
# sanity on expected backend support
@ -552,9 +435,10 @@ _def_msgspec_codec: MsgCodec = mk_codec(ipc_pld_spec=Any)
# https://jcristharif.com/msgspec/supported-types.html
#
_def_tractor_codec: MsgCodec = mk_codec(
ipc_pld_spec=Any,
# TODO: use this for debug mode locking prot?
# ipc_pld_spec=Any,
ipc_pld_spec=Raw,
# ipc_pld_spec=Raw,
)
# TODO: IDEALLY provides for per-`trio.Task` specificity of the
# IPC msging codec used by the transport layer when doing
@ -654,12 +538,12 @@ def apply_codec(
yield var.get()
finally:
var.reset(token)
log.info(
'Reverted to last msg-spec codec\n\n'
f'{orig}\n'
)
assert var.get() is orig
assert var.get() is orig
log.info(
'Reverted to last msg-spec codec\n\n'
f'{orig}\n'
)
def current_codec() -> MsgCodec:
'''
@ -690,7 +574,7 @@ def limit_msg_spec(
'''
__tracebackhide__: bool = True
curr_codec: MsgCodec = current_codec()
curr_codec = current_codec()
msgspec_codec: MsgCodec = mk_codec(
ipc_pld_spec=payload_spec,
**codec_kwargs,

View File

@ -38,7 +38,6 @@ from typing import (
from msgspec import (
defstruct,
# field,
Raw,
Struct,
# UNSET,
# UnsetType,
@ -106,7 +105,7 @@ class Msg(
# TODO: could also be set to `msgspec.Raw` if the sub-decoders
# approach is preferred over the generic parameterization
# approach as take by `mk_msg_spec()` below.
pld: PayloadT|Raw
pld: PayloadT
class Aid(
@ -266,7 +265,35 @@ class Start(
# TODO: enforcing a msg-spec in terms `Msg.pld`
# parameterizable msgs to be used in the appls IPC dialog.
# => SEE `._codec.MsgDec` for more <=
#
# -[ ] both as part of the `.open_context()` call AND as part of the
# immediate ack-reponse (see similar below)
# we should do spec matching and fail if anything is awry?
#
# -[ ] eventually spec should be generated/parsed from the
# type-annots as # desired in GH issue:
# https://github.com/goodboy/tractor/issues/365
#
# -[ ] semantics of the mismatch case
# - when caller-callee specs we should raise
# a `MsgTypeError` or `MsgSpecError` or similar?
#
# -[ ] wrapper types for both spec types such that we can easily
# IPC transport them?
# - `TypeSpec: Union[Type]`
# * also a `.__contains__()` for doing `None in
# TypeSpec[None|int]` since rn you need to do it on
# `.__args__` for unions..
# - `MsgSpec: Union[Type[Msg]]
#
# -[ ] auto-genning this from new (in 3.12) type parameter lists Bo
# |_ https://docs.python.org/3/reference/compound_stmts.html#type-params
# |_ historical pep 695: https://peps.python.org/pep-0695/
# |_ full lang spec: https://typing.readthedocs.io/en/latest/spec/
# |_ on annotation scopes:
# https://docs.python.org/3/reference/executionmodel.html#annotation-scopes
# |_ 3.13 will have subscriptable funcs Bo
# https://peps.python.org/pep-0718/
pld_spec: str = str(Any)
@ -305,7 +332,7 @@ class Started(
decorated IPC endpoint.
'''
pld: PayloadT|Raw
pld: PayloadT
# TODO: instead of using our existing `Start`
@ -322,7 +349,7 @@ class Yield(
Per IPC transmission of a value from `await MsgStream.send(<value>)`.
'''
pld: PayloadT|Raw
pld: PayloadT
class Stop(
@ -350,12 +377,11 @@ class Return(
func-as-`trio.Task`.
'''
pld: PayloadT|Raw
pld: PayloadT
class CancelAck(
Msg,
Generic[PayloadT],
Return,
):
'''
Deliver the `bool` return-value from a cancellation `Actor`
@ -684,9 +710,7 @@ def mk_msg_spec(
)
return (
ipc_spec,
msgtypes_table[spec_build_method]
+
ipc_msg_types,
msgtypes_table[spec_build_method] + ipc_msg_types,
)