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

Move sync log msg back to info

epoch_index_backup
Tyler Goodlet 2023-01-05 19:32:34 -05:00
parent bfeebba734
commit acef3505fd
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
piker/fsp/_engine.py
View File

@ -21,7 +21,9 @@ core task logic for processing chains
from dataclasses import dataclass from dataclasses import dataclass
from functools import partial from functools import partial
from typing import ( from typing import (
AsyncIterator, Callable, Optional, AsyncIterator,
Callable,
Optional,
Union, Union,
) )
@ -386,7 +388,7 @@ async def cascade(
) -> tuple[TaskTracker, int]: ) -> tuple[TaskTracker, int]:
# TODO: adopt an incremental update engine/approach # TODO: adopt an incremental update engine/approach
# where possible here eventually! # where possible here eventually!
log.debug(f're-syncing fsp {func_name} to source') log.info(f're-syncing fsp {func_name} to source')
tracker.cs.cancel() tracker.cs.cancel()
await tracker.complete.wait() await tracker.complete.wait()
tracker, index = await n.start(fsp_target) tracker, index = await n.start(fsp_target)
@ -429,6 +431,7 @@ async def cascade(
tracker: TaskTracker, tracker: TaskTracker,
src: ShmArray, src: ShmArray,
dst: ShmArray, dst: ShmArray,
) -> tuple[TaskTracker, int]: ) -> tuple[TaskTracker, int]:
synced, step_diff, _ = is_synced(src, dst) synced, step_diff, _ = is_synced(src, dst)
@ -444,7 +447,8 @@ async def cascade(
# signal # signal
times = src.array['time'] times = src.array['time']
if len(times) > 1: if len(times) > 1:
delay_s = float(times[-1] - times[times != times[-1]][-1]) last_ts = times[-1]
delay_s = float(last_ts - times[times != last_ts][-1])
else: else:
# our default "HFT" sample rate. # our default "HFT" sample rate.
delay_s = _default_delay_s delay_s = _default_delay_s