hat.event.backends.lmdb.backend

  1from collections.abc import Collection
  2from pathlib import Path
  3import asyncio
  4import collections
  5import contextlib
  6import logging
  7import typing
  8
  9from hat import aio
 10from hat import json
 11
 12from hat.event.backends.lmdb import common
 13from hat.event.backends.lmdb import environment
 14from hat.event.backends.lmdb import latestdb
 15from hat.event.backends.lmdb import refdb
 16from hat.event.backends.lmdb import systemdb
 17from hat.event.backends.lmdb import timeseriesdb
 18from hat.event.backends.lmdb.conditions import Conditions
 19
 20
 21mlog = logging.getLogger(__name__)
 22
 23cleanup_max_results = 1024
 24
 25flush_queue_size = 4096
 26
 27max_registered_count = 1024 * 256
 28
 29default_timeseries_max_results = 4096
 30
 31version = '0.9'
 32
 33
 34class Databases(typing.NamedTuple):
 35    system: systemdb.SystemDb
 36    latest: latestdb.LatestDb
 37    timeseries: timeseriesdb.TimeseriesDb
 38    ref: refdb.RefDb
 39
 40
 41class Changes(typing.NamedTuple):
 42    system: systemdb.Changes
 43    latest: latestdb.Changes
 44    timeseries: timeseriesdb.Changes
 45    ref: refdb.Changes
 46
 47
 48async def create(conf: json.Data,
 49                 registered_events_cb: common.BackendRegisteredEventsCb | None,
 50                 flushed_events_cb: common.BackendFlushedEventsCb | None
 51                 ) -> 'LmdbBackend':
 52    backend = LmdbBackend()
 53    backend._registered_events_cb = registered_events_cb
 54    backend._flushed_events_cb = flushed_events_cb
 55    backend._conditions = Conditions(conf['conditions'])
 56    backend._loop = asyncio.get_running_loop()
 57    backend._flush_queue = aio.Queue(flush_queue_size)
 58    backend._registered_count = 0
 59    backend._registered_queue = collections.deque()
 60    backend._async_group = aio.Group()
 61
 62    backend._env = await environment.create(Path(conf['db_path']))
 63    backend.async_group.spawn(aio.call_on_done, backend._env.wait_closing(),
 64                              backend.close)
 65
 66    try:
 67        latest_subscription = common.create_subscription(
 68            tuple(i) for i in conf['latest']['subscriptions'])
 69
 70        timeseries_partitions = (
 71            timeseriesdb.Partition(
 72                order_by=common.OrderBy[i['order_by']],
 73                subscription=common.create_subscription(
 74                    tuple(event_type) for event_type in i['subscriptions']),
 75                limit=(
 76                    timeseriesdb.Limit(
 77                        min_entries=i['limit'].get('min_entries'),
 78                        max_entries=i['limit'].get('max_entries'),
 79                        duration=i['limit'].get('duration'),
 80                        size=i['limit'].get('size'))
 81                    if 'limit' in i else None))
 82            for i in conf['timeseries'])
 83
 84        timeseries_max_results = conf.get('timeseries_max_results',
 85                                          default_timeseries_max_results)
 86
 87        backend._dbs = await backend._env.execute(
 88            _ext_create_dbs, backend._env, conf['identifier'],
 89            backend._conditions, latest_subscription, timeseries_partitions,
 90            timeseries_max_results)
 91
 92        backend.async_group.spawn(backend._flush_loop, conf['flush_period'])
 93        backend.async_group.spawn(backend._cleanup_loop,
 94                                  conf['cleanup_period'])
 95
 96    except BaseException:
 97        await aio.uncancellable(backend._env.async_close())
 98        raise
 99
100    return backend
101
102
103def _ext_create_dbs(env, identifier, conditions, latest_subscription,
104                    timeseries_partitions, timeseries_max_results):
105    with env.ext_begin(write=True) as txn:
106        system_db = systemdb.ext_create(env, txn, version, identifier)
107        latest_db = latestdb.ext_create(env, txn, conditions,
108                                        latest_subscription)
109        timeseries_db = timeseriesdb.ext_create(env, txn, conditions,
110                                                timeseries_partitions,
111                                                timeseries_max_results)
112        ref_db = refdb.RefDb(env)
113
114    return Databases(system=system_db,
115                     latest=latest_db,
116                     timeseries=timeseries_db,
117                     ref=ref_db)
118
119
120class LmdbBackend(common.Backend):
121
122    @property
123    def async_group(self) -> aio.Group:
124        return self._async_group
125
126    async def get_last_event_id(self,
127                                server_id: int
128                                ) -> common.EventId:
129        if not self.is_open:
130            raise common.BackendClosedError()
131
132        return self._dbs.system.get_last_event_id(server_id)
133
134    async def register(self,
135                       events: Collection[common.Event]
136                       ) -> Collection[common.Event] | None:
137        if not self.is_open:
138            raise common.BackendClosedError()
139
140        for event in events:
141            server_id = event.id.server
142
143            last_event_id = self._dbs.system.get_last_event_id(server_id)
144            last_timestamp = self._dbs.system.get_last_timestamp(server_id)
145
146            if last_event_id >= event.id:
147                mlog.warning("event registration skipped: invalid event id")
148                continue
149
150            if last_timestamp > event.timestamp:
151                mlog.warning("event registration skipped: invalid timestamp")
152                continue
153
154            if not self._conditions.matches(event):
155                mlog.warning("event registration skipped: invalid conditions")
156                continue
157
158            refs = collections.deque()
159
160            latest_result = self._dbs.latest.add(event)
161
162            if latest_result.added_ref:
163                refs.append(latest_result.added_ref)
164
165            if latest_result.removed_ref:
166                self._dbs.ref.remove(*latest_result.removed_ref)
167
168            refs.extend(self._dbs.timeseries.add(event))
169
170            if not refs:
171                continue
172
173            self._dbs.ref.add(event, refs)
174            self._dbs.system.set_last_event_id(event.id)
175            self._dbs.system.set_last_timestamp(server_id, event.timestamp)
176
177        self._registered_queue.append(events)
178        self._registered_count += len(events)
179
180        if self._registered_count > max_registered_count:
181            await self._flush_queue.put(None)
182
183        if self._registered_events_cb:
184            await aio.call(self._registered_events_cb, events)
185
186        return events
187
188    async def query(self,
189                    params: common.QueryParams
190                    ) -> common.QueryResult:
191        if not self.is_open:
192            raise common.BackendClosedError()
193
194        if isinstance(params, common.QueryLatestParams):
195            return self._dbs.latest.query(params)
196
197        if isinstance(params, common.QueryTimeseriesParams):
198            return await self._dbs.timeseries.query(params)
199
200        if isinstance(params, common.QueryServerParams):
201            return await self._dbs.ref.query(params)
202
203        raise ValueError('unsupported params type')
204
205    async def flush(self):
206        try:
207            future = self._loop.create_future()
208            await self._flush_queue.put(future)
209            await future
210
211        except aio.QueueClosedError:
212            raise common.BackendClosedError()
213
214    async def _flush_loop(self, flush_period):
215        futures = collections.deque()
216
217        async def cleanup():
218            with contextlib.suppress(Exception):
219                await self._flush()
220
221            await self._env.async_close()
222
223        try:
224            while True:
225                try:
226                    future = await aio.wait_for(self._flush_queue.get(),
227                                                flush_period)
228                    futures.append(future)
229
230                except asyncio.TimeoutError:
231                    pass
232
233                except aio.CancelledWithResultError as e:
234                    if e.result:
235                        futures.append(e.result)
236
237                    raise
238
239                while not self._flush_queue.empty():
240                    futures.append(self._flush_queue.get_nowait())
241
242                await aio.uncancellable(self._flush())
243
244                while futures:
245                    future = futures.popleft()
246                    if future and not future.done():
247                        future.set_result(None)
248
249        except Exception as e:
250            mlog.error('backend flush error: %s', e, exc_info=e)
251
252        finally:
253            self.close()
254            self._flush_queue.close()
255
256            while not self._flush_queue.empty():
257                futures.append(self._flush_queue.get_nowait())
258
259            for future in futures:
260                if future and not future.done():
261                    future.set_exception(common.BackendClosedError())
262
263            await aio.uncancellable(cleanup())
264
265    async def _cleanup_loop(self, cleanup_period):
266        try:
267            while True:
268                await asyncio.sleep(0)
269
270                repeat = await self._env.execute(_ext_cleanup, self._env,
271                                                 self._dbs, common.now())
272                if repeat:
273                    continue
274
275                await asyncio.sleep(cleanup_period)
276
277        except Exception as e:
278            mlog.error('backend cleanup error: %s', e, exc_info=e)
279
280        finally:
281            self.close()
282
283    async def _flush(self):
284        if not self._env.is_open:
285            return
286
287        self._registered_count = 0
288        registered_queue, self._registered_queue = (self._registered_queue,
289                                                    collections.deque())
290
291        changes = Changes(system=self._dbs.system.create_changes(),
292                          latest=self._dbs.latest.create_changes(),
293                          timeseries=self._dbs.timeseries.create_changes(),
294                          ref=self._dbs.ref.create_changes())
295
296        # TODO lock period between create_changes and locking executor
297        #      (timeseries and ref must write changes before new queries are
298        #      allowed)
299
300        await self._env.execute(_ext_flush, self._env, self._dbs, changes)
301
302        if not self._flushed_events_cb:
303            return
304
305        while registered_queue:
306            events = registered_queue.popleft()
307            await aio.call(self._flushed_events_cb, events)
308
309
310def _ext_flush(env, dbs, changes):
311    with env.ext_begin(write=True) as txn:
312        dbs.system.ext_write(txn, changes.system)
313        dbs.latest.ext_write(txn, changes.latest)
314        dbs.timeseries.ext_write(txn, changes.timeseries)
315        dbs.ref.ext_write(txn, changes.ref)
316
317
318def _ext_cleanup(env, dbs, now):
319    with env.ext_begin(write=True) as txn:
320        result = dbs.timeseries.ext_cleanup(txn, now, cleanup_max_results)
321        if not result:
322            return False
323
324        dbs.ref.ext_cleanup(txn, result)
325
326    return len(result) >= cleanup_max_results