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

`.tsp._history`: drop `feed_is_live` syncing, another seg flag 

The `await feed_is_live.wait()` is more or less pointless and would only
cause slower startup afaig (as-far-as-i-grok) so i'm masking it here.
This also removes the final `strict_exception_groups=False` use from the
non-tests code base, flipping to the `tractor.trionics` collapser once
and for all!
hist_backfill_fixes
Tyler Goodlet 2025-10-02 19:53:08 -04:00
parent 4d50bebf11
commit cca2b14548
1 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
piker/tsp/_history.py
View File

@ -886,7 +886,7 @@ async def load_tsdb_hist(
np.ndarray, np.ndarray,
DateTime, DateTime,
DateTime, DateTime,
] | None: ]|None:
# loads a (large) frame of data from the tsdb depending # loads a (large) frame of data from the tsdb depending
# on the db's query size limit; our "nativedb" (using # on the db's query size limit; our "nativedb" (using
# parquet) generally can load the entire history into mem # parquet) generally can load the entire history into mem
@ -899,7 +899,7 @@ async def load_tsdb_hist(
DateTime, DateTime,
] ]
try: try:
tsdb_entry: tuple | None = await storage.load( tsdb_entry: tuple|None = await storage.load(
fqme, fqme,
timeframe=timeframe, timeframe=timeframe,
) )
@ -1046,12 +1046,15 @@ async def tsdb_backfill(
last_tsdb_dt, last_tsdb_dt,
) = tsdb_entry ) = tsdb_entry
# await tractor.pause()
# if there is a gap to backfill from the first # if there is a gap to backfill from the first
# history frame until the last datum loaded from the tsdb # history frame until the last datum loaded from the tsdb
# continue that now in the background # continue that now in the background
async with trio.open_nursery( async with (
strict_exception_groups=False, tractor.trionics.collapse_eg(),
) as tn: trio.open_nursery() as tn,
):
bf_done = await tn.start( bf_done = await tn.start(
partial( partial(
@ -1322,8 +1325,14 @@ async def manage_history(
# TODO: maybe it should be a subpkg of `.data`? # TODO: maybe it should be a subpkg of `.data`?
from piker import storage from piker import storage
storemod: ModuleType
client: StorageClient
tn: trio.Nursery
async with ( async with (
storage.open_storage_client() as (storemod, client), storage.open_storage_client() as (
storemod,
client,
),
# NOTE: this nursery spawns a task per "timeframe" (aka # NOTE: this nursery spawns a task per "timeframe" (aka
# sampling period) data set since normally differently # sampling period) data set since normally differently
@ -1392,7 +1401,7 @@ async def manage_history(
some_data_ready.set() some_data_ready.set()
# wait for a live feed before starting the sampler. # wait for a live feed before starting the sampler.
await feed_is_live.wait() # await feed_is_live.wait()
# yield back after client connect with filled shm # yield back after client connect with filled shm
task_status.started(( task_status.started((