Compare commits

..

No commits in common. "67f673bf36bf2625d1c1aaa488910f5b710ec30a" and "3ba46362a94175d8b3a1650c14b9104fac8bd7e9" have entirely different histories.

7 changed files with 173 additions and 600 deletions

View File

@ -7,6 +7,7 @@ B~)
'''
from typing import (
Any,
_GenericAlias,
Type,
Union,
)
@ -25,23 +26,20 @@ from msgspec import (
import pytest
import tractor
from tractor.msg import (
_codec,
_def_msgspec_codec,
_ctxvar_MsgCodec,
NamespacePath,
MsgCodec,
mk_codec,
apply_codec,
current_codec,
current_msgspec_codec,
)
from tractor.msg import (
types,
)
from tractor import _state
from tractor.msg import types
from tractor.msg.types import (
# PayloadT,
Msg,
Started,
# Started,
mk_msg_spec,
)
import trio
@ -62,110 +60,56 @@ def test_msg_spec_xor_pld_spec():
)
# TODO: wrap these into `._codec` such that user can just pass
# a type table of some sort?
def enc_hook(obj: Any) -> Any:
if isinstance(obj, NamespacePath):
return str(obj)
else:
raise NotImplementedError(
f'Objects of type {type(obj)} are not supported'
)
def dec_hook(type: Type, obj: Any) -> Any:
print(f'type is: {type}')
if type is NamespacePath:
return NamespacePath(obj)
else:
raise NotImplementedError(
f'Objects of type {type(obj)} are not supported'
)
def ex_func(*args):
print(f'ex_func({args})')
def mk_custom_codec(
pld_spec: Union[Type]|Any,
ipc_msg_spec: Type[Any] = Any,
) -> MsgCodec:
'''
Create custom `msgpack` enc/dec-hooks and set a `Decoder`
which only loads `NamespacePath` types.
'''
uid: tuple[str, str] = tractor.current_actor().uid
# XXX NOTE XXX: despite defining `NamespacePath` as a type
# field on our `Msg.pld`, we still need a enc/dec_hook() pair
# to cast to/from that type on the wire. See the docs:
# https://jcristharif.com/msgspec/extending.html#mapping-to-from-native-types
def enc_nsp(obj: Any) -> Any:
match obj:
case NamespacePath():
print(
f'{uid}: `NamespacePath`-Only ENCODE?\n'
f'type: {type(obj)}\n'
f'obj: {obj}\n'
)
return str(obj)
logmsg: str = (
f'{uid}: Encoding `{obj}: <{type(obj)}>` not supported'
f'type: {type(obj)}\n'
f'obj: {obj}\n'
)
print(logmsg)
raise NotImplementedError(logmsg)
def dec_nsp(
type: Type,
obj: Any,
) -> Any:
print(
f'{uid}: CUSTOM DECODE\n'
f'input type: {type}\n'
f'obj: {obj}\n'
f'type(obj): `{type(obj).__class__}`\n'
)
nsp = None
# This never seems to hit?
if isinstance(obj, Msg):
print(f'Msg type: {obj}')
if (
type is NamespacePath
and isinstance(obj, str)
and ':' in obj
):
nsp = NamespacePath(obj)
if nsp:
print(f'Returning NSP instance: {nsp}')
return nsp
logmsg: str = (
f'{uid}: Decoding `{obj}: <{type(obj)}>` not supported'
f'input type: {type(obj)}\n'
f'obj: {obj}\n'
f'type(obj): `{type(obj).__class__}`\n'
)
print(logmsg)
raise NotImplementedError(logmsg)
# apply custom hooks and set a `Decoder` which only
# loads `NamespacePath` types.
nsp_codec: MsgCodec = mk_codec(
ipc_pld_spec=pld_spec,
# NOTE XXX: the encode hook MUST be used no matter what since
# our `NamespacePath` is not any of a `Any` native type nor
# a `msgspec.Struct` subtype - so `msgspec` has no way to know
# how to encode it unless we provide the custom hook.
#
# AGAIN that is, regardless of whether we spec an
# `Any`-decoded-pld the enc has no knowledge (by default)
# how to enc `NamespacePath` (nsp), so we add a custom
# hook to do that ALWAYS.
enc_hook=enc_nsp,
# XXX NOTE: pretty sure this is mutex with the `type=` to
# `Decoder`? so it won't work in tandem with the
# `ipc_pld_spec` passed above?
dec_hook=dec_nsp,
ipc_msg_spec=ipc_msg_spec,
enc_hook=enc_hook,
dec_hook=dec_hook,
)
# TODO: validate `MsgCodec` interface/semantics?
# -[ ] simple field tests to ensure caching + reset is workin?
# -[ ] custom / changing `.decoder()` calls?
#
# dec = nsp_codec.decoder(
# types=NamespacePath,
# )
# assert nsp_codec.dec is dec
return nsp_codec
@tractor.context
async def send_back_nsp(
ctx: Context,
expect_debug: bool,
use_any_spec: bool,
ctx: tractor.Context,
) -> None:
'''
@ -173,65 +117,28 @@ async def send_back_nsp(
and ensure we can round trip a func ref with our parent.
'''
# debug mode sanity check
assert expect_debug == _state.debug_mode()
task: trio.Task = trio.lowlevel.current_task()
task_ctx: Context = task.context
assert _ctxvar_MsgCodec not in task_ctx
# task: trio.Task = trio.lowlevel.current_task()
# TreeVar
# curr_codec = _ctxvar_MsgCodec.get_in(task)
# ContextVar
# task_ctx: Context = task.context
# assert _ctxvar_MsgCodec not in task_ctx
curr_codec = _ctxvar_MsgCodec.get()
assert curr_codec is _codec._def_tractor_codec
if use_any_spec:
pld_spec = Any
else:
# NOTE: don't need the |None here since
# the parent side will never send `None` like
# we do here in the implicit return at the end of this
# `@context` body.
pld_spec = NamespacePath # |None
nsp_codec: MsgCodec = mk_custom_codec(
pld_spec=pld_spec,
)
nsp_codec: MsgCodec = mk_custom_codec()
with apply_codec(nsp_codec) as codec:
chk_codec_applied(
custom_codec=nsp_codec,
enter_value=codec,
)
# ensure roundtripping works locally
nsp = NamespacePath.from_ref(ex_func)
wire_bytes: bytes = nsp_codec.encode(
Started(
cid=ctx.cid,
pld=nsp
)
)
msg: Started = nsp_codec.decode(wire_bytes)
pld = msg.pld
assert pld == nsp
await ctx.started(nsp)
async with ctx.open_stream() as ipc:
async for msg in ipc:
if use_any_spec:
assert msg == f'{__name__}:ex_func'
assert msg == f'{__name__}:ex_func'
# TODO: as per below
# assert isinstance(msg, NamespacePath)
assert isinstance(msg, str)
else:
assert isinstance(msg, NamespacePath)
await ipc.send(msg)
# TODO: as per below
# assert isinstance(msg, NamespacePath)
assert isinstance(msg, str)
def chk_codec_applied(
@ -239,20 +146,11 @@ def chk_codec_applied(
enter_value: MsgCodec,
) -> MsgCodec:
# task: trio.Task = trio.lowlevel.current_task()
task: trio.Task = trio.lowlevel.current_task()
task_ctx: Context = task.context
# TreeVar
# curr_codec = _ctxvar_MsgCodec.get_in(task)
# ContextVar
# task_ctx: Context = task.context
# assert _ctxvar_MsgCodec in task_ctx
# curr_codec: MsgCodec = task.context[_ctxvar_MsgCodec]
# RunVar
curr_codec: MsgCodec = _ctxvar_MsgCodec.get()
last_read_codec = _ctxvar_MsgCodec.get()
assert curr_codec is last_read_codec
assert _ctxvar_MsgCodec in task_ctx
curr_codec: MsgCodec = task.context[_ctxvar_MsgCodec]
assert (
# returned from `mk_codec()`
@ -265,31 +163,14 @@ def chk_codec_applied(
curr_codec is
# public API for all of the above
current_codec()
current_msgspec_codec()
# the default `msgspec` settings
is not _codec._def_msgspec_codec
is not _codec._def_tractor_codec
is not _def_msgspec_codec
)
@pytest.mark.parametrize(
'ipc_pld_spec',
[
# _codec._def_msgspec_codec,
Any,
# _codec._def_tractor_codec,
NamespacePath|None,
],
ids=[
'any_type',
'nsp_type',
]
)
def test_codec_hooks_mod(
debug_mode: bool,
ipc_pld_spec: Union[Type]|Any,
):
def test_codec_hooks_mod():
'''
Audit the `.msg.MsgCodec` override apis details given our impl
uses `contextvars` to accomplish per `trio` task codec
@ -297,21 +178,11 @@ def test_codec_hooks_mod(
'''
async def main():
task: trio.Task = trio.lowlevel.current_task()
task_ctx: Context = task.context
assert _ctxvar_MsgCodec not in task_ctx
# task: trio.Task = trio.lowlevel.current_task()
# ContextVar
# task_ctx: Context = task.context
# assert _ctxvar_MsgCodec not in task_ctx
# TreeVar
# def_codec: MsgCodec = _ctxvar_MsgCodec.get_in(task)
def_codec = _ctxvar_MsgCodec.get()
assert def_codec is _codec._def_tractor_codec
async with tractor.open_nursery(
debug_mode=debug_mode,
) as an:
async with tractor.open_nursery() as an:
p: tractor.Portal = await an.start_actor(
'sub',
enable_modules=[__name__],
@ -321,9 +192,7 @@ def test_codec_hooks_mod(
# - codec not modified -> decode nsp as `str`
# - codec modified with hooks -> decode nsp as
# `NamespacePath`
nsp_codec: MsgCodec = mk_custom_codec(
pld_spec=ipc_pld_spec,
)
nsp_codec: MsgCodec = mk_custom_codec()
with apply_codec(nsp_codec) as codec:
chk_codec_applied(
custom_codec=nsp_codec,
@ -333,22 +202,9 @@ def test_codec_hooks_mod(
async with (
p.open_context(
send_back_nsp,
# TODO: send the original nsp here and
# test with `limit_msg_spec()` above?
expect_debug=debug_mode,
use_any_spec=(ipc_pld_spec==Any),
) as (ctx, first),
ctx.open_stream() as ipc,
):
if ipc_pld_spec is NamespacePath:
assert isinstance(first, NamespacePath)
print(
'root: ENTERING CONTEXT BLOCK\n'
f'type(first): {type(first)}\n'
f'first: {first}\n'
)
# ensure codec is still applied across
# `tractor.Context` + its embedded nursery.
chk_codec_applied(
@ -356,46 +212,23 @@ def test_codec_hooks_mod(
enter_value=codec,
)
first_nsp = NamespacePath(first)
# ensure roundtripping works
wire_bytes: bytes = nsp_codec.encode(
Started(
cid=ctx.cid,
pld=first_nsp
)
)
msg: Started = nsp_codec.decode(wire_bytes)
pld = msg.pld
assert pld == first_nsp
# try a manual decode of the started msg+pld
assert first == f'{__name__}:ex_func'
# TODO: actually get the decoder loading
# to native once we spec our SCIPP msgspec
# (structurred-conc-inter-proc-protocol)
# implemented as per,
# https://github.com/goodboy/tractor/issues/36
#
if ipc_pld_spec is NamespacePath:
assert isinstance(first, NamespacePath)
# `Any`-payload-spec case
else:
assert isinstance(first, str)
assert first == f'{__name__}:ex_func'
# assert isinstance(first, NamespacePath)
assert isinstance(first, str)
await ipc.send(first)
with trio.move_on_after(.6):
with trio.move_on_after(1):
async for msg in ipc:
print(msg)
# TODO: as per above
# assert isinstance(msg, NamespacePath)
assert isinstance(msg, str)
await ipc.send(msg)
await trio.sleep(0.1)
await p.cancel_actor()

View File

@ -845,10 +845,7 @@ async def keep_sending_from_callee(
('caller', 1, never_open_stream),
('callee', 0, keep_sending_from_callee),
],
ids=[
('caller_1buf_never_open_stream'),
('callee_0buf_keep_sending_from_callee'),
]
ids='overrun_condition={}'.format,
)
def test_one_end_stream_not_opened(
overrun_by: tuple[str, int, Callable],
@ -872,30 +869,29 @@ def test_one_end_stream_not_opened(
enable_modules=[__name__],
)
with trio.fail_after(1):
async with portal.open_context(
entrypoint,
) as (ctx, sent):
assert sent is None
async with portal.open_context(
entrypoint,
) as (ctx, sent):
assert sent is None
if 'caller' in overrunner:
if 'caller' in overrunner:
async with ctx.open_stream() as stream:
async with ctx.open_stream() as stream:
# itersend +1 msg more then the buffer size
# to cause the most basic overrun.
for i in range(buf_size):
print(f'sending {i}')
await stream.send(i)
# itersend +1 msg more then the buffer size
# to cause the most basic overrun.
for i in range(buf_size):
print(f'sending {i}')
await stream.send(i)
else:
# expect overrun error to be relayed back
# and this sleep interrupted
await trio.sleep_forever()
else:
# expect overrun error to be relayed back
# and this sleep interrupted
await trio.sleep_forever()
else:
# callee overruns caller case so we do nothing here
await trio.sleep_forever()
else:
# callee overruns caller case so we do nothing here
await trio.sleep_forever()
await portal.cancel_actor()

View File

@ -190,14 +190,11 @@ class Lock:
is_trio_main = (
# TODO: since this is private, @oremanj says
# we should just copy the impl for now..
(is_main_thread := trio._util.is_main_thread())
trio._util.is_main_thread()
and
(async_lib := sniffio.current_async_library()) == 'trio'
)
if (
not is_trio_main
and is_main_thread
):
if not is_trio_main:
log.warning(
f'Current async-lib detected by `sniffio`: {async_lib}\n'
)

View File

@ -31,24 +31,25 @@ from ._codec import (
apply_codec as apply_codec,
mk_codec as mk_codec,
MsgCodec as MsgCodec,
current_codec as current_codec,
current_msgspec_codec as current_msgspec_codec,
)
from .types import (
Msg as Msg,
Aid as Aid,
SpawnSpec as SpawnSpec,
Start as Start, # with pld
FuncSpec as FuncSpec,
Start as Start,
StartAck as StartAck,
StartAck as StartAck, # with pld
IpcCtxSpec as IpcCtxSpec,
Started as Started,
Yield as Yield,
Stop as Stop,
Return as Return,
Error as Error,
Error as Error, # with pld
ErrorData as ErrorData,
# full msg spec set
__spec__ as __spec__,

View File

@ -30,13 +30,13 @@ ToDo: backends we prolly should offer:
'''
from __future__ import annotations
from contextvars import (
ContextVar,
Token,
)
from contextlib import (
contextmanager as cm,
)
# from contextvars import (
# ContextVar,
# Token,
# )
from typing import (
Any,
Callable,
@ -47,12 +47,6 @@ from types import ModuleType
import msgspec
from msgspec import msgpack
from trio.lowlevel import (
RunVar,
RunVarToken,
)
# TODO: see notes below from @mikenerone..
# from tricycle import TreeVar
from tractor.msg.pretty_struct import Struct
from tractor.msg.types import (
@ -78,9 +72,6 @@ class MsgCodec(Struct):
'''
A IPC msg interchange format lib's encoder + decoder pair.
Pretty much nothing more then delegation to underlying
`msgspec.<interchange-protocol>.Encoder/Decoder`s for now.
'''
_enc: msgpack.Encoder
_dec: msgpack.Decoder
@ -95,6 +86,11 @@ class MsgCodec(Struct):
lib: ModuleType = msgspec
# ad-hoc type extensions
# https://jcristharif.com/msgspec/extending.html#mapping-to-from-native-types
enc_hook: Callable[[Any], Any]|None = None # coder
dec_hook: Callable[[type, Any], Any]|None = None # decoder
# TODO: a sub-decoder system as well?
# payload_msg_specs: Union[Type[Struct]] = Any
# see related comments in `.msg.types`
@ -308,8 +304,7 @@ def mk_codec(
libname: str = 'msgspec',
# proxy as `Struct(**kwargs)` for ad-hoc type extensions
# https://jcristharif.com/msgspec/extending.html#mapping-to-from-native-types
# proxy as `Struct(**kwargs)`
# ------ - ------
dec_hook: Callable|None = None,
enc_hook: Callable|None = None,
@ -394,52 +389,14 @@ def mk_codec(
# no custom structs, hooks or other special types.
_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
# to use any `msgspec` supported type as the `Msg.pld` payload,
# https://jcristharif.com/msgspec/supported-types.html
#
_def_tractor_codec: MsgCodec = mk_codec(
ipc_pld_spec=Any,
)
# TODO: IDEALLY provides for per-`trio.Task` specificity of the
# NOTE: provides for per-`trio.Task` specificity of the
# IPC msging codec used by the transport layer when doing
# `Channel.send()/.recv()` of wire data.
# ContextVar-TODO: DIDN'T WORK, kept resetting in every new task to default!?
# _ctxvar_MsgCodec: ContextVar[MsgCodec] = ContextVar(
# TreeVar-TODO: DIDN'T WORK, kept resetting in every new embedded nursery
# even though it's supposed to inherit from a parent context ???
#
# _ctxvar_MsgCodec: TreeVar[MsgCodec] = TreeVar(
#
# ^-NOTE-^: for this to work see the mods by @mikenerone from `trio` gitter:
#
# 22:02:54 <mikenerone> even for regular contextvars, all you have to do is:
# `task: Task = trio.lowlevel.current_task()`
# `task.parent_nursery.parent_task.context.run(my_ctx_var.set, new_value)`
#
# From a comment in his prop code he couldn't share outright:
# 1. For every TreeVar set in the current task (which covers what
# we need from SynchronizerFacade), walk up the tree until the
# root or finding one where the TreeVar is already set, setting
# it in all of the contexts along the way.
# 2. For each of those, we also forcibly set the values that are
# pending for child nurseries that have not yet accessed the
# TreeVar.
# 3. We similarly set the pending values for the child nurseries
# of the *current* task.
#
# TODO: STOP USING THIS, since it's basically a global and won't
# allow sub-IPC-ctxs to limit the msg-spec however desired..
_ctxvar_MsgCodec: MsgCodec = RunVar(
_ctxvar_MsgCodec: ContextVar[MsgCodec] = ContextVar(
'msgspec_codec',
# TODO: move this to our new `Msg`-spec!
# default=_def_msgspec_codec,
default=_def_tractor_codec,
default=_def_msgspec_codec,
)
@ -453,36 +410,15 @@ def apply_codec(
runtime context such that all IPC msgs are processed
with it for that task.
Uses a `tricycle.TreeVar` to ensure the scope of the codec
matches the `@cm` block and DOES NOT change to the original
(default) value in new tasks (as it does for `ContextVar`).
See the docs:
- https://tricycle.readthedocs.io/en/latest/reference.html#tree-variables
- https://github.com/oremanj/tricycle/blob/master/tricycle/_tests/test_tree_var.py
'''
orig: MsgCodec = _ctxvar_MsgCodec.get()
assert orig is not codec
token: RunVarToken = _ctxvar_MsgCodec.set(codec)
# TODO: for TreeVar approach, see docs for @cm `.being()` API:
# https://tricycle.readthedocs.io/en/latest/reference.html#tree-variables
# try:
# with _ctxvar_MsgCodec.being(codec):
# new = _ctxvar_MsgCodec.get()
# assert new is codec
# yield codec
token: Token = _ctxvar_MsgCodec.set(codec)
try:
yield _ctxvar_MsgCodec.get()
finally:
_ctxvar_MsgCodec.reset(token)
assert _ctxvar_MsgCodec.get() is orig
def current_codec() -> MsgCodec:
def current_msgspec_codec() -> MsgCodec:
'''
Return the current `trio.Task.context`'s value
for `msgspec_codec` used by `Channel.send/.recv()`
@ -513,6 +449,5 @@ def limit_msg_spec(
payload_types=payload_types,
**codec_kwargs,
)
with apply_codec(msgspec_codec) as applied_codec:
assert applied_codec is msgspec_codec
with apply_codec(msgspec_codec):
yield msgspec_codec

View File

@ -80,28 +80,6 @@ class DiffDump(UserList):
return repstr
def iter_fields(struct: Struct) -> Iterator[
tuple[
structs.FieldIinfo,
str,
Any,
]
]:
'''
Iterate over all non-@property fields of this struct.
'''
fi: structs.FieldInfo
for fi in structs.fields(struct):
key: str = fi.name
val: Any = getattr(struct, key)
yield (
fi,
key,
val,
)
class Struct(
_Struct,
@ -113,6 +91,23 @@ class Struct(
A "human friendlier" (aka repl buddy) struct subtype.
'''
def _sin_props(self) -> Iterator[
tuple[
structs.FieldIinfo,
str,
Any,
]
]:
'''
Iterate over all non-@property fields of this struct.
'''
fi: structs.FieldInfo
for fi in structs.fields(self):
key: str = fi.name
val: Any = getattr(self, key)
yield fi, key, val
def to_dict(
self,
include_non_members: bool = True,
@ -135,7 +130,7 @@ class Struct(
# added as type-defined `@property` methods!
sin_props: dict = {}
fi: structs.FieldInfo
for fi, k, v in iter_fields(self):
for fi, k, v in self._sin_props():
sin_props[k] = asdict[k]
return sin_props
@ -164,7 +159,7 @@ class Struct(
fi: structs.FieldInfo
k: str
v: Any
for fi, k, v in iter_fields(self):
for fi, k, v in self._sin_props():
# TODO: how can we prefer `Literal['option1', 'option2,
# ..]` over .__name__ == `Literal` but still get only the

View File

@ -26,7 +26,6 @@ from __future__ import annotations
import types
from typing import (
Any,
Callable,
Generic,
Literal,
Type,
@ -38,12 +37,8 @@ from msgspec import (
defstruct,
# field,
Struct,
# UNSET,
# UnsetType,
)
from tractor.msg import (
pretty_struct,
UNSET,
UnsetType,
)
# type variable for the boxed payload field `.pld`
@ -53,19 +48,11 @@ PayloadT = TypeVar('PayloadT')
class Msg(
Struct,
Generic[PayloadT],
# https://jcristharif.com/msgspec/structs.html#tagged-unions
tag=True,
tag_field='msg_type',
# https://jcristharif.com/msgspec/structs.html#field-ordering
# kw_only=True,
# https://jcristharif.com/msgspec/structs.html#equality-and-order
# eq=True,
# order=True,
# https://jcristharif.com/msgspec/structs.html#encoding-decoding-as-arrays
# as_array=True,
):
'''
The "god" boxing msg type.
@ -103,53 +90,6 @@ class Msg(
pld: PayloadT
class Aid(
Struct,
tag=True,
tag_field='msg_type',
):
'''
Actor-identity msg.
Initial contact exchange enabling an actor "mailbox handshake"
delivering the peer identity (and maybe eventually contact)
info.
Used by discovery protocol to register actors as well as
conduct the initial comms (capability) filtering.
'''
name: str
uuid: str
# TODO: use built-in support for UUIDs?
# -[ ] `uuid.UUID` which has multi-protocol support
# https://jcristharif.com/msgspec/supported-types.html#uuid
class SpawnSpec(
pretty_struct.Struct,
tag=True,
tag_field='msg_type',
):
'''
Initial runtime spec handed down from a spawning parent to its
child subactor immediately following first contact via an
`Aid` msg.
'''
_parent_main_data: dict
_runtime_vars: dict[str, Any]
# module import capability
enable_modules: dict[str, str]
# TODO: not just sockaddr pairs?
# -[ ] abstract into a `TransportAddr` type?
reg_addrs: list[tuple[str, int]]
bind_addrs: list[tuple[str, int]]
# TODO: caps based RPC support in the payload?
#
# -[ ] integration with our ``enable_modules: list[str]`` caps sys.
@ -165,31 +105,18 @@ class SpawnSpec(
#
# -[ ] can we combine .ns + .func into a native `NamespacePath` field?
#
# -[ ] better name, like `Call/TaskInput`?
# -[ ]better name, like `Call/TaskInput`?
#
# -[ ] XXX a debugger lock msg transaction with payloads like,
# child -> `.pld: DebugLock` -> root
# child <- `.pld: DebugLocked` <- root
# child -> `.pld: DebugRelease` -> root
#
# WHY => when a pld spec is provided it might not allow for
# debug mode msgs as they currently are (using plain old `pld.
# str` payloads) so we only when debug_mode=True we need to
# union in this debugger payload set?
#
# mk_msg_spec(
# MyPldSpec,
# debug_mode=True,
# ) -> (
# Union[MyPldSpec]
# | Union[DebugLock, DebugLocked, DebugRelease]
# )
class FuncSpec(Struct):
ns: str
func: str
kwargs: dict
uid: str # (calling) actor-id
class Start(
Struct,
tag=True,
tag_field='msg_type',
Msg,
):
'''
Initial request to remotely schedule an RPC `trio.Task` via
@ -207,26 +134,14 @@ class Start(
- `Context.open_context()`
'''
cid: str
ns: str
func: str
kwargs: dict
uid: tuple[str, str] # (calling) actor-id
pld: FuncSpec
class StartAck(
Struct,
tag=True,
tag_field='msg_type',
):
class IpcCtxSpec(Struct):
'''
Init response to a `Cmd` request indicating the far
end's RPC spec, namely its callable "type".
An inter-actor-`trio.Task`-comms `Context` spec.
'''
cid: str
# TODO: maybe better names for all these?
# -[ ] obvi ^ would need sync with `._rpc`
functype: Literal[
@ -245,6 +160,18 @@ class StartAck(
# msgspec: MsgSpec
class StartAck(
Msg,
Generic[PayloadT],
):
'''
Init response to a `Cmd` request indicating the far
end's RPC callable "type".
'''
pld: IpcCtxSpec
class Started(
Msg,
Generic[PayloadT],
@ -275,19 +202,13 @@ class Yield(
pld: PayloadT
class Stop(
Struct,
tag=True,
tag_field='msg_type',
):
class Stop(Msg):
'''
Stream termination signal much like an IPC version
of `StopAsyncIteration`.
'''
cid: str
# TODO: do we want to support a payload on stop?
# pld: UnsetType = UNSET
pld: UnsetType = UNSET
class Return(
@ -302,33 +223,32 @@ class Return(
pld: PayloadT
class Error(
Struct,
tag=True,
tag_field='msg_type',
):
class ErrorData(Struct):
'''
A pkt that wraps `RemoteActorError`s for relay and raising.
Fields are 1-to-1 meta-data as needed originally by
Remote actor error meta-data as needed originally by
`RemoteActorError.msgdata: dict`.
'''
src_uid: tuple[str, str]
src_uid: str
src_type_str: str
boxed_type_str: str
relay_path: list[tuple[str, str]]
relay_path: list[str]
tb_str: str
cid: str|None = None
# TODO: use UNSET or don't include them via
#
# `ContextCancelled`
canceller: tuple[str, str]|None = None
canceller: str|None = None
# `StreamOverrun`
sender: tuple[str, str]|None = None
sender: str|None = None
class Error(Msg):
'''
A pkt that wraps `RemoteActorError`s for relay.
'''
pld: ErrorData
# TODO: should be make a msg version of `ContextCancelled?`
@ -345,12 +265,6 @@ class Error(
# approx order of the IPC txn-state spaces.
__spec__: list[Msg] = [
# identity handshake
Aid,
# spawn specification from parent
SpawnSpec,
# inter-actor RPC initiation
Start,
StartAck,
@ -366,8 +280,6 @@ __spec__: list[Msg] = [
]
_runtime_spec_msgs: list[Msg] = [
Aid,
SpawnSpec,
Start,
StartAck,
Stop,
@ -531,99 +443,3 @@ def mk_msg_spec(
pld_spec | runtime_spec,
msgtypes_table[spec_build_method] + ipc_msg_types,
)
# TODO: make something similar to this inside `._codec` such that
# user can just pass a type table of some sort?
# def mk_dict_msg_codec_hooks() -> tuple[Callable, Callable]:
# '''
# Deliver a `enc_hook()`/`dec_hook()` pair which does
# manual convertion from our above native `Msg` set
# to `dict` equivalent (wire msgs) in order to keep legacy compat
# with the original runtime implementation.
# Note: this is is/was primarly used while moving the core
# runtime over to using native `Msg`-struct types wherein we
# start with the send side emitting without loading
# a typed-decoder and then later flipping the switch over to
# load to the native struct types once all runtime usage has
# been adjusted appropriately.
# '''
# def enc_to_dict(msg: Any) -> Any:
# '''
# Encode `Msg`-structs to `dict` msgs instead
# of using `msgspec.msgpack.Decoder.type`-ed
# features.
# '''
# match msg:
# case Start():
# dctmsg: dict = pretty_struct.Struct.to_dict(
# msg
# )['pld']
# case Error():
# dctmsg: dict = pretty_struct.Struct.to_dict(
# msg
# )['pld']
# return {'error': dctmsg}
# def dec_from_dict(
# type: Type,
# obj: Any,
# ) -> Any:
# '''
# Decode to `Msg`-structs from `dict` msgs instead
# of using `msgspec.msgpack.Decoder.type`-ed
# features.
# '''
# cid: str = obj.get('cid')
# match obj:
# case {'cmd': pld}:
# return Start(
# cid=cid,
# pld=pld,
# )
# case {'functype': pld}:
# return StartAck(
# cid=cid,
# functype=pld,
# # pld=IpcCtxSpec(
# # functype=pld,
# # ),
# )
# case {'started': pld}:
# return Started(
# cid=cid,
# pld=pld,
# )
# case {'yield': pld}:
# return Yield(
# cid=obj['cid'],
# pld=pld,
# )
# case {'stop': pld}:
# return Stop(
# cid=cid,
# )
# case {'return': pld}:
# return Return(
# cid=cid,
# pld=pld,
# )
# case {'error': pld}:
# return Error(
# cid=cid,
# pld=ErrorData(
# **pld
# ),
# )
# return (
# # enc_to_dict,
# dec_from_dict,
# )