Support `tractor.pause_from_sync()` in `brokerd`s

By passing down the `tractor.hilevel.ServiceMngr.debug_mode: bool`
(normally proxied in from the `--pdb` CLI flag) to `spawn_brokerd()` and
adjusting the `_setup_persistent_brokerd()` endpoint to do the
`tractor.devx._debug.maybe_init_greenback()` if needed.

Also in the `broker_init()` factory merge all `tractor` related `kwargs`
(i.e. `start_actor_kwargs | datad_kwargs | spawn_kws`) into the 2nd
element returned as to be passed to `ActorNursery.start_actor()`. Start
re-naming some internal vars/fields as `datad` as well.
Tyler Goodlet 2025-02-11 11:01:52 -05:00
parent 8ab2feba3e
commit 01cbd0a775
1 changed files with 39 additions and 20 deletions

View File

@ -61,12 +61,13 @@ async def _setup_persistent_brokerd(
ctx: tractor.Context, ctx: tractor.Context,
brokername: str, brokername: str,
loglevel: str | None = None, loglevel: str | None = None,
debug_mode: bool = False,
) -> None: ) -> None:
''' '''
Allocate a actor-wide service nursery in ``brokerd`` Allocate a actor-wide service nursery in `brokerd` such that
such that feeds can be run in the background persistently by feeds can be run in the background persistently by the broker
the broker backend as needed. backend as needed.
''' '''
# NOTE: we only need to setup logging once (and only) here # NOTE: we only need to setup logging once (and only) here
@ -87,6 +88,18 @@ async def _setup_persistent_brokerd(
from piker.data import feed from piker.data import feed
assert not feed._bus assert not feed._bus
if (
debug_mode
and
tractor.current_actor().is_infected_aio()
):
# NOTE, whenever running `asyncio` in provider's actor
# runtime be sure we enabled `breakpoint()` support
# for non-`trio.Task` usage.
from tractor.devx._debug import maybe_init_greenback
await maybe_init_greenback()
# breakpoint() # XXX, SHOULD WORK from `trio.Task`!
# allocate a nursery to the bus for spawning background # allocate a nursery to the bus for spawning background
# tasks to service client IPC requests, normally # tasks to service client IPC requests, normally
# `tractor.Context` connections to explicitly required # `tractor.Context` connections to explicitly required
@ -146,18 +159,21 @@ def broker_init(
above. above.
''' '''
from ..brokers import get_brokermod brokermod: ModuleType = get_brokermod(brokername)
brokermod = get_brokermod(brokername)
modpath: str = brokermod.__name__ modpath: str = brokermod.__name__
spawn_kws: dict = getattr(
start_actor_kwargs['name'] = f'brokerd.{brokername}'
start_actor_kwargs.update(
getattr(
brokermod, brokermod,
'_spawn_kwargs', '_spawn_kwargs',
{}, {},
) )
) # ^^ NOTE, here we pull any runtime parameters specific
# to spawning the sub-actor for the backend. For ex.
# both `ib` and `deribit` rely on,
# `'infect_asyncio': True,` since they both
# use `tractor`'s "infected `asyncio` mode"
# for their libs but you could also do something like
# `'debug_mode: True` which would be like passing
# `--pdb` for just that provider backend.
# XXX TODO: make this not so hacky/monkeypatched.. # XXX TODO: make this not so hacky/monkeypatched..
# -> we need a sane way to configure the logging level for all # -> we need a sane way to configure the logging level for all
@ -167,8 +183,7 @@ def broker_init(
# lookup actor-enabled modules declared by the backend offering the # lookup actor-enabled modules declared by the backend offering the
# `brokerd` endpoint(s). # `brokerd` endpoint(s).
enabled: list[str] enabled: list[str] = [
enabled = start_actor_kwargs['enable_modules'] = [
__name__, # so that eps from THIS mod can be invoked __name__, # so that eps from THIS mod can be invoked
modpath, modpath,
] ]
@ -180,9 +195,13 @@ def broker_init(
subpath: str = f'{modpath}.{submodname}' subpath: str = f'{modpath}.{submodname}'
enabled.append(subpath) enabled.append(subpath)
datad_kwargs: dict = {
'name': f'brokerd.{brokername}',
'enable_modules': enabled,
}
return ( return (
brokermod, brokermod,
start_actor_kwargs, # to `ActorNursery.start_actor()` start_actor_kwargs | datad_kwargs | spawn_kws, # to `ActorNursery.start_actor()`
# XXX see impl above; contains all (actor global) # XXX see impl above; contains all (actor global)
# setup/teardown expected in all `brokerd` actor instances. # setup/teardown expected in all `brokerd` actor instances.
@ -215,10 +234,6 @@ async def spawn_brokerd(
**tractor_kwargs, **tractor_kwargs,
) )
brokermod = get_brokermod(brokername)
extra_tractor_kwargs = getattr(brokermod, '_spawn_kwargs', {})
tractor_kwargs.update(extra_tractor_kwargs)
# ask `pikerd` to spawn a new sub-actor and manage it under its # ask `pikerd` to spawn a new sub-actor and manage it under its
# actor nursery # actor nursery
from piker.service import ( from piker.service import (
@ -236,8 +251,12 @@ async def spawn_brokerd(
# passed to daemon_fixture_ep(**kwargs) # passed to daemon_fixture_ep(**kwargs)
brokername=brokername, brokername=brokername,
loglevel=loglevel, loglevel=loglevel,
debug_mode=mngr.debug_mode,
), ),
debug_mode=mngr.debug_mode, debug_mode=mngr.debug_mode,
# ^TODO, allow overriding this per-daemon from client side?
# |_ it's already supported in `tractor` so..
loglevel=loglevel, loglevel=loglevel,
enable_modules=( enable_modules=(
_data_mods _data_mods