jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

Use `NamespacePath` in `Context` mgmt internals 

The only case where we can't is in `Portal.run_from_ns()` usage (since we
pass a path with `self:<Actor.meth>`) and because `.to_tuple()`
internally uses `.load_ref()` which will of course fail on such a path..

So or now impl as,
- mk `Actor.start_remote_task()` take a `nsf: NamespacePath` but also
  offer a `load_nsf: bool = False` such that by default we bypass ref
  loading (maybe this is fine for perf long run as well?) for the
  `Actor`/'self:'` case mentioned above.
- mk `.get_context()` take an instance `nsf` obvi.

More logging msg format tweaks:
- change msg-flow related content to show the `Context._nsf`, which,
  right, is coming follow up commit..
- bunch more `.runtime()` format updates to show `msg: dict` contents
  and internal primitives with trailing `'\n'` for easier reading.
- report import loading `stackscope` in subactors.
modden_spawn_from_client_req
Tyler Goodlet 2024-02-20 15:59:55 -05:00
parent 20a089c331
commit 621b252b0c
1 changed files with 124 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
tractor/_runtime.py
View File

@ -48,15 +48,12 @@ import trio
from trio import ( from trio import (
CancelScope, CancelScope,
) )
from trio.lowlevel import (
current_task,
Task,
)
from trio_typing import ( from trio_typing import (
Nursery, Nursery,
TaskStatus, TaskStatus,
) )
from .msg import NamespacePath
from ._ipc import Channel from ._ipc import Channel
from ._context import ( from ._context import (
mk_context, mk_context,
@ -145,8 +142,9 @@ async def _invoke(
cs: CancelScope | None = None cs: CancelScope | None = None
ctx = actor.get_context( ctx = actor.get_context(
chan, chan=chan,
cid, cid=cid,
nsf=NamespacePath.from_ref(func),
# We shouldn't ever need to pass this through right? # We shouldn't ever need to pass this through right?
# it's up to the soon-to-be called rpc task to # it's up to the soon-to-be called rpc task to
# open the stream with this option. # open the stream with this option.
@ -276,8 +274,8 @@ async def _invoke(
# TODO: should would be nice to have our # TODO: should would be nice to have our
# `TaskMngr` nursery here! # `TaskMngr` nursery here!
# res: Any = await coro res: Any = await coro
res = await coro ctx._result = res
# deliver final result to caller side. # deliver final result to caller side.
await chan.send({ await chan.send({
@ -315,11 +313,13 @@ async def _invoke(
# associated child isn't in debug any more # associated child isn't in debug any more
await maybe_wait_for_debugger() await maybe_wait_for_debugger()
ctx: Context = actor._contexts.pop((chan.uid, cid)) ctx: Context = actor._contexts.pop((chan.uid, cid))
log.cancel( res_msg: str = (
f'Context task was terminated:\n' 'IPC context terminated with result:\n'
f'func: {func}\n' f'result={ctx._result}\n'
f'ctx: {pformat(ctx)}' f'error={ctx._local_error}\n'
f'|_{pformat(ctx)}\n\n'
) )
log.cancel(res_msg)
if ctx.cancelled_caught: if ctx.cancelled_caught:
@ -331,7 +331,6 @@ async def _invoke(
ctx._maybe_raise_remote_err(re) ctx._maybe_raise_remote_err(re)
# fname: str = func.__name__ # fname: str = func.__name__
task: Task = current_task()
cs: CancelScope = ctx._scope cs: CancelScope = ctx._scope
if cs.cancel_called: if cs.cancel_called:
our_uid: tuple = actor.uid our_uid: tuple = actor.uid
@ -378,16 +377,16 @@ async def _invoke(
div_str + div_str +
f'<= canceller: {canceller}\n' f'<= canceller: {canceller}\n'
f'=> uid: {our_uid}\n' f'=> uid: {our_uid}\n'
f' |_ task: `{task.name}()`' f' |_{ctx._task}()\n'
) )
# TODO: does this ever get set any more or can # TODO: does this ever get set any more or can
# we remove it? # we remove it?
if ctx._cancel_msg: if ctx._cancel_msg:
msg += ( msg += (
'------ - ------\n' # '------ - ------\n'
'IPC msg:\n' # 'IPC msg:\n'
f'{ctx._cancel_msg}' f'\n{ctx._cancel_msg}'
) )
# task-contex was either cancelled by request using # task-contex was either cancelled by request using
@ -435,7 +434,12 @@ async def _invoke(
task_status.started(ctx) task_status.started(ctx)
result = await coro result = await coro
fname: str = func.__name__ fname: str = func.__name__
log.runtime(f'{fname}() result: {result}') log.runtime(
'RPC complete:\n'
f'task: {ctx._task}\n'
f'|_cid={ctx.cid}\n'
f'|_{fname}() -> {pformat(result)}\n'
)
# NOTE: only send result if we know IPC isn't down # NOTE: only send result if we know IPC isn't down
if ( if (
@ -965,7 +969,7 @@ class Actor:
# and bail after timeout (2-generals on closure). # and bail after timeout (2-generals on closure).
assert chan.msgstream assert chan.msgstream
log.runtime( log.warning(
f'Draining lingering msgs from stream {chan.msgstream}' f'Draining lingering msgs from stream {chan.msgstream}'
) )
@ -977,13 +981,24 @@ class Actor:
# making sure any RPC response to that call is # making sure any RPC response to that call is
# delivered the local calling task. # delivered the local calling task.
# TODO: factor this into a helper? # TODO: factor this into a helper?
log.runtime(f'drained {msg} for {chan.uid}') log.warning(
'Draining msg from disconnected\n'
f'peer: {chan.uid}]\n\n'
f'{pformat(msg)}\n'
)
cid = msg.get('cid') cid = msg.get('cid')
if cid: if cid:
# deliver response to local caller/waiter # deliver response to local caller/waiter
await self._push_result(chan, cid, msg) await self._push_result(
chan,
cid,
msg,
)
log.runtime('Waiting on actor nursery to exit..') log.runtime(
'Waiting on local actor nursery to exit..\n'
f'|_{local_nursery}\n'
)
await local_nursery.exited.wait() await local_nursery.exited.wait()
if disconnected: if disconnected:
@ -1167,6 +1182,7 @@ class Actor:
self, self,
chan: Channel, chan: Channel,
cid: str, cid: str,
nsf: NamespacePath,
msg_buffer_size: int | None = None, msg_buffer_size: int | None = None,
allow_overruns: bool = False, allow_overruns: bool = False,
@ -1180,11 +1196,15 @@ class Actor:
task-as-function invocation. task-as-function invocation.
''' '''
log.runtime(f"Getting result queue for {chan.uid} cid {cid}")
actor_uid = chan.uid actor_uid = chan.uid
assert actor_uid assert actor_uid
try: try:
ctx = self._contexts[(actor_uid, cid)] ctx = self._contexts[(actor_uid, cid)]
log.runtime(
f'Retreived cached IPC ctx for\n'
f'peer: {chan.uid}\n'
f'cid:{cid}\n'
)
ctx._allow_overruns = allow_overruns ctx._allow_overruns = allow_overruns
# adjust buffer size if specified # adjust buffer size if specified
@ -1193,9 +1213,15 @@ class Actor:
state.max_buffer_size = msg_buffer_size state.max_buffer_size = msg_buffer_size
except KeyError: except KeyError:
log.runtime(
f'Creating NEW IPC ctx for\n'
f'peer: {chan.uid}\n'
f'cid: {cid}\n'
)
ctx = mk_context( ctx = mk_context(
chan, chan,
cid, cid,
nsf=nsf,
msg_buffer_size=msg_buffer_size or self.msg_buffer_size, msg_buffer_size=msg_buffer_size or self.msg_buffer_size,
_allow_overruns=allow_overruns, _allow_overruns=allow_overruns,
) )
@ -1206,11 +1232,13 @@ class Actor:
async def start_remote_task( async def start_remote_task(
self, self,
chan: Channel, chan: Channel,
ns: str, nsf: NamespacePath,
func: str,
kwargs: dict, kwargs: dict,
# IPC channel config
msg_buffer_size: int | None = None, msg_buffer_size: int | None = None,
allow_overruns: bool = False, allow_overruns: bool = False,
load_nsf: bool = False,
) -> Context: ) -> Context:
''' '''
@ -1225,20 +1253,43 @@ class Actor:
cid = str(uuid.uuid4()) cid = str(uuid.uuid4())
assert chan.uid assert chan.uid
ctx = self.get_context( ctx = self.get_context(
chan, chan=chan,
cid, cid=cid,
nsf=nsf,
msg_buffer_size=msg_buffer_size, msg_buffer_size=msg_buffer_size,
allow_overruns=allow_overruns, allow_overruns=allow_overruns,
) )
log.runtime(f"Sending cmd to {chan.uid}: {ns}.{func}({kwargs})")
if (
'self' in nsf
or not load_nsf
):
ns, _, func = nsf.partition(':')
else:
# TODO: pass nsf directly over wire!
# -[ ] but, how to do `self:<Actor.meth>`??
ns, func = nsf.to_tuple()
log.runtime(
'Sending cmd to\n'
f'peer: {chan.uid} => \n'
'\n'
f'=> {ns}.{func}({kwargs})\n'
)
await chan.send( await chan.send(
{'cmd': (ns, func, kwargs, self.uid, cid)} {'cmd': (
ns,
func,
kwargs,
self.uid,
cid,
)}
) )
# Wait on first response msg and validate; this should be # Wait on first response msg and validate; this should be
# immediate. # immediate.
first_msg = await ctx._recv_chan.receive() first_msg: dict = await ctx._recv_chan.receive()
functype = first_msg.get('functype') functype: str = first_msg.get('functype')
if 'error' in first_msg: if 'error' in first_msg:
raise unpack_error(first_msg, chan) raise unpack_error(first_msg, chan)
@ -1280,14 +1331,19 @@ class Actor:
parent_data: dict[str, Any] parent_data: dict[str, Any]
parent_data = await chan.recv() parent_data = await chan.recv()
log.runtime( log.runtime(
"Received state from parent:\n" 'Received state from parent:\n\n'
f"{parent_data}" # TODO: eventually all these msgs as
# `msgspec.Struct` with a special mode that
# pformats them in multi-line mode, BUT only
# if "trace"/"util" mode is enabled?
f'{pformat(parent_data)}\n'
) )
accept_addrs: list[tuple[str, int]] = parent_data.pop('bind_addrs') accept_addrs: list[tuple[str, int]] = parent_data.pop('bind_addrs')
rvs = parent_data.pop('_runtime_vars') rvs = parent_data.pop('_runtime_vars')
if rvs['_debug_mode']: if rvs['_debug_mode']:
try: try:
log.info('Enabling `stackscope` traces on SIGUSR1')
from .devx import enable_stack_on_sig from .devx import enable_stack_on_sig
enable_stack_on_sig() enable_stack_on_sig()
except ImportError: except ImportError:
@ -1368,7 +1424,8 @@ class Actor:
for listener in listeners for listener in listeners
] ]
log.runtime( log.runtime(
f'Started tcp server(s) on {sockets}' 'Started TCP server(s)\n'
f'|_{sockets}\n'
) )
self._listeners.extend(listeners) self._listeners.extend(listeners)
@ -1923,7 +1980,7 @@ async def process_messages(
log.runtime( log.runtime(
'Entering IPC msg loop:\n' 'Entering IPC msg loop:\n'
f'peer: {chan.uid}\n' f'peer: {chan.uid}\n'
f'|_{chan}' f'|_{chan}\n'
) )
nursery_cancelled_before_task: bool = False nursery_cancelled_before_task: bool = False
msg: dict | None = None msg: dict | None = None
@ -1969,12 +2026,17 @@ async def process_messages(
if cid: if cid:
# deliver response to local caller/waiter # deliver response to local caller/waiter
# via its per-remote-context memory channel. # via its per-remote-context memory channel.
await actor._push_result(chan, cid, msg) await actor._push_result(
chan,
cid,
msg,
)
log.runtime( log.runtime(
f'Waiting on next IPC msg from {chan.uid}:\n' 'Waiting on next IPC msg from\n'
f'peer: {chan.uid}:\n'
f'|_{chan}\n'
# f'last msg: {msg}\n' # f'last msg: {msg}\n'
f'|_{chan}'
) )
continue continue
@ -1994,9 +2056,11 @@ async def process_messages(
raise exc raise exc
log.runtime( log.runtime(
f"Processing request from {actorid}\n" 'Handling RPC cmd from\n'
f"{ns}.{funcname}({kwargs})") f'peer: {actorid}\n'
'\n'
f'=> {ns}.{funcname}({kwargs})\n'
)
if ns == 'self': if ns == 'self':
if funcname == 'cancel': if funcname == 'cancel':
func: Callable = actor.cancel func: Callable = actor.cancel
@ -2105,17 +2169,18 @@ async def process_messages(
# in the lone case where a ``Context`` is not # in the lone case where a ``Context`` is not
# delivered, it's likely going to be a locally # delivered, it's likely going to be a locally
# scoped exception from ``_invoke()`` itself. # scoped exception from ``_invoke()`` itself.
if isinstance(ctx, Exception): if isinstance(err := ctx, Exception):
log.warning( log.warning(
f"Task for RPC func {func} failed with" 'Task for RPC failed?'
f"{ctx}" f'|_ {func}()\n\n'
f'{err}'
) )
continue continue
else: else:
# mark that we have ongoing rpc tasks # mark that we have ongoing rpc tasks
actor._ongoing_rpc_tasks = trio.Event() actor._ongoing_rpc_tasks = trio.Event()
log.runtime(f"RPC func is {func}")
# store cancel scope such that the rpc task can be # store cancel scope such that the rpc task can be
# cancelled gracefully if requested # cancelled gracefully if requested
@ -2126,7 +2191,10 @@ async def process_messages(
) )
log.runtime( log.runtime(
f"Waiting on next msg for {chan} from {chan.uid}") 'Waiting on next IPC msg from\n'
f'peer: {chan.uid}\n'
f'|_{chan}\n'
)
# end of async for, channel disconnect vis # end of async for, channel disconnect vis
# ``trio.EndOfChannel`` # ``trio.EndOfChannel``
@ -2143,9 +2211,12 @@ async def process_messages(
# handshake for them (yet) and instead we simply bail out of # handshake for them (yet) and instead we simply bail out of
# the message loop and expect the teardown sequence to clean # the message loop and expect the teardown sequence to clean
# up. # up.
# TODO: don't show this msg if it's an emphemeral
# discovery ep call?
log.runtime( log.runtime(
f'channel from {chan.uid} closed abruptly:\n' f'channel closed abruptly with\n'
f'-> {chan.raddr}\n' f'peer: {chan.uid}\n'
f'|_{chan.raddr}\n'
) )
# transport **was** disconnected # transport **was** disconnected
@ -2187,9 +2258,11 @@ async def process_messages(
finally: finally:
# msg debugging for when he machinery is brokey # msg debugging for when he machinery is brokey
log.runtime( log.runtime(
f'Exiting IPC msg loop with {chan.uid} ' 'Exiting IPC msg loop with\n'
f'final msg: {msg}\n' f'peer: {chan.uid}\n'
f'|_{chan}' f'|_{chan}\n\n'
'final msg:\n'
f'{pformat(msg)}\n'
) )
# transport **was not** disconnected # transport **was not** disconnected