Compare commits

..

No commits in common. "e6d4ec43b99461c58ff97b50fc5ddf37516dedf4" and "a6058d14ae8be5d9de718be5dfcfac4fc399c837" have entirely different histories.

4 changed files with 35 additions and 66 deletions

View File

@ -58,7 +58,6 @@ from typing import (
import warnings import warnings
# ------ - ------ # ------ - ------
import trio import trio
from trio.lowlevel import Task
# ------ - ------ # ------ - ------
from ._exceptions import ( from ._exceptions import (
ContextCancelled, ContextCancelled,
@ -122,7 +121,7 @@ class Unresolved:
@dataclass @dataclass
class Context: class Context:
''' '''
An inter-actor, SC transitive, `Task` communication context. An inter-actor, SC transitive, `trio.Task` communication context.
NB: This class should **never be instatiated directly**, it is allocated NB: This class should **never be instatiated directly**, it is allocated
by the runtime in 2 ways: by the runtime in 2 ways:
@ -135,7 +134,7 @@ class Context:
Allows maintaining task or protocol specific state between Allows maintaining task or protocol specific state between
2 cancel-scope-linked, communicating and parallel executing 2 cancel-scope-linked, communicating and parallel executing
`Task`s. Contexts are allocated on each side of any task `trio.Task`s. Contexts are allocated on each side of any task
RPC-linked msg dialog, i.e. for every request to a remote RPC-linked msg dialog, i.e. for every request to a remote
actor from a `Portal`. On the "callee" side a context is actor from a `Portal`. On the "callee" side a context is
always allocated inside `._rpc._invoke()`. always allocated inside `._rpc._invoke()`.
@ -215,7 +214,7 @@ class Context:
# which is exactly the primitive that allows for # which is exactly the primitive that allows for
# cross-actor-task-supervision and thus SC. # cross-actor-task-supervision and thus SC.
_scope: trio.CancelScope|None = None _scope: trio.CancelScope|None = None
_task: Task|None = None _task: trio.lowlevel.Task|None = None
# TODO: cs around result waiting so we can cancel any # TODO: cs around result waiting so we can cancel any
# permanently blocking `._rx_chan.receive()` call in # permanently blocking `._rx_chan.receive()` call in
@ -259,14 +258,14 @@ class Context:
# a call to `.cancel()` which triggers `ContextCancelled`. # a call to `.cancel()` which triggers `ContextCancelled`.
_cancel_msg: str|dict|None = None _cancel_msg: str|dict|None = None
# NOTE: this state-var is used by the runtime to determine if the # NOTE: this state var used by the runtime to determine if the
# `pdbp` REPL is allowed to engage on contexts terminated via # `pdbp` REPL is allowed to engage on contexts terminated via
# a `ContextCancelled` due to a call to `.cancel()` triggering # a `ContextCancelled` due to a call to `.cancel()` triggering
# "graceful closure" on either side: # "graceful closure" on either side:
# - `._runtime._invoke()` will check this flag before engaging # - `._runtime._invoke()` will check this flag before engaging
# the crash handler REPL in such cases where the "callee" # the crash handler REPL in such cases where the "callee"
# raises the cancellation, # raises the cancellation,
# - `.devx._debug.lock_stdio_for_peer()` will set it to `False` if # - `.devx._debug.lock_tty_for_child()` will set it to `False` if
# the global tty-lock has been configured to filter out some # the global tty-lock has been configured to filter out some
# actors from being able to acquire the debugger lock. # actors from being able to acquire the debugger lock.
_enter_debugger_on_cancel: bool = True _enter_debugger_on_cancel: bool = True
@ -862,7 +861,7 @@ class Context:
) -> None: ) -> None:
''' '''
Cancel this inter-actor IPC context by requestng the Cancel this inter-actor IPC context by requestng the
remote side's cancel-scope-linked `Task` by calling remote side's cancel-scope-linked `trio.Task` by calling
`._scope.cancel()` and delivering an `ContextCancelled` `._scope.cancel()` and delivering an `ContextCancelled`
ack msg in reponse. ack msg in reponse.
@ -1031,7 +1030,7 @@ class Context:
# XXX NOTE XXX: `ContextCancelled`/`StreamOverrun` absorption # XXX NOTE XXX: `ContextCancelled`/`StreamOverrun` absorption
# for "graceful cancellation" case: # for "graceful cancellation" case:
# #
# Whenever a "side" of a context (a `Task` running in # Whenever a "side" of a context (a `trio.Task` running in
# an actor) **is** the side which requested ctx # an actor) **is** the side which requested ctx
# cancellation (likekly via ``Context.cancel()``), we # cancellation (likekly via ``Context.cancel()``), we
# **don't** want to re-raise any eventually received # **don't** want to re-raise any eventually received
@ -1090,8 +1089,7 @@ class Context:
else: else:
log.warning( log.warning(
'Local error already set for ctx?\n' 'Local error already set for ctx?\n'
f'{self._local_error}\n\n' f'{self._local_error}\n'
f'{self}'
) )
return remote_error return remote_error
@ -2119,9 +2117,8 @@ async def open_context_from_portal(
# the `ContextCancelled` "self cancellation absorbed" case # the `ContextCancelled` "self cancellation absorbed" case
# handled in the block above ^^^ !! # handled in the block above ^^^ !!
# await _debug.pause() # await _debug.pause()
# log.cancel( log.cancel(
log.exception( 'Context terminated due to\n\n'
f'{ctx.side}-side of `Context` terminated with '
f'.outcome => {ctx.repr_outcome()}\n' f'.outcome => {ctx.repr_outcome()}\n'
) )
@ -2322,7 +2319,7 @@ async def open_context_from_portal(
# type_only=True, # type_only=True,
) )
log.cancel( log.cancel(
f'Context terminated due to local {ctx.side!r}-side error:\n\n' f'Context terminated due to local scope error:\n\n'
f'{ctx.chan.uid} => {outcome_str}\n' f'{ctx.chan.uid} => {outcome_str}\n'
) )
@ -2388,25 +2385,15 @@ def mk_context(
# TODO: use the new type-parameters to annotate this in 3.13? # TODO: use the new type-parameters to annotate this in 3.13?
# -[ ] https://peps.python.org/pep-0718/#unknown-types # -[ ] https://peps.python.org/pep-0718/#unknown-types
# -[ ] allow for `pld_spec` input(s) ideally breaking down,
# |_ `start: ParameterSpec`,
# |_ `started: TypeAlias`,
# |_ `yields: TypeAlias`,
# |_ `return: TypeAlias`,
# |_ `invalid_policy: str|Callable` ?
# -[ ] prolly implement the `@acm` wrapper using
# a `contextlib.ContextDecorator`?
#
def context( def context(
func: Callable, func: Callable,
) -> Callable: ) -> Callable:
''' '''
Mark an async function as an SC-supervised, inter-`Actor`, RPC Mark an (async) function as an SC-supervised, inter-`Actor`,
scheduled child-side `Task`, IPC endpoint otherwise child-`trio.Task`, IPC endpoint otherwise known more
known more colloquially as a (RPC) "context". colloquially as a (RPC) "context".
Functions annotated the fundamental IPC endpoint type offered by Functions annotated the fundamental IPC endpoint type offered by `tractor`.
`tractor`.
''' '''
# TODO: apply whatever solution ``mypy`` ends up picking for this: # TODO: apply whatever solution ``mypy`` ends up picking for this:

View File

@ -80,7 +80,6 @@ async def open_root_actor(
# enables the multi-process debugger support # enables the multi-process debugger support
debug_mode: bool = False, debug_mode: bool = False,
maybe_enable_greenback: bool = False, # `.pause_from_sync()/breakpoint()` support maybe_enable_greenback: bool = False, # `.pause_from_sync()/breakpoint()` support
enable_stack_on_sig: bool = False,
# internal logging # internal logging
loglevel: str|None = None, loglevel: str|None = None,
@ -120,7 +119,7 @@ async def open_root_actor(
) )
): ):
os.environ['PYTHONBREAKPOINT'] = ( os.environ['PYTHONBREAKPOINT'] = (
'tractor.devx._debug._sync_pause_from_builtin' 'tractor.devx._debug.pause_from_sync'
) )
_state._runtime_vars['use_greenback'] = True _state._runtime_vars['use_greenback'] = True
@ -221,11 +220,7 @@ async def open_root_actor(
assert _log assert _log
# TODO: factor this into `.devx._stackscope`!! # TODO: factor this into `.devx._stackscope`!!
if ( if debug_mode:
debug_mode
and
enable_stack_on_sig
):
try: try:
logger.info('Enabling `stackscope` traces on SIGUSR1') logger.info('Enabling `stackscope` traces on SIGUSR1')
from .devx import enable_stack_on_sig from .devx import enable_stack_on_sig

View File

@ -26,7 +26,6 @@ from contextlib import (
from functools import partial from functools import partial
import inspect import inspect
from pprint import pformat from pprint import pformat
import traceback
from typing import ( from typing import (
Any, Any,
Callable, Callable,
@ -48,7 +47,6 @@ from ._context import (
) )
from ._exceptions import ( from ._exceptions import (
ContextCancelled, ContextCancelled,
RemoteActorError,
ModuleNotExposed, ModuleNotExposed,
MsgTypeError, MsgTypeError,
TransportClosed, TransportClosed,
@ -200,8 +198,7 @@ async def _invoke_non_context(
raise ipc_err raise ipc_err
else: else:
log.exception( log.exception(
f'Failed to ack runtime RPC request\n\n' f'Failed to respond to runtime RPC request for\n\n'
f'{func} x=> {ctx.chan}\n\n'
f'{ack}\n' f'{ack}\n'
) )
@ -418,6 +415,7 @@ async def _errors_relayed_via_ipc(
async def _invoke( async def _invoke(
actor: Actor, actor: Actor,
cid: str, cid: str,
chan: Channel, chan: Channel,
@ -693,6 +691,10 @@ async def _invoke(
boxed_type=trio.Cancelled, boxed_type=trio.Cancelled,
canceller=canceller, canceller=canceller,
) )
# does this matter other then for
# consistentcy/testing? |_ no user code should be
# in this scope at this point..
# ctx._local_error = ctxc
raise ctxc raise ctxc
# XXX: do we ever trigger this block any more? # XXX: do we ever trigger this block any more?
@ -713,11 +715,6 @@ async def _invoke(
# always set this (child) side's exception as the # always set this (child) side's exception as the
# local error on the context # local error on the context
ctx._local_error: BaseException = scope_error ctx._local_error: BaseException = scope_error
# ^-TODO-^ question,
# does this matter other then for
# consistentcy/testing?
# |_ no user code should be in this scope at this point
# AND we already set this in the block below?
# if a remote error was set then likely the # if a remote error was set then likely the
# exception group was raised due to that, so # exception group was raised due to that, so
@ -744,32 +741,22 @@ async def _invoke(
logmeth: Callable = log.runtime logmeth: Callable = log.runtime
merr: Exception|None = ctx.maybe_error merr: Exception|None = ctx.maybe_error
message: str = 'IPC context terminated ' descr_str: str = 'with final result `{repr(ctx.outcome)}`'
descr_str: str = ( message: str = (
f'after having {ctx.repr_state!r}\n' f'IPC context terminated {descr_str}\n\n'
) )
if merr: if merr:
descr_str: str = (
logmeth: Callable = log.error f'with ctx having {ctx.repr_state!r}\n'
f'{ctx.repr_outcome()}\n'
)
if isinstance(merr, ContextCancelled): if isinstance(merr, ContextCancelled):
logmeth: Callable = log.runtime logmeth: Callable = log.runtime
if not isinstance(merr, RemoteActorError):
tb_str: str = ''.join(traceback.format_exception(merr))
descr_str += (
f'\n{merr!r}\n' # needed?
f'{tb_str}\n'
)
else: else:
descr_str += f'\n{merr!r}\n' logmeth: Callable = log.error
else: message += f'\n{merr!r}\n'
descr_str += f'\nand final result {ctx.outcome!r}\n'
logmeth( logmeth(message)
message
+
descr_str
)
async def try_ship_error_to_remote( async def try_ship_error_to_remote(

View File

@ -57,8 +57,8 @@ DATE_FORMAT = '%b %d %H:%M:%S'
CUSTOM_LEVELS: dict[str, int] = { CUSTOM_LEVELS: dict[str, int] = {
'TRANSPORT': 5, 'TRANSPORT': 5,
'RUNTIME': 15, 'RUNTIME': 15,
'CANCEL': 16,
'DEVX': 17, 'DEVX': 17,
'CANCEL': 18,
'PDB': 500, 'PDB': 500,
} }
STD_PALETTE = { STD_PALETTE = {
@ -111,7 +111,7 @@ class StackLevelAdapter(LoggerAdapter):
''' '''
return self.log( return self.log(
level=22, level=16,
msg=msg, msg=msg,
# stacklevel=4, # stacklevel=4,
) )