diff --git a/tractor/_root.py b/tractor/_root.py index 711bd442..50773056 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -18,7 +18,9 @@ Root actor runtime ignition(s). ''' -from contextlib import asynccontextmanager as acm +from contextlib import ( + asynccontextmanager as acm, +) from functools import partial import importlib import inspect @@ -26,7 +28,10 @@ import logging import os import signal import sys -from typing import Callable +from typing import ( + Any, + Callable, +) import warnings @@ -47,21 +52,95 @@ from .ipc import ( _connect_chan, ) from ._addr import ( + Address, UnwrappedAddress, default_lo_addrs, mk_uuid, preferred_transport, wrap_address, ) -from ._exceptions import is_multi_cancelled +from ._exceptions import ( + ActorFailure, + is_multi_cancelled, +) logger = log.get_logger('tractor') +# TODO: stick this in a `@acm` defined in `devx._debug`? +# -[ ] also maybe consider making this a `wrapt`-deco to +# save an indent level? +# +@acm +async def maybe_block_bp( + debug_mode: bool, + maybe_enable_greenback: bool, +) -> bool: + # Override the global debugger hook to make it play nice with + # ``trio``, see much discussion in: + # https://github.com/python-trio/trio/issues/1155#issuecomment-742964018 + builtin_bp_handler: Callable = sys.breakpointhook + orig_bp_path: str|None = os.environ.get( + 'PYTHONBREAKPOINT', + None, + ) + bp_blocked: bool + if ( + debug_mode + and maybe_enable_greenback + and ( + maybe_mod := await _debug.maybe_init_greenback( + raise_not_found=False, + ) + ) + ): + logger.info( + f'Found `greenback` installed @ {maybe_mod}\n' + 'Enabling `tractor.pause_from_sync()` support!\n' + ) + os.environ['PYTHONBREAKPOINT'] = ( + 'tractor.devx._debug._sync_pause_from_builtin' + ) + _state._runtime_vars['use_greenback'] = True + bp_blocked = False + + else: + # TODO: disable `breakpoint()` by default (without + # `greenback`) since it will break any multi-actor + # usage by a clobbered TTY's stdstreams! + def block_bps(*args, **kwargs): + raise RuntimeError( + 'Trying to use `breakpoint()` eh?\n\n' + 'Welp, `tractor` blocks `breakpoint()` built-in calls by default!\n' + 'If you need to use it please install `greenback` and set ' + '`debug_mode=True` when opening the runtime ' + '(either via `.open_nursery()` or `open_root_actor()`)\n' + ) + + sys.breakpointhook = block_bps + # lol ok, + # https://docs.python.org/3/library/sys.html#sys.breakpointhook + os.environ['PYTHONBREAKPOINT'] = "0" + bp_blocked = True + + try: + yield bp_blocked + finally: + # restore any prior built-in `breakpoint()` hook state + if builtin_bp_handler is not None: + sys.breakpointhook = builtin_bp_handler + + if orig_bp_path is not None: + os.environ['PYTHONBREAKPOINT'] = orig_bp_path + + else: + # clear env back to having no entry + os.environ.pop('PYTHONBREAKPOINT', None) + + @acm async def open_root_actor( - *, # defaults are above registry_addrs: list[UnwrappedAddress]|None = None, @@ -111,355 +190,323 @@ async def open_root_actor( Runtime init entry point for ``tractor``. ''' - _debug.hide_runtime_frames() - __tracebackhide__: bool = hide_tb - - # TODO: stick this in a `@cm` defined in `devx._debug`? - # - # Override the global debugger hook to make it play nice with - # ``trio``, see much discussion in: - # https://github.com/python-trio/trio/issues/1155#issuecomment-742964018 - builtin_bp_handler: Callable = sys.breakpointhook - orig_bp_path: str|None = os.environ.get( - 'PYTHONBREAKPOINT', - None, - ) - if ( - debug_mode - and maybe_enable_greenback - and ( - maybe_mod := await _debug.maybe_init_greenback( - raise_not_found=False, - ) + # XXX NEVER allow nested actor-trees! + if already_actor := _state.current_actor(err_on_no_runtime=False): + rtvs: dict[str, Any] = _state._runtime_vars + root_mailbox: list[str, int] = rtvs['_root_mailbox'] + registry_addrs: list[list[str, int]] = rtvs['_registry_addrs'] + raise ActorFailure( + f'A current actor already exists !?\n' + f'({already_actor}\n' + f'\n' + f'You can NOT open a second root actor from within ' + f'an existing tree and the current root of this ' + f'already exists !!\n' + f'\n' + f'_root_mailbox: {root_mailbox!r}\n' + f'_registry_addrs: {registry_addrs!r}\n' ) + + async with maybe_block_bp( + debug_mode=debug_mode, + maybe_enable_greenback=maybe_enable_greenback, ): - logger.info( - f'Found `greenback` installed @ {maybe_mod}\n' - 'Enabling `tractor.pause_from_sync()` support!\n' - ) - os.environ['PYTHONBREAKPOINT'] = ( - 'tractor.devx._debug._sync_pause_from_builtin' - ) - _state._runtime_vars['use_greenback'] = True + _debug.hide_runtime_frames() + __tracebackhide__: bool = hide_tb - else: - # TODO: disable `breakpoint()` by default (without - # `greenback`) since it will break any multi-actor - # usage by a clobbered TTY's stdstreams! - def block_bps(*args, **kwargs): - raise RuntimeError( - 'Trying to use `breakpoint()` eh?\n\n' - 'Welp, `tractor` blocks `breakpoint()` built-in calls by default!\n' - 'If you need to use it please install `greenback` and set ' - '`debug_mode=True` when opening the runtime ' - '(either via `.open_nursery()` or `open_root_actor()`)\n' + # attempt to retreive ``trio``'s sigint handler and stash it + # on our debugger lock state. + _debug.DebugStatus._trio_handler = signal.getsignal(signal.SIGINT) + + # mark top most level process as root actor + _state._runtime_vars['_is_root'] = True + + # caps based rpc list + enable_modules = ( + enable_modules + or + [] + ) + + if rpc_module_paths: + warnings.warn( + "`rpc_module_paths` is now deprecated, use " + " `enable_modules` instead.", + DeprecationWarning, + stacklevel=2, + ) + enable_modules.extend(rpc_module_paths) + + if start_method is not None: + _spawn.try_set_start_method(start_method) + + # TODO! remove this ASAP! + if arbiter_addr is not None: + warnings.warn( + '`arbiter_addr` is now deprecated\n' + 'Use `registry_addrs: list[tuple]` instead..', + DeprecationWarning, + stacklevel=2, + ) + registry_addrs = [arbiter_addr] + + if not registry_addrs: + registry_addrs: list[UnwrappedAddress] = default_lo_addrs( + enable_transports ) - sys.breakpointhook = block_bps - # lol ok, - # https://docs.python.org/3/library/sys.html#sys.breakpointhook - os.environ['PYTHONBREAKPOINT'] = "0" + assert registry_addrs - # attempt to retreive ``trio``'s sigint handler and stash it - # on our debugger lock state. - _debug.DebugStatus._trio_handler = signal.getsignal(signal.SIGINT) + loglevel = ( + loglevel + or log._default_loglevel + ).upper() - # mark top most level process as root actor - _state._runtime_vars['_is_root'] = True - - # caps based rpc list - enable_modules = ( - enable_modules - or - [] - ) - - if rpc_module_paths: - warnings.warn( - "`rpc_module_paths` is now deprecated, use " - " `enable_modules` instead.", - DeprecationWarning, - stacklevel=2, - ) - enable_modules.extend(rpc_module_paths) - - if start_method is not None: - _spawn.try_set_start_method(start_method) - - if arbiter_addr is not None: - warnings.warn( - '`arbiter_addr` is now deprecated\n' - 'Use `registry_addrs: list[tuple]` instead..', - DeprecationWarning, - stacklevel=2, - ) - registry_addrs = [arbiter_addr] - - if not registry_addrs: - registry_addrs: list[UnwrappedAddress] = default_lo_addrs( - enable_transports - ) - - assert registry_addrs - - loglevel = ( - loglevel - or log._default_loglevel - ).upper() - - if ( - debug_mode - and _spawn._spawn_method == 'trio' - ): - _state._runtime_vars['_debug_mode'] = True - - # expose internal debug module to every actor allowing for - # use of ``await tractor.pause()`` - enable_modules.append('tractor.devx._debug') - - # if debug mode get's enabled *at least* use that level of - # logging for some informative console prompts. if ( - logging.getLevelName( - # lul, need the upper case for the -> int map? - # sweet "dynamic function behaviour" stdlib... - loglevel, - ) > logging.getLevelName('PDB') + debug_mode + and _spawn._spawn_method == 'trio' ): - loglevel = 'PDB' + _state._runtime_vars['_debug_mode'] = True + + # expose internal debug module to every actor allowing for + # use of ``await tractor.pause()`` + enable_modules.append('tractor.devx._debug') + + # if debug mode get's enabled *at least* use that level of + # logging for some informative console prompts. + if ( + logging.getLevelName( + # lul, need the upper case for the -> int map? + # sweet "dynamic function behaviour" stdlib... + loglevel, + ) > logging.getLevelName('PDB') + ): + loglevel = 'PDB' - elif debug_mode: - raise RuntimeError( - "Debug mode is only supported for the `trio` backend!" - ) - - assert loglevel - _log = log.get_console_log(loglevel) - assert _log - - # TODO: factor this into `.devx._stackscope`!! - if ( - debug_mode - and - enable_stack_on_sig - ): - from .devx._stackscope import enable_stack_on_sig - enable_stack_on_sig() - - # closed into below ping task-func - ponged_addrs: list[UnwrappedAddress] = [] - - async def ping_tpt_socket( - addr: UnwrappedAddress, - timeout: float = 1, - ) -> None: - ''' - Attempt temporary connection to see if a registry is - listening at the requested address by a tranport layer - ping. - - If a connection can't be made quickly we assume none no - server is listening at that addr. - - ''' - try: - # TODO: this connect-and-bail forces us to have to - # carefully rewrap TCP 104-connection-reset errors as - # EOF so as to avoid propagating cancel-causing errors - # to the channel-msg loop machinery. Likely it would - # be better to eventually have a "discovery" protocol - # with basic handshake instead? - with trio.move_on_after(timeout): - async with _connect_chan(addr): - ponged_addrs.append(addr) - - except OSError: - # TODO: make this a "discovery" log level? - logger.info( - f'No actor registry found @ {addr}\n' - ) - - async with trio.open_nursery() as tn: - for addr in registry_addrs: - tn.start_soon( - ping_tpt_socket, - addr, - ) - - trans_bind_addrs: list[UnwrappedAddress] = [] - - # Create a new local root-actor instance which IS NOT THE - # REGISTRAR - if ponged_addrs: - if ensure_registry: + elif debug_mode: raise RuntimeError( - f'Failed to open `{name}`@{ponged_addrs}: ' - 'registry socket(s) already bound' + "Debug mode is only supported for the `trio` backend!" ) - # we were able to connect to an arbiter - logger.info( - f'Registry(s) seem(s) to exist @ {ponged_addrs}' - ) + assert loglevel + _log = log.get_console_log(loglevel) + assert _log - actor = Actor( - name=name or 'anonymous', - uuid=mk_uuid(), - registry_addrs=ponged_addrs, - loglevel=loglevel, - enable_modules=enable_modules, - ) - # DO NOT use the registry_addrs as the transport server - # addrs for this new non-registar, root-actor. - for addr in ponged_addrs: - waddr = wrap_address(addr) - print(waddr) - trans_bind_addrs.append( - waddr.get_random(namespace=waddr.namespace) - ) - - # Start this local actor as the "registrar", aka a regular - # actor who manages the local registry of "mailboxes" of - # other process-tree-local sub-actors. - else: - - # NOTE that if the current actor IS THE REGISTAR, the - # following init steps are taken: - # - the tranport layer server is bound to each addr - # pair defined in provided registry_addrs, or the default. - trans_bind_addrs = registry_addrs - - # - it is normally desirable for any registrar to stay up - # indefinitely until either all registered (child/sub) - # actors are terminated (via SC supervision) or, - # a re-election process has taken place. - # NOTE: all of ^ which is not implemented yet - see: - # https://github.com/goodboy/tractor/issues/216 - # https://github.com/goodboy/tractor/pull/348 - # https://github.com/goodboy/tractor/issues/296 - - actor = Arbiter( - name=name or 'registrar', - uuid=mk_uuid(), - registry_addrs=registry_addrs, - loglevel=loglevel, - enable_modules=enable_modules, - ) - # XXX, in case the root actor runtime was actually run from - # `tractor.to_asyncio.run_as_asyncio_guest()` and NOt - # `.trio.run()`. - actor._infected_aio = _state._runtime_vars['_is_infected_aio'] - - # Start up main task set via core actor-runtime nurseries. - try: - # assign process-local actor - _state._current_actor = actor - - # start local channel-server and fake the portal API - # NOTE: this won't block since we provide the nursery - ml_addrs_str: str = '\n'.join( - f'@{addr}' for addr in trans_bind_addrs - ) - logger.info( - f'Starting local {actor.uid} on the following transport addrs:\n' - f'{ml_addrs_str}' - ) - - # start the actor runtime in a new task - async with trio.open_nursery( - strict_exception_groups=False, - # ^XXX^ TODO? instead unpack any RAE as per "loose" style? - ) as nursery: - - # ``_runtime.async_main()`` creates an internal nursery - # and blocks here until any underlying actor(-process) - # tree has terminated thereby conducting so called - # "end-to-end" structured concurrency throughout an - # entire hierarchical python sub-process set; all - # "actor runtime" primitives are SC-compat and thus all - # transitively spawned actors/processes must be as - # well. - await nursery.start( - partial( - async_main, - actor, - accept_addrs=trans_bind_addrs, - parent_addr=None - ) - ) - try: - yield actor - except ( - Exception, - BaseExceptionGroup, - ) as err: - - # TODO, in beginning to handle the subsubactor with - # crashed grandparent cases.. - # - # was_locked: bool = await _debug.maybe_wait_for_debugger( - # child_in_debug=True, - # ) - # XXX NOTE XXX see equiv note inside - # `._runtime.Actor._stream_handler()` where in the - # non-root or root-that-opened-this-mahually case we - # wait for the local actor-nursery to exit before - # exiting the transport channel handler. - entered: bool = await _debug._maybe_enter_pm( - err, - api_frame=inspect.currentframe(), - debug_filter=debug_filter, - ) - - if ( - not entered - and - not is_multi_cancelled( - err, - ) - ): - logger.exception('Root actor crashed\n') - - # ALWAYS re-raise any error bubbled up from the - # runtime! - raise - - finally: - # NOTE: not sure if we'll ever need this but it's - # possibly better for even more determinism? - # logger.cancel( - # f'Waiting on {len(nurseries)} nurseries in root..') - # nurseries = actor._actoruid2nursery.values() - # async with trio.open_nursery() as tempn: - # for an in nurseries: - # tempn.start_soon(an.exited.wait) - - logger.info( - 'Closing down root actor' - ) - await actor.cancel(None) # self cancel - finally: - _state._current_actor = None - _state._last_actor_terminated = actor - - # restore built-in `breakpoint()` hook state + # TODO: factor this into `.devx._stackscope`!! if ( debug_mode and - maybe_enable_greenback + enable_stack_on_sig ): - if builtin_bp_handler is not None: - sys.breakpointhook = builtin_bp_handler + from .devx._stackscope import enable_stack_on_sig + enable_stack_on_sig() - if orig_bp_path is not None: - os.environ['PYTHONBREAKPOINT'] = orig_bp_path + # closed into below ping task-func + ponged_addrs: list[UnwrappedAddress] = [] - else: - # clear env back to having no entry - os.environ.pop('PYTHONBREAKPOINT', None) + async def ping_tpt_socket( + addr: UnwrappedAddress, + timeout: float = 1, + ) -> None: + ''' + Attempt temporary connection to see if a registry is + listening at the requested address by a tranport layer + ping. - logger.runtime("Root actor terminated") + If a connection can't be made quickly we assume none no + server is listening at that addr. + + ''' + try: + # TODO: this connect-and-bail forces us to have to + # carefully rewrap TCP 104-connection-reset errors as + # EOF so as to avoid propagating cancel-causing errors + # to the channel-msg loop machinery. Likely it would + # be better to eventually have a "discovery" protocol + # with basic handshake instead? + with trio.move_on_after(timeout): + async with _connect_chan(addr): + ponged_addrs.append(addr) + + except OSError: + # TODO: make this a "discovery" log level? + logger.info( + f'No actor registry found @ {addr}\n' + ) + + async with trio.open_nursery() as tn: + for addr in registry_addrs: + tn.start_soon( + ping_tpt_socket, + addr, + ) + + trans_bind_addrs: list[UnwrappedAddress] = [] + + # Create a new local root-actor instance which IS NOT THE + # REGISTRAR + if ponged_addrs: + if ensure_registry: + raise RuntimeError( + f'Failed to open `{name}`@{ponged_addrs}: ' + 'registry socket(s) already bound' + ) + + # we were able to connect to an arbiter + logger.info( + f'Registry(s) seem(s) to exist @ {ponged_addrs}' + ) + + actor = Actor( + name=name or 'anonymous', + uuid=mk_uuid(), + registry_addrs=ponged_addrs, + loglevel=loglevel, + enable_modules=enable_modules, + ) + # DO NOT use the registry_addrs as the transport server + # addrs for this new non-registar, root-actor. + for addr in ponged_addrs: + waddr: Address = wrap_address(addr) + trans_bind_addrs.append( + waddr.get_random(bindspace=waddr.bindspace) + ) + + # Start this local actor as the "registrar", aka a regular + # actor who manages the local registry of "mailboxes" of + # other process-tree-local sub-actors. + else: + + # NOTE that if the current actor IS THE REGISTAR, the + # following init steps are taken: + # - the tranport layer server is bound to each addr + # pair defined in provided registry_addrs, or the default. + trans_bind_addrs = registry_addrs + + # - it is normally desirable for any registrar to stay up + # indefinitely until either all registered (child/sub) + # actors are terminated (via SC supervision) or, + # a re-election process has taken place. + # NOTE: all of ^ which is not implemented yet - see: + # https://github.com/goodboy/tractor/issues/216 + # https://github.com/goodboy/tractor/pull/348 + # https://github.com/goodboy/tractor/issues/296 + + actor = Arbiter( + name=name or 'registrar', + uuid=mk_uuid(), + registry_addrs=registry_addrs, + loglevel=loglevel, + enable_modules=enable_modules, + ) + # XXX, in case the root actor runtime was actually run from + # `tractor.to_asyncio.run_as_asyncio_guest()` and NOt + # `.trio.run()`. + actor._infected_aio = _state._runtime_vars['_is_infected_aio'] + + # Start up main task set via core actor-runtime nurseries. + try: + # assign process-local actor + _state._current_actor = actor + + # start local channel-server and fake the portal API + # NOTE: this won't block since we provide the nursery + ml_addrs_str: str = '\n'.join( + f'@{addr}' for addr in trans_bind_addrs + ) + logger.info( + f'Starting local {actor.uid} on the following transport addrs:\n' + f'{ml_addrs_str}' + ) + + # start the actor runtime in a new task + async with trio.open_nursery( + strict_exception_groups=False, + # ^XXX^ TODO? instead unpack any RAE as per "loose" style? + ) as nursery: + + # ``_runtime.async_main()`` creates an internal nursery + # and blocks here until any underlying actor(-process) + # tree has terminated thereby conducting so called + # "end-to-end" structured concurrency throughout an + # entire hierarchical python sub-process set; all + # "actor runtime" primitives are SC-compat and thus all + # transitively spawned actors/processes must be as + # well. + await nursery.start( + partial( + async_main, + actor, + accept_addrs=trans_bind_addrs, + parent_addr=None + ) + ) + try: + yield actor + except ( + Exception, + BaseExceptionGroup, + ) as err: + + # TODO, in beginning to handle the subsubactor with + # crashed grandparent cases.. + # + # was_locked: bool = await _debug.maybe_wait_for_debugger( + # child_in_debug=True, + # ) + # XXX NOTE XXX see equiv note inside + # `._runtime.Actor._stream_handler()` where in the + # non-root or root-that-opened-this-mahually case we + # wait for the local actor-nursery to exit before + # exiting the transport channel handler. + entered: bool = await _debug._maybe_enter_pm( + err, + api_frame=inspect.currentframe(), + debug_filter=debug_filter, + ) + + if ( + not entered + and + not is_multi_cancelled( + err, + ) + ): + logger.exception( + 'Root actor crashed\n' + f'>x)\n' + f' |_{actor}\n' + ) + + # ALWAYS re-raise any error bubbled up from the + # runtime! + raise + + finally: + # NOTE: not sure if we'll ever need this but it's + # possibly better for even more determinism? + # logger.cancel( + # f'Waiting on {len(nurseries)} nurseries in root..') + # nurseries = actor._actoruid2nursery.values() + # async with trio.open_nursery() as tempn: + # for an in nurseries: + # tempn.start_soon(an.exited.wait) + + logger.info( + f'Closing down root actor\n' + f'>)\n' + f'|_{actor}\n' + ) + await actor.cancel(None) # self cancel + finally: + _state._current_actor = None + _state._last_actor_terminated = actor + logger.runtime( + f'Root actor terminated\n' + f')>\n' + f' |_{actor}\n' + ) def run_daemon(