pikers
/
piker
4
1
Fork 0
Code Issues 10 Pull Requests 14 Packages Projects Releases Wiki Activity

Append broker name to symbols before quotes broadcast in sampler task

mkts_backup
Tyler Goodlet 2022-03-18 15:05:32 -04:00
parent d03cd23571
commit c0d1facf3b
1 changed files with 15 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
piker/data/_sampling.py
View File

@ -184,6 +184,7 @@ async def sample_and_broadcast(
bus: '_FeedsBus', # noqa
shm: ShmArray,
quote_stream: trio.abc.ReceiveChannel,
brokername: str,
sum_tick_vlm: bool = True,
) -> None:
@ -193,8 +194,7 @@ async def sample_and_broadcast(
# iterate stream delivered by broker
async for quotes in quote_stream:
# TODO: ``numba`` this!
for sym, quote in quotes.items():
for broker_symbol, quote in quotes.items():
# TODO: in theory you can send the IPC msg *before* writing
# to the sharedmem array to decrease latency, however, that
# will require at least some way to prevent task switching
@ -258,7 +258,13 @@ async def sample_and_broadcast(
# end up triggering backpressure which which will
# eventually block this producer end of the feed and
# thus other consumers still attached.
subs = bus._subscribers[sym.lower()]
subs = bus._subscribers[broker_symbol.lower()]
# NOTE: by default the broker backend doesn't append
# it's own "name" into the fqsn schema (but maybe it
# should?) so we have to manually generate the correct
# key here.
bsym = f'{broker_symbol}.{brokername}'
lags = 0
for (stream, tick_throttle) in subs:
@ -269,7 +275,9 @@ async def sample_and_broadcast(
# this is a send mem chan that likely
# pushes to the ``uniform_rate_send()`` below.
try:
stream.send_nowait((sym, quote))
stream.send_nowait(
(bsym, quote)
)
except trio.WouldBlock:
ctx = getattr(stream, '_ctx', None)
if ctx:
@ -283,7 +291,9 @@ async def sample_and_broadcast(
f'feed @ {tick_throttle} Hz'
)
else:
await stream.send({sym: quote})
await stream.send(
{bsym: quote}
)
if cs.cancelled_caught:
lags += 1