mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 01:04:32 +02:00
fix(json_helpers): serialize file access per path within the process
The fasteners lock is inter-process only. On Linux fcntl record locks belong to the process, so two threads writing the same file were never serialized and could tear it; on Windows they were, until the lock switched itself off. Every writer of these files runs inside one multi-threaded process, so that is the case that matters. readfile(lock=True) and writefile now take a re-entrant lock per normalized path, held from the snapshot through the replace so writes to one path land in call order. Other processes are covered by the atomic replace. A .lock file left next to a JSON file by the old lock is removed the first time that path is used.
This commit is contained in:
CalamitousFelicitousness
+71
-103
@@ -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}')
|
||||
|
||||
@@ -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,
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user