Compare commits
5 Commits
5449bd5673
...
8477919fc9
Author | SHA1 | Date |
---|---|---|
Tyler Goodlet | 8477919fc9 | |
Tyler Goodlet | 872feef24b | |
Tyler Goodlet | affc210033 | |
Tyler Goodlet | 04bd111037 | |
Tyler Goodlet | a0ee0cc713 |
|
@ -7,9 +7,6 @@ related settings around IPC contexts.
|
||||||
from contextlib import (
|
from contextlib import (
|
||||||
asynccontextmanager as acm,
|
asynccontextmanager as acm,
|
||||||
)
|
)
|
||||||
from contextvars import (
|
|
||||||
Context,
|
|
||||||
)
|
|
||||||
|
|
||||||
from msgspec import (
|
from msgspec import (
|
||||||
Struct,
|
Struct,
|
||||||
|
@ -19,6 +16,7 @@ import trio
|
||||||
|
|
||||||
import tractor
|
import tractor
|
||||||
from tractor import (
|
from tractor import (
|
||||||
|
Context,
|
||||||
MsgTypeError,
|
MsgTypeError,
|
||||||
current_ipc_ctx,
|
current_ipc_ctx,
|
||||||
Portal,
|
Portal,
|
||||||
|
@ -35,7 +33,17 @@ from tractor.msg.types import (
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class PldMsg(Struct):
|
class PldMsg(
|
||||||
|
Struct,
|
||||||
|
|
||||||
|
# TODO: with multiple structs in-spec we need to tag them!
|
||||||
|
# -[ ] offer a built-in `PldMsg` type to inherit from which takes
|
||||||
|
# case of these details?
|
||||||
|
#
|
||||||
|
# https://jcristharif.com/msgspec/structs.html#tagged-unions
|
||||||
|
# tag=True,
|
||||||
|
# tag_field='msg_type',
|
||||||
|
):
|
||||||
field: str
|
field: str
|
||||||
|
|
||||||
|
|
||||||
|
@ -96,7 +104,9 @@ async def maybe_expect_raises(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@tractor.context
|
@tractor.context(
|
||||||
|
pld_spec=maybe_msg_spec,
|
||||||
|
)
|
||||||
async def child(
|
async def child(
|
||||||
ctx: Context,
|
ctx: Context,
|
||||||
started_value: int|PldMsg|None,
|
started_value: int|PldMsg|None,
|
||||||
|
@ -116,21 +126,19 @@ async def child(
|
||||||
assert ctx is curr_ctx
|
assert ctx is curr_ctx
|
||||||
|
|
||||||
rx: msgops.PldRx = ctx._pld_rx
|
rx: msgops.PldRx = ctx._pld_rx
|
||||||
orig_pldec: _codec.MsgDec = rx.pld_dec
|
curr_pldec: _codec.MsgDec = rx.pld_dec
|
||||||
# senity that default pld-spec should be set
|
|
||||||
assert (
|
|
||||||
rx.pld_dec
|
|
||||||
is
|
|
||||||
msgops._def_any_pldec
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
ctx_meta: dict = getattr(
|
||||||
with msgops.limit_plds(
|
child,
|
||||||
spec=maybe_msg_spec,
|
'_tractor_context_meta',
|
||||||
) as pldec:
|
None,
|
||||||
# sanity on `MsgDec` state
|
)
|
||||||
assert rx.pld_dec is pldec
|
if ctx_meta:
|
||||||
assert pldec.spec is maybe_msg_spec
|
assert (
|
||||||
|
ctx_meta['pld_spec']
|
||||||
|
is curr_pldec.spec
|
||||||
|
is curr_pldec.pld_spec
|
||||||
|
)
|
||||||
|
|
||||||
# 2 cases: hdndle send-side and recv-only validation
|
# 2 cases: hdndle send-side and recv-only validation
|
||||||
# - when `raise_on_started_mte == True`, send validate
|
# - when `raise_on_started_mte == True`, send validate
|
||||||
|
@ -211,18 +219,6 @@ async def child(
|
||||||
# msg-type-error from this RPC task ;)
|
# msg-type-error from this RPC task ;)
|
||||||
return return_value
|
return return_value
|
||||||
|
|
||||||
finally:
|
|
||||||
# sanity on `limit_plds()` reversion
|
|
||||||
assert (
|
|
||||||
rx.pld_dec
|
|
||||||
is
|
|
||||||
msgops._def_any_pldec
|
|
||||||
)
|
|
||||||
log.runtime(
|
|
||||||
'Reverted to previous pld-spec\n\n'
|
|
||||||
f'{orig_pldec}\n'
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
'return_value',
|
'return_value',
|
||||||
|
@ -321,7 +317,6 @@ def test_basic_payload_spec(
|
||||||
child,
|
child,
|
||||||
return_value=return_value,
|
return_value=return_value,
|
||||||
started_value=started_value,
|
started_value=started_value,
|
||||||
pld_spec=maybe_msg_spec,
|
|
||||||
validate_pld_spec=pld_check_started_value,
|
validate_pld_spec=pld_check_started_value,
|
||||||
) as (ctx, first),
|
) as (ctx, first),
|
||||||
):
|
):
|
||||||
|
|
|
@ -2399,7 +2399,11 @@ def mk_context(
|
||||||
# |_ `return: TypeAlias`,
|
# |_ `return: TypeAlias`,
|
||||||
# |_ `invalid_policy: str|Callable` ?
|
# |_ `invalid_policy: str|Callable` ?
|
||||||
# -[ ] prolly implement the `@acm` wrapper using
|
# -[ ] prolly implement the `@acm` wrapper using
|
||||||
# a `contextlib.ContextDecorator`?
|
# a `contextlib.ContextDecorator`, i guess not if
|
||||||
|
# we don't need an `__aexit__` block right?
|
||||||
|
# |_ de hecho, @acm can already be used as a decorator as of 3.10
|
||||||
|
# but i dunno how that's gonna play with `trio.Nursery.start[_soon]()`
|
||||||
|
# |_ https://docs.python.org/3/library/contextlib.html#using-a-context-manager-as-a-function-decorator
|
||||||
#
|
#
|
||||||
def context(
|
def context(
|
||||||
func: Callable|None = None,
|
func: Callable|None = None,
|
||||||
|
|
|
@ -441,7 +441,8 @@ class RemoteActorError(Exception):
|
||||||
|
|
||||||
for key in fields:
|
for key in fields:
|
||||||
if (
|
if (
|
||||||
key == 'relay_uid' and not self.is_inception()
|
key == 'relay_uid'
|
||||||
|
and not self.is_inception()
|
||||||
):
|
):
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
@ -1291,8 +1292,7 @@ def _mk_msg_type_err(
|
||||||
|
|
||||||
msgtyperr = MsgTypeError(
|
msgtyperr = MsgTypeError(
|
||||||
message=message,
|
message=message,
|
||||||
ipc_msg=msg,
|
_bad_msg=msg,
|
||||||
bad_msg=msg,
|
|
||||||
)
|
)
|
||||||
# ya, might be `None`
|
# ya, might be `None`
|
||||||
msgtyperr.__cause__ = src_type_error
|
msgtyperr.__cause__ = src_type_error
|
||||||
|
|
|
@ -68,6 +68,7 @@ from trio import (
|
||||||
)
|
)
|
||||||
import tractor
|
import tractor
|
||||||
from tractor.log import get_logger
|
from tractor.log import get_logger
|
||||||
|
from tractor._context import Context
|
||||||
from tractor._state import (
|
from tractor._state import (
|
||||||
current_actor,
|
current_actor,
|
||||||
is_root_process,
|
is_root_process,
|
||||||
|
@ -83,7 +84,6 @@ if TYPE_CHECKING:
|
||||||
from trio.lowlevel import Task
|
from trio.lowlevel import Task
|
||||||
from threading import Thread
|
from threading import Thread
|
||||||
from tractor._ipc import Channel
|
from tractor._ipc import Channel
|
||||||
from tractor._context import Context
|
|
||||||
from tractor._runtime import (
|
from tractor._runtime import (
|
||||||
Actor,
|
Actor,
|
||||||
)
|
)
|
||||||
|
@ -529,7 +529,10 @@ class Lock:
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@tractor.context
|
@tractor.context(
|
||||||
|
# enable the locking msgspec
|
||||||
|
pld_spec=__pld_spec__,
|
||||||
|
)
|
||||||
async def lock_stdio_for_peer(
|
async def lock_stdio_for_peer(
|
||||||
ctx: Context,
|
ctx: Context,
|
||||||
subactor_task_uid: tuple[str, int],
|
subactor_task_uid: tuple[str, int],
|
||||||
|
@ -597,12 +600,6 @@ async def lock_stdio_for_peer(
|
||||||
# scope despite the shielding we apply below.
|
# scope despite the shielding we apply below.
|
||||||
debug_lock_cs: CancelScope = ctx._scope
|
debug_lock_cs: CancelScope = ctx._scope
|
||||||
|
|
||||||
# TODO: use `.msg._ops.maybe_limit_plds()` here instead so we
|
|
||||||
# can merge into a single async with, with the
|
|
||||||
# `Lock.acquire_for_ctx()` enter below?
|
|
||||||
#
|
|
||||||
# enable the locking msgspec
|
|
||||||
with apply_debug_pldec():
|
|
||||||
async with Lock.acquire_for_ctx(ctx=ctx):
|
async with Lock.acquire_for_ctx(ctx=ctx):
|
||||||
debug_lock_cs.shield = True
|
debug_lock_cs.shield = True
|
||||||
|
|
||||||
|
@ -1037,48 +1034,6 @@ class PdbREPL(pdbp.Pdb):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
# TODO: prolly remove this and instead finally get our @context API
|
|
||||||
# supporting a msg/pld-spec via type annots as per,
|
|
||||||
# https://github.com/goodboy/tractor/issues/365
|
|
||||||
@cm
|
|
||||||
def apply_debug_pldec() -> _codec.MsgCodec:
|
|
||||||
'''
|
|
||||||
Apply the subactor TTY `Lock`-ing protocol's msgspec temporarily
|
|
||||||
(only in the current task).
|
|
||||||
|
|
||||||
'''
|
|
||||||
from tractor.msg import (
|
|
||||||
_ops as msgops,
|
|
||||||
)
|
|
||||||
cctx: Context = current_ipc_ctx()
|
|
||||||
rx: msgops.PldRx = cctx.pld_rx
|
|
||||||
orig_pldec: msgops.MsgDec = rx.pld_dec
|
|
||||||
|
|
||||||
try:
|
|
||||||
with msgops.limit_plds(
|
|
||||||
spec=__pld_spec__,
|
|
||||||
) as debug_dec:
|
|
||||||
assert (
|
|
||||||
debug_dec
|
|
||||||
is
|
|
||||||
rx.pld_dec
|
|
||||||
)
|
|
||||||
log.runtime(
|
|
||||||
'Applied `.devx._debug` pld-spec\n\n'
|
|
||||||
f'{debug_dec}\n'
|
|
||||||
)
|
|
||||||
yield debug_dec
|
|
||||||
|
|
||||||
finally:
|
|
||||||
assert (
|
|
||||||
rx.pld_dec is orig_pldec
|
|
||||||
)
|
|
||||||
log.runtime(
|
|
||||||
'Reverted to previous pld-spec\n\n'
|
|
||||||
f'{orig_pldec}\n'
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
async def request_root_stdio_lock(
|
async def request_root_stdio_lock(
|
||||||
actor_uid: tuple[str, str],
|
actor_uid: tuple[str, str],
|
||||||
task_uid: tuple[str, int],
|
task_uid: tuple[str, int],
|
||||||
|
@ -1147,6 +1102,7 @@ async def request_root_stdio_lock(
|
||||||
async with portal.open_context(
|
async with portal.open_context(
|
||||||
lock_stdio_for_peer,
|
lock_stdio_for_peer,
|
||||||
subactor_task_uid=task_uid,
|
subactor_task_uid=task_uid,
|
||||||
|
|
||||||
# NOTE: set it here in the locker request task bc it's
|
# NOTE: set it here in the locker request task bc it's
|
||||||
# possible for multiple such requests for the lock in any
|
# possible for multiple such requests for the lock in any
|
||||||
# single sub-actor AND there will be a race between when the
|
# single sub-actor AND there will be a race between when the
|
||||||
|
@ -1159,7 +1115,7 @@ async def request_root_stdio_lock(
|
||||||
# this IPC-ctx request task, NOT any other task(s)
|
# this IPC-ctx request task, NOT any other task(s)
|
||||||
# including the one that actually enters the REPL. This
|
# including the one that actually enters the REPL. This
|
||||||
# is oc desired bc ow the debugged task will msg-type-error.
|
# is oc desired bc ow the debugged task will msg-type-error.
|
||||||
pld_spec=__pld_spec__,
|
# pld_spec=__pld_spec__,
|
||||||
|
|
||||||
) as (req_ctx, status):
|
) as (req_ctx, status):
|
||||||
|
|
||||||
|
@ -2856,7 +2812,9 @@ def open_crash_handler(
|
||||||
|
|
||||||
|
|
||||||
@cm
|
@cm
|
||||||
def maybe_open_crash_handler(pdb: bool = False):
|
def maybe_open_crash_handler(
|
||||||
|
pdb: bool = False,
|
||||||
|
):
|
||||||
'''
|
'''
|
||||||
Same as `open_crash_handler()` but with bool input flag
|
Same as `open_crash_handler()` but with bool input flag
|
||||||
to allow conditional handling.
|
to allow conditional handling.
|
||||||
|
|
|
@ -27,6 +27,7 @@ from contextlib import (
|
||||||
)
|
)
|
||||||
from typing import (
|
from typing import (
|
||||||
Any,
|
Any,
|
||||||
|
Callable,
|
||||||
Type,
|
Type,
|
||||||
TYPE_CHECKING,
|
TYPE_CHECKING,
|
||||||
Union,
|
Union,
|
||||||
|
@ -138,6 +139,7 @@ class PldRx(Struct):
|
||||||
def limit_plds(
|
def limit_plds(
|
||||||
self,
|
self,
|
||||||
spec: Union[Type[Struct]],
|
spec: Union[Type[Struct]],
|
||||||
|
**dec_kwargs,
|
||||||
|
|
||||||
) -> MsgDec:
|
) -> MsgDec:
|
||||||
'''
|
'''
|
||||||
|
@ -147,7 +149,10 @@ class PldRx(Struct):
|
||||||
|
|
||||||
'''
|
'''
|
||||||
orig_dec: MsgDec = self._pld_dec
|
orig_dec: MsgDec = self._pld_dec
|
||||||
limit_dec: MsgDec = mk_dec(spec=spec)
|
limit_dec: MsgDec = mk_dec(
|
||||||
|
spec=spec,
|
||||||
|
**dec_kwargs,
|
||||||
|
)
|
||||||
try:
|
try:
|
||||||
self._pld_dec = limit_dec
|
self._pld_dec = limit_dec
|
||||||
yield limit_dec
|
yield limit_dec
|
||||||
|
@ -449,7 +454,7 @@ class PldRx(Struct):
|
||||||
@cm
|
@cm
|
||||||
def limit_plds(
|
def limit_plds(
|
||||||
spec: Union[Type[Struct]],
|
spec: Union[Type[Struct]],
|
||||||
**kwargs,
|
**dec_kwargs,
|
||||||
|
|
||||||
) -> MsgDec:
|
) -> MsgDec:
|
||||||
'''
|
'''
|
||||||
|
@ -467,7 +472,7 @@ def limit_plds(
|
||||||
|
|
||||||
with rx.limit_plds(
|
with rx.limit_plds(
|
||||||
spec=spec,
|
spec=spec,
|
||||||
**kwargs,
|
**dec_kwargs,
|
||||||
) as pldec:
|
) as pldec:
|
||||||
log.runtime(
|
log.runtime(
|
||||||
'Applying payload-decoder\n\n'
|
'Applying payload-decoder\n\n'
|
||||||
|
@ -487,7 +492,9 @@ def limit_plds(
|
||||||
async def maybe_limit_plds(
|
async def maybe_limit_plds(
|
||||||
ctx: Context,
|
ctx: Context,
|
||||||
spec: Union[Type[Struct]]|None = None,
|
spec: Union[Type[Struct]]|None = None,
|
||||||
|
dec_hook: Callable|None = None,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
|
|
||||||
) -> MsgDec|None:
|
) -> MsgDec|None:
|
||||||
'''
|
'''
|
||||||
Async compat maybe-payload type limiter.
|
Async compat maybe-payload type limiter.
|
||||||
|
@ -497,7 +504,11 @@ async def maybe_limit_plds(
|
||||||
used.
|
used.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
if spec is None:
|
if (
|
||||||
|
spec is None
|
||||||
|
and
|
||||||
|
dec_hook is None
|
||||||
|
):
|
||||||
yield None
|
yield None
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -505,7 +516,11 @@ async def maybe_limit_plds(
|
||||||
curr_ctx: Context = current_ipc_ctx()
|
curr_ctx: Context = current_ipc_ctx()
|
||||||
assert ctx is curr_ctx
|
assert ctx is curr_ctx
|
||||||
|
|
||||||
with ctx._pld_rx.limit_plds(spec=spec) as msgdec:
|
with ctx._pld_rx.limit_plds(
|
||||||
|
spec=spec,
|
||||||
|
dec_hook=dec_hook,
|
||||||
|
**kwargs,
|
||||||
|
) as msgdec:
|
||||||
yield msgdec
|
yield msgdec
|
||||||
|
|
||||||
curr_ctx: Context = current_ipc_ctx()
|
curr_ctx: Context = current_ipc_ctx()
|
||||||
|
|
Loading…
Reference in New Issue