playground lmdb timeseries

Author: bozokopic

playground/lmdb/timeseries/create.py

@@ -0,0 +1,63 @@
+from pathlib import Path
+import asyncio
+import contextlib
+import sys
+
+from hat import aio
+
+from hat.event import common
+import hat.event.backends.lmdb
+
+
+def main():
+    with contextlib.suppress(asyncio.CancelledError):
+        aio.run_asyncio(async_main())
+
+
+async def async_main():
+    db_path = Path(sys.argv[1])
+
+    server_id = 1
+    group_id = 0
+    point_count = 2000
+    interval = 5
+    retention = 24 * 60 * 60
+
+    conf = {'db_path': str(db_path),
+            'identifier': None,
+            'flush_period': 1,
+            'cleanup_period': 1,
+            'conditions': [],
+            'latest': {'subscriptions': []},
+            'timeseries': [{'order_by': 'SOURCE_TIMESTAMP',
+                            'subscriptions': [['eds', 'timeseries',
+                                               str(group_id), '?']],
+                            'limit': {'duration': retention}}]}
+
+    backend = await aio.call(hat.event.backends.lmdb.info.create, conf, None,
+                             None)
+
+    try:
+        t = 0
+        session_id = 0
+
+        while t < retention:
+            session_id += 1
+            events = [common.Event(id=common.EventId(server_id, session_id,
+                                                     point_id + 1),
+                                   type=('eds', 'timeseries', str(group_id),
+                                         str(point_id)),
+                                   timestamp=common.Timestamp(t, 0),
+                                   source_timestamp=common.Timestamp(t, 0),
+                                   payload=None)
+                      for point_id in range(point_count)]
+            await backend.register(events)
+
+            t += interval
+
+    finally:
+        await aio.uncancellable(backend.async_close())
+
+
+if __name__ == '__main__':
+    main()

playground/lmdb/timeseries/create.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+
+set -e
+
+RUN_PATH=$(dirname "$(realpath "$0")")
+PLAYGROUND_PATH=$RUN_PATH/../..
+. $PLAYGROUND_PATH/env.sh
+
+
+db_path=$DATA_PATH/lmdb_timeseries.db
+
+rm -f $db_path
+exec $PYTHON $RUN_PATH/create.py $db_path

playground/lmdb/timeseries/query.py

@@ -0,0 +1,60 @@
+from pathlib import Path
+import asyncio
+import contextlib
+import cProfile
+import sys
+import time
+
+from hat import aio
+
+from hat.event import common
+import hat.event.backends.lmdb
+
+
+def main():
+    with contextlib.suppress(asyncio.CancelledError):
+        aio.run_asyncio(async_main())
+
+
+async def async_main():
+    db_path = Path(sys.argv[1])
+
+    group_id = 0
+    max_results = 10_000
+
+    conf = {'db_path': str(db_path),
+            'identifier': None,
+            'flush_period': 1,
+            'cleanup_period': 1,
+            'conditions': [],
+            'latest': {'subscriptions': []},
+            'timeseries': [{'order_by': 'SOURCE_TIMESTAMP',
+                            'subscriptions': [['eds', 'timeseries',
+                                               str(group_id), '?']]}]}
+
+    backend = await aio.call(hat.event.backends.lmdb.info.create, conf, None,
+                             None)
+
+    try:
+        t1 = time.monotonic()
+
+        with cProfile.Profile() as pr:
+            await backend.query(
+                common.QueryTimeseriesParams(
+                    event_types=[('eds', 'timeseries', str(group_id), '0')],
+                    order_by=common.OrderBy.SOURCE_TIMESTAMP,
+                    order=common.Order.DESCENDING,
+                    max_results=max_results + 1))
+
+        t2 = time.monotonic()
+
+        print(t2 - t1)
+
+        pr.dump_stats(str(db_path.with_suffix('.profile')))
+
+    finally:
+        await aio.uncancellable(backend.async_close())
+
+
+if __name__ == '__main__':
+    main()

playground/lmdb/timeseries/query.sh

@@ -0,0 +1,12 @@
+#!/bin/sh
+
+set -e
+
+RUN_PATH=$(dirname "$(realpath "$0")")
+PLAYGROUND_PATH=$RUN_PATH/../..
+. $PLAYGROUND_PATH/env.sh
+
+
+db_path=$DATA_PATH/lmdb_timeseries.db
+
+exec $PYTHON $RUN_PATH/query.py $db_path

src_py/hat/event/backends/lmdb/backend.py

@@ -171,7 +171,7 @@ async def register(self,
         self._registered_count += len(events)
 
         if self._registered_count > max_registered_count:
-            await self._flush_queue.put(self._loop.create_future())
+            await self._flush_queue.put(None)
 
         if self._registered_events_cb:
             await aio.call(self._registered_events_cb, events)
@@ -236,7 +236,7 @@ async def cleanup():
 
                 while futures:
                     future = futures.popleft()
-                    if not future.done():
+                    if future and not future.done():
                         future.set_result(None)
 
         except Exception as e:
@@ -250,7 +250,7 @@ async def cleanup():
                 futures.append(self._flush_queue.get_nowait())
 
             for future in futures:
-                if not future.done():
+                if future and not future.done():
                     future.set_exception(common.BackendClosedError())
 
             await aio.uncancellable(cleanup())