perf: load the store once and serve reads from memory

T3-17: every get/set/delete re-read and re-parsed the whole file (get 40 ms,
set 63 ms at 100k keys). the file is now loaded once into memory; reads
return deep copies, set stores the value's json round-trip so memory always
equals a reload, and a failed write rolls the in-memory change back. the
stale-temp sweep runs once per instance. an external edit after first load
is no longer seen (single-process by contract). folds into 1.1.0.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
dsql committed 2026-10-16 04:34:30 -04:00
1 parent 27b4fcce99
commit 0af38e28dd
2 files changed
+53 -11

No files matched your search

+10
View File
@@ -59,6 +59,16 @@ kv = aiocache("state.json")
Prefer `AioKV` in new code. Prefer `AioKV` in new code.
## Memory model
The file is read once, on first use, and then served from memory: `get` / `get_all` don't
touch the disk, and each `set` / `delete` writes the whole store back atomically. `get` and
`get_all` return deep copies, and `set` stores the value's JSON round-trip (exactly what a
reload would give back), so mutating a value you passed in or got back never changes the
store. Because it is single-process, an edit to the file made by something else after the
first read is not seen (and is overwritten by the next write). Orphaned temp files are swept
once per instance, on its first save.
## Durability ## Durability
Writes are **atomic**: data is written to a temp file in the same directory and Writes are **atomic**: data is written to a temp file in the same directory and
+43 -11
View File
@@ -18,12 +18,13 @@ cross-process storage.
import os import os
import re import re
import copy
import shutil import shutil
import json import json
import time import time
import asyncio import asyncio
import logging import logging
from typing import Any, Dict from typing import Any, Dict, Optional
import aiofiles import aiofiles
@@ -41,31 +42,53 @@ class AioKV:
""" """
self.file = file self.file = file
self.lock = asyncio.Lock() self.lock = asyncio.Lock()
# the file is read once, then served from memory (single-process by contract);
# None = not loaded yet
self._data: Optional[Dict[str, Any]] = None
self._swept = False
async def set(self, key: str, value: Any = None) -> None: async def set(self, key: str, value: Any = None) -> None:
"""set a value; if value is omitted or None, stores int(time.time())""" """set a value; if value is omitted or None, stores int(time.time())
the value is stored as its json round-trip (what a reload from the file would give
back), and the in-memory store is rolled back if the write fails"""
self._check_key(key) self._check_key(key)
value = value if value is not None else int(time.time())
stored = json.loads(json.dumps(value, allow_nan=False))
async with self.lock: async with self.lock:
cache = await self._load() cache = await self._cache()
cache[key] = value if value is not None else int(time.time()) missing = object()
previous = cache.get(key, missing)
cache[key] = stored
try:
await self._save(cache) await self._save(cache)
except BaseException:
if previous is missing:
del cache[key]
else:
cache[key] = previous
raise
async def get(self, key: str, default: Any = None) -> Any: async def get(self, key: str, default: Any = None) -> Any:
"""return the value for key, or default if absent""" """return a deep copy of the value for key, or default if absent"""
self._check_key(key) self._check_key(key)
async with self.lock: async with self.lock:
cache = await self._load() cache = await self._cache()
return cache.get(key, default) return copy.deepcopy(cache[key]) if key in cache else default
async def delete(self, key: str) -> bool: async def delete(self, key: str) -> bool:
"""delete a key; returns True if removed or absent, False on error""" """delete a key; returns True if removed or absent, False on error"""
try: try:
self._check_key(key) self._check_key(key)
async with self.lock: async with self.lock:
cache = await self._load() cache = await self._cache()
if key in cache: if key in cache:
del cache[key] value = cache.pop(key)
try:
await self._save(cache) await self._save(cache)
except BaseException:
cache[key] = value
raise
return True return True
except Exception: except Exception:
log.exception("aiokv.delete() failed for key %s", key) log.exception("aiokv.delete() failed for key %s", key)
@@ -83,15 +106,16 @@ class AioKV:
await asyncio.to_thread(os.remove, target) await asyncio.to_thread(os.remove, target)
except FileNotFoundError: except FileNotFoundError:
pass pass
self._data = {}
return True return True
except Exception: except Exception:
log.exception("aiokv.clear() failed") log.exception("aiokv.clear() failed")
return False return False
async def get_all(self) -> Dict[str, Any]: async def get_all(self) -> Dict[str, Any]:
"""return a dict of every key/value in the store""" """return a deep copy of every key/value in the store"""
async with self.lock: async with self.lock:
return await self._load() return copy.deepcopy(await self._cache())
@staticmethod @staticmethod
def _check_key(key: str) -> None: def _check_key(key: str) -> None:
@@ -99,6 +123,12 @@ class AioKV:
if not isinstance(key, str): if not isinstance(key, str):
raise ValueError(f"aiokv keys must be str, got {type(key).__name__}") raise ValueError(f"aiokv keys must be str, got {type(key).__name__}")
async def _cache(self) -> Dict[str, Any]:
"""the in-memory store, loading the file on first use; caller must hold the lock"""
if self._data is None:
self._data = await self._load()
return self._data
async def _load(self) -> Dict[str, Any]: async def _load(self) -> Dict[str, Any]:
"""load the store, returning {} if absent or blank; raises JSONDecodeError on """load the store, returning {} if absent or blank; raises JSONDecodeError on
corrupt content and ValueError on valid-but-non-object JSON rather than masking it""" corrupt content and ValueError on valid-but-non-object JSON rather than masking it"""
@@ -124,7 +154,9 @@ class AioKV:
target = await asyncio.to_thread(os.path.realpath, self.file) target = await asyncio.to_thread(os.path.realpath, self.file)
directory = os.path.dirname(target) or "." directory = os.path.dirname(target) or "."
await asyncio.to_thread(os.makedirs, directory, exist_ok=True) await asyncio.to_thread(os.makedirs, directory, exist_ok=True)
if not self._swept:
await self._sweep_stale_tmp(directory, os.path.basename(target)) await self._sweep_stale_tmp(directory, os.path.basename(target))
self._swept = True
payload = await asyncio.to_thread(json.dumps, cache, allow_nan=False) payload = await asyncio.to_thread(json.dumps, cache, allow_nan=False)
tmp = f"{target}.{os.getpid()}.tmp" tmp = f"{target}.{os.getpid()}.tmp"