From 0af38e28dde12a44bb7af9ea656f7f6de3a2e8bb Mon Sep 17 00:00:00 2001 From: disqualifier Date: Fri, 16 Oct 2026 04:34:30 -0400 Subject: [PATCH] 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 --- README.md | 10 ++++++++ src/aiokv/aiokv.py | 60 +++++++++++++++++++++++++++++++++++----------- 2 files changed, 56 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index e35b801..c7ef3bd 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,16 @@ kv = aiocache("state.json") 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 Writes are **atomic**: data is written to a temp file in the same directory and diff --git a/src/aiokv/aiokv.py b/src/aiokv/aiokv.py index cb5fc9f..a5f9c94 100644 --- a/src/aiokv/aiokv.py +++ b/src/aiokv/aiokv.py @@ -18,12 +18,13 @@ cross-process storage. import os import re +import copy import shutil import json import time import asyncio import logging -from typing import Any, Dict +from typing import Any, Dict, Optional import aiofiles @@ -41,31 +42,53 @@ class AioKV: """ self.file = file 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: - """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) + value = value if value is not None else int(time.time()) + stored = json.loads(json.dumps(value, allow_nan=False)) async with self.lock: - cache = await self._load() - cache[key] = value if value is not None else int(time.time()) - await self._save(cache) + cache = await self._cache() + missing = object() + previous = cache.get(key, missing) + cache[key] = stored + try: + 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: - """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) async with self.lock: - cache = await self._load() - return cache.get(key, default) + cache = await self._cache() + return copy.deepcopy(cache[key]) if key in cache else default async def delete(self, key: str) -> bool: """delete a key; returns True if removed or absent, False on error""" try: self._check_key(key) async with self.lock: - cache = await self._load() + cache = await self._cache() if key in cache: - del cache[key] - await self._save(cache) + value = cache.pop(key) + try: + await self._save(cache) + except BaseException: + cache[key] = value + raise return True except Exception: log.exception("aiokv.delete() failed for key %s", key) @@ -83,15 +106,16 @@ class AioKV: await asyncio.to_thread(os.remove, target) except FileNotFoundError: pass + self._data = {} return True except Exception: log.exception("aiokv.clear() failed") return False 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: - return await self._load() + return copy.deepcopy(await self._cache()) @staticmethod def _check_key(key: str) -> None: @@ -99,6 +123,12 @@ class AioKV: if not isinstance(key, str): 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]: """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""" @@ -124,7 +154,9 @@ class AioKV: target = await asyncio.to_thread(os.path.realpath, self.file) directory = os.path.dirname(target) or "." await asyncio.to_thread(os.makedirs, directory, exist_ok=True) - await self._sweep_stale_tmp(directory, os.path.basename(target)) + if not self._swept: + await self._sweep_stale_tmp(directory, os.path.basename(target)) + self._swept = True payload = await asyncio.to_thread(json.dumps, cache, allow_nan=False) tmp = f"{target}.{os.getpid()}.tmp"