pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 5 Packages Projects Releases Wiki Activity

More detailed rt feed drop logging

incremental_update_paths
Tyler Goodlet 2022-06-01 15:01:30 -04:00
parent e6d03ba97f
commit 80835d4e04
1 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
piker/data/_sampling.py
View File

@ -336,7 +336,8 @@ async def sample_and_broadcast(
key = id(stream) key = id(stream)
overruns[key] += 1 overruns[key] += 1
log.warning( log.warning(
f'Feed overrun {bus.brokername} -> ' f'Feed overrun {broker_symbol}'
'@{bus.brokername} -> '
f'feed @ {tick_throttle} Hz' f'feed @ {tick_throttle} Hz'
) )
if overruns[key] > 6: if overruns[key] > 6:
@ -375,11 +376,12 @@ async def sample_and_broadcast(
trio.ClosedResourceError, trio.ClosedResourceError,
trio.EndOfChannel, trio.EndOfChannel,
): ):
ctx = getattr(stream, '_ctx', None) chan = ctx.chan
if ctx: if ctx:
log.warning( log.warning(
f'{ctx.chan.uid} dropped ' 'Dropped `brokerd`-quotes-feed connection:\n'
'`brokerd`-quotes-feed connection' f'{broker_symbol}:'
f'{ctx.cid}@{chan.uid}'
) )
if tick_throttle: if tick_throttle:
assert stream._closed assert stream._closed
@ -392,7 +394,11 @@ async def sample_and_broadcast(
try: try:
subs.remove((stream, tick_throttle)) subs.remove((stream, tick_throttle))
except ValueError: except ValueError:
log.error(f'{stream} was already removed from subs!?') log.error(
f'Stream was already removed from subs!?\n'
f'{broker_symbol}:'
f'{ctx.cid}@{chan.uid}'
)
# TODO: a less naive throttler, here's some snippets: # TODO: a less naive throttler, here's some snippets: