mirror of
https://github.com/vladmandic/automatic
synced 2026-09-18 16:54:33 +02:00
fix(json_helpers): make atomic saves work on windows, keep lock file
writefile(atomic=True) renamed the temp file while it was still open, which Windows refuses, so every atomic save failed there and left a temp file behind. The temp file is now created with mkstemp, closed before os.replace and removed when the save fails; the replace retries briefly when another handle holds the target, which Windows reports as a permission error. Both helpers removed the .lock file after releasing it. That removal raced other holders on both platforms, and any failure switched locking off for the whole process without a log line. The lock file now stays in place; a failed release or a lock timeout is logged instead. writefile deep-copied and validated the live object in Python, so a concurrent insert into a shared dict raised "dictionary changed size during iteration" and dropped the save. dict.copy and list.copy run under the GIL, so the deep copy and the per-key validation now walk that snapshot; nested containers remain the caller's responsibility. test/test-json-helpers.py covers all three on Linux and Windows.
This commit is contained in:
+37
-17
@@ -1,5 +1,6 @@
|
||||
import os
|
||||
import sys
|
||||
import contextlib
|
||||
import time
|
||||
import json
|
||||
from typing import overload, Literal
|
||||
@@ -10,6 +11,18 @@ from modules.logger import log
|
||||
locking_available = True # used by file read/write locking
|
||||
|
||||
|
||||
def replace_file(source: str, target: str, attempts: int = 10, delay: float = 0.05):
|
||||
"""os.replace that waits out a target another handle holds open, which Windows reports as a permission error."""
|
||||
for attempt in range(attempts):
|
||||
try:
|
||||
os.replace(source, target)
|
||||
return
|
||||
except PermissionError:
|
||||
if attempt == attempts - 1:
|
||||
raise
|
||||
time.sleep(delay)
|
||||
|
||||
|
||||
@overload
|
||||
def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool = False, *, as_type: Literal["dict"]) -> dict: ...
|
||||
@overload
|
||||
@@ -27,6 +40,8 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool
|
||||
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')
|
||||
except Exception as err:
|
||||
lock_file = None
|
||||
locking_available = False
|
||||
@@ -57,11 +72,9 @@ def readfile(filename: str | os.PathLike[str], silent: bool = False, lock: bool
|
||||
|
||||
try:
|
||||
if locking_available and lock_file is not None:
|
||||
lock_file.release_read_lock()
|
||||
if locked and os.path.exists(f"{filename}.lock"):
|
||||
os.remove(f"{filename}.lock")
|
||||
except Exception:
|
||||
locking_available = False
|
||||
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 isinstance(data, list) and as_type == "dict":
|
||||
if not data:
|
||||
@@ -93,8 +106,9 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile
|
||||
|
||||
try:
|
||||
t0 = time.time()
|
||||
data = copy.deepcopy(obj) # ensure keys/items aren't added/deleted during json.dumps
|
||||
for k, v in obj.items() if isinstance(obj, dict) else []: # validate each key-by-key to avoid global exceptions
|
||||
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:
|
||||
@@ -111,6 +125,8 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile
|
||||
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
|
||||
@@ -119,11 +135,17 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile
|
||||
|
||||
try:
|
||||
if atomic:
|
||||
with tempfile.NamedTemporaryFile(mode=mode, encoding="utf8", delete=False, dir=os.path.dirname(filename)) as f:
|
||||
f.write(output)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(f.name, filename)
|
||||
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)
|
||||
@@ -136,8 +158,6 @@ def writefile(obj: dict | list, filename: str | os.PathLike[str], mode="w", sile
|
||||
|
||||
try:
|
||||
if locking_available and lock_file is not None:
|
||||
lock_file.release_write_lock()
|
||||
if locked and os.path.exists(f"{filename}.lock"):
|
||||
os.remove(f"{filename}.lock")
|
||||
except Exception:
|
||||
locking_available = False
|
||||
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}')
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Offline tests for modules.json_helpers on Linux and Windows.
|
||||
|
||||
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
|
||||
- concurrent inserts into a shared dict never drop a save
|
||||
- readfile returns what writefile wrote, as dict and as list
|
||||
|
||||
No running server required.
|
||||
|
||||
Usage:
|
||||
python test/test-json-helpers.py
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
|
||||
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, script_dir)
|
||||
os.chdir(script_dir)
|
||||
sys.argv = [sys.argv[0]]
|
||||
|
||||
from modules.logger import log # pylint: disable=wrong-import-position
|
||||
from modules import json_helpers # pylint: disable=wrong-import-position
|
||||
|
||||
|
||||
class ErrorLog:
|
||||
"""Counts error lines from json_helpers while passing everything else to the real logger."""
|
||||
|
||||
def __init__(self, inner):
|
||||
self.inner = inner
|
||||
self.errors = []
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def error(self, message, *args, **kwargs):
|
||||
with self.lock:
|
||||
self.errors.append(str(message))
|
||||
|
||||
def count(self, needle):
|
||||
return sum(1 for m in self.errors if needle in m)
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self.inner, name)
|
||||
|
||||
|
||||
def fresh_helper():
|
||||
importlib.reload(json_helpers)
|
||||
json_helpers.log = ErrorLog(log)
|
||||
return json_helpers
|
||||
|
||||
|
||||
def run_threads(worker, count):
|
||||
pool = [threading.Thread(target=worker, args=(n,)) for n in range(count)]
|
||||
for t in pool:
|
||||
t.start()
|
||||
for t in pool:
|
||||
t.join()
|
||||
|
||||
|
||||
def leftovers(folder, target):
|
||||
keep = {os.path.basename(target), os.path.basename(target) + '.lock'}
|
||||
return sorted(set(os.listdir(folder)) - keep)
|
||||
|
||||
|
||||
def test_atomic_save_replaces_target(folder):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'atomic.json')
|
||||
jh.writefile({'first': 1}, target, silent=True, atomic=True)
|
||||
assert jh.readfile(target, silent=True, as_type='dict') == {'first': 1}, 'fresh target not written'
|
||||
jh.writefile({'second': 2}, target, silent=True, atomic=True)
|
||||
assert jh.readfile(target, silent=True, as_type='dict') == {'second': 2}, 'existing target not replaced'
|
||||
assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}'
|
||||
assert jh.log.errors == [], jh.log.errors
|
||||
|
||||
|
||||
def test_failed_atomic_save_leaves_no_temp(folder):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'blocked.json')
|
||||
os.mkdir(target) # a directory cannot be replaced by a file
|
||||
jh.writefile({'x': 1}, target, silent=True, atomic=True)
|
||||
assert jh.log.count('Save failed') == 1, jh.log.errors
|
||||
assert os.path.isdir(target), 'target directory was replaced'
|
||||
assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}'
|
||||
|
||||
|
||||
def test_atomic_save_waits_for_open_target(folder):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'open.json')
|
||||
jh.writefile({'v': 0}, target, silent=True)
|
||||
opened = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
def holder(_n):
|
||||
with open(target, 'rb') as f:
|
||||
f.read(1)
|
||||
opened.set()
|
||||
release.wait(5)
|
||||
|
||||
thread = threading.Thread(target=holder, args=(0,))
|
||||
thread.start()
|
||||
opened.wait(5)
|
||||
threading.Timer(0.2, release.set).start()
|
||||
jh.writefile({'v': 1}, target, silent=True, atomic=True)
|
||||
thread.join()
|
||||
assert jh.readfile(target, silent=True, as_type='dict') == {'v': 1}, 'save did not wait for the open target'
|
||||
assert jh.log.errors == [], jh.log.errors
|
||||
assert leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}'
|
||||
|
||||
|
||||
def test_locking_survives_overlap(folder, threads=8, iterations=25):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'locked.json')
|
||||
jh.writefile({'seed': 0}, target, silent=True)
|
||||
|
||||
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')
|
||||
|
||||
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 leftovers(folder, target) == [], f'temp files left: {leftovers(folder, target)}'
|
||||
|
||||
|
||||
def test_concurrent_inserts_keep_every_save(folder, threads=4, per_thread=40):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'cache.json')
|
||||
cache = {}
|
||||
payload = {'metadata': {f'k{i}': 'x' * 64 for i in range(200)}, 'tensors': list(range(500))}
|
||||
|
||||
def worker(n):
|
||||
for i in range(per_thread):
|
||||
cache[f'{n}-{i}'] = payload
|
||||
jh.writefile(cache, target, silent=True, atomic=True)
|
||||
|
||||
run_threads(worker, threads)
|
||||
assert jh.log.count('changed size') == 0, f'{jh.log.count("changed size")} saves dropped'
|
||||
assert jh.log.errors == [], jh.log.errors
|
||||
with open(target, encoding='utf8') as f:
|
||||
assert len(json.load(f)) == threads * per_thread, 'last save is incomplete'
|
||||
|
||||
|
||||
def test_roundtrip(folder):
|
||||
jh = fresh_helper()
|
||||
target = os.path.join(folder, 'roundtrip.json')
|
||||
data = {'a': 1, 'b': [1, 2, 3], 'c': {'d': 'e'}, 'f': None}
|
||||
jh.writefile(data, target, silent=True)
|
||||
assert jh.readfile(target, silent=True, as_type='dict') == data
|
||||
jh.writefile([1, 'two', {'three': 3}], target, silent=True, atomic=True)
|
||||
assert jh.readfile(target, silent=True, as_type='list') == [1, 'two', {'three': 3}]
|
||||
assert jh.readfile(os.path.join(folder, 'missing.json'), silent=True, as_type='dict') == {}
|
||||
assert jh.log.errors == [], jh.log.errors
|
||||
|
||||
|
||||
def run_all():
|
||||
tests = [
|
||||
test_atomic_save_replaces_target,
|
||||
test_failed_atomic_save_leaves_no_temp,
|
||||
test_atomic_save_waits_for_open_target,
|
||||
test_locking_survives_overlap,
|
||||
test_concurrent_inserts_keep_every_save,
|
||||
test_roundtrip,
|
||||
]
|
||||
passed = 0
|
||||
failed = 0
|
||||
for fn in tests:
|
||||
folder = tempfile.mkdtemp(prefix='sdnext-json-')
|
||||
try:
|
||||
fn(folder)
|
||||
log.info(f' PASS {fn.__name__}')
|
||||
passed += 1
|
||||
except Exception as e:
|
||||
log.error(f' FAIL {fn.__name__}: {type(e).__name__}: {e}')
|
||||
failed += 1
|
||||
finally:
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
log.warning(f'Total: {passed} passed, {failed} failed')
|
||||
return failed == 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import time
|
||||
t0 = time.time()
|
||||
ok = run_all()
|
||||
log.warning(f'Total time: {time.time() - t0:.2f}s')
|
||||
sys.exit(0 if ok else 1)
|
||||
Reference in New Issue
Block a user