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

Don't open stream before starting the fsp context..

fsp_hotfixes
Tyler Goodlet 2021-11-05 10:04:10 -04:00
parent ea9b66d1c3
commit 2b97f98151
1 changed files with 18 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
piker/fsp/_engine.py
View File

@ -79,7 +79,7 @@ async def filter_quotes_by_sym(
async def fsp_compute(
stream: tractor.MsgStream,
ctx: tractor.Context,
symbol: str,
feed: Feed,
quote_stream: trio.abc.ReceiveChannel,
@ -147,6 +147,7 @@ async def fsp_compute(
# setup a respawn handle
with trio.CancelScope() as cs:
tracker = TaskTracker(trio.Event(), cs)
await ctx.started(index)
task_status.started((tracker, index))
profiler(f'{func_name} yield last index')
@ -155,23 +156,24 @@ async def fsp_compute(
try:
# rt stream
async for processed in out_stream:
async with ctx.open_stream() as stream:
async for processed in out_stream:
log.debug(f"{func_name}: {processed}")
index = src.index
dst.array[-1][func_name] = processed
log.debug(f"{func_name}: {processed}")
index = src.index
dst.array[-1][func_name] = processed
# NOTE: for now we aren't streaming this to the consumer
# stream latest array index entry which basically just acts
# as trigger msg to tell the consumer to read from shm
if attach_stream:
await stream.send(index)
# NOTE: for now we aren't streaming this to the consumer
# stream latest array index entry which basically just acts
# as trigger msg to tell the consumer to read from shm
if attach_stream:
await stream.send(index)
# period = time.time() - last
# hz = 1/period if period else float('nan')
# if hz > 60:
# log.info(f'FSP quote too fast: {hz}')
# last = time.time()
# period = time.time() - last
# hz = 1/period if period else float('nan')
# if hz > 60:
# log.info(f'FSP quote too fast: {hz}')
# last = time.time()
finally:
tracker.complete.set()
@ -229,14 +231,13 @@ async def cascade(
# last_len = new_len = len(src.array)
async with (
ctx.open_stream() as stream,
trio.open_nursery() as n,
):
fsp_target = partial(
fsp_compute,
stream=stream,
ctx=ctx,
symbol=symbol,
feed=feed,
quote_stream=quote_stream,
@ -255,7 +256,6 @@ async def cascade(
last = dst.array[-1:]
zeroed = np.zeros(last.shape, dtype=last.dtype)
await ctx.started(index)
profiler(f'{func_name}: fsp up')
async def resync(tracker: TaskTracker) -> tuple[TaskTracker, int]: