piker/data/_ahab.py

@@ -19,7 +19,6 @@ Supervisor for docker with included specific-image service helpers.
 
 '''
 import os
-import time
 from typing import (
     Optional,
     Callable,
@@ -187,65 +186,45 @@ class Container:
 
     async def cancel(
         self,
-        stop_msg: str,
     ) -> None:
 
         cid = self.cntr.id
-        # first try a graceful cancel
-        log.cancel(
-            f'SIGINT cancelling container: {cid}\n'
-            f'waiting on stop msg: "{stop_msg}"'
-        )
         self.try_signal('SIGINT')
 
-        start = time.time()
-        for _ in range(30):
-
         with trio.move_on_after(0.5) as cs:
             cs.shield = True
-                await self.process_logs_until(stop_msg)
+            await self.process_logs_until('initiating graceful shutdown')
+            await self.process_logs_until('exiting...',)
 
-                # if we aren't cancelled on above checkpoint then we
+        for _ in range(10):
-                # assume we read the expected stop msg and terminated.
+            with trio.move_on_after(0.5) as cs:
+                cs.shield = True
+                await self.process_logs_until('exiting...',)
                 break
 
-            try:
+            if cs.cancelled_caught:
-                log.info(f'Polling for container shutdown:\n{cid}')
-
-                if self.cntr.status not in {'exited', 'not-running'}:
-                    self.cntr.wait(
-                        timeout=0.1,
-                        condition='not-running',
-                    )
-
-                break
-
-            except (
-                ReadTimeout,
-            ):
-                log.info(f'Still waiting on container:\n{cid}')
-                continue
-
-            except (
-                docker.errors.APIError,
-                ConnectionError,
-            ):
-                log.exception('Docker connection failure')
-                break
-        else:
-            delay = time.time() - start
-            log.error(
-                f'Failed to kill container {cid} after {delay}s\n'
-                'sending SIGKILL..'
-            )
                 # get out the big guns, bc apparently marketstore
                 # doesn't actually know how to terminate gracefully
                 # :eyeroll:...
                 self.try_signal('SIGKILL')
+
+                try:
+                    log.info('Waiting on container shutdown: {cid}')
                     self.cntr.wait(
-                timeout=3,
+                        timeout=0.1,
                         condition='not-running',
                     )
+                    break
+
+                except (
+                    ReadTimeout,
+                    ConnectionError,
+                ):
+                    log.error(f'failed to wait on container {cid}')
+                    raise
+
+        else:
+            raise RuntimeError('Failed to cancel container {cid}')
 
         log.cancel(f'Container stopped: {cid}')
 
@@ -266,16 +245,13 @@ async def open_ahabd(
         # params, etc. passing to ``Containter.run()``?
         # call into endpoint for container config/init
         ep_func = NamespacePath(endpoint).load_ref()
-        (
+        dcntr, cntr_config = ep_func(client)
-            dcntr,
-            cntr_config,
-            start_msg,
-            stop_msg,
-        ) = ep_func(client)
         cntr = Container(dcntr)
 
         with trio.move_on_after(1):
-            found = await cntr.process_logs_until(start_msg)
+            found = await cntr.process_logs_until(
+                "launching tcp listener for all services...",
+            )
 
             if not found and cntr not in client.containers.list():
                 raise RuntimeError(
@@ -295,9 +271,16 @@ async def open_ahabd(
             # callers to have root perms?
             await trio.sleep_forever()
 
-        finally:
+        except (
+            BaseException,
+            # trio.Cancelled,
+            # KeyboardInterrupt,
+        ):
+
             with trio.CancelScope(shield=True):
-                await cntr.cancel(stop_msg)
+                await cntr.cancel()
+
+            raise
 
 
 async def start_ahab(

piker/data/marketstore.py

@@ -127,15 +127,10 @@ def start_marketstore(
     import os
     import docker
     from .. import config
+
     get_console_log('info', name=__name__)
 
-    mktsdir = os.path.join(config._config_dir, 'marketstore')
+    yml_file = os.path.join(config._config_dir, 'mkts.yml')
-
-    # create when dne
-    if not os.path.isdir(mktsdir):
-        os.mkdir(mktsdir)
-
-    yml_file = os.path.join(mktsdir, 'mkts.yml')
     if not os.path.isfile(yml_file):
         log.warning(
             f'No `marketstore` config exists?: {yml_file}\n'
@@ -148,14 +143,14 @@ def start_marketstore(
     # create a mount from user's local piker config dir into container
     config_dir_mnt = docker.types.Mount(
         target='/etc',
-        source=mktsdir,
+        source=config._config_dir,
         type='bind',
     )
 
     # create a user config subdir where the marketstore
     # backing filesystem database can be persisted.
     persistent_data_dir = os.path.join(
-        mktsdir, 'data',
+        config._config_dir, 'data',
     )
     if not os.path.isdir(persistent_data_dir):
         os.mkdir(persistent_data_dir)
@@ -185,14 +180,7 @@ def start_marketstore(
         init=True,
         # remove=True,
     )
-    return (
+    return dcntr, _config
-        dcntr,
-        _config,
-
-        # expected startup and stop msgs
-        "launching tcp listener for all services...",
-        "exiting...",
-    )
 
 
 _tick_tbk_ids: tuple[str, str] = ('1Sec', 'TICK')
@@ -395,12 +383,7 @@ class Storage:
     ]:
 
         first_tsdb_dt, last_tsdb_dt = None, None
-        tsdb_arrays = await self.read_ohlcv(
+        tsdb_arrays = await self.read_ohlcv(fqsn)
-            fqsn,
-            # on first load we don't need to pull the max
-            # history per request size worth.
-            limit=3000,
-        )
         log.info(f'Loaded tsdb history {tsdb_arrays}')
 
         if tsdb_arrays:
@@ -418,7 +401,6 @@ class Storage:
         fqsn: str,
         timeframe: Optional[Union[int, str]] = None,
         end: Optional[int] = None,
-        limit: int = int(800e3),
 
     ) -> tuple[
         MarketstoreClient,
@@ -441,7 +423,7 @@ class Storage:
 
             # TODO: figure the max limit here given the
             # ``purepc`` msg size limit of purerpc: 33554432
-            limit=limit,
+            limit=int(800e3),
         )
 
         if timeframe is None:

setup.py

@@ -57,7 +57,6 @@ setup(
         # from github currently (see requirements.txt)
         # 'trimeter',  # not released yet..
         # 'tractor',
-        # asyncvnc,
 
         # brokers
         'asks==2.4.8',
@@ -72,34 +71,32 @@ setup(
 
         # UI
         'PyQt5',
-        # 'pyqtgraph',  from our fork see reqs.txt
+        'pyqtgraph',
-        'qdarkstyle >= 3.0.2',  # themeing
+        'qdarkstyle >= 3.0.2',
-        'fuzzywuzzy[speedup]',  # fuzzy search
+        # fuzzy search
+        'fuzzywuzzy[speedup]',
 
         # tsdbs
-        # anyio-marketstore  # from gh see reqs.txt
+        'pymarketstore',
     ],
     extras_require={
+
+        # serialization
         'tsdb': [
             'docker',
         ],
 
     },
     tests_require=['pytest'],
-    python_requires=">=3.10",
+    python_requires=">=3.9",  # literally for ``datetime.datetime.fromisoformat``...
-    keywords=[
+    keywords=["async", "trading", "finance", "quant", "charting"],
-        "async",
-        "trading",
-        "finance",
-        "quant",
-        "charting",
-    ],
     classifiers=[
         'Development Status :: 3 - Alpha',
         'License :: OSI Approved :: ',
         'Operating System :: POSIX :: Linux',
         "Programming Language :: Python :: Implementation :: CPython",
         "Programming Language :: Python :: 3 :: Only",
+        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         'Intended Audience :: Financial and Insurance Industry',
         'Intended Audience :: Science/Research',