From 243575719737a0e2c80919a133e271f1fe49d3a7 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 18 Jun 2026 21:25:19 -0700 Subject: [PATCH] feat: enhance SQLiteHandler and StorageCollector for improved caching and async performance - Added caching for packet type stats and cumulative counts in SQLiteHandler to reduce database load. - Implemented a dedicated writer thread in StorageCollector to handle blocking storage operations, preventing asyncio event loop stalls. - Updated record_packet method to utilize the new writer thread for efficient packet processing. --- repeater/data_acquisition/sqlite_handler.py | 28 ++++++++- .../data_acquisition/storage_collector.py | 62 +++++++++++++------ 2 files changed, 69 insertions(+), 21 deletions(-) diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index 8ea4362..b36bfc3 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -19,7 +19,16 @@ class SQLiteHandler: self._api_token_last_used_interval_sec = 300 self._hot_cache_ttl_sec = 60 self._packet_stats_cache = {} + self._packet_type_stats_cache = {} self._neighbors_cache = {"timestamp": 0.0, "value": None} + # Short time-based cache for the per-packet cumulative-counts aggregate + # (two full-table scans). The storage writer thread calls this once per + # recorded packet/duplicate; a few seconds of staleness is fine for the + # RRD/UI counters and stops a full scan running on every packet. + # Intentionally NOT cleared by _invalidate_hot_caches() — that runs on + # every write, which would defeat the cache under load. + self._cumulative_counts_cache = {"timestamp": 0.0, "value": None} + self._cumulative_counts_ttl_sec = 3.0 # Thread-local storage for persistent SQLite connections. # Opening a new connection on every DB call is expensive on SD-card # storage: each sqlite3.connect() call triggers file-system operations @@ -1186,7 +1195,11 @@ class SQLiteHandler: def get_packet_type_stats(self, hours: int = 24) -> dict: try: - cutoff = time.time() - (hours * 3600) + now = time.time() + cached = self._packet_type_stats_cache.get(hours) + if cached and (now - cached["timestamp"]) < self._hot_cache_ttl_sec: + return cached["value"] + cutoff = now - (hours * 3600) # Align with pyMC_core feat/newRadios PAYLOAD_TYPES (0x0B = CONTROL) try: @@ -1262,13 +1275,15 @@ class SQLiteHandler: if other_count > 0: type_counts["Other Types (>15)"] = other_count - return { + result = { "hours": hours, "packet_type_totals": type_counts, "total_packets": sum(type_counts.values()), "period": f"{hours} hours", "data_source": "sqlite", } + self._packet_type_stats_cache[hours] = {"timestamp": now, "value": result} + return result except Exception as e: logger.error(f"Failed to get packet type stats from SQLite: {e}") @@ -1602,6 +1617,11 @@ class SQLiteHandler: logger.error(f"Failed to cleanup old data: {e}") def get_cumulative_counts(self) -> dict: + now = time.time() + cached = self._cumulative_counts_cache.get("value") + cached_ts = float(self._cumulative_counts_cache.get("timestamp", 0.0)) + if cached is not None and (now - cached_ts) < self._cumulative_counts_ttl_sec: + return cached try: with self._connect() as conn: conn.row_factory = sqlite3.Row @@ -1630,12 +1650,14 @@ class SQLiteHandler: """ ).fetchone() - return { + result = { "rx_total": int(totals["rx_total"] or 0), "tx_total": int(totals["tx_total"] or 0), "drop_total": int(totals["drop_total"] or 0), "type_counts": type_counts, } + self._cumulative_counts_cache = {"timestamp": now, "value": result} + return result except Exception as e: logger.error(f"Failed to get cumulative counts: {e}") diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index f004e0d..32e7506 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -1,4 +1,5 @@ import asyncio +import concurrent.futures import logging import time from typing import Optional @@ -20,6 +21,17 @@ class StorageCollector: self.glass_publish_callback = None self._pending_tasks = set() + # Dedicated single writer thread for all blocking storage work (the SQLite + # write, the cumulative-counts aggregate, RRD updates, and network + # publishing). This keeps that work off the asyncio event loop, which it + # was previously stalling for seconds per packet on a busy mesh — starving + # every other coroutine (e.g. send_advert would time out). One worker + # preserves packet write ordering and reuses a single thread-local SQLite + # connection (no WAL writer contention, no connection fan-out). + self._db_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=1, thread_name_prefix="storage-writer" + ) + self.storage_dir = resolve_storage_dir(config) self.storage_dir.mkdir(parents=True, exist_ok=True) @@ -145,7 +157,12 @@ class StorageCollector: return stats def record_packet(self, packet_record: dict, skip_mqtt_if_invalid: bool = True): - """Record packet to storage and publish to MQTT + """Record a packet to storage and publish it. + + All blocking work — the SQLite write, the cumulative-counts aggregate, the + RRD update, and network publishing — runs on the dedicated writer thread so + it never blocks the asyncio event loop. Callers treat this as + fire-and-forget (the previous synchronous version blocked the loop). Args: packet_record: Dictionary containing packet information @@ -155,27 +172,32 @@ class StorageCollector: f"Recording packet: type={packet_record.get('type')}, " f"transmitted={packet_record.get('transmitted')}" ) + self._submit_db(self._record_packet_blocking, packet_record, skip_mqtt_if_invalid) - # HOT PATH: Store to local databases only (fast, non-blocking) + def _submit_db(self, fn, *args): + """Run a blocking storage operation on the dedicated writer thread. + + Falls back to running inline only if the executor has already been shut + down (process teardown), so late records are not silently dropped. + """ + try: + self._db_executor.submit(self._run_db_task, fn, *args) + except RuntimeError: + self._run_db_task(fn, *args) + + def _run_db_task(self, fn, *args): + """Execute a writer-thread task, logging (not raising) on failure.""" + try: + fn(*args) + except Exception as e: + logger.error(f"Storage writer task failed: {e}", exc_info=True) + + def _record_packet_blocking(self, packet_record: dict, skip_mqtt: bool): + """Store, aggregate, update metrics, and publish one packet (writer thread).""" self.sqlite_handler.store_packet(packet_record) cumulative_counts = self.sqlite_handler.get_cumulative_counts() self.rrd_handler.update_packet_metrics(packet_record, cumulative_counts) - - # DEFERRED: Publish to network sinks and WebSocket in background tasks - # This prevents network latency from blocking packet processing - self._schedule_background( - self._deferred_publish, - packet_record, - skip_mqtt_if_invalid, - sync_fallback=self._publish_packet_sync, - ) - - async def _deferred_publish(self, packet_record: dict, skip_mqtt: bool): - """Deferred background task for all network publishing operations.""" - try: - self._publish_packet_sync(packet_record, skip_mqtt) - except Exception as e: - logger.error(f"Deferred publish failed: {e}", exc_info=True) + self._publish_packet_sync(packet_record, skip_mqtt) def _publish_packet_sync(self, packet_record: dict, skip_mqtt: bool): """Publish packet updates synchronously (used when no asyncio loop is active).""" @@ -421,6 +443,10 @@ class StorageCollector: return self.sqlite_handler.get_noise_floor_stats(hours) def close(self): + # Drain and stop the storage writer thread first so pending writes and + # publishes complete before MQTT and the DB connections are torn down. + self._db_executor.shutdown(wait=True) + # Cancel all pending background tasks for task in self._pending_tasks: if not task.done():