diff --git a/modules/json_helpers.py b/modules/json_helpers.py index 5d9a6d707..4b866ee24 100644 --- a/modules/json_helpers.py +++ b/modules/json_helpers.py @@ -1,14 +1,27 @@ import os import sys import contextlib +import threading import time import json from typing import overload, Literal -import fasteners import orjson from modules.logger import log -locking_available = True # used by file read/write locking +path_locks: dict[str, threading.RLock] = {} +path_locks_guard = threading.Lock() + + +def path_lock(filename: str | os.PathLike[str]) -> threading.RLock: + """One lock per file path; threads of this process serialize on it, other processes are covered by atomic replace.""" + key = os.path.normcase(os.path.abspath(filename)) + with path_locks_guard: + lock = path_locks.get(key) + if lock is None: + lock = path_locks[key] = threading.RLock() + with contextlib.suppress(OSError): + os.remove(f"{key}.lock") # left behind by the file lock this replaces + return lock def replace_file(source: str, target: str, attempts: int = 10, delay: float = 0.05): @@ -30,51 +43,27 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool @overload def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool = False) -> dict | list: ... def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool = False, *, as_type="") -> dict | list: - global locking_available # pylint: disable=global-statement + """Read a JSON file; lock=True serializes with writers of the same path in this process.""" data = {} if as_type == "dict" else [] - lock_file = None - locked = False - if lock and locking_available: + with path_lock(filename) if lock else contextlib.nullcontext(): try: - lock_file = fasteners.InterProcessReaderWriterLock(f"{filename}.lock") - lock_file.logger.disabled = True # type: ignore - False positive. Bad typing in Fasteners. - locked = lock_file.acquire_read_lock(blocking=True, timeout=3) - if not locked: - log.warning(f'File read lock: file="{filename}" timeout') + t0 = time.time() + with open(filename, "rb") as file: + b = file.read() + if len(b) == 0: + return {} if as_type == "dict" else [] + data = orjson.loads(b) # pylint: disable=no-member + t1 = time.time() + if not silent: + fn = f"{sys._getframe(2).f_code.co_name}:{sys._getframe(1).f_code.co_name}" # pylint: disable=protected-access + log.debug(f'Read: file="{filename}" json={len(data)} bytes={os.path.getsize(filename)} time={t1 - t0:.3f} fn={fn}') + except FileNotFoundError as err: + if not silent: + log.debug(f'Read failed: file="{filename}" {err}') except Exception as err: - lock_file = None - locking_available = False - log.error(f'File read lock: file="{filename}" {err}') - locked = False - - try: - # if not os.path.exists(filename): - # return {} - t0 = time.time() - with open(filename, "rb") as file: - b = file.read() - if len(b) == 0: - return {} if as_type == "dict" else [] - data = orjson.loads(b) # pylint: disable=no-member - # if type(data) is str: - # data = json.loads(data) - t1 = time.time() - if not silent: - fn = f"{sys._getframe(2).f_code.co_name}:{sys._getframe(1).f_code.co_name}" # pylint: disable=protected-access - log.debug(f'Read: file="{filename}" json={len(data)} bytes={os.path.getsize(filename)} time={t1 - t0:.3f} fn={fn}') - except FileNotFoundError as err: - if not silent: - log.debug(f'Read failed: file="{filename}" {err}') - except Exception as err: - if not silent: - log.error(f'Read failed: file="{filename}" {err}') - - try: - if locking_available and lock_file is not None: - lock_file.release_read_lock() # the lock file stays: removing it after release races other holders - except Exception as err: - log.error(f'File read lock release: file="{filename}" {err}') + if not silent: + log.error(f'Read failed: file="{filename}" {err}') if isinstance(data, list) and as_type == "dict": if not data: @@ -93,71 +82,50 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", silent=False, atomic=False): + """Write obj as JSON; writes to the same path from this process run one at a time, in call order.""" import copy import tempfile - global locking_available # pylint: disable=global-statement - lock_file = None - locked = False - def default(obj): log.error(f'Save: file="{filename}" not a valid object: {obj}') return str(obj) - try: - t0 = time.time() - snapshot = obj.copy() if isinstance(obj, (dict, list)) else obj # dict.copy and list.copy run under the GIL, so a concurrent insert cannot interrupt them - data = copy.deepcopy(snapshot) - for k, v in list(data.items()) if isinstance(data, dict) else []: # validate each key-by-key to avoid global exceptions - try: - _tmp = json.dumps(v, indent=2, default=default, allow_nan=False, ensure_ascii=False) - except Exception as err: - if not silent: - log.error(f'Save: file="{filename}" key="{k}" value="{v}" {err}') - del data[k] - output = json.dumps(data, indent=2, default=default) - except Exception as err: - log.error(f'Save failed: file="{filename}" {err}') - return + with path_lock(filename): + try: + t0 = time.time() + snapshot = obj.copy() if isinstance(obj, (dict, list)) else obj # dict.copy and list.copy run under the GIL, so a concurrent insert cannot interrupt them + data = copy.deepcopy(snapshot) + for k, v in list(data.items()) if isinstance(data, dict) else []: # validate each key-by-key to avoid global exceptions + try: + _tmp = json.dumps(v, indent=2, default=default, allow_nan=False, ensure_ascii=False) + except Exception as err: + if not silent: + log.error(f'Save: file="{filename}" key="{k}" value="{v}" {err}') + del data[k] + output = json.dumps(data, indent=2, default=default) + except Exception as err: + log.error(f'Save failed: file="{filename}" {err}') + return - try: - if locking_available: - lock_file = fasteners.InterProcessReaderWriterLock(f"{filename}.lock") if locking_available else None - lock_file.logger.disabled = True # type: ignore - False positive. Bad typing in Fasteners. - locked = lock_file.acquire_write_lock(blocking=True, timeout=3) if lock_file is not None else False - if not locked: - log.warning(f'File write lock: file="{filename}" timeout') - except Exception as err: - locking_available = False - lock_file = None - log.error(f'File write lock: file="{filename}" {err}') - locked = False - - try: - if atomic: - fd, temp_name = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(filename)), prefix=f"{os.path.basename(filename)}.", suffix=".tmp") - try: - with os.fdopen(fd, mode, encoding="utf8") as f: - f.write(output) - f.flush() - os.fsync(f.fileno()) - replace_file(temp_name, filename) - except BaseException: - with contextlib.suppress(OSError): - os.remove(temp_name) - raise - else: - with open(filename, mode=mode, encoding="utf8") as file: - file.write(output) - t1 = time.time() - if not silent: - datalength = len(data) - log.debug(f'Save: file="{filename}" json={datalength} bytes={len(output)} time={t1 - t0:.3f}') - except Exception as err: - log.error(f'Save failed: file="{filename}" {err}') - - try: - if locking_available and lock_file is not None: - lock_file.release_write_lock() # the lock file stays: removing it after release races other holders - except Exception as err: - log.error(f'File write lock release: file="{filename}" {err}') + try: + if atomic: + fd, temp_name = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(filename)), prefix=f"{os.path.basename(filename)}.", suffix=".tmp") + try: + with os.fdopen(fd, mode, encoding="utf8") as f: + f.write(output) + f.flush() + os.fsync(f.fileno()) + replace_file(temp_name, filename) + except BaseException: + with contextlib.suppress(OSError): + os.remove(temp_name) + raise + else: + with open(filename, mode=mode, encoding="utf8") as file: + file.write(output) + t1 = time.time() + if not silent: + datalength = len(data) + log.debug(f'Save: file="{filename}" json={datalength} bytes={len(output)} time={t1 - t0:.3f}') + except Exception as err: + log.error(f'Save failed: file="{filename}" {err}') diff --git a/test/test-json-helpers.py b/test/test-json-helpers.py index 95e91ce31..f21f2a6e8 100644 --- a/test/test-json-helpers.py +++ b/test/test-json-helpers.py @@ -6,7 +6,9 @@ Covers: - atomic saves replace a fresh or existing target and leave no temp file, including when the save fails - an atomic save waits out a target another thread briefly holds open, which Windows otherwise refuses -- locking stays available under overlapping locked reads and writes and the lock file is left in place +- overlapping locked reads and atomic writes never fail or read a torn file, and no lock file is created +- locked readers never see a torn file while unlocked writers rewrite it in place +- a lock file left behind by the former file lock is removed - concurrent inserts into a shared dict never drop a save - readfile returns what writefile wrote, as dict and as list @@ -116,25 +118,65 @@ def test_atomic_save_waits_for_open_target(folder): assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' -def test_locking_survives_overlap(folder, threads=8, iterations=25): +def test_overlapping_locked_access(folder, threads=8, iterations=25): jh = fresh_helper() target = os.path.join(folder, 'locked.json') jh.writefile({'seed': 0}, target, silent=True) + torn = [] def worker(n): for i in range(iterations): if (n + i) % 2: jh.writefile({'n': n, 'i': i}, target, silent=True, atomic=True) - else: - jh.readfile(target, silent=True, lock=True, as_type='dict') + elif not jh.readfile(target, silent=True, lock=True, as_type='dict'): + torn.append((n, i)) run_threads(worker, threads) - assert jh.locking_available, 'locking switched itself off' assert jh.log.errors == [], jh.log.errors - assert os.path.exists(target + '.lock'), 'lock file was removed' + assert torn == [], f'{len(torn)} reads returned nothing' + assert not os.path.exists(target + '.lock'), 'a lock file was created' assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}' +def test_locked_readers_never_see_torn_writes(folder, writers=8, per_writer=15, readers=2): + jh = fresh_helper() + target = os.path.join(folder, 'config.json') + jh.writefile({'seed': 0}, target, silent=True) + stop = threading.Event() + torn = [] + + def reader(_n): + while not stop.is_set(): + if not jh.readfile(target, silent=True, lock=True, as_type='dict'): + torn.append(1) + + def writer(n): + for i in range(per_writer): + jh.writefile({'writer': n, 'i': i, 'blob': 'x' * (40000 + 1000 * n)}, target, silent=True, atomic=False) + + pool = [threading.Thread(target=reader, args=(r,)) for r in range(readers)] + for t in pool: + t.start() + run_threads(writer, writers) + stop.set() + for t in pool: + t.join() + assert torn == [], f'{len(torn)} reads returned nothing' + with open(target, encoding='utf8') as f: + json.load(f) + assert jh.log.errors == [], jh.log.errors + + +def test_legacy_lock_file_removed(folder): + jh = fresh_helper() + target = os.path.join(folder, 'legacy.json') + with open(target + '.lock', 'w', encoding='utf8'): + pass + jh.writefile({'x': 1}, target, silent=True) + assert not os.path.exists(target + '.lock'), 'legacy lock file kept' + assert jh.log.errors == [], jh.log.errors + + def test_concurrent_inserts_keep_every_save(folder, threads=4, per_thread=40): jh = fresh_helper() target = os.path.join(folder, 'cache.json') @@ -170,7 +212,9 @@ def run_all(): test_atomic_save_replaces_target, test_failed_atomic_save_leaves_no_temp, test_atomic_save_waits_for_open_target, - test_locking_survives_overlap, + test_overlapping_locked_access, + test_locked_readers_never_see_torn_writes, + test_legacy_lock_file_removed, test_concurrent_inserts_keep_every_save, test_roundtrip, ]