Compare commits
13 Commits
e69796ad6d
...
7eacab0ac5
Author | SHA1 | Date |
---|---|---|
|
7eacab0ac5 | |
|
16e8acf3c4 | |
|
78d76f64cd | |
|
77477bf6a2 | |
|
d4067b31fb | |
|
7928d88b5f | |
|
40602d40ca | |
|
f148114adb | |
|
9498e5f102 | |
|
53eb6b8f91 | |
|
c7d11a68c1 | |
|
c36c38f432 | |
|
41b0584588 |
|
@ -30,7 +30,8 @@ from types import ModuleType
|
|||
from typing import (
|
||||
Any,
|
||||
Iterator,
|
||||
Generator
|
||||
Generator,
|
||||
TYPE_CHECKING,
|
||||
)
|
||||
|
||||
import pendulum
|
||||
|
@ -59,8 +60,10 @@ from ..clearing._messages import (
|
|||
BrokerdPosition,
|
||||
)
|
||||
from piker.types import Struct
|
||||
from piker.data._symcache import SymbologyCache
|
||||
from ..log import get_logger
|
||||
from piker.log import get_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from piker.data._symcache import SymbologyCache
|
||||
|
||||
log = get_logger(__name__)
|
||||
|
||||
|
@ -493,6 +496,17 @@ class Account(Struct):
|
|||
|
||||
_mktmap_table: dict[str, MktPair] | None = None,
|
||||
|
||||
only_require: list[str]|True = True,
|
||||
# ^list of fqmes that are "required" to be processed from
|
||||
# this ledger pass; we often don't care about others and
|
||||
# definitely shouldn't always error in such cases.
|
||||
# (eg. broker backend loaded that doesn't yet supsport the
|
||||
# symcache but also, inside the paper engine we don't ad-hoc
|
||||
# request `get_mkt_info()` for every symbol in the ledger,
|
||||
# only the one for which we're simulating against).
|
||||
# TODO, not sure if there's a better soln for this, ideally
|
||||
# all backends get symcache support afap i guess..
|
||||
|
||||
) -> dict[str, Position]:
|
||||
'''
|
||||
Update the internal `.pps[str, Position]` table from input
|
||||
|
@ -535,11 +549,32 @@ class Account(Struct):
|
|||
if _mktmap_table is None:
|
||||
raise
|
||||
|
||||
required: bool = (
|
||||
only_require is True
|
||||
or (
|
||||
only_require is not True
|
||||
and
|
||||
fqme in only_require
|
||||
)
|
||||
)
|
||||
# XXX: caller is allowed to provide a fallback
|
||||
# mktmap table for the case where a new position is
|
||||
# being added and the preloaded symcache didn't
|
||||
# have this entry prior (eg. with frickin IB..)
|
||||
mkt = _mktmap_table[fqme]
|
||||
if (
|
||||
not (mkt := _mktmap_table.get(fqme))
|
||||
and
|
||||
required
|
||||
):
|
||||
raise
|
||||
|
||||
elif not required:
|
||||
continue
|
||||
|
||||
else:
|
||||
# should be an entry retreived somewhere
|
||||
assert mkt
|
||||
|
||||
|
||||
if not (pos := pps.get(bs_mktid)):
|
||||
|
||||
|
@ -656,7 +691,7 @@ class Account(Struct):
|
|||
def write_config(self) -> None:
|
||||
'''
|
||||
Write the current account state to the user's account TOML file, normally
|
||||
something like ``pps.toml``.
|
||||
something like `pps.toml`.
|
||||
|
||||
'''
|
||||
# TODO: show diff output?
|
||||
|
|
|
@ -23,6 +23,7 @@ from __future__ import annotations
|
|||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
)
|
||||
from functools import partial
|
||||
from types import ModuleType
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
|
@ -190,14 +191,17 @@ 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')
|
||||
|
||||
|
@ -217,27 +221,35 @@ async def spawn_brokerd(
|
|||
|
||||
# ask `pikerd` to spawn a new sub-actor and manage it under its
|
||||
# actor nursery
|
||||
from piker.service import Services
|
||||
|
||||
from piker.service import (
|
||||
get_service_mngr,
|
||||
ServiceMngr,
|
||||
)
|
||||
dname: str = tractor_kwargs.pop('name') # f'brokerd.{brokername}'
|
||||
portal = await Services.actor_n.start_actor(
|
||||
dname,
|
||||
enable_modules=_data_mods + tractor_kwargs.pop('enable_modules'),
|
||||
debug_mode=Services.debug_mode,
|
||||
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)
|
||||
brokername=brokername,
|
||||
loglevel=loglevel,
|
||||
),
|
||||
debug_mode=mngr.debug_mode,
|
||||
loglevel=loglevel,
|
||||
enable_modules=(
|
||||
_data_mods
|
||||
+
|
||||
tractor_kwargs.pop('enable_modules')
|
||||
),
|
||||
**tractor_kwargs
|
||||
)
|
||||
|
||||
# NOTE: the service mngr expects an already spawned actor + its
|
||||
# portal ref in order to do non-blocking setup of brokerd
|
||||
# service nursery.
|
||||
await Services.start_service_task(
|
||||
dname,
|
||||
portal,
|
||||
|
||||
# signature of target root-task endpoint
|
||||
daemon_fixture_ep,
|
||||
brokername=brokername,
|
||||
loglevel=loglevel,
|
||||
assert (
|
||||
not ctx.cancel_called
|
||||
and ctx.portal # parent side
|
||||
and dname in ctx.chan.uid # subactor is named as desired
|
||||
)
|
||||
return True
|
||||
|
||||
|
@ -262,8 +274,7 @@ async def maybe_spawn_brokerd(
|
|||
from piker.service import maybe_spawn_daemon
|
||||
|
||||
async with maybe_spawn_daemon(
|
||||
|
||||
f'brokerd.{brokername}',
|
||||
service_name=f'brokerd.{brokername}',
|
||||
service_task_target=spawn_brokerd,
|
||||
spawn_args={
|
||||
'brokername': brokername,
|
||||
|
|
|
@ -567,6 +567,7 @@ class Client:
|
|||
) -> str:
|
||||
return {
|
||||
'USDTM': 'usdtm_futes',
|
||||
'SPOT': 'spot',
|
||||
# 'COINM': 'coin_futes',
|
||||
# ^-TODO-^ bc someone might want it..?
|
||||
}[pair.venue]
|
||||
|
|
|
@ -181,7 +181,6 @@ class FutesPair(Pair):
|
|||
quoteAsset: str # 'USDT',
|
||||
quotePrecision: int # 8,
|
||||
requiredMarginPercent: float # '5.0000',
|
||||
settlePlan: int # 0,
|
||||
timeInForce: list[str] # ['GTC', 'IOC', 'FOK', 'GTX'],
|
||||
triggerProtect: float # '0.0500',
|
||||
underlyingSubType: list[str] # ['PoW'],
|
||||
|
|
|
@ -111,6 +111,10 @@ class KucoinMktPair(Struct, frozen=True):
|
|||
quoteMaxSize: float
|
||||
quoteMinSize: float
|
||||
symbol: str # our bs_mktid, kucoin's internal id
|
||||
feeCategory: int
|
||||
makerFeeCoefficient: float
|
||||
takerFeeCoefficient: float
|
||||
st: bool
|
||||
|
||||
|
||||
class AccountTrade(Struct, frozen=True):
|
||||
|
@ -593,7 +597,7 @@ async def get_client() -> AsyncGenerator[Client, None]:
|
|||
'''
|
||||
async with (
|
||||
httpx.AsyncClient(
|
||||
base_url=f'https://api.kucoin.com/api',
|
||||
base_url='https://api.kucoin.com/api',
|
||||
) as trio_client,
|
||||
):
|
||||
client = Client(httpx_client=trio_client)
|
||||
|
@ -637,7 +641,7 @@ async def open_ping_task(
|
|||
await trio.sleep((ping_interval - 1000) / 1000)
|
||||
await ws.send_msg({'id': connect_id, 'type': 'ping'})
|
||||
|
||||
log.info('Starting ping task for kucoin ws connection')
|
||||
log.warning('Starting ping task for kucoin ws connection')
|
||||
n.start_soon(ping_server)
|
||||
|
||||
yield
|
||||
|
@ -649,9 +653,14 @@ async def open_ping_task(
|
|||
async def get_mkt_info(
|
||||
fqme: str,
|
||||
|
||||
) -> tuple[MktPair, KucoinMktPair]:
|
||||
) -> tuple[
|
||||
MktPair,
|
||||
KucoinMktPair,
|
||||
]:
|
||||
'''
|
||||
Query for and return a `MktPair` and `KucoinMktPair`.
|
||||
Query for and return both a `piker.accounting.MktPair` and
|
||||
`KucoinMktPair` from provided `fqme: str`
|
||||
(fully-qualified-market-endpoint).
|
||||
|
||||
'''
|
||||
async with open_cached_client('kucoin') as client:
|
||||
|
@ -726,6 +735,8 @@ async def stream_quotes(
|
|||
|
||||
log.info(f'Starting up quote stream(s) for {symbols}')
|
||||
for sym_str in symbols:
|
||||
mkt: MktPair
|
||||
pair: KucoinMktPair
|
||||
mkt, pair = await get_mkt_info(sym_str)
|
||||
init_msgs.append(
|
||||
FeedInit(mkt_info=mkt)
|
||||
|
@ -733,7 +744,11 @@ async def stream_quotes(
|
|||
|
||||
ws: NoBsWs
|
||||
token, ping_interval = await client._get_ws_token()
|
||||
connect_id = str(uuid4())
|
||||
log.info('API reported ping_interval: {ping_interval}\n')
|
||||
|
||||
connect_id: str = str(uuid4())
|
||||
typ: str
|
||||
quote: dict
|
||||
async with (
|
||||
open_autorecon_ws(
|
||||
(
|
||||
|
@ -747,20 +762,37 @@ async def stream_quotes(
|
|||
),
|
||||
) as ws,
|
||||
open_ping_task(ws, ping_interval, connect_id),
|
||||
aclosing(stream_messages(ws, sym_str)) as msg_gen,
|
||||
aclosing(
|
||||
iter_normed_quotes(
|
||||
ws, sym_str
|
||||
)
|
||||
) as iter_quotes,
|
||||
):
|
||||
typ, quote = await anext(msg_gen)
|
||||
typ, quote = await anext(iter_quotes)
|
||||
|
||||
while typ != 'trade':
|
||||
# take care to not unblock here until we get a real
|
||||
# trade quote
|
||||
typ, quote = await anext(msg_gen)
|
||||
# take care to not unblock here until we get a real
|
||||
# trade quote?
|
||||
# ^TODO, remove this right?
|
||||
# -[ ] what often blocks chart boot/new-feed switching
|
||||
# since we'ere waiting for a live quote instead of just
|
||||
# loading history afap..
|
||||
# |_ XXX, not sure if we require a bit of rework to core
|
||||
# feed init logic or if backends justg gotta be
|
||||
# changed up.. feel like there was some causality
|
||||
# dilema prolly only seen with IB too..
|
||||
# while typ != 'trade':
|
||||
# typ, quote = await anext(iter_quotes)
|
||||
|
||||
task_status.started((init_msgs, quote))
|
||||
feed_is_live.set()
|
||||
|
||||
async for typ, msg in msg_gen:
|
||||
await send_chan.send({sym_str: msg})
|
||||
# XXX NOTE, DO NOT include the `.<backend>` suffix!
|
||||
# OW the sampling loop will not broadcast correctly..
|
||||
# since `bus._subscribers.setdefault(bs_fqme, set())`
|
||||
# is used inside `.data.open_feed_bus()` !!!
|
||||
topic: str = mkt.bs_fqme
|
||||
async for typ, quote in iter_quotes:
|
||||
await send_chan.send({topic: quote})
|
||||
|
||||
|
||||
@acm
|
||||
|
@ -815,7 +847,7 @@ async def subscribe(
|
|||
)
|
||||
|
||||
|
||||
async def stream_messages(
|
||||
async def iter_normed_quotes(
|
||||
ws: NoBsWs,
|
||||
sym: str,
|
||||
|
||||
|
@ -846,6 +878,9 @@ async def stream_messages(
|
|||
|
||||
yield 'trade', {
|
||||
'symbol': sym,
|
||||
# TODO, is 'last' even used elsewhere/a-good
|
||||
# semantic? can't we just read the ticks with our
|
||||
# .data.ticktools.frame_ticks()`/
|
||||
'last': trade_data.price,
|
||||
'brokerd_ts': last_trade_ts,
|
||||
'ticks': [
|
||||
|
@ -938,7 +973,7 @@ async def open_history_client(
|
|||
if end_dt is None:
|
||||
inow = round(time.time())
|
||||
|
||||
print(
|
||||
log.debug(
|
||||
f'difference in time between load and processing'
|
||||
f'{inow - times[-1]}'
|
||||
)
|
||||
|
|
|
@ -653,7 +653,11 @@ class Router(Struct):
|
|||
flume = feed.flumes[fqme]
|
||||
first_quote: dict = flume.first_quote
|
||||
book: DarkBook = self.get_dark_book(broker)
|
||||
book.lasts[fqme]: float = float(first_quote['last'])
|
||||
|
||||
if not (last := first_quote.get('last')):
|
||||
last: float = flume.rt_shm.array[-1]['close']
|
||||
|
||||
book.lasts[fqme]: float = float(last)
|
||||
|
||||
async with self.maybe_open_brokerd_dialog(
|
||||
brokermod=brokermod,
|
||||
|
@ -716,7 +720,7 @@ class Router(Struct):
|
|||
subs = self.subscribers[sub_key]
|
||||
|
||||
sent_some: bool = False
|
||||
for client_stream in subs:
|
||||
for client_stream in subs.copy():
|
||||
try:
|
||||
await client_stream.send(msg)
|
||||
sent_some = True
|
||||
|
@ -1010,10 +1014,14 @@ async def translate_and_relay_brokerd_events(
|
|||
status_msg.brokerd_msg = msg
|
||||
status_msg.src = msg.broker_details['name']
|
||||
|
||||
await router.client_broadcast(
|
||||
status_msg.req.symbol,
|
||||
status_msg,
|
||||
)
|
||||
if not status_msg.req:
|
||||
# likely some order change state?
|
||||
await tractor.pause()
|
||||
else:
|
||||
await router.client_broadcast(
|
||||
status_msg.req.symbol,
|
||||
status_msg,
|
||||
)
|
||||
|
||||
if status == 'closed':
|
||||
log.info(f'Execution for {oid} is complete!')
|
||||
|
|
|
@ -653,6 +653,7 @@ async def open_trade_dialog(
|
|||
# in) use manually constructed table from calling
|
||||
# the `.get_mkt_info()` provider EP above.
|
||||
_mktmap_table=mkt_by_fqme,
|
||||
only_require=list(mkt_by_fqme),
|
||||
)
|
||||
|
||||
pp_msgs: list[BrokerdPosition] = []
|
||||
|
|
|
@ -335,7 +335,7 @@ def services(config, tl, ports):
|
|||
name='service_query',
|
||||
loglevel=config['loglevel'] if tl else None,
|
||||
),
|
||||
tractor.get_arbiter(
|
||||
tractor.get_registry(
|
||||
host=host,
|
||||
port=ports[0]
|
||||
) as portal
|
||||
|
|
|
@ -25,6 +25,7 @@ from collections import (
|
|||
defaultdict,
|
||||
)
|
||||
from contextlib import asynccontextmanager as acm
|
||||
from functools import partial
|
||||
import time
|
||||
from typing import (
|
||||
Any,
|
||||
|
@ -42,7 +43,7 @@ from tractor.trionics import (
|
|||
maybe_open_nursery,
|
||||
)
|
||||
import trio
|
||||
from trio_typing import TaskStatus
|
||||
from trio import TaskStatus
|
||||
|
||||
from .ticktools import (
|
||||
frame_ticks,
|
||||
|
@ -70,6 +71,7 @@ if TYPE_CHECKING:
|
|||
_default_delay_s: float = 1.0
|
||||
|
||||
|
||||
# TODO: use new `tractor.singleton_acm` API for this!
|
||||
class Sampler:
|
||||
'''
|
||||
Global sampling engine registry.
|
||||
|
@ -79,9 +81,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 (real-time) quote feeds, see
|
||||
``.service.maybe_open_samplerd()`` and the below
|
||||
``register_with_sampler()``.
|
||||
time-step-sample a (real-time) quote feeds, see
|
||||
`.service.maybe_open_samplerd()` and the below
|
||||
`register_with_sampler()`.
|
||||
|
||||
'''
|
||||
service_nursery: None | trio.Nursery = None
|
||||
|
@ -375,7 +377,10 @@ async def register_with_sampler(
|
|||
assert Sampler.ohlcv_shms
|
||||
|
||||
# unblock caller
|
||||
await ctx.started(set(Sampler.ohlcv_shms.keys()))
|
||||
await ctx.started(
|
||||
# XXX bc msgpack only allows one array type!
|
||||
list(Sampler.ohlcv_shms.keys())
|
||||
)
|
||||
|
||||
if open_index_stream:
|
||||
try:
|
||||
|
@ -419,7 +424,6 @@ async def register_with_sampler(
|
|||
|
||||
|
||||
async def spawn_samplerd(
|
||||
|
||||
loglevel: str | None = None,
|
||||
**extra_tractor_kwargs
|
||||
|
||||
|
@ -429,7 +433,10 @@ async def spawn_samplerd(
|
|||
update and increment count write and stream broadcasting.
|
||||
|
||||
'''
|
||||
from piker.service import Services
|
||||
from piker.service import (
|
||||
get_service_mngr,
|
||||
ServiceMngr,
|
||||
)
|
||||
|
||||
dname = 'samplerd'
|
||||
log.info(f'Spawning `{dname}`')
|
||||
|
@ -437,26 +444,33 @@ 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?
|
||||
async with Services.locks[dname + '_singleton']:
|
||||
mngr: ServiceMngr = get_service_mngr()
|
||||
already_started: bool = dname in mngr.service_tasks
|
||||
|
||||
if dname not in Services.service_tasks:
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
await Services.start_service_task(
|
||||
dname,
|
||||
portal,
|
||||
async with mngr._locks[dname + '_singleton']:
|
||||
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
|
||||
enable_modules=[
|
||||
'piker.data._sampling',
|
||||
],
|
||||
loglevel=loglevel,
|
||||
**extra_tractor_kwargs
|
||||
)
|
||||
if not already_started:
|
||||
assert (
|
||||
ctx
|
||||
and
|
||||
ctx.portal
|
||||
and
|
||||
not ctx.cancel_called
|
||||
)
|
||||
return True
|
||||
|
||||
|
@ -889,6 +903,7 @@ 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,
|
||||
|
|
|
@ -273,7 +273,7 @@ async def _reconnect_forever(
|
|||
nobsws._connected.set()
|
||||
await trio.sleep_forever()
|
||||
except HandshakeError:
|
||||
log.exception(f'Retrying connection')
|
||||
log.exception('Retrying connection')
|
||||
|
||||
# ws & nursery block ends
|
||||
|
||||
|
@ -359,8 +359,8 @@ async def open_autorecon_ws(
|
|||
|
||||
|
||||
'''
|
||||
JSONRPC response-request style machinery for transparent multiplexing of msgs
|
||||
over a NoBsWs.
|
||||
JSONRPC response-request style machinery for transparent multiplexing
|
||||
of msgs over a NoBsWs.
|
||||
|
||||
'''
|
||||
|
||||
|
@ -377,16 +377,20 @@ async def open_jsonrpc_session(
|
|||
url: str,
|
||||
start_id: int = 0,
|
||||
response_type: type = JSONRPCResult,
|
||||
request_type: Optional[type] = None,
|
||||
request_hook: Optional[Callable] = None,
|
||||
error_hook: Optional[Callable] = None,
|
||||
# request_type: Optional[type] = None,
|
||||
# request_hook: Optional[Callable] = None,
|
||||
# error_hook: Optional[Callable] = None,
|
||||
) -> Callable[[str, dict], dict]:
|
||||
|
||||
# NOTE, store all request msgs so we can raise errors on the
|
||||
# caller side!
|
||||
req_msgs: dict[int, dict] = {}
|
||||
|
||||
async with (
|
||||
trio.open_nursery() as n,
|
||||
open_autorecon_ws(url) as ws
|
||||
):
|
||||
rpc_id: Iterable = count(start_id)
|
||||
rpc_id: Iterable[int] = count(start_id)
|
||||
rpc_results: dict[int, dict] = {}
|
||||
|
||||
async def json_rpc(method: str, params: dict) -> dict:
|
||||
|
@ -394,26 +398,40 @@ async def open_jsonrpc_session(
|
|||
perform a json rpc call and wait for the result, raise exception in
|
||||
case of error field present on response
|
||||
'''
|
||||
nonlocal req_msgs
|
||||
|
||||
req_id: int = next(rpc_id)
|
||||
msg = {
|
||||
'jsonrpc': '2.0',
|
||||
'id': next(rpc_id),
|
||||
'id': req_id,
|
||||
'method': method,
|
||||
'params': params
|
||||
}
|
||||
_id = msg['id']
|
||||
|
||||
rpc_results[_id] = {
|
||||
result = rpc_results[_id] = {
|
||||
'result': None,
|
||||
'event': trio.Event()
|
||||
'error': None,
|
||||
'event': trio.Event(), # signal caller resp arrived
|
||||
}
|
||||
req_msgs[_id] = msg
|
||||
|
||||
await ws.send_msg(msg)
|
||||
|
||||
# wait for reponse before unblocking requester code
|
||||
await rpc_results[_id]['event'].wait()
|
||||
|
||||
ret = rpc_results[_id]['result']
|
||||
if (maybe_result := result['result']):
|
||||
ret = maybe_result
|
||||
del rpc_results[_id]
|
||||
|
||||
del rpc_results[_id]
|
||||
else:
|
||||
err = result['error']
|
||||
raise Exception(
|
||||
f'JSONRPC request failed\n'
|
||||
f'req: {msg}\n'
|
||||
f'resp: {err}\n'
|
||||
)
|
||||
|
||||
if ret.error is not None:
|
||||
raise Exception(json.dumps(ret.error, indent=4))
|
||||
|
@ -428,6 +446,7 @@ async def open_jsonrpc_session(
|
|||
the server side.
|
||||
|
||||
'''
|
||||
nonlocal req_msgs
|
||||
async for msg in ws:
|
||||
match msg:
|
||||
case {
|
||||
|
@ -451,15 +470,29 @@ async def open_jsonrpc_session(
|
|||
'params': _,
|
||||
}:
|
||||
log.debug(f'Recieved\n{msg}')
|
||||
if request_hook:
|
||||
await request_hook(request_type(**msg))
|
||||
# if request_hook:
|
||||
# await request_hook(request_type(**msg))
|
||||
|
||||
case {
|
||||
'error': error
|
||||
}:
|
||||
log.warning(f'Recieved\n{error}')
|
||||
if error_hook:
|
||||
await error_hook(response_type(**msg))
|
||||
# if error_hook:
|
||||
# await error_hook(response_type(**msg))
|
||||
|
||||
# retreive orig request msg, set error
|
||||
# response in original "result" msg,
|
||||
# THEN FINALLY set the event to signal caller
|
||||
# to raise the error in the parent task.
|
||||
req_id: int = error['id']
|
||||
req_msg: dict = req_msgs[req_id]
|
||||
result: dict = rpc_results[req_id]
|
||||
result['error'] = error
|
||||
result['event'].set()
|
||||
log.error(
|
||||
f'JSONRPC request failed\n'
|
||||
f'req: {req_msg}\n'
|
||||
f'resp: {error}\n'
|
||||
)
|
||||
|
||||
case _:
|
||||
log.warning(f'Unhandled JSON-RPC msg!?\n{msg}')
|
||||
|
|
|
@ -30,7 +30,11 @@ Actor runtime primtives and (distributed) service APIs for,
|
|||
=> TODO: maybe to (re)move elsewhere?
|
||||
|
||||
'''
|
||||
from ._mngr import Services as Services
|
||||
from ._mngr import (
|
||||
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,
|
||||
|
|
|
@ -21,7 +21,6 @@
|
|||
from __future__ import annotations
|
||||
import os
|
||||
from typing import (
|
||||
Optional,
|
||||
Any,
|
||||
ClassVar,
|
||||
)
|
||||
|
@ -30,13 +29,13 @@ from contextlib import (
|
|||
)
|
||||
|
||||
import tractor
|
||||
import trio
|
||||
|
||||
from ._util import (
|
||||
get_console_log,
|
||||
)
|
||||
from ._mngr import (
|
||||
Services,
|
||||
open_service_mngr,
|
||||
ServiceMngr,
|
||||
)
|
||||
from ._registry import ( # noqa
|
||||
_tractor_kwargs,
|
||||
|
@ -59,7 +58,7 @@ async def open_piker_runtime(
|
|||
registry_addrs: list[tuple[str, int]] = [],
|
||||
|
||||
enable_modules: list[str] = [],
|
||||
loglevel: Optional[str] = None,
|
||||
loglevel: str|None = None,
|
||||
|
||||
# XXX NOTE XXX: you should pretty much never want debug mode
|
||||
# for data daemons when running in production.
|
||||
|
@ -69,7 +68,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[
|
||||
|
@ -119,6 +118,10 @@ 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,
|
||||
|
||||
|
@ -167,12 +170,13 @@ async def open_pikerd(
|
|||
|
||||
**kwargs,
|
||||
|
||||
) -> Services:
|
||||
) -> ServiceMngr:
|
||||
'''
|
||||
Start a root piker daemon with an indefinite lifetime.
|
||||
Start a root piker daemon actor (aka `pikerd`) with an indefinite
|
||||
lifetime.
|
||||
|
||||
A root actor nursery is created which can be used to create and keep
|
||||
alive underling services (see below).
|
||||
A root actor-nursery is created which can be used to spawn and
|
||||
supervise underling service sub-actors (see below).
|
||||
|
||||
'''
|
||||
# NOTE: for the root daemon we always enable the root
|
||||
|
@ -199,8 +203,6 @@ 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:
|
||||
|
@ -209,25 +211,17 @@ async def open_pikerd(
|
|||
'Maybe you have another daemon already running?'
|
||||
)
|
||||
|
||||
# assign globally for future daemon/task creation
|
||||
Services.actor_n = actor_nursery
|
||||
Services.service_n = service_nursery
|
||||
Services.debug_mode = debug_mode
|
||||
|
||||
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()
|
||||
mngr: ServiceMngr
|
||||
async with open_service_mngr(
|
||||
debug_mode=debug_mode,
|
||||
) as mngr:
|
||||
yield mngr
|
||||
|
||||
|
||||
# TODO: do we even need this?
|
||||
# @acm
|
||||
# async def maybe_open_runtime(
|
||||
# loglevel: Optional[str] = None,
|
||||
# loglevel: str|None = None,
|
||||
# **kwargs,
|
||||
|
||||
# ) -> None:
|
||||
|
@ -256,7 +250,7 @@ async def maybe_open_pikerd(
|
|||
loglevel: str | None = None,
|
||||
**kwargs,
|
||||
|
||||
) -> tractor._portal.Portal | ClassVar[Services]:
|
||||
) -> tractor._portal.Portal | ClassVar[ServiceMngr]:
|
||||
'''
|
||||
If no ``pikerd`` daemon-root-actor can be found start it and
|
||||
yield up (we should probably figure out returning a portal to self
|
||||
|
|
|
@ -49,7 +49,7 @@ from requests.exceptions import (
|
|||
ReadTimeout,
|
||||
)
|
||||
|
||||
from ._mngr import Services
|
||||
from ._mngr import ServiceMngr
|
||||
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: Services,
|
||||
services: ServiceMngr,
|
||||
service_name: str,
|
||||
|
||||
# endpoint config passed as **kwargs
|
||||
|
@ -549,7 +549,8 @@ 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):
|
||||
|
|
|
@ -26,14 +26,17 @@ 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 (
|
||||
Services,
|
||||
get_service_mngr,
|
||||
ServiceMngr,
|
||||
)
|
||||
from ._actor_runtime import maybe_open_pikerd
|
||||
from ._registry import find_service
|
||||
|
@ -41,15 +44,14 @@ 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:
|
||||
|
@ -67,7 +69,7 @@ async def maybe_spawn_daemon(
|
|||
'''
|
||||
# serialize access to this section to avoid
|
||||
# 2 or more tasks racing to create a daemon
|
||||
lock = Services.locks[service_name]
|
||||
lock = _locks[service_name]
|
||||
await lock.acquire()
|
||||
|
||||
async with find_service(
|
||||
|
@ -102,6 +104,12 @@ async def maybe_spawn_daemon(
|
|||
# service task for that actor.
|
||||
started: bool
|
||||
if pikerd_portal is None:
|
||||
|
||||
# await tractor.pause()
|
||||
if tractor_kwargs.get('debug_mode', False):
|
||||
from tractor.devx._debug import maybe_init_greenback
|
||||
await maybe_init_greenback()
|
||||
|
||||
started = await service_task_target(
|
||||
loglevel=loglevel,
|
||||
**spawn_args,
|
||||
|
@ -132,7 +140,65 @@ async def maybe_spawn_daemon(
|
|||
async with tractor.wait_for_actor(service_name) as portal:
|
||||
lock.release()
|
||||
yield portal
|
||||
await portal.cancel_actor()
|
||||
# --- ---- ---
|
||||
# XXX NOTE XXX
|
||||
# --- ---- ---
|
||||
# DO NOT PUT A `portal.cancel_actor()` here (as was prior)!
|
||||
#
|
||||
# Doing so will cause an "out-of-band" ctxc
|
||||
# (`tractor.ContextCancelled`) to be raised inside the
|
||||
# `ServiceMngr.open_context_in_task()`'s call to
|
||||
# `ctx.wait_for_result()` AND the internal self-ctxc
|
||||
# "graceful capture" WILL NOT CATCH IT!
|
||||
#
|
||||
# This can cause certain types of operations to raise
|
||||
# that ctxc BEFORE THEY `return`, resulting in
|
||||
# a "false-negative" ctxc being raised when really
|
||||
# nothing actually failed, other then our semantic
|
||||
# "failure" to suppress an expected, graceful,
|
||||
# self-cancel scenario..
|
||||
#
|
||||
# bUt wHy duZ It WorK lIKe dis..
|
||||
# ------------------------------
|
||||
# from the perspective of the `tractor.Context` this
|
||||
# cancel request was conducted "out of band" since
|
||||
# `Context.cancel()` was never called and thus the
|
||||
# `._cancel_called: bool` was never set. Despite the
|
||||
# remote `.canceller` being set to `pikerd` (i.e. the
|
||||
# same `Actor.uid` of the raising service-mngr task) the
|
||||
# service-task's ctx itself was never marked as having
|
||||
# requested cancellation and thus still raises the ctxc
|
||||
# bc it was unaware of any such request.
|
||||
#
|
||||
# How to make grokin these cases easier tho?
|
||||
# ------------------------------------------
|
||||
# Because `Portal.cancel_actor()` was called it requests
|
||||
# "full-`Actor`-runtime-cancellation" of it's peer
|
||||
# process which IS NOT THE SAME as a single inter-actor
|
||||
# RPC task cancelling its local context with a remote
|
||||
# peer `Task` in that same peer process.
|
||||
#
|
||||
# ?TODO? It might be better if we do one (or all) of the
|
||||
# following:
|
||||
#
|
||||
# -[ ] at least set a special message for the
|
||||
# `ContextCancelled` when raised locally by the
|
||||
# unaware ctx task such that we check for the
|
||||
# `.canceller` being *our `Actor`* and in the case
|
||||
# where `Context._cancel_called == False` we specially
|
||||
# note that this is likely an "out-of-band"
|
||||
# runtime-cancel request triggered by some call to
|
||||
# `Portal.cancel_actor()`, possibly even reporting the
|
||||
# exact LOC of that caller by tracking it inside our
|
||||
# portal-type?
|
||||
# -[ ] possibly add another field `ContextCancelled` like
|
||||
# maybe a,
|
||||
# `.request_type: Literal['os', 'proc', 'actor',
|
||||
# 'ctx']` type thing which would allow immediately
|
||||
# being able to tell what kind of cancellation caused
|
||||
# the unexpected ctxc?
|
||||
# -[ ] REMOVE THIS COMMENT, once we've settled on how to
|
||||
# better augment `tractor` to be more explicit on this!
|
||||
|
||||
|
||||
async def spawn_emsd(
|
||||
|
@ -147,26 +213,25 @@ async def spawn_emsd(
|
|||
"""
|
||||
log.info('Spawning emsd')
|
||||
|
||||
portal = await Services.actor_n.start_actor(
|
||||
smngr: ServiceMngr = get_service_mngr()
|
||||
portal = await smngr.an.start_actor(
|
||||
'emsd',
|
||||
enable_modules=[
|
||||
'piker.clearing._ems',
|
||||
'piker.clearing._client',
|
||||
],
|
||||
loglevel=loglevel,
|
||||
debug_mode=Services.debug_mode, # set by pikerd flag
|
||||
debug_mode=smngr.debug_mode, # set by pikerd flag
|
||||
**extra_tractor_kwargs
|
||||
)
|
||||
|
||||
# non-blocking setup of clearing service
|
||||
from ..clearing._ems import _setup_persistent_emsd
|
||||
|
||||
await Services.start_service_task(
|
||||
'emsd',
|
||||
portal,
|
||||
|
||||
# signature of target root-task endpoint
|
||||
_setup_persistent_emsd,
|
||||
await smngr.start_service_ctx(
|
||||
name='emsd',
|
||||
portal=portal,
|
||||
ctx_fn=_setup_persistent_emsd,
|
||||
loglevel=loglevel,
|
||||
)
|
||||
return True
|
||||
|
|
|
@ -18,148 +18,36 @@
|
|||
daemon-service management API.
|
||||
|
||||
"""
|
||||
from collections import defaultdict
|
||||
from typing import (
|
||||
Callable,
|
||||
Any,
|
||||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
)
|
||||
|
||||
import trio
|
||||
from trio_typing import TaskStatus
|
||||
import tractor
|
||||
from tractor import (
|
||||
current_actor,
|
||||
ContextCancelled,
|
||||
Context,
|
||||
Portal,
|
||||
from tractor.hilevel import (
|
||||
ServiceMngr,
|
||||
# open_service_mngr as _open_service_mngr,
|
||||
get_service_mngr as get_service_mngr,
|
||||
)
|
||||
# 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
|
||||
|
||||
from ._util import (
|
||||
log, # sub-sys logger
|
||||
)
|
||||
Services: ServiceMngr|None = None
|
||||
|
||||
@acm
|
||||
async def open_service_mngr(
|
||||
**kwargs,
|
||||
) -> ServiceMngr:
|
||||
|
||||
# TODO: we need remote wrapping and a general soln:
|
||||
# - factor this into a ``tractor.highlevel`` extension # pack for the
|
||||
# library.
|
||||
# - wrap a "remote api" wherein you can get a method proxy
|
||||
# to the pikerd actor for starting services remotely!
|
||||
# - prolly rename this to ActorServicesNursery since it spawns
|
||||
# new actors and supervises them to completion?
|
||||
class Services:
|
||||
|
||||
actor_n: tractor._supervise.ActorNursery
|
||||
service_n: trio.Nursery
|
||||
debug_mode: bool # tractor sub-actor debug mode flag
|
||||
service_tasks: dict[
|
||||
str,
|
||||
tuple[
|
||||
trio.CancelScope,
|
||||
Portal,
|
||||
trio.Event,
|
||||
]
|
||||
] = {}
|
||||
locks = defaultdict(trio.Lock)
|
||||
|
||||
@classmethod
|
||||
async def start_service_task(
|
||||
self,
|
||||
name: str,
|
||||
portal: Portal,
|
||||
target: Callable,
|
||||
allow_overruns: bool = False,
|
||||
**ctx_kwargs,
|
||||
|
||||
) -> (trio.CancelScope, Context):
|
||||
'''
|
||||
Open a context in a service sub-actor, add to a stack
|
||||
that gets unwound at ``pikerd`` teardown.
|
||||
|
||||
This allows for allocating long-running sub-services in our main
|
||||
daemon and explicitly controlling their lifetimes.
|
||||
|
||||
'''
|
||||
async def open_context_in_task(
|
||||
task_status: TaskStatus[
|
||||
tuple[
|
||||
trio.CancelScope,
|
||||
trio.Event,
|
||||
Any,
|
||||
]
|
||||
] = trio.TASK_STATUS_IGNORED,
|
||||
|
||||
) -> Any:
|
||||
|
||||
with trio.CancelScope() as cs:
|
||||
|
||||
async with portal.open_context(
|
||||
target,
|
||||
allow_overruns=allow_overruns,
|
||||
**ctx_kwargs,
|
||||
|
||||
) as (ctx, first):
|
||||
|
||||
# unblock once the remote context has started
|
||||
complete = trio.Event()
|
||||
task_status.started((cs, complete, first))
|
||||
log.info(
|
||||
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.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 (await portal.result(), ctx_res)
|
||||
except ContextCancelled as ctxe:
|
||||
canceller: tuple[str, str] = ctxe.canceller
|
||||
our_uid: tuple[str, str] = current_actor().uid
|
||||
if (
|
||||
canceller != portal.channel.uid
|
||||
and
|
||||
canceller != our_uid
|
||||
):
|
||||
log.cancel(
|
||||
f'Actor-service {name} was remotely cancelled?\n'
|
||||
f'remote canceller: {canceller}\n'
|
||||
f'Keeping {our_uid} alive, ignoring sub-actor cancel..\n'
|
||||
)
|
||||
else:
|
||||
raise
|
||||
|
||||
|
||||
|
||||
finally:
|
||||
await portal.cancel_actor()
|
||||
complete.set()
|
||||
self.service_tasks.pop(name)
|
||||
|
||||
cs, complete, first = await self.service_n.start(open_context_in_task)
|
||||
|
||||
# store the cancel scope and portal for later cancellation or
|
||||
# retstart if needed.
|
||||
self.service_tasks[name] = (cs, portal, complete)
|
||||
|
||||
return cs, first
|
||||
|
||||
@classmethod
|
||||
async def cancel_service(
|
||||
self,
|
||||
name: str,
|
||||
|
||||
) -> Any:
|
||||
'''
|
||||
Cancel the service task and actor for the given ``name``.
|
||||
|
||||
'''
|
||||
log.info(f'Cancelling `pikerd` service {name}')
|
||||
cs, portal, complete = self.service_tasks[name]
|
||||
cs.cancel()
|
||||
await complete.wait()
|
||||
assert name not in self.service_tasks, \
|
||||
f'Serice task for {name} not terminated?'
|
||||
global Services
|
||||
async with tractor.hilevel.open_service_mngr(
|
||||
**kwargs,
|
||||
) as mngr:
|
||||
# Services = proxy(mngr)
|
||||
Services = mngr
|
||||
yield mngr
|
||||
Services = None
|
||||
|
|
|
@ -21,11 +21,13 @@ 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 (
|
||||
|
@ -127,7 +129,7 @@ def start_elasticsearch(
|
|||
|
||||
@acm
|
||||
async def start_ahab_daemon(
|
||||
service_mngr: Services,
|
||||
service_mngr: ServiceMngr,
|
||||
user_config: dict | None = None,
|
||||
loglevel: str | None = None,
|
||||
|
||||
|
|
|
@ -53,7 +53,7 @@ import pendulum
|
|||
# import purerpc
|
||||
|
||||
from ..data.feed import maybe_open_feed
|
||||
from . import Services
|
||||
from . import ServiceMngr
|
||||
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: Services,
|
||||
service_mngr: ServiceMngr,
|
||||
user_config: dict | None = None,
|
||||
loglevel: str | None = None,
|
||||
|
||||
|
|
|
@ -458,13 +458,15 @@ async def start_backfill(
|
|||
'bf_until <- last_start_dt:\n'
|
||||
f'{backfill_until_dt} <- {last_start_dt}\n'
|
||||
)
|
||||
|
||||
# ugh, what's a better way?
|
||||
# TODO: fwiw, we probably want a way to signal a throttle
|
||||
# condition (eg. with ib) so that we can halt the
|
||||
# request loop until the condition is resolved?
|
||||
if timeframe > 1:
|
||||
await tractor.pause()
|
||||
# UGH: what's a better way?
|
||||
# TODO: backends are responsible for being correct on
|
||||
# this right!?
|
||||
# -[ ] in the `ib` case we could maybe offer some way
|
||||
# to halt the request loop until the condition is
|
||||
# resolved or should the backend be entirely in
|
||||
# charge of solving such faults? yes, right?
|
||||
# if timeframe > 1:
|
||||
# await tractor.pause()
|
||||
return
|
||||
|
||||
assert (
|
||||
|
@ -578,6 +580,7 @@ async def start_backfill(
|
|||
'crypto',
|
||||
'crypto_currency',
|
||||
'fiat', # a "forex pair"
|
||||
'perpetual_future', # stupid "perps" from cex land
|
||||
}:
|
||||
# for now, our table key schema is not including
|
||||
# the dst[/src] source asset token.
|
||||
|
|
|
@ -10,7 +10,7 @@ from piker import (
|
|||
config,
|
||||
)
|
||||
from piker.service import (
|
||||
Services,
|
||||
get_service_mngr,
|
||||
)
|
||||
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 Services
|
||||
assert service_manager is get_service_mngr()
|
||||
|
||||
async with tractor.wait_for_actor(
|
||||
'pikerd',
|
||||
|
|
|
@ -26,7 +26,7 @@ import pytest
|
|||
import tractor
|
||||
from uuid import uuid4
|
||||
|
||||
from piker.service import Services
|
||||
from piker.service import ServiceMngr
|
||||
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: Services,
|
||||
open_test_pikerd: ServiceMngr,
|
||||
loglevel: str,
|
||||
):
|
||||
async def load_bad_fqme():
|
||||
|
|
|
@ -15,7 +15,7 @@ import tractor
|
|||
|
||||
from piker.service import (
|
||||
find_service,
|
||||
Services,
|
||||
ServiceMngr,
|
||||
)
|
||||
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: Services
|
||||
services: ServiceMngr
|
||||
|
||||
async with (
|
||||
open_test_pikerd(
|
||||
|
|
Loading…
Reference in New Issue