piker/brokers/_daemon.py

@@ -23,7 +23,6 @@ from __future__ import annotations
 from contextlib import (
     asynccontextmanager as acm,
 )
-from functools import partial
 from types import ModuleType
 from typing import (
     TYPE_CHECKING,
@@ -191,17 +190,14 @@ def broker_init(
 
 
 async def spawn_brokerd(
+
     brokername: str,
     loglevel: str | None = None,
 
     **tractor_kwargs,
 
 ) -> bool:
-    '''
-    Spawn a `brokerd.<backendname>` subactor service daemon
-    using `pikerd`'s service mngr.
 
-    '''
     from piker.service._util import log  # use service mngr log
     log.info(f'Spawning {brokername} broker daemon')
 
@@ -221,35 +217,27 @@ async def spawn_brokerd(
 
     # ask `pikerd` to spawn a new sub-actor and manage it under its
     # actor nursery
-    from piker.service import (
+    from piker.service import Services
-        get_service_mngr,
-        ServiceMngr,
-    )
-    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
-    mngr: ServiceMngr = get_service_mngr()
-    ctx: tractor.Context = await mngr.start_service(
-        daemon_name=dname,
-        ctx_ep=partial(
-            # signature of target root-task endpoint
-            daemon_fixture_ep,
 
-            # passed to daemon_fixture_ep(**kwargs)
+    dname: str = tractor_kwargs.pop('name')  # f'brokerd.{brokername}'
-            brokername=brokername,
+    portal = await Services.actor_n.start_actor(
-            loglevel=loglevel,
+        dname,
-        ),
+        enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
-        debug_mode=mngr.debug_mode,
+        debug_mode=Services.debug_mode,
-        loglevel=loglevel,
-        enable_modules=(
-            _data_mods
-            +
-            tractor_kwargs.pop('enable_modules')
-        ),
         **tractor_kwargs
     )
-    assert (
+
-        not ctx.cancel_called
+    # NOTE: the service mngr expects an already spawned actor + its
-        and ctx.portal  # parent side
+    # portal ref in order to do non-blocking setup of brokerd
-        and dname in ctx.chan.uid  # subactor is named as desired
+    # service nursery.
+    await Services.start_service_task(
+        dname,
+        portal,
+
+        # signature of target root-task endpoint
+        daemon_fixture_ep,
+        brokername=brokername,
+        loglevel=loglevel,
     )
     return True
 
@@ -274,7 +262,8 @@ async def maybe_spawn_brokerd(
     from piker.service import maybe_spawn_daemon
 
     async with maybe_spawn_daemon(
-        service_name=f'brokerd.{brokername}',
+
+        f'brokerd.{brokername}',
         service_task_target=spawn_brokerd,
         spawn_args={
             'brokername': brokername,

piker/brokers/ib/_util.py

@@ -100,7 +100,7 @@ async def data_reset_hack(
         log.warning(
             no_setup_msg
             +
-            'REQUIRES A `vnc_addrs: array` ENTRY'
+            f'REQUIRES A `vnc_addrs: array` ENTRY'
         )
 
     vnc_host, vnc_port = vnc_sockaddr.get(

piker/brokers/ib/api.py

@@ -287,31 +287,9 @@ class Client:
         self.conf = config
 
         # NOTE: the ib.client here is "throttled" to 45 rps by default
-        self.ib: IB = ib
+        self.ib = ib
         self.ib.RaiseRequestErrors: bool = True
 
-        # self._acnt_names: set[str] = {}
-        self._acnt_names: list[str] = []
-
-    @property
-    def acnts(self) -> list[str]:
-        # return list(self._acnt_names)
-        return self._acnt_names
-
-    def __repr__(self) -> str:
-        return (
-            f'<{type(self).__name__}('
-            f'ib={self.ib} '
-            f'acnts={self.acnts}'
-
-            # TODO: we need to mask out acnt-#s and other private
-            # infos if we're going to console this!
-            # f' |_.conf:\n'
-            # f'    {pformat(self.conf)}\n'
-
-            ')>'
-        )
-
     async def get_fills(self) -> list[Fill]:
         '''
         Return list of rents `Fills` from trading session.
@@ -398,21 +376,19 @@ class Client:
             # whatToShow='MIDPOINT',
             # whatToShow='TRADES',
         )
+        log.info(
+            f'REQUESTING {ib_duration_str} worth {bar_size} BARS\n'
+            f'fqme: {fqme}\n'
+            f'global _enters: {_enters}\n'
+            f'kwargs: {pformat(kwargs)}\n'
+        )
+
         bars = await self.ib.reqHistoricalDataAsync(
             **kwargs,
         )
 
-        query_info: str = (
-            f'REQUESTING IB history BARS\n'
-            f'    ------ - ------\n'
-            f'dt_duration: {dt_duration}\n'
-            f'ib_duration_str: {ib_duration_str}\n'
-            f'bar_size: {bar_size}\n'
-            f'fqme: {fqme}\n'
-            f'actor-global _enters: {_enters}\n'
-            f'kwargs: {pformat(kwargs)}\n'
-        )
         # tail case if no history for range or none prior.
+        if not bars:
             # NOTE: there's actually 3 cases here to handle (and
             # this should be read alongside the implementation of
             # `.reqHistoricalDataAsync()`):
@@ -422,39 +398,33 @@ class Client:
             #   a weekend, holiday or other non-trading period prior to
             #   ``end_dt`` which exceeds the ``duration``,
             # - LITERALLY this is the start of the mkt's history!
-        if not bars:
-            # TODO: figure out wut's going on here.
 
-            # TODO: is this handy, a sync requester for tinkering
-            # with empty frame cases?
-            # def get_hist():
-            #     return self.ib.reqHistoricalData(**kwargs)
-            # import pdbp
-            # pdbp.set_trace()
 
-            log.critical(
+            # sync requester for debugging empty frame cases
-                'STUPID IB SAYS NO HISTORY\n\n'
+            def get_hist():
-                + query_info
+                return self.ib.reqHistoricalData(**kwargs)
-            )
+
+            assert get_hist
+            import pdbp
+            pdbp.set_trace()
 
-            # TODO: we could maybe raise ``NoData`` instead if we
-            # rewrite the method in the first case?
-            # right now there's no way to detect a timeout..
             return [], np.empty(0), dt_duration
+            # TODO: we could maybe raise ``NoData`` instead if we
+            # rewrite the method in the first case? right now there's no
+            # way to detect a timeout.
 
-        log.info(query_info)
+        # NOTE XXX: ensure minimum duration in bars B)
-        # NOTE XXX: ensure minimum duration in bars?
+        # => we recursively call this method until we get at least
-        # => recursively call this method until we get at least as
+        # as many bars such that they sum in aggregate to the the
-        #   many bars such that they sum in aggregate to the the
         # desired total time (duration) at most.
+        # XXX XXX XXX
+        # WHY DID WE EVEN NEED THIS ORIGINALLY!?
+        # XXX XXX XXX
         # - if you query over a gap and get no data
         #   that may short circuit the history 
         if (
-            # XXX XXX XXX
+            end_dt
-            # => WHY DID WE EVEN NEED THIS ORIGINALLY!? <=
+            and False
-            # XXX XXX XXX
-            False
-            and end_dt
         ):
             nparr: np.ndarray = bars_to_np(bars)
             times: np.ndarray = nparr['time']
@@ -957,10 +927,7 @@ class Client:
                         warnset = True
 
             else:
-                log.info(
+                log.info(f'Got first quote for {contract}')
-                    'Got first quote for contract\n'
-                    f'{contract}\n'
-                )
                 break
         else:
             if timeouterr and raise_on_timeout:
@@ -1024,12 +991,8 @@ class Client:
                     outsideRth=True,
 
                     optOutSmartRouting=True,
-                    # TODO: need to understand this setting better as
-                    # it pertains to shit ass mms..
                     routeMarketableToBbo=True,
-
                     designatedLocation='SMART',
-
                     # TODO: make all orders GTC?
                     # https://interactivebrokers.github.io/tws-api/classIBApi_1_1Order.html#a95539081751afb9980f4c6bd1655a6ba
                     # goodTillDate=f"yyyyMMdd-HH:mm:ss",
@@ -1157,8 +1120,8 @@ def get_config() -> dict[str, Any]:
     names = list(accounts.keys())
     accts = section['accounts'] = bidict(accounts)
     log.info(
-        f'{path} defines {len(accts)} account aliases:\n'
+        f'brokers.toml defines {len(accts)} accounts: '
-        f'{pformat(names)}\n'
+        f'{pformat(names)}'
     )
 
     if section is None:
@@ -1225,7 +1188,7 @@ async def load_aio_clients(
         try_ports = list(try_ports.values())
 
     _err = None
-    accounts_def: dict[str, str] = config.load_accounts(['ib'])
+    accounts_def = config.load_accounts(['ib'])
     ports = try_ports if port is None else [port]
     combos = list(itertools.product(hosts, ports))
     accounts_found: dict[str, Client] = {}
@@ -1264,9 +1227,7 @@ async def load_aio_clients(
                 client = Client(ib=ib, config=conf)
 
                 # update all actor-global caches
-                log.runtime(
+                log.info(f"Caching client for {sockaddr}")
-                    f'Connected and caching `Client` @ {sockaddr!r}'
-                )
                 _client_cache[sockaddr] = client
                 break
 
@@ -1281,54 +1242,32 @@ async def load_aio_clients(
                 OSError,
             ) as ce:
                 _err = ce
-                message: str = (
+                log.warning(
-                    f'Failed to connect on {host}:{port} after {i} tries with\n'
+                    f'Failed to connect on {host}:{port} for {i} time with,\n'
-                    f'{ib.client.apiError.value()!r}\n\n'
+                    f'{ib.client.apiError.value()}\n'
-                    'Retrying with a new client id..\n'
+                    'retrying with a new client id..')
-                )
-                log.runtime(message)
-        else:
-            # XXX report loudly if we never established after all
-            # re-tries
-            log.warning(message)
 
         # Pre-collect all accounts available for this
         # connection and map account names to this client
         # instance.
         for value in ib.accountValues():
-            acct_number: str = value.account
+            acct_number = value.account
 
-            acnt_alias: str = accounts_def.inverse.get(acct_number)
+            entry = accounts_def.inverse.get(acct_number)
-            if not acnt_alias:
+            if not entry:
-
-                # TODO: should we constuct the below reco-ex from
-                # the existing config content?
-                _, path = config.load(
-                    conf_name='brokers',
-                )
                 raise ValueError(
-                    'No alias in account section for account!\n'
+                    'No section in brokers.toml for account:'
-                    f'Please add an acnt alias entry to your {path}\n'
+                    f' {acct_number}\n'
-                    'For example,\n\n'
+                    f'Please add entry to continue using this API client'
-
-                    '[ib.accounts]\n'
-                    'margin = {accnt_number!r}\n'
-                    '^^^^^^ <- you need this part!\n\n'
-
-                    'This ensures `piker` will not leak private acnt info '
-                    'to console output by default!\n'
                 )
 
             # surjection of account names to operating clients.
-            if acnt_alias not in accounts_found:
+            if acct_number not in accounts_found:
-                accounts_found[acnt_alias] = client
+                accounts_found[entry] = client
-                # client._acnt_names.add(acnt_alias)
-                client._acnt_names.append(acnt_alias)
 
-        if accounts_found:
         log.info(
-                f'Loaded accounts for api client\n\n'
+            f'Loaded accounts for client @ {host}:{port}\n'
-                f'{pformat(accounts_found)}\n'
+            f'{pformat(accounts_found)}'
         )
 
         # XXX: why aren't we just updating this directy above
@@ -1533,7 +1472,7 @@ async def open_aio_client_method_relay(
         msg: tuple[str, dict] | dict | None = await from_trio.get()
         match msg:
             case None:  # termination sentinel
-                log.info('asyncio `Client` method-proxy SHUTDOWN!')
+                print('asyncio PROXY-RELAY SHUTDOWN')
                 break
 
             case (meth_name, kwargs):

piker/brokers/ib/broker.py

@@ -1183,14 +1183,7 @@ async def deliver_trade_events(
                         pos
                         and fill
                     ):
-                        now_cr: CommissionReport = fill.commissionReport
+                        assert fill.commissionReport == cr
-                        if (now_cr != cr):
-                            log.warning(
-                                'UhhHh ib updated the commission report mid-fill..?\n'
-                                f'was: {pformat(cr)}\n'
-                                f'now: {pformat(now_cr)}\n'
-                            )
-
                         await emit_pp_update(
                             ems_stream,
                             accounts_def,

piker/brokers/ib/feed.py

@@ -671,8 +671,8 @@ async def _setup_quote_stream(
         # making them mostly useless and explains why the scanner
         # is always slow XD
         # '293',  # Trade count for day
-        # '294',  # Trade rate / minute
+        '294',  # Trade rate / minute
-        # '295',  # Vlm rate / minute
+        '295',  # Vlm rate / minute
     ),
     contract: Contract | None = None,
 
@@ -915,13 +915,9 @@ async def stream_quotes(
 
         if first_ticker:
             first_quote: dict = normalize(first_ticker)
-
+            log.info(
-            # TODO: we need a stack-oriented log levels filters for
+                'Rxed init quote:\n'
-            # this!
+                f'{pformat(first_quote)}'
-            # log.info(message, filter={'stack': 'live_feed'}) ?
-            log.runtime(
-                'Rxed init quote:\n\n'
-                f'{pformat(first_quote)}\n'
             )
 
         # NOTE: it might be outside regular trading hours for
@@ -973,11 +969,7 @@ async def stream_quotes(
             raise_on_timeout=True,
         )
         first_quote: dict = normalize(first_ticker)
-
+        log.info(
-        # TODO: we need a stack-oriented log levels filters for
-        # this!
-        # log.info(message, filter={'stack': 'live_feed'}) ?
-        log.runtime(
             'Rxed init quote:\n'
             f'{pformat(first_quote)}'
         )

piker/brokers/ib/ledger.py

@@ -31,11 +31,7 @@ from typing import (
 )
 
 from bidict import bidict
-from pendulum import (
+import pendulum
-    DateTime,
-    parse,
-    from_timestamp,
-)
 from ib_insync import (
     Contract,
     Commodity,
@@ -70,11 +66,10 @@ tx_sort: Callable = partial(
     iter_by_dt,
     parsers={
         'dateTime': parse_flex_dt,
-        'datetime': parse,
+        'datetime': pendulum.parse,
-
+        # for some some fucking 2022 and
-        # XXX: for some some fucking 2022 and
+        # back options records...fuck me.
-        # back options records.. f@#$ me..
+        'date': pendulum.parse,
-        'date': parse,
     }
 )
 
@@ -94,38 +89,15 @@ def norm_trade(
 
     conid: int = str(record.get('conId') or record['conid'])
     bs_mktid: str = str(conid)
+    comms = record.get('commission')
+    if comms is None:
+        comms = -1*record['ibCommission']
 
-    # NOTE: sometimes weird records (like BTTX?)
+    price = record.get('price') or record['tradePrice']
-    # have no field for this?
-    comms: float = -1 * (
-        record.get('commission')
-        or record.get('ibCommission')
-        or 0
-    )
-    if not comms:
-        log.warning(
-            'No commissions found for record?\n'
-            f'{pformat(record)}\n'
-        )
-
-    price: float = (
-        record.get('price')
-        or record.get('tradePrice')
-    )
-    if price is None:
-        log.warning(
-            'No `price` field found in record?\n'
-            'Skipping normalization..\n'
-            f'{pformat(record)}\n'
-        )
-        return None
 
     # the api doesn't do the -/+ on the quantity for you but flex
     # records do.. are you fucking serious ib...!?
-    size: float|int = (
+    size = record.get('quantity') or record['shares'] * {
-        record.get('quantity')
-        or record['shares']
-    ) * {
         'BOT': 1,
         'SLD': -1,
     }[record['side']]
@@ -156,31 +128,26 @@ def norm_trade(
         # otype = tail[6]
         # strike = tail[7:]
 
-        log.warning(
+        print(f'skipping opts contract {symbol}')
-            f'Skipping option contract -> NO SUPPORT YET!\n'
-            f'{symbol}\n'
-        )
         return None
 
     # timestamping is way different in API records
-    dtstr: str = record.get('datetime')
+    dtstr = record.get('datetime')
-    date: str = record.get('date')
+    date = record.get('date')
-    flex_dtstr: str = record.get('dateTime')
+    flex_dtstr = record.get('dateTime')
 
     if dtstr or date:
-        dt: DateTime = parse(dtstr or date)
+        dt = pendulum.parse(dtstr or date)
 
     elif flex_dtstr:
         # probably a flex record with a wonky non-std timestamp..
-        dt: DateTime = parse_flex_dt(record['dateTime'])
+        dt = parse_flex_dt(record['dateTime'])
 
     # special handling of symbol extraction from
     # flex records using some ad-hoc schema parsing.
-    asset_type: str = (
+    asset_type: str = record.get(
-        record.get('assetCategory')
+        'assetCategory'
-        or record.get('secType')
+    ) or record.get('secType', 'STK')
-        or 'STK'
-    )
 
     if (expiry := (
             record.get('lastTradeDateOrContractMonth')
@@ -390,7 +357,6 @@ def norm_trade_records(
         if txn is None:
             continue
 
-        # inject txns sorted by datetime
         insort(
             records,
             txn,
@@ -439,7 +405,7 @@ def api_trades_to_ledger_entries(
                     txn_dict[attr_name] = val
 
         tid = str(txn_dict['execId'])
-        dt = from_timestamp(txn_dict['time'])
+        dt = pendulum.from_timestamp(txn_dict['time'])
         txn_dict['datetime'] = str(dt)
         acctid = accounts[txn_dict['acctNumber']]
 

piker/brokers/ib/symbols.py

@@ -209,10 +209,7 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                 break
 
             ib_client = proxy._aio_ns.ib
-            log.info(
+            log.info(f'Using {ib_client} for symbol search')
-                f'Using API client for symbol-search\n'
-                f'{ib_client}\n'
-            )
 
             last = time.time()
             async for pattern in stream:
@@ -297,7 +294,7 @@ async def open_symbol_search(ctx: tractor.Context) -> None:
                     elif stock_results:
                         break
                     # else:
-                    # await tractor.pause()
+                    await tractor.pause()
 
                     # # match against our ad-hoc set immediately
                     # adhoc_matches = fuzzy.extract(
@@ -525,21 +522,7 @@ async def get_mkt_info(
         venue = con.primaryExchange or con.exchange
 
     price_tick: Decimal = Decimal(str(details.minTick))
-    ib_min_tick_gt_2: Decimal = Decimal('0.01')
+    # price_tick: Decimal = Decimal('0.01')
-    if (
-        price_tick < ib_min_tick_gt_2
-    ):
-        # TODO: we need to add some kinda dynamic rounding sys
-        # to our MktPair i guess?
-        # not sure where the logic should sit, but likely inside
-        # the `.clearing._ems` i suppose...
-        log.warning(
-            'IB seems to disallow a min price tick < 0.01 '
-            'when the price is > 2.0..?\n'
-            f'Decreasing min tick precision for {fqme} to 0.01'
-        )
-        # price_tick = ib_min_tick
-        # await tractor.pause()
 
     if atype == 'stock':
         # XXX: GRRRR they don't support fractional share sizes for

piker/data/_sampling.py

@@ -25,7 +25,6 @@ from collections import (
     defaultdict,
 )
 from contextlib import asynccontextmanager as acm
-from functools import partial
 import time
 from typing import (
     Any,
@@ -43,7 +42,7 @@ from tractor.trionics import (
     maybe_open_nursery,
 )
 import trio
-from trio import TaskStatus
+from trio_typing import TaskStatus
 
 from .ticktools import (
     frame_ticks,
@@ -71,7 +70,6 @@ if TYPE_CHECKING:
 _default_delay_s: float = 1.0
 
 
-# TODO: use new `tractor.singleton_acm` API for this!
 class Sampler:
     '''
     Global sampling engine registry.
@@ -81,9 +79,9 @@ class Sampler:
 
     This non-instantiated type is meant to be a singleton within
     a `samplerd` actor-service spawned once by the user wishing to
-    time-step-sample a (real-time) quote feeds, see
+    time-step-sample (real-time) quote feeds, see
-    `.service.maybe_open_samplerd()` and the below
+    ``.service.maybe_open_samplerd()`` and the below
-    `register_with_sampler()`.
+    ``register_with_sampler()``.
 
     '''
     service_nursery: None | trio.Nursery = None
@@ -377,10 +375,7 @@ async def register_with_sampler(
                 assert Sampler.ohlcv_shms
 
             # unblock caller
-            await ctx.started(
+            await ctx.started(set(Sampler.ohlcv_shms.keys()))
-                # XXX bc msgpack only allows one array type!
-                list(Sampler.ohlcv_shms.keys())
-            )
 
             if open_index_stream:
                 try:
@@ -424,6 +419,7 @@ async def register_with_sampler(
 
 
 async def spawn_samplerd(
+
     loglevel: str | None = None,
     **extra_tractor_kwargs
 
@@ -433,10 +429,7 @@ async def spawn_samplerd(
     update and increment count write and stream broadcasting.
 
     '''
-    from piker.service import (
+    from piker.service import Services
-        get_service_mngr,
-        ServiceMngr,
-    )
 
     dname = 'samplerd'
     log.info(f'Spawning `{dname}`')
@@ -444,33 +437,26 @@ async def spawn_samplerd(
     # singleton lock creation of ``samplerd`` since we only ever want
     # one daemon per ``pikerd`` proc tree.
     # TODO: make this built-into the service api?
-    mngr: ServiceMngr = get_service_mngr()
+    async with Services.locks[dname + '_singleton']:
-    already_started: bool = dname in mngr.service_tasks
 
-    async with mngr._locks[dname + '_singleton']:
+        if dname not in Services.service_tasks:
-        ctx: Context = await mngr.start_service(
-            daemon_name=dname,
-            ctx_ep=partial(
-                register_with_sampler,
-                period_s=1,
-                sub_for_broadcasts=False,
-            ),
-            debug_mode=mngr.debug_mode,  # set by pikerd flag
 
-            # proxy-through to tractor
+            portal = await Services.actor_n.start_actor(
+                dname,
                 enable_modules=[
                     'piker.data._sampling',
                 ],
                 loglevel=loglevel,
+                debug_mode=Services.debug_mode,  # set by pikerd flag
                 **extra_tractor_kwargs
             )
-        if not already_started:
+
-            assert (
+            await Services.start_service_task(
-                ctx
+                dname,
-                and
+                portal,
-                ctx.portal
+                register_with_sampler,
-                and
+                period_s=1,
-                not ctx.cancel_called
+                sub_for_broadcasts=False,
             )
             return True
 
@@ -903,7 +889,6 @@ async def uniform_rate_send(
             # to consumers which crash or lose network connection.
             # I.e. we **DO NOT** want to crash and propagate up to
             # ``pikerd`` these kinds of errors!
-            trio.EndOfChannel,
             trio.ClosedResourceError,
             trio.BrokenResourceError,
             ConnectionResetError,

piker/service/__init__.py

@@ -30,11 +30,7 @@ Actor runtime primtives and (distributed) service APIs for,
   => TODO: maybe to (re)move elsewhere?
 
 '''
-from ._mngr import (
+from ._mngr import Services as Services
-    get_service_mngr as get_service_mngr,
-    open_service_mngr as open_service_mngr,
-    ServiceMngr as ServiceMngr,
-)
 from ._registry import (
     _tractor_kwargs as _tractor_kwargs,
     _default_reg_addr as _default_reg_addr,

piker/service/_actor_runtime.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 import os
 from typing import (
+    Optional,
     Any,
     ClassVar,
 )
@@ -29,13 +30,13 @@ from contextlib import (
 )
 
 import tractor
+import trio
 
 from ._util import (
     get_console_log,
 )
 from ._mngr import (
-    open_service_mngr,
+    Services,
-    ServiceMngr,
 )
 from ._registry import (  # noqa
     _tractor_kwargs,
@@ -58,7 +59,7 @@ async def open_piker_runtime(
     registry_addrs: list[tuple[str, int]] = [],
 
     enable_modules: list[str] = [],
-    loglevel: str|None = None,
+    loglevel: Optional[str] = None,
 
     # XXX NOTE XXX: you should pretty much never want debug mode
     # for data daemons when running in production.
@@ -68,7 +69,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_runtime_overrides: dict|None = None,
+    tractor_runtime_overrides: dict | None = None,
     **tractor_kwargs,
 
 ) -> tuple[
@@ -118,10 +119,6 @@ async def open_piker_runtime(
                 # spawn other specialized daemons I think?
                 enable_modules=enable_modules,
 
-                # TODO: how to configure this?
-                # keep it on by default if debug mode is set?
-                # maybe_enable_greenback=debug_mode,
-
                 **tractor_kwargs,
             ) as actor,
 
@@ -170,13 +167,12 @@ async def open_pikerd(
 
     **kwargs,
 
-) -> ServiceMngr:
+) -> Services:
     '''
-    Start a root piker daemon actor (aka `pikerd`) with an indefinite
+    Start a root piker daemon with an indefinite lifetime.
-    lifetime.
 
-    A root actor-nursery is created which can be used to spawn and
+    A root actor nursery is created which can be used to create and keep
-    supervise underling service sub-actors (see below).
+    alive underling services (see below).
 
     '''
     # NOTE: for the root daemon we always enable the root
@@ -203,6 +199,8 @@ async def open_pikerd(
             root_actor,
             reg_addrs,
         ),
+        tractor.open_nursery() as actor_nursery,
+        trio.open_nursery() as service_nursery,
     ):
         for addr in reg_addrs:
             if addr not in root_actor.accept_addrs:
@@ -211,17 +209,25 @@ async def open_pikerd(
                     'Maybe you have another daemon already running?'
                 )
 
-        mngr: ServiceMngr
+        # assign globally for future daemon/task creation
-        async with open_service_mngr(
+        Services.actor_n = actor_nursery
-            debug_mode=debug_mode,
+        Services.service_n = service_nursery
-        ) as mngr:
+        Services.debug_mode = debug_mode
-            yield mngr
+
+        try:
+            yield Services
+
+        finally:
+            # TODO: is this more clever/efficient?
+            # if 'samplerd' in Services.service_tasks:
+            #     await Services.cancel_service('samplerd')
+            service_nursery.cancel_scope.cancel()
 
 
 # TODO: do we even need this?
 # @acm
 # async def maybe_open_runtime(
-#     loglevel: str|None = None,
+#     loglevel: Optional[str] = None,
 #     **kwargs,
 
 # ) -> None:
@@ -250,7 +256,7 @@ async def maybe_open_pikerd(
     loglevel: str | None = None,
     **kwargs,
 
-) -> tractor._portal.Portal | ClassVar[ServiceMngr]:
+) -> tractor._portal.Portal | ClassVar[Services]:
     '''
     If no ``pikerd`` daemon-root-actor can be found start it and
     yield up (we should probably figure out returning a portal to self

piker/service/_ahab.py

@@ -49,7 +49,7 @@ from requests.exceptions import (
     ReadTimeout,
 )
 
-from ._mngr import ServiceMngr
+from ._mngr import Services
 from ._util import (
     log,  # sub-sys logger
     get_console_log,
@@ -453,7 +453,7 @@ async def open_ahabd(
 
 @acm
 async def start_ahab_service(
-    services: ServiceMngr,
+    services: Services,
     service_name: str,
 
     # endpoint config passed as **kwargs
@@ -549,8 +549,7 @@ async def start_ahab_service(
         log.warning('Failed to cancel root permsed container')
 
     except (
-        # trio.MultiError,
+        trio.MultiError,
-        ExceptionGroup,
     ) as err:
         for subexc in err.exceptions:
             if isinstance(subexc, PermissionError):

piker/service/_daemon.py

@@ -26,17 +26,14 @@ from typing import (
 from contextlib import (
     asynccontextmanager as acm,
 )
-from collections import defaultdict
 
 import tractor
-import trio
 
 from ._util import (
     log,  # sub-sys logger
 )
 from ._mngr import (
-    get_service_mngr,
+    Services,
-    ServiceMngr,
 )
 from ._actor_runtime import maybe_open_pikerd
 from ._registry import find_service
@@ -44,14 +41,15 @@ from ._registry import find_service
 
 @acm
 async def maybe_spawn_daemon(
+
     service_name: str,
     service_task_target: Callable,
+
     spawn_args: dict[str, Any],
 
     loglevel: str | None = None,
     singleton: bool = False,
 
-    _locks = defaultdict(trio.Lock),
     **pikerd_kwargs,
 
 ) -> tractor.Portal:
@@ -69,7 +67,7 @@ async def maybe_spawn_daemon(
     '''
     # serialize access to this section to avoid
     # 2 or more tasks racing to create a daemon
-    lock = _locks[service_name]
+    lock = Services.locks[service_name]
     await lock.acquire()
 
     async with find_service(
@@ -149,22 +147,21 @@ async def spawn_emsd(
     """
     log.info('Spawning emsd')
 
-    smngr: ServiceMngr = get_service_mngr()
+    portal = await Services.actor_n.start_actor(
-    portal = await smngr.actor_n.start_actor(
         'emsd',
         enable_modules=[
             'piker.clearing._ems',
             'piker.clearing._client',
         ],
         loglevel=loglevel,
-        debug_mode=smngr.debug_mode,  # set by pikerd flag
+        debug_mode=Services.debug_mode,  # set by pikerd flag
         **extra_tractor_kwargs
     )
 
     # non-blocking setup of clearing service
     from ..clearing._ems import _setup_persistent_emsd
 
-    await smngr.start_service_task(
+    await Services.start_service_task(
         'emsd',
         portal,
 

piker/service/_mngr.py

@@ -18,29 +18,16 @@
 daemon-service management API.
 
 """
-from __future__ import annotations
-from contextlib import (
-    asynccontextmanager as acm,
-    # contextmanager as cm,
-)
 from collections import defaultdict
-from dataclasses import (
-    dataclass,
-    field,
-)
-import functools
-import inspect
 from typing import (
     Callable,
     Any,
 )
 
-import msgspec
-import tractor
 import trio
-from trio import TaskStatus
+from trio_typing import TaskStatus
+import tractor
 from tractor import (
-    ActorNursery,
     current_actor,
     ContextCancelled,
     Context,
@@ -52,130 +39,6 @@ from ._util import (
 )
 
 
-# TODO: implement a singleton deco-API for wrapping the below
-# factory's impl for general actor-singleton use?
-#
-# @singleton
-# async def open_service_mngr(
-#     **init_kwargs,
-# ) -> ServiceMngr:
-#     '''
-#     Note this function body is invoke IFF no existing singleton instance already
-#     exists in this proc's memory.
-
-#     '''
-#     # setup
-#     yield ServiceMngr(**init_kwargs)
-#     # teardown
-
-
-
-# TODO: singleton factory API instead of a class API
-@acm
-async def open_service_mngr(
-    *,
-    debug_mode: bool = False,
-
-    # impl deat which ensures a single global instance
-    _singleton: list[ServiceMngr|None] = [None],
-    **init_kwargs,
-
-) -> ServiceMngr:
-    '''
-    Open a multi-subactor-as-service-daemon tree supervisor.
-
-    The delivered `ServiceMngr` is a singleton instance for each
-    actor-process and is allocated on first open and never
-    de-allocated unless explicitly deleted by al call to
-    `del_service_mngr()`.
-
-    '''
-    # TODO: factor this an allocation into
-    # a `._mngr.open_service_mngr()` and put in the
-    # once-n-only-once setup/`.__aenter__()` part!
-    # -[ ] how to make this only happen on the `mngr == None` case?
-    #  |_ use `.trionics.maybe_open_context()` (for generic
-    #     async-with-style-only-once of the factory impl, though
-    #     what do we do for the allocation case?
-    #    / `.maybe_open_nursery()` (since for this specific case
-    #    it's simpler?) to activate
-    async with (
-        tractor.open_nursery() as an,
-        trio.open_nursery() as tn,
-    ):
-        # impl specific obvi..
-        init_kwargs.update({
-            'actor_n': an,
-            'service_n': tn,
-        })
-
-        mngr: ServiceMngr|None
-        if (mngr := _singleton[0]) is None:
-
-            log.info('Allocating a new service mngr!')
-            mngr = _singleton[0] = ServiceMngr(**init_kwargs)
-
-            # TODO: put into `.__aenter__()` section of
-            # eventual `@singleton_acm` API wrapper.
-            #
-            # assign globally for future daemon/task creation
-            mngr.actor_n = an
-            mngr.service_n = tn
-
-        else:
-            assert (
-                mngr.actor_n
-                and
-                mngr.service_tn
-            )
-            log.info(
-                'Using extant service mngr!\n\n'
-                f'{mngr!r}\n'  # it has a nice `.__repr__()` of services state
-            )
-
-        try:
-            # NOTE: this is a singleton factory impl specific detail
-            # which should be supported in the condensed
-            # `@singleton_acm` API?
-            mngr.debug_mode = debug_mode
-
-            yield mngr
-        finally:
-            # TODO: is this more clever/efficient?
-            # if 'samplerd' in mngr.service_tasks:
-            #     await mngr.cancel_service('samplerd')
-            tn.cancel_scope.cancel()
-
-
-
-def get_service_mngr() -> ServiceMngr:
-    '''
-    Try to get the singleton service-mngr for this actor presuming it
-    has already been allocated using,
-
-    .. code:: python
-
-        async with open_<@singleton_acm(func)>() as mngr`
-            ... this block kept open ...
-
-    If not yet allocated raise a `ServiceError`.
-
-    '''
-    # https://stackoverflow.com/a/12627202
-    # https://docs.python.org/3/library/inspect.html#inspect.Signature
-    maybe_mngr: ServiceMngr|None = inspect.signature(
-        open_service_mngr
-    ).parameters['_singleton'].default[0]
-
-    if maybe_mngr is None:
-        raise RuntimeError(
-            'Someone must allocate a `ServiceMngr` using\n\n'
-            '`async with open_service_mngr()` beforehand!!\n'
-        )
-
-    return maybe_mngr
-
-
 # TODO: we need remote wrapping and a general soln:
 # - factor this into a ``tractor.highlevel`` extension # pack for the
 #   library.
@@ -183,46 +46,31 @@ def get_service_mngr() -> ServiceMngr:
 #   to the pikerd actor for starting services remotely!
 # - prolly rename this to ActorServicesNursery since it spawns
 #   new actors and supervises them to completion?
-@dataclass
+class Services:
-class ServiceMngr:
-# class ServiceMngr(msgspec.Struct):
-    '''
-    A multi-subactor-as-service manager.
 
-    Spawn, supervise and monitor service/daemon subactors in a SC
+    actor_n: tractor._supervise.ActorNursery
-    process tree.
-
-    '''
-    actor_n: ActorNursery
     service_n: trio.Nursery
-    debug_mode: bool = False # tractor sub-actor debug mode flag
+    debug_mode: bool  # tractor sub-actor debug mode flag
-
     service_tasks: dict[
         str,
         tuple[
             trio.CancelScope,
-            Context,
             Portal,
             trio.Event,
         ]
-    ] = field(default_factory=dict)
+    ] = {}
-
+    locks = defaultdict(trio.Lock)
-    # internal per-service task mutexs
-    _locks = defaultdict(trio.Lock)
 
+    @classmethod
     async def start_service_task(
         self,
         name: str,
         portal: Portal,
-
-        # TODO: typevar for the return type of the target and then
-        # use it below for `ctx_res`?
         target: Callable,
-
         allow_overruns: bool = False,
         **ctx_kwargs,
 
-    ) -> (trio.CancelScope, Context, Any):
+    ) -> (trio.CancelScope, Context):
         '''
         Open a context in a service sub-actor, add to a stack
         that gets unwound at ``pikerd`` teardown.
@@ -235,7 +83,6 @@ class ServiceMngr:
             task_status: TaskStatus[
                 tuple[
                     trio.CancelScope,
-                    Context,
                     trio.Event,
                     Any,
                 ]
@@ -243,87 +90,64 @@ class ServiceMngr:
 
         ) -> Any:
 
-            # TODO: use the ctx._scope directly here instead?
-            # -[ ] actually what semantics do we expect for this
-            #   usage!?
             with trio.CancelScope() as cs:
-                try:
+
                 async with portal.open_context(
                     target,
                     allow_overruns=allow_overruns,
                     **ctx_kwargs,
 
-                    ) as (ctx, started):
+                ) as (ctx, first):
 
                     # unblock once the remote context has started
                     complete = trio.Event()
-                        task_status.started((
+                    task_status.started((cs, complete, first))
-                            cs,
-                            ctx,
-                            complete,
-                            started,
-                        ))
                     log.info(
-                            f'`pikerd` service {name} started with value {started}'
+                        f'`pikerd` service {name} started with value {first}'
                     )
+                    try:
                         # wait on any context's return value
                         # and any final portal result from the
                         # sub-actor.
-                        ctx_res: Any = await ctx.wait_for_result()
+                        ctx_res: Any = await ctx.result()
 
                         # NOTE: blocks indefinitely until cancelled
                         # either by error from the target context
                         # function or by being cancelled here by the
                         # surrounding cancel scope.
-                        return (
+                        return (await portal.result(), ctx_res)
-                            await portal.wait_for_result(),
-                            ctx_res,
-                        )
-
                     except ContextCancelled as ctxe:
                         canceller: tuple[str, str] = ctxe.canceller
                         our_uid: tuple[str, str] = current_actor().uid
                         if (
-                        canceller != portal.chan.uid
+                            canceller != portal.channel.uid
                             and
                             canceller != our_uid
                         ):
                             log.cancel(
-                            f'Actor-service `{name}` was remotely cancelled by a peer?\n'
+                                f'Actor-service {name} was remotely cancelled?\n'
-
+                                f'remote canceller: {canceller}\n'
-                            # TODO: this would be a good spot to use
+                                f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
-                            # a respawn feature Bo
-                            f'-> Keeping `pikerd` service manager alive despite this inter-peer cancel\n\n'
-
-                            f'cancellee: {portal.chan.uid}\n'
-                            f'canceller: {canceller}\n'
                             )
                         else:
                             raise
 
+
+
                     finally:
-                    # NOTE: the ctx MUST be cancelled first if we
-                    # don't want the above `ctx.wait_for_result()` to
-                    # raise a self-ctxc. WHY, well since from the ctx's
-                    # perspective the cancel request will have
-                    # arrived out-out-of-band at the `Actor.cancel()`
-                    # level, thus `Context.cancel_called == False`,
-                    # meaning `ctx._is_self_cancelled() == False`.
-                    # with trio.CancelScope(shield=True):
-                    # await ctx.cancel()
                         await portal.cancel_actor()
                         complete.set()
                         self.service_tasks.pop(name)
 
-        cs, sub_ctx, complete, started = await self.service_n.start(
+        cs, complete, first = await self.service_n.start(open_context_in_task)
-            open_context_in_task
-        )
 
         # store the cancel scope and portal for later cancellation or
         # retstart if needed.
-        self.service_tasks[name] = (cs, sub_ctx, portal, complete)
+        self.service_tasks[name] = (cs, portal, complete)
-        return cs, sub_ctx, started
 
+        return cs, first
+
+    @classmethod
     async def cancel_service(
         self,
         name: str,
@@ -334,80 +158,8 @@ class ServiceMngr:
 
         '''
         log.info(f'Cancelling `pikerd` service {name}')
-        cs, sub_ctx, portal, complete = self.service_tasks[name]
+        cs, portal, complete = self.service_tasks[name]
-
+        cs.cancel()
-        # cs.cancel()
-        await sub_ctx.cancel()
         await complete.wait()
-
+        assert name not in self.service_tasks, \
-        if name in self.service_tasks:
-            # TODO: custom err?
-            # raise ServiceError(
-            raise RuntimeError(
             f'Serice task for {name} not terminated?'
-            )
-
-        # assert name not in self.service_tasks, \
-        #     f'Serice task for {name} not terminated?'
-
-    async def start_service(
-        self,
-        daemon_name: str,
-        ctx_ep: Callable,  # kwargs must `partial`-ed in!
-
-        debug_mode: bool = False,
-        **tractor_actor_kwargs,
-
-    ) -> Context:
-        '''
-        Start a "service" task in a new sub-actor (daemon) and manage it's lifetime
-        indefinitely.
-
-        Services can be cancelled/shutdown using `.cancel_service()`.
-
-        '''
-        entry: tuple|None = self.service_tasks.get(daemon_name)
-        if entry:
-            (cs, sub_ctx, portal, complete) = entry
-            return sub_ctx
-
-        if daemon_name not in self.service_tasks:
-            portal = await self.actor_n.start_actor(
-                daemon_name,
-                debug_mode=(  # maybe set globally during allocate
-                    debug_mode
-                    or
-                    self.debug_mode
-                ),
-                **tractor_actor_kwargs,
-            )
-            ctx_kwargs: dict[str, Any] = {}
-            if isinstance(ctx_ep, functools.partial):
-                ctx_kwargs: dict[str, Any] = ctx_ep.keywords
-                ctx_ep: Callable = ctx_ep.func
-
-            (cs, sub_ctx, started) = await self.start_service_task(
-                daemon_name,
-                portal,
-                ctx_ep,
-                **ctx_kwargs,
-            )
-
-            return sub_ctx
-
-
-# TODO:
-# -[ ] factor all the common shit from `.data._sampling`
-#   and `.brokers._daemon` into here / `ServiceMngr`
-#   in terms of allocating the `Portal` as part of the
-#   "service-in-subactor" starting!
-# -[ ] move to `tractor.hilevel._service`, import and use here!
-# NOTE: purposely leaks the ref to the mod-scope Bo
-# import tractor
-# from tractor.hilevel import (
-#     open_service_mngr,
-#     ServiceMngr,
-# )
-# mngr: ServiceMngr|None = None
-# with tractor.hilevel.open_service_mngr() as mngr:
-#     Services = proxy(mngr)

piker/service/elastic.py

@@ -21,13 +21,11 @@ from typing import (
     TYPE_CHECKING,
 )
 
-# TODO: oof, needs to be changed to `httpx`!
 import asks
 
 if TYPE_CHECKING:
     import docker
     from ._ahab import DockerContainer
-    from . import ServiceMngr
 
 from ._util import log  # sub-sys logger
 from ._util import (
@@ -129,7 +127,7 @@ def start_elasticsearch(
 
 @acm
 async def start_ahab_daemon(
-    service_mngr: ServiceMngr,
+    service_mngr: Services,
     user_config: dict | None = None,
     loglevel: str | None = None,
 

piker/service/marketstore.py

@@ -53,7 +53,7 @@ import pendulum
 # import purerpc
 
 from ..data.feed import maybe_open_feed
-from . import ServiceMngr
+from . import Services
 from ._util import (
     log,  # sub-sys logger
     get_console_log,
@@ -233,7 +233,7 @@ def start_marketstore(
 
 @acm
 async def start_ahab_daemon(
-    service_mngr: ServiceMngr,
+    service_mngr: Services,
     user_config: dict | None = None,
     loglevel: str | None = None,
 

tests/conftest.py

@@ -10,7 +10,7 @@ from piker import (
     config,
 )
 from piker.service import (
-    get_service_mngr,
+    Services,
 )
 from piker.log import get_console_log
 
@@ -129,7 +129,7 @@ async def _open_test_pikerd(
         ) as service_manager,
     ):
         # this proc/actor is the pikerd
-        assert service_manager is get_service_mngr()
+        assert service_manager is Services
 
         async with tractor.wait_for_actor(
             'pikerd',

tests/test_ems.py

@@ -26,7 +26,7 @@ import pytest
 import tractor
 from uuid import uuid4
 
-from piker.service import ServiceMngr
+from piker.service import Services
 from piker.log import get_logger
 from piker.clearing._messages import (
     Order,
@@ -158,7 +158,7 @@ def load_and_check_pos(
 
 
 def test_ems_err_on_bad_broker(
-    open_test_pikerd: ServiceMngr,
+    open_test_pikerd: Services,
     loglevel: str,
 ):
     async def load_bad_fqme():

tests/test_services.py

@@ -15,7 +15,7 @@ import tractor
 
 from piker.service import (
     find_service,
-    ServiceMngr,
+    Services,
 )
 from piker.data import (
     open_feed,
@@ -44,7 +44,7 @@ def test_runtime_boot(
     async def main():
         port = 6666
         daemon_addr = ('127.0.0.1', port)
-        services: ServiceMngr
+        services: Services
 
         async with (
             open_test_pikerd(