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

Support not registering for sample-index msgs via `sub_for_broadcasts: bool` flag

samplerd_service
Tyler Goodlet 2023-01-05 10:12:42 -05:00
parent e0ca5d5200
commit 2778ee1401
1 changed files with 13 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
piker/data/_sampling.py
View File

@ -162,7 +162,6 @@ class Sampler:
if shm_period_s not in broadcasted: if shm_period_s not in broadcasted:
sub_pair = self.subscribers[shm_period_s] sub_pair = self.subscribers[shm_period_s]
sub_pair[0] = i_epoch sub_pair[0] = i_epoch
print(f'skipping `{shm_period_s}s` sample update')
broadcasted.add(shm_period_s) broadcasted.add(shm_period_s)
# TODO: ``numba`` this! # TODO: ``numba`` this!
@ -174,8 +173,8 @@ class Sampler:
array = shm.array array = shm.array
last = array[-1:][shm._write_fields].copy() last = array[-1:][shm._write_fields].copy()
# guard against startup backfilling race with # guard against startup backfilling races where
# empty buffers. # the buffer has not yet been filled.
if not last.size: if not last.size:
continue continue
@ -288,7 +287,9 @@ async def register_with_sampler(
ctx: tractor.Context, ctx: tractor.Context,
period_s: float, period_s: float,
shms_by_period: dict[float, dict] | None = None, shms_by_period: dict[float, dict] | None = None,
open_index_stream: bool = True,
open_index_stream: bool = True, # open a 2way stream for sample step msgs?
sub_for_broadcasts: bool = True, # sampler side to send step updates?
) -> None: ) -> None:
@ -341,15 +342,16 @@ async def register_with_sampler(
if open_index_stream: if open_index_stream:
try: try:
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
subs.add(stream) if sub_for_broadcasts:
subs.add(stream)
# except broadcast requests from the subscriber # except broadcast requests from the subscriber
async for msg in stream: async for msg in stream:
if msg == 'broadcast_all': if msg == 'broadcast_all':
await Sampler.broadcast_all() await Sampler.broadcast_all()
finally: finally:
subs.remove(stream) if sub_for_broadcasts:
subs.remove(stream)
else: else:
# if no shms are passed in we just wait until cancelled # if no shms are passed in we just wait until cancelled
# by caller. # by caller.
@ -404,6 +406,7 @@ async def spawn_samplerd(
portal, portal,
register_with_sampler, register_with_sampler,
period_s=1, period_s=1,
sub_for_broadcasts=False,
) )
return True return True
@ -437,9 +440,10 @@ async def maybe_open_samplerd(
@acm @acm
async def open_sample_stream( async def open_sample_stream(
period_s: int, period_s: float,
shms_by_period: dict[float, dict] | None = None, shms_by_period: dict[float, dict] | None = None,
open_index_stream: bool = True, open_index_stream: bool = True,
sub_for_broadcasts: bool = True,
cache_key: str | None = None, cache_key: str | None = None,
allow_new_sampler: bool = True, allow_new_sampler: bool = True,
@ -482,6 +486,7 @@ async def open_sample_stream(
'period_s': period_s, 'period_s': period_s,
'shms_by_period': shms_by_period, 'shms_by_period': shms_by_period,
'open_index_stream': open_index_stream, 'open_index_stream': open_index_stream,
'sub_for_broadcasts': sub_for_broadcasts,
}, },
) as (ctx, first) ) as (ctx, first)
): ):