Compare commits

...

7 Commits

Author SHA1 Message Date
Tyler Goodlet ac0f43dc98 Go Python 3.10+ in anticipation of upcoming feature PRs 2022-06-28 10:02:09 -04:00
goodboy 3977f1cc7e
Merge pull request #341 from pikers/contain_mkts
Contain mkts
2022-06-26 13:49:23 -04:00
Tyler Goodlet e45cb9d08a Always cancel container on teardown 2022-06-26 13:36:29 -04:00
Tyler Goodlet 27c523ca74 Speedup: only load a "views worth" of datums on first query 2022-06-23 15:21:09 -04:00
Tyler Goodlet b8b76a32a6 Harden container cancel-and-wait supervisor loop
This should hopefully make teardown more reliable and includes better
logic to fail over to a hard kill path after a 3 second timeout waiting
for the instance to complete using the `docker-py` wait API. Also
generalize the supervisor teardown loop by allowing the container config
endpoint to return 2 msgs to expect:
- a startup message that can be read from the container's internal
  process logging that indicates it is fully up and ready.
- a teardown msg that can be polled for that indicates the container has
  gracefully terminated after a cancellation request which is passed to
  our container wrappers `.cancel()` method.

Make the marketstore config endpoint return the 2 messages we previously
had hard coded and use this new api.
2022-06-23 10:23:14 -04:00
Tyler Goodlet dcee0ddd55 Move/expect all marketstore configs under a `<configdir>/piker/marketstore` subdir 2022-06-23 09:48:32 -04:00
goodboy 67eab85f06
Merge pull request #340 from pikers/slic_fix_v2
Slice fix v2
2022-06-22 19:55:39 -04:00
3 changed files with 89 additions and 51 deletions

View File

@ -19,6 +19,7 @@ Supervisor for docker with included specific-image service helpers.
'''
import os
import time
from typing import (
Optional,
Callable,
@ -186,45 +187,65 @@ 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')
with trio.move_on_after(0.5) as cs:
cs.shield = True
await self.process_logs_until('initiating graceful shutdown')
await self.process_logs_until('exiting...',)
start = time.time()
for _ in range(30):
for _ in range(10):
with trio.move_on_after(0.5) as cs:
cs.shield = True
await self.process_logs_until('exiting...',)
await self.process_logs_until(stop_msg)
# if we aren't cancelled on above checkpoint then we
# assume we read the expected stop msg and terminated.
break
if cs.cancelled_caught:
# 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}')
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.error(f'failed to wait on container {cid}')
raise
log.exception('Docker connection failure')
break
else:
raise RuntimeError('Failed to cancel container {cid}')
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')
self.cntr.wait(
timeout=3,
condition='not-running',
)
log.cancel(f'Container stopped: {cid}')
@ -245,13 +266,16 @@ 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(
"launching tcp listener for all services...",
)
found = await cntr.process_logs_until(start_msg)
if not found and cntr not in client.containers.list():
raise RuntimeError(
@ -271,16 +295,9 @@ async def open_ahabd(
# callers to have root perms?
await trio.sleep_forever()
except (
BaseException,
# trio.Cancelled,
# KeyboardInterrupt,
):
finally:
with trio.CancelScope(shield=True):
await cntr.cancel()
raise
await cntr.cancel(stop_msg)
async def start_ahab(

View File

@ -127,10 +127,15 @@ def start_marketstore(
import os
import docker
from .. import config
get_console_log('info', name=__name__)
yml_file = os.path.join(config._config_dir, 'mkts.yml')
mktsdir = os.path.join(config._config_dir, 'marketstore')
# 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'
@ -143,14 +148,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=config._config_dir,
source=mktsdir,
type='bind',
)
# create a user config subdir where the marketstore
# backing filesystem database can be persisted.
persistent_data_dir = os.path.join(
config._config_dir, 'data',
mktsdir, 'data',
)
if not os.path.isdir(persistent_data_dir):
os.mkdir(persistent_data_dir)
@ -180,7 +185,14 @@ def start_marketstore(
init=True,
# remove=True,
)
return dcntr, _config
return (
dcntr,
_config,
# expected startup and stop msgs
"launching tcp listener for all services...",
"exiting...",
)
_tick_tbk_ids: tuple[str, str] = ('1Sec', 'TICK')
@ -383,7 +395,12 @@ class Storage:
]:
first_tsdb_dt, last_tsdb_dt = None, None
tsdb_arrays = await self.read_ohlcv(fqsn)
tsdb_arrays = await self.read_ohlcv(
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:
@ -401,6 +418,7 @@ class Storage:
fqsn: str,
timeframe: Optional[Union[int, str]] = None,
end: Optional[int] = None,
limit: int = int(800e3),
) -> tuple[
MarketstoreClient,
@ -423,7 +441,7 @@ class Storage:
# TODO: figure the max limit here given the
# ``purepc`` msg size limit of purerpc: 33554432
limit=int(800e3),
limit=limit,
)
if timeframe is None:

View File

@ -57,6 +57,7 @@ setup(
# from github currently (see requirements.txt)
# 'trimeter', # not released yet..
# 'tractor',
# asyncvnc,
# brokers
'asks==2.4.8',
@ -71,32 +72,34 @@ setup(
# UI
'PyQt5',
'pyqtgraph',
'qdarkstyle >= 3.0.2',
# fuzzy search
'fuzzywuzzy[speedup]',
# 'pyqtgraph', from our fork see reqs.txt
'qdarkstyle >= 3.0.2', # themeing
'fuzzywuzzy[speedup]', # fuzzy search
# tsdbs
'pymarketstore',
# anyio-marketstore # from gh see reqs.txt
],
extras_require={
# serialization
'tsdb': [
'docker',
],
},
tests_require=['pytest'],
python_requires=">=3.9", # literally for ``datetime.datetime.fromisoformat``...
keywords=["async", "trading", "finance", "quant", "charting"],
python_requires=">=3.10",
keywords=[
"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',