Use `tractor.ipc._shm` types directly across codebase

Port all 16 internal import sites from re-exporting
via `piker.data._sharedmem` shim to importing core
shm types directly from `tractor.ipc._shm`.

Deats,
- `ShmArray` now imported from tractor in 10 files.
- `_Token` renamed to `NDToken` everywhere (5 files).
- `attach_shm_array` → `attach_shm_ndarray` at all
  call sites.
- `data/__init__.py` sources `ShmArray`,
  `get_shm_token` from tractor; keeps
  `open/attach_shm_array` as public API aliases.
- Trim shim to only piker-specific wrappers:
  `_make_token()`, `maybe_open_shm_array()`,
  `try_read()`.
- Drop `Optional` usage in shim, use `|None`.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

piker/brokers/binance/feed.py

@@ -203,9 +203,13 @@ async def stream_messages(
                 yield 'trade', piker_quote
 
 
-def make_sub(pairs: list[str], sub_name: str, uid: int) -> dict[str, str]:
+def make_sub(
+    pairs: list[str],
+    sub_name: str,
+    uid: int,
+) -> dict[str, str]:
     '''
-    Create a request subscription packet dict.
+    Create a request subscription packet `dict`.
 
     - spot:
       https://binance-docs.github.io/apidocs/spot/en/#live-subscribing-unsubscribing-to-streams
@@ -332,7 +336,8 @@ async def get_mkt_info(
         # TODO: handle coinm futes which have a margin asset that
         # is some crypto token!
         # https://binance-docs.github.io/apidocs/delivery/en/#exchange-information
-        or 'btc' in venue_lower
+        or
+        'btc' in venue_lower
     ):
         return None
 
@@ -343,12 +348,14 @@ async def get_mkt_info(
 
     if (
         venue
-        and 'spot' not in venue_lower
+        and
+        'spot' not in venue_lower
 
         # XXX: catch all in case user doesn't know which
         # venue they want (usdtm vs. coinm) and we can choose
         # a default (via config?) once we support coin-m APIs.
-        or 'perp' in venue_lower
+        or
+        'perp' in venue_lower
     ):
         if not mkt_mode:
             mkt_mode: str = f'{venue_lower}_futes'

piker/brokers/deribit/feed.py

@@ -32,7 +32,7 @@ import tractor
 
 from piker.brokers import open_cached_client
 from piker.log import get_logger, get_console_log
-from piker.data import ShmArray
+from tractor.ipc._shm import ShmArray
 from piker.brokers._util import (
     BrokerError,
     DataUnavailable,

piker/data/__init__.py

@@ -23,13 +23,13 @@ sharing live streams over a network.
 
 """
 from .ticktools import iterticks
-from ._sharedmem import (
+from tractor.ipc._shm import (
-    maybe_open_shm_array,
-    attach_shm_array,
-    open_shm_array,
-    get_shm_token,
     ShmArray,
+    get_shm_token,
+    open_shm_ndarray as open_shm_array,
+    attach_shm_ndarray as attach_shm_array,
 )
+from ._sharedmem import maybe_open_shm_array
 from ._source import (
     def_iohlcv_fields,
     def_ohlcv_fields,

piker/data/_formatters.py

@@ -28,9 +28,7 @@ from msgspec import field
 import numpy as np
 from numpy.lib import recfunctions as rfn
 
-from ._sharedmem import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from ._pathops import (
     path_arrays_from_ohlc,
 )

piker/data/_sampling.py

@@ -55,9 +55,7 @@ from ._util import (
 from ..service import maybe_spawn_daemon
 
 if TYPE_CHECKING:
-    from ._sharedmem import (
+    from tractor.ipc._shm import ShmArray
-        ShmArray,
-    )
     from .feed import (
         _FeedsBus,
         Sub,
@@ -378,16 +376,16 @@ async def register_with_sampler(
                 # feed_is_live.is_set()
                 # ^TODO? pass it in instead?
             ):
-                from ._sharedmem import (
+                from tractor.ipc._shm import (
-                    attach_shm_array,
+                    attach_shm_ndarray,
-                    _Token,
+                    NDToken,
                 )
                 for period in shms_by_period:
 
                     # load and register shm handles
                     shm_token_msg = shms_by_period[period]
-                    shm = attach_shm_array(
+                    shm = attach_shm_ndarray(
-                        _Token.from_msg(shm_token_msg),
+                        NDToken.from_msg(shm_token_msg),
                         readonly=False,
                     )
                     shms_by_period[period] = shm

piker/data/_sharedmem.py

@@ -1,622 +1,67 @@
 # piker: trading gear for hackers
 # Copyright (C) Tyler Goodlet (in stewardship for pikers)
 
-# This program is free software: you can redistribute it and/or modify
+# This program is free software: you can redistribute it
-# it under the terms of the GNU Affero General Public License as published by
+# and/or modify it under the terms of the GNU Affero General
-# the Free Software Foundation, either version 3 of the License, or
+# Public License as published by the Free Software
-# (at your option) any later version.
+# Foundation, either version 3 of the License, or (at your
+# option) any later version.
 
-# This program is distributed in the hope that it will be useful,
+# This program is distributed in the hope that it will be
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# useful, but WITHOUT ANY WARRANTY; without even the implied
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
-# GNU Affero General Public License for more details.
+# PURPOSE.  See the GNU Affero General Public License for
+# more details.
 
-# You should have received a copy of the GNU Affero General Public License
+# You should have received a copy of the GNU Affero General
-# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+# Public License along with this program.  If not, see
+# <https://www.gnu.org/licenses/>.
 
-"""
+'''
-NumPy compatible shared memory buffers for real-time IPC streaming.
+Piker-specific shared memory helpers.
 
-"""
+Thin shim providing piker-only wrappers around
-from __future__ import annotations
+``tractor.ipc._shm``; all core types and functions
-from sys import byteorder
+are now imported directly from tractor throughout
-import time
+the codebase.
-from typing import Optional
-from multiprocessing.shared_memory import SharedMemory, _USE_POSIX
 
-if _USE_POSIX:
+'''
-    from _posixshmem import shm_unlink
-
-# import msgspec
 import numpy as np
-from numpy.lib import recfunctions as rfn
+
-import tractor
+from tractor.ipc._shm import (
+    NDToken,
+    ShmArray,
+    _known_tokens,
+    _make_token as _tractor_make_token,
+    open_shm_ndarray,
+    attach_shm_ndarray,
+)
 
 from ._util import log
-from ._source import def_iohlcv_fields
-from piker.types import Struct
-
-
-def cuckoff_mantracker():
-    '''
-    Disable all ``multiprocessing``` "resource tracking" machinery since
-    it's an absolute multi-threaded mess of non-SC madness.
-
-    '''
-    from multiprocessing import resource_tracker as mantracker
-
-    # Tell the "resource tracker" thing to fuck off.
-    class ManTracker(mantracker.ResourceTracker):
-        def register(self, name, rtype):
-            pass
-
-        def unregister(self, name, rtype):
-            pass
-
-        def ensure_running(self):
-            pass
-
-    # "know your land and know your prey"
-    # https://www.dailymotion.com/video/x6ozzco
-    mantracker._resource_tracker = ManTracker()
-    mantracker.register = mantracker._resource_tracker.register
-    mantracker.ensure_running = mantracker._resource_tracker.ensure_running
-    mantracker.unregister = mantracker._resource_tracker.unregister
-    mantracker.getfd = mantracker._resource_tracker.getfd
-
-
-cuckoff_mantracker()
-
-
-class SharedInt:
-    """Wrapper around a single entry shared memory array which
-    holds an ``int`` value used as an index counter.
-
-    """
-    def __init__(
-        self,
-        shm: SharedMemory,
-    ) -> None:
-        self._shm = shm
-
-    @property
-    def value(self) -> int:
-        return int.from_bytes(self._shm.buf, byteorder)
-
-    @value.setter
-    def value(self, value) -> None:
-        self._shm.buf[:] = value.to_bytes(self._shm.size, byteorder)
-
-    def destroy(self) -> None:
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            name = self._shm.name
-            try:
-                shm_unlink(name)
-            except FileNotFoundError:
-                # might be a teardown race here?
-                log.warning(f'Shm for {name} already unlinked?')
-
-
-class _Token(Struct, frozen=True):
-    '''
-    Internal represenation of a shared memory "token"
-    which can be used to key a system wide post shm entry.
-
-    '''
-    shm_name: str  # this servers as a "key" value
-    shm_first_index_name: str
-    shm_last_index_name: str
-    dtype_descr: tuple
-    size: int  # in struct-array index / row terms
-
-    @property
-    def dtype(self) -> np.dtype:
-        return np.dtype(list(map(tuple, self.dtype_descr))).descr
-
-    def as_msg(self):
-        return self.to_dict()
-
-    @classmethod
-    def from_msg(cls, msg: dict) -> _Token:
-        if isinstance(msg, _Token):
-            return msg
-
-        # TODO: native struct decoding
-        # return _token_dec.decode(msg)
-
-        msg['dtype_descr'] = tuple(map(tuple, msg['dtype_descr']))
-        return _Token(**msg)
-
-
-# _token_dec = msgspec.msgpack.Decoder(_Token)
-
-# TODO: this api?
-# _known_tokens = tractor.ActorVar('_shm_tokens', {})
-# _known_tokens = tractor.ContextStack('_known_tokens', )
-# _known_tokens = trio.RunVar('shms', {})
-
-# process-local store of keys to tokens
-_known_tokens = {}
-
-
-def get_shm_token(key: str) -> _Token:
-    """Convenience func to check if a token
-    for the provided key is known by this process.
-    """
-    return _known_tokens.get(key)
 
 
 def _make_token(
     key: str,
     size: int,
-    dtype: Optional[np.dtype] = None,
-) -> _Token:
-    '''
-    Create a serializable token that can be used
-    to access a shared array.
-
-    '''
-    dtype = def_iohlcv_fields if dtype is None else dtype
-    return _Token(
-        shm_name=key,
-        shm_first_index_name=key + "_first",
-        shm_last_index_name=key + "_last",
-        dtype_descr=tuple(np.dtype(dtype).descr),
-        size=size,
-    )
-
-
-class ShmArray:
-    '''
-    A shared memory ``numpy`` (compatible) array API.
-
-    An underlying shared memory buffer is allocated based on
-    a user specified ``numpy.ndarray``. This fixed size array
-    can be read and written to by pushing data both onto the "front"
-    or "back" of a set index range. The indexes for the "first" and
-    "last" index are themselves stored in shared memory (accessed via
-    ``SharedInt`` interfaces) values such that multiple processes can
-    interact with the same array using a synchronized-index.
-
-    '''
-    def __init__(
-        self,
-        shmarr: np.ndarray,
-        first: SharedInt,
-        last: SharedInt,
-        shm: SharedMemory,
-        # readonly: bool = True,
-    ) -> None:
-        self._array = shmarr
-
-        # indexes for first and last indices corresponding
-        # to fille data
-        self._first = first
-        self._last = last
-
-        self._len = len(shmarr)
-        self._shm = shm
-        self._post_init: bool = False
-
-        # pushing data does not write the index (aka primary key)
-        dtype = shmarr.dtype
-        if dtype.fields:
-            self._write_fields = list(shmarr.dtype.fields.keys())[1:]
-        else:
-            self._write_fields = None
-
-    # TODO: ringbuf api?
-
-    @property
-    def _token(self) -> _Token:
-        return _Token(
-            shm_name=self._shm.name,
-            shm_first_index_name=self._first._shm.name,
-            shm_last_index_name=self._last._shm.name,
-            dtype_descr=tuple(self._array.dtype.descr),
-            size=self._len,
-        )
-
-    @property
-    def token(self) -> dict:
-        """Shared memory token that can be serialized and used by
-        another process to attach to this array.
-        """
-        return self._token.as_msg()
-
-    @property
-    def index(self) -> int:
-        return self._last.value % self._len
-
-    @property
-    def array(self) -> np.ndarray:
-        '''
-        Return an up-to-date ``np.ndarray`` view of the
-        so-far-written data to the underlying shm buffer.
-
-        '''
-        a = self._array[self._first.value:self._last.value]
-
-        # first, last = self._first.value, self._last.value
-        # a = self._array[first:last]
-
-        # TODO: eventually comment this once we've not seen it in the
-        # wild in a long time..
-        # XXX: race where first/last indexes cause a reader
-        # to load an empty array..
-        if len(a) == 0 and self._post_init:
-            raise RuntimeError('Empty array race condition hit!?')
-
-        return a
-
-    def ustruct(
-        self,
-        fields: Optional[list[str]] = None,
-
-        # type that all field values will be cast to
-        # in the returned view.
-        common_dtype: np.dtype = float,
-
-    ) -> np.ndarray:
-
-        array = self._array
-
-        if fields:
-            selection = array[fields]
-            # fcount = len(fields)
-        else:
-            selection = array
-            # fcount = len(array.dtype.fields)
-
-        # XXX: manual ``.view()`` attempt that also doesn't work.
-        # uview = selection.view(
-        #     dtype='<f16',
-        # ).reshape(-1, 4, order='A')
-
-        # assert len(selection) == len(uview)
-
-        u = rfn.structured_to_unstructured(
-            selection,
-            # dtype=float,
-            copy=True,
-        )
-
-        # unstruct = np.ndarray(u.shape, dtype=a.dtype, buffer=shm.buf)
-        # array[:] = a[:]
-        return u
-        # return ShmArray(
-        #     shmarr=u,
-        #     first=self._first,
-        #     last=self._last,
-        #     shm=self._shm
-        # )
-
-    def last(
-        self,
-        length: int = 1,
-
-    ) -> np.ndarray:
-        '''
-        Return the last ``length``'s worth of ("row") entries from the
-        array.
-
-        '''
-        return self.array[-length:]
-
-    def push(
-        self,
-        data: np.ndarray,
-
-        field_map: Optional[dict[str, str]] = None,
-        prepend: bool = False,
-        update_first: bool = True,
-        start: int | None = None,
-
-    ) -> int:
-        '''
-        Ring buffer like "push" to append data
-        into the buffer and return updated "last" index.
-
-        NB: no actual ring logic yet to give a "loop around" on overflow
-        condition, lel.
-
-        '''
-        length = len(data)
-
-        if prepend:
-            index = (start or self._first.value) - length
-
-            if index < 0:
-                raise ValueError(
-                    f'Array size of {self._len} was overrun during prepend.\n'
-                    f'You have passed {abs(index)} too many datums.'
-                )
-
-        else:
-            index = start if start is not None else self._last.value
-
-        end = index + length
-
-        if field_map:
-            src_names, dst_names = zip(*field_map.items())
-        else:
-            dst_names = src_names = self._write_fields
-
-        try:
-            self._array[
-                list(dst_names)
-            ][index:end] = data[list(src_names)][:]
-
-            # NOTE: there was a race here between updating
-            # the first and last indices and when the next reader
-            # tries to access ``.array`` (which due to the index
-            # overlap will be empty). Pretty sure we've fixed it now
-            # but leaving this here as a reminder.
-            if (
-                prepend
-                and update_first
-                and length
-            ):
-                assert index < self._first.value
-
-            if (
-                index < self._first.value
-                and update_first
-            ):
-                assert prepend, 'prepend=True not passed but index decreased?'
-                self._first.value = index
-
-            elif not prepend:
-                self._last.value = end
-
-            self._post_init = True
-            return end
-
-        except ValueError as err:
-            if field_map:
-                raise
-
-            # should raise if diff detected
-            self.diff_err_fields(data)
-            raise err
-
-    def diff_err_fields(
-        self,
-        data: np.ndarray,
-    ) -> None:
-        # reraise with any field discrepancy
-        our_fields, their_fields = (
-            set(self._array.dtype.fields),
-            set(data.dtype.fields),
-        )
-
-        only_in_ours = our_fields - their_fields
-        only_in_theirs = their_fields - our_fields
-
-        if only_in_ours:
-            raise TypeError(
-                f"Input array is missing field(s): {only_in_ours}"
-            )
-        elif only_in_theirs:
-            raise TypeError(
-                f"Input array has unknown field(s): {only_in_theirs}"
-            )
-
-    # TODO: support "silent" prepends that don't update ._first.value?
-    def prepend(
-        self,
-        data: np.ndarray,
-    ) -> int:
-        end = self.push(data, prepend=True)
-        assert end
-
-    def close(self) -> None:
-        self._first._shm.close()
-        self._last._shm.close()
-        self._shm.close()
-
-    def destroy(self) -> None:
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            shm_unlink(self._shm.name)
-
-        self._first.destroy()
-        self._last.destroy()
-
-    def flush(self) -> None:
-        # TODO: flush to storage backend like markestore?
-        ...
-
-
-def open_shm_array(
-    size: int,
-    key: str | None = None,
     dtype: np.dtype|None = None,
-    append_start_index: int | None = None,
+) -> NDToken:
-    readonly: bool = False,
+    '''
-
+    Wrap tractor's ``_make_token()`` with piker's
-) -> ShmArray:
+    default dtype fallback to ``def_iohlcv_fields``.
-    '''Open a memory shared ``numpy`` using the standard library.
-
-    This call unlinks (aka permanently destroys) the buffer on teardown
-    and thus should be used from the parent-most accessor (process).
 
     '''
-    # create new shared mem segment for which we
+    from ._source import def_iohlcv_fields
-    # have write permission
+    dtype = (
-    a = np.zeros(size, dtype=dtype)
+        def_iohlcv_fields
-    a['index'] = np.arange(len(a))
+        if dtype is None
-
+        else dtype
-    shm = SharedMemory(
-        name=key,
-        create=True,
-        size=a.nbytes
     )
-    array = np.ndarray(
+    return _tractor_make_token(
-        a.shape,
-        dtype=a.dtype,
-        buffer=shm.buf
-    )
-    array[:] = a[:]
-    array.setflags(write=int(not readonly))
-
-    token = _make_token(
         key=key,
         size=size,
         dtype=dtype,
     )
 
-    # create single entry arrays for storing an first and last indices
-    first = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_first_index_name,
-            create=True,
-            size=4,  # std int
-        )
-    )
-
-    last = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_last_index_name,
-            create=True,
-            size=4,  # std int
-        )
-    )
-
-    # start the "real-time" updated section after 3-days worth of 1s
-    # sampled OHLC. this allows appending up to a days worth from
-    # tick/quote feeds before having to flush to a (tsdb) storage
-    # backend, and looks something like,
-    # -------------------------
-    # |              |        i
-    # _________________________
-    # <-------------> <------->
-    #  history         real-time
-    #
-    # Once fully "prepended", the history section will leave the
-    # ``ShmArray._start.value: int = 0`` and the yet-to-be written
-    # real-time section will start at ``ShmArray.index: int``.
-
-    # this sets the index to nearly 2/3rds into the the length of
-    # the buffer leaving at least a "days worth of second samples"
-    # for the real-time section.
-    if append_start_index is None:
-        append_start_index = round(size * 0.616)
-
-    last.value = first.value = append_start_index
-
-    shmarr = ShmArray(
-        array,
-        first,
-        last,
-        shm,
-    )
-
-    assert shmarr._token == token
-    _known_tokens[key] = shmarr.token
-
-    # "unlink" created shm on process teardown by
-    # pushing teardown calls onto actor context stack
-    stack = tractor.current_actor(
-        err_on_no_runtime=False,
-    ).lifetime_stack
-    if stack:
-        stack.callback(shmarr.close)
-        stack.callback(shmarr.destroy)
-
-    return shmarr
-
-
-def attach_shm_array(
-    token: tuple[str, str, tuple[str, str]],
-    readonly: bool = True,
-
-) -> ShmArray:
-    '''
-    Attach to an existing shared memory array previously
-    created by another process using ``open_shared_array``.
-
-    No new shared mem is allocated but wrapper types for read/write
-    access are constructed.
-
-    '''
-    token = _Token.from_msg(token)
-    key = token.shm_name
-
-    if key in _known_tokens:
-        assert _Token.from_msg(_known_tokens[key]) == token, "WTF"
-
-    # XXX: ugh, looks like due to the ``shm_open()`` C api we can't
-    # actually place files in a subdir, see discussion here:
-    # https://stackoverflow.com/a/11103289
-
-    # attach to array buffer and view as per dtype
-    _err: Optional[Exception] = None
-    for _ in range(3):
-        try:
-            shm = SharedMemory(
-                name=key,
-                create=False,
-            )
-            break
-        except OSError as oserr:
-            _err = oserr
-            time.sleep(0.1)
-    else:
-        if _err:
-            raise _err
-
-    shmarr = np.ndarray(
-        (token.size,),
-        dtype=token.dtype,
-        buffer=shm.buf
-    )
-    shmarr.setflags(write=int(not readonly))
-
-    first = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_first_index_name,
-            create=False,
-            size=4,  # std int
-        ),
-    )
-    last = SharedInt(
-        shm=SharedMemory(
-            name=token.shm_last_index_name,
-            create=False,
-            size=4,  # std int
-        ),
-    )
-
-    # make sure we can read
-    first.value
-
-    sha = ShmArray(
-        shmarr,
-        first,
-        last,
-        shm,
-    )
-    # read test
-    sha.array
-
-    # Stash key -> token knowledge for future queries
-    # via `maybe_opepn_shm_array()` but only after we know
-    # we can attach.
-    if key not in _known_tokens:
-        _known_tokens[key] = token
-
-    # "close" attached shm on actor teardown
-    if (actor := tractor.current_actor(
-        err_on_no_runtime=False,
-    )):
-        actor.lifetime_stack.callback(sha.close)
-
-    return sha
-
 
 def maybe_open_shm_array(
     key: str,
@@ -625,37 +70,37 @@ def maybe_open_shm_array(
     append_start_index: int|None = None,
     readonly: bool = False,
     **kwargs,
-
 ) -> tuple[ShmArray, bool]:
     '''
-    Attempt to attach to a shared memory block using a "key" lookup
+    Attempt to attach to a shared memory block
-    to registered blocks in the users overall "system" registry
+    using a "key" lookup to registered blocks in
-    (presumes you don't have the block's explicit token).
+    the user's overall "system" registry (presumes
+    you don't have the block's explicit token).
 
-    This function is meant to solve the problem of discovering whether
+    This is a thin wrapper around tractor's
-    a shared array token has been allocated or discovered by the actor
+    ``maybe_open_shm_ndarray()`` preserving piker's
-    running in **this** process. Systems where multiple actors may seek
+    historical defaults (``readonly=False``,
-    to access a common block can use this function to attempt to acquire
+    ``append_start_index=None``).
-    a token as discovered by the actors who have previously stored
-    a "key" -> ``_Token`` map in an actor local (aka python global)
-    variable.
 
-    If you know the explicit ``_Token`` for your memory segment instead
+    If you know the explicit ``NDToken`` for your
-    use ``attach_shm_array``.
+    memory segment instead use
+    ``tractor.ipc._shm.attach_shm_ndarray()``.
 
     '''
     try:
         # see if we already know this key
         token = _known_tokens[key]
         return (
-            attach_shm_array(
+            attach_shm_ndarray(
                 token=token,
                 readonly=readonly,
             ),
             False,
         )
     except KeyError:
-        log.debug(f"Could not find {key} in shms cache")
+        log.debug(
+            f'Could not find {key} in shms cache'
+        )
         if dtype:
             token = _make_token(
                 key,
@@ -663,9 +108,18 @@ def maybe_open_shm_array(
                 dtype=dtype,
             )
             try:
-                return attach_shm_array(token=token, **kwargs), False
+                return (
+                    attach_shm_ndarray(
+                        token=token,
+                        **kwargs,
+                    ),
+                    False,
+                )
             except FileNotFoundError:
-                log.debug(f"Could not attach to shm with token {token}")
+                log.debug(
+                    f'Could not attach to shm'
+                    f' with token {token}'
+                )
 
         # This actor does not know about memory
         # associated with the provided "key".
@@ -673,7 +127,7 @@ def maybe_open_shm_array(
         # to fail if a block has been allocated
         # on the OS by someone else.
         return (
-            open_shm_array(
+            open_shm_ndarray(
                 key=key,
                 size=size,
                 dtype=dtype,
@@ -683,18 +137,20 @@ def maybe_open_shm_array(
             True,
         )
 
+
 def try_read(
-    array: np.ndarray
+    array: np.ndarray,
-
+) -> np.ndarray|None:
-) -> Optional[np.ndarray]:
     '''
-    Try to read the last row from a shared mem array or ``None``
+    Try to read the last row from a shared mem
-    if the array read returns a zero-length array result.
+    array or ``None`` if the array read returns
+    a zero-length array result.
 
-    Can be used to check for backfilling race conditions where an array
+    Can be used to check for backfilling race
-    is currently being (re-)written by a writer actor but the reader is
+    conditions where an array is currently being
-    unaware and reads during the window where the first and last indexes
+    (re-)written by a writer actor but the reader
-    are being updated.
+    is unaware and reads during the window where
+    the first and last indexes are being updated.
 
     '''
     try:
@@ -702,14 +158,13 @@ def try_read(
     except IndexError:
         # XXX: race condition with backfilling shm.
         #
-        # the underlying issue is that a backfill (aka prepend) and subsequent
+        # the underlying issue is that a backfill
-        # shm array first/last index update could result in an empty array
+        # (aka prepend) and subsequent shm array
-        # read here since the indices may be updated in such a way that
+        # first/last index update could result in an
-        # a read delivers an empty array (though it seems like we
+        # empty array read here since the indices may
-        # *should* be able to prevent that?). also, as and alt and
+        # be updated in such a way that a read delivers
-        # something we need anyway, maybe there should be some kind of
+        # an empty array (though it seems like we
-        # signal that a prepend is taking place and this consumer can
+        # *should* be able to prevent that?).
-        # respond (eg. redrawing graphics) accordingly.
 
-        # the array read was emtpy
+        # the array read was empty
         return None

piker/data/feed.py

@@ -973,9 +973,6 @@ async def open_feed(
                     # assert flume.mkt.fqme == fqme
                     feed.flumes[fqme] = flume
 
-                    # TODO: do we need this?
-                    flume.feed = feed
-
                     # attach and cache shm handles
                     rt_shm = flume.rt_shm
                     assert rt_shm

piker/data/flows.py

@@ -22,25 +22,19 @@ real-time data processing data-structures.
 
 """
 from __future__ import annotations
-from typing import (
-    TYPE_CHECKING,
-)
 
 import tractor
 import pendulum
 import numpy as np
 
 from piker.types import Struct
-from ._sharedmem import (
+from tractor.ipc._shm import (
-    attach_shm_array,
     ShmArray,
-    _Token,
+    NDToken,
+    attach_shm_ndarray,
 )
 from piker.accounting import MktPair
 
-if TYPE_CHECKING:
-    from piker.data.feed import Feed
-
 
 class Flume(Struct):
     '''
@@ -64,11 +58,11 @@ class Flume(Struct):
     '''
     mkt: MktPair
     first_quote: dict
-    _rt_shm_token: _Token
+    _rt_shm_token: NDToken
 
     # optional since some data flows won't have a "downsampled" history
     # buffer/stream (eg. FSPs).
-    _hist_shm_token: _Token | None = None
+    _hist_shm_token: NDToken|None = None
 
     # private shm refs loaded dynamically from tokens
     _hist_shm: ShmArray | None = None
@@ -80,15 +74,11 @@ class Flume(Struct):
     izero_rt: int = 0
     throttle_rate: int | None = None
 
-    # TODO: do we need this really if we can pull the `Portal` from
-    # ``tractor``'s internals?
-    feed: Feed|None = None
-
     @property
     def rt_shm(self) -> ShmArray:
 
         if self._rt_shm is None:
-            self._rt_shm = attach_shm_array(
+            self._rt_shm = attach_shm_ndarray(
                 token=self._rt_shm_token,
                 readonly=self._readonly,
             )
@@ -104,7 +94,7 @@ class Flume(Struct):
             )
 
         if self._hist_shm is None:
-            self._hist_shm = attach_shm_array(
+            self._hist_shm = attach_shm_ndarray(
                 token=self._hist_shm_token,
                 readonly=self._readonly,
             )
@@ -156,7 +146,6 @@ class Flume(Struct):
         # will get instead some kind of msg-compat version
         # that it can load.
         msg.pop('stream')
-        msg.pop('feed')
         msg.pop('_rt_shm')
         msg.pop('_hist_shm')
 

piker/fsp/_api.py

@@ -37,12 +37,12 @@ import numpy as np
 import tractor
 from tractor.msg import NamespacePath
 
-from ..data._sharedmem import (
+from tractor.ipc._shm import (
     ShmArray,
-    maybe_open_shm_array,
+    NDToken,
-    attach_shm_array,
+    attach_shm_ndarray,
-    _Token,
 )
+from ..data._sharedmem import maybe_open_shm_array
 from ..log import get_logger
 
 log = get_logger(__name__)
@@ -78,8 +78,8 @@ class Fsp:
     # + the consuming fsp *to* the consumers output
     # shm flow.
     _flow_registry: dict[
-        tuple[_Token, str],
+        tuple[NDToken, str],
-        tuple[_Token, Optional[ShmArray]],
+        tuple[NDToken, Optional[ShmArray]],
     ] = {}
 
     def __init__(
@@ -148,7 +148,7 @@ class Fsp:
                 # times as possible as per:
                 # - https://github.com/pikers/piker/issues/359
                 # - https://github.com/pikers/piker/issues/332
-                maybe_array := attach_shm_array(dst_token)
+                maybe_array := attach_shm_ndarray(dst_token)
             )
 
         return maybe_array

piker/fsp/_engine.py

@@ -40,7 +40,7 @@ from ..log import (
 )
 from .. import data
 from ..data.flows import Flume
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ..data._sampling import (
     _default_delay_s,
     open_sample_stream,
@@ -49,7 +49,7 @@ from ..accounting import MktPair
 from ._api import (
     Fsp,
     _load_builtins,
-    _Token,
+    NDToken,
 )
 from ..toolz import Profiler
 
@@ -414,7 +414,7 @@ async def cascade(
     dst_flume_addr: dict,
     ns_path: NamespacePath,
 
-    shm_registry: dict[str, _Token],
+    shm_registry: dict[str, NDToken],
 
     zero_on_step: bool = False,
     loglevel: str|None = None,
@@ -465,9 +465,9 @@ async def cascade(
     # not sure how else to do it.
     for (token, fsp_name, dst_token) in shm_registry:
         Fsp._flow_registry[(
-            _Token.from_msg(token),
+            NDToken.from_msg(token),
             fsp_name,
-        )] = _Token.from_msg(dst_token), None
+        )] = NDToken.from_msg(dst_token), None
 
     fsp: Fsp = reg.get(
         NamespacePath(ns_path)

piker/fsp/_momo.py

@@ -25,7 +25,7 @@ from numba import jit, float64, optional, int64
 
 from ._api import fsp
 from ..data import iterticks
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 
 
 @jit(

piker/fsp/_volume.py

@@ -21,7 +21,7 @@ from tractor.trionics._broadcast import AsyncReceiver
 
 from ._api import fsp
 from ..data import iterticks
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ._momo import _wma
 from ..log import get_logger
 

piker/storage/cli.py

@@ -37,9 +37,7 @@ import typer
 
 from piker.service import open_piker_runtime
 from piker.cli import cli
-from piker.data import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from piker import tsp
 from . import log
 from . import (

piker/storage/nativedb.py

@@ -64,10 +64,8 @@ from pendulum import (
 
 from piker import config
 from piker import tsp
-from piker.data import (
+from tractor.ipc._shm import ShmArray
-    def_iohlcv_fields,
+from piker.data import def_iohlcv_fields
-    ShmArray,
-)
 from piker.log import get_logger
 from . import TimeseriesNotFound
 

piker/tsp/_anal.py

@@ -276,14 +276,41 @@ def get_null_segs(
         absi_zdiff: np.ndarray = np.diff(absi_zeros)
 
         if zero_t.size < 2:
-            try:
+            idx: int = zero_t['index'][0]
-                breakpoint()
+            idx_before: int = idx - 1
-            except RuntimeError:
+            idx_after: int = idx + 1
-                # XXX, if greenback not active from
+            index = frame['index']
-                # piker store ldshm cmd..
+            before_cond = idx_before <= index
-                log.exception(
+            after_cond = index <= idx_after
-                    "Can't debug single-sample null!\n"
+            bars: np.ndarray = frame[
+                before_cond
+                &
+                after_cond
+            ]
+            time: np.ndarray = bars['time']
+            from pendulum import (
+                from_timestamp,
+                Interval,
             )
+            gap: Interval = (
+                from_timestamp(time[-1])
+                -
+                from_timestamp(time[0])
+            )
+            log.warning(
+                f'Single OHLCV-bar null-segment detected??\n'
+                f'gap -> {gap}\n'
+            )
+
+            # ^^XXX, if you want to debug the above bar-gap^^
+            # try:
+            #     breakpoint()
+            # except RuntimeError:
+            #     # XXX, if greenback not active from
+            #     # piker store ldshm cmd..
+            #     log.exception(
+            #         "Can't debug single-sample null!\n"
+            #     )
 
             return None
 

piker/tsp/_history.py

@@ -59,11 +59,12 @@ from piker.brokers import NoData
 from piker.accounting import (
     MktPair,
 )
-from piker.log import get_logger
+from piker.log import (
-from ..data._sharedmem import (
+    get_logger,
-    maybe_open_shm_array,
+    get_console_log,
-    ShmArray,
 )
+from tractor.ipc._shm import ShmArray
+from ..data._sharedmem import maybe_open_shm_array
 from piker.data._source import (
     def_iohlcv_fields,
 )
@@ -737,12 +738,21 @@ async def start_backfill(
                 # including the dst[/src] source asset token. SO,
                 # 'tsla.nasdaq.ib' over 'tsla/usd.nasdaq.ib' for
                 # historical reasons ONLY.
-                if mkt.dst.atype not in {
+                if (
+                    mkt.dst.atype not in {
                         'crypto',
                         'crypto_currency',
                         'fiat',  # a "forex pair"
                         'perpetual_future',  # stupid "perps" from cex land
-                }:
+                    }
+                    and not (
+                        mkt.src.atype == 'crypto_currency'
+                        and
+                        mkt.dst.atype in {
+                            'future',
+                        }
+                    )
+                ):
                     col_sym_key: str = mkt.get_fqme(
                         delim_char='',
                         without_src=True,
@@ -1386,6 +1396,10 @@ async def manage_history(
         engages.
 
     '''
+    get_console_log(
+        name=__name__,
+        level=loglevel,
+    )
     # TODO: is there a way to make each shm file key
     # actor-tree-discovery-addr unique so we avoid collisions
     # when doing tests which also allocate shms for certain instruments

piker/ui/_chart.py

@@ -49,7 +49,7 @@ from ._cursor import (
     Cursor,
     ContentsLabel,
 )
-from ..data._sharedmem import ShmArray
+from tractor.ipc._shm import ShmArray
 from ._ohlc import BarItems
 from ._curve import (
     Curve,

piker/ui/_dataviz.py

@@ -42,9 +42,7 @@ from numpy import (
 import pyqtgraph as pg
 
 from piker.ui.qt import QLineF
-from ..data._sharedmem import (
+from tractor.ipc._shm import ShmArray
-    ShmArray,
-)
 from ..data.flows import Flume
 from ..data._formatters import (
     IncrementalFormatter,

piker/ui/_editors.py

@@ -168,7 +168,7 @@ class ArrowEditor(Struct):
         '''
         uid: str = arrow._uid
         arrows: list[pg.ArrowItem] = self._arrows[uid]
-        log.info(
+        log.debug(
             f'Removing arrow from views\n'
             f'uid: {uid!r}\n'
             f'{arrow!r}\n'
@@ -286,7 +286,9 @@ class LineEditor(Struct):
             for line in lines:
                 line.show_labels()
                 line.hide_markers()
-                log.debug(f'Level active for level: {line.value()}')
+                log.debug(
+                    f'Line active @ level: {line.value()!r}'
+                )
                 # TODO: other flashy things to indicate the order is active
 
         return lines
@@ -329,7 +331,11 @@ class LineEditor(Struct):
                     if line in hovered:
                         hovered.remove(line)
 
-                    log.debug(f'deleting {line} with oid: {uuid}')
+                    log.debug(
+                        f'Deleting level-line\n'
+                        f'line: {line!r}\n'
+                        f'oid: {uuid!r}\n'
+                    )
                     line.delete()
 
                     # make sure the xhair doesn't get left off
@@ -337,7 +343,11 @@ class LineEditor(Struct):
                     cursor.show_xhair()
 
         else:
-            log.warning(f'Could not find line for {line}')
+            log.warning(
+                f'Could not find line for removal ??\n'
+                f'\n'
+                f'{line!r}\n'
+            )
 
         return lines
 
@@ -569,11 +579,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
         if update_label:
             self.init_label(view_rect)
 
-        print(
+        log.debug(
-            'SelectRect modify:\n'
+            f'SelectRect modify,\n'
             f'QRectF: {view_rect}\n'
-            f'start_pos: {start_pos}\n'
+            f'start_pos: {start_pos!r}\n'
-            f'end_pos: {end_pos}\n'
+            f'end_pos: {end_pos!r}\n'
         )
         self.show()
 
@@ -640,8 +650,11 @@ class SelectRect(QtWidgets.QGraphicsRectItem):
             dmn=dmn,
         ))
 
-        # print(f'x2, y2: {(x2, y2)}')
+        # tracing
-        # print(f'xmn, ymn: {(xmn, ymx)}')
+        # log.info(
+        #     f'x2, y2: {(x2, y2)}\n'
+        #     f'xmn, ymn: {(xmn, ymx)}\n'
+        # )
 
         label_anchor = Point(
             xmx + 2,

piker/ui/_fsp.py

@@ -44,14 +44,12 @@ from piker.fsp import (
     dolla_vlm,
     flow_rates,
 )
-from piker.data import (
+from tractor.ipc._shm import (
-    Flume,
     ShmArray,
+    NDToken,
 )
-from piker.data._sharedmem import (
+from piker.data import Flume
-    _Token,
+from piker.data._sharedmem import try_read
-    try_read,
-)
 from piker.log import get_logger
 from piker.toolz import Profiler
 from piker.types import Struct
@@ -382,7 +380,7 @@ class FspAdmin:
             tuple,
             tuple[tractor.MsgStream, ShmArray]
         ] = {}
-        self._flow_registry: dict[_Token, str] = {}
+        self._flow_registry: dict[NDToken, str] = {}
 
         # TODO: make this a `.src_flume` and add
         # a `dst_flume`?

piker/ui/_lines.py

@@ -38,7 +38,6 @@ from piker.ui.qt import (
     QtGui,
     QGraphicsPathItem,
     QStyleOptionGraphicsItem,
-    QGraphicsItem,
     QGraphicsScene,
     QWidget,
     QPointF,

pyproject.toml

@@ -106,7 +106,7 @@ default-groups = [
 
 [dependency-groups]
 uis = [
-  "pyqtgraph",
+  "pyqtgraph >= 0.14.0",
   "qdarkstyle >=3.0.2, <4.0.0",
   "pyqt6 >=6.7.0, <7.0.0",
 
@@ -193,9 +193,12 @@ include = ["piker"]
 
 
 [tool.uv.sources]
-# pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
 tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
 pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
+# pyqtgraph = { git = "https://github.com/pyqtgraph/pyqtgraph.git", branch = 'master' }
+# pyqtgraph = { path = '../pyqtgraph', editable = true }
+# ?TODO, resync our fork?
+# pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
 
 # to get fancy next-cmd/suggestion feats prior to 0.22.2 B)
 # https://github.com/xonsh/xonsh/pull/6037

uv.lock

@@ -1055,7 +1055,7 @@ dev = [
     { name = "prompt-toolkit", specifier = "==3.0.40" },
     { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
-    { name = "pyqtgraph" },
+    { name = "pyqtgraph", specifier = ">=0.14.0" },
     { name = "pytest" },
     { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
     { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
@@ -1073,7 +1073,7 @@ repl = [
 testing = [{ name = "pytest" }]
 uis = [
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
-    { name = "pyqtgraph" },
+    { name = "pyqtgraph", specifier = ">=0.14.0" },
     { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
     { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
 ]