mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-09 19:43:53 -06:00
fix: hold the data dir lock while packing it
This commit is contained in:
1 parent
c9f8f4b230
commit
a1ce1d2fa0
3 files changed
+58
-22
No files matched your search
+20
-16
@@ -62,6 +62,7 @@ from common import (
|
|||||||
yaml_load,
|
yaml_load,
|
||||||
)
|
)
|
||||||
import packresolve
|
import packresolve
|
||||||
|
from refresh_data_dirs import cache_lock
|
||||||
import region as region_mod
|
import region as region_mod
|
||||||
import slot as slot_mod
|
import slot as slot_mod
|
||||||
import slots
|
import slots
|
||||||
@@ -345,22 +346,25 @@ def _data_directory_members(
|
|||||||
dd_prefix = base_dest
|
dd_prefix = base_dest
|
||||||
else:
|
else:
|
||||||
dd_prefix = dd_dest
|
dd_prefix = dd_dest
|
||||||
for root, _dirs, filenames in os.walk(local_path):
|
# Held while the caller writes what is yielded: a refresh swapping
|
||||||
for fname in filenames:
|
# the tree waits until the walk is done.
|
||||||
src = os.path.join(root, fname)
|
with cache_lock(local_path, shared=True):
|
||||||
rel = os.path.relpath(src, local_path)
|
for root, _dirs, filenames in os.walk(local_path):
|
||||||
full = f"{dd_prefix}/{rel}"
|
for fname in filenames:
|
||||||
if full in seen_destinations or (
|
src = os.path.join(root, fname)
|
||||||
full.lower() in seen_lower and case_insensitive
|
rel = os.path.relpath(src, local_path)
|
||||||
):
|
full = f"{dd_prefix}/{rel}"
|
||||||
continue
|
if full in seen_destinations or (
|
||||||
if _has_path_conflict(full, seen_destinations, seen_parents):
|
full.lower() in seen_lower and case_insensitive
|
||||||
continue
|
):
|
||||||
seen_destinations.add(full)
|
continue
|
||||||
_register_path(full, seen_destinations, seen_parents)
|
if _has_path_conflict(full, seen_destinations, seen_parents):
|
||||||
if case_insensitive:
|
continue
|
||||||
seen_lower.add(full.lower())
|
seen_destinations.add(full)
|
||||||
yield src, full
|
_register_path(full, seen_destinations, seen_parents)
|
||||||
|
if case_insensitive:
|
||||||
|
seen_lower.add(full.lower())
|
||||||
|
yield src, full
|
||||||
|
|
||||||
|
|
||||||
def _pack_data_directories(
|
def _pack_data_directories(
|
||||||
|
|||||||
@@ -52,8 +52,8 @@ def load_registry(registry_path: str = DEFAULT_REGISTRY) -> dict[str, dict]:
|
|||||||
|
|
||||||
|
|
||||||
@contextlib.contextmanager
|
@contextlib.contextmanager
|
||||||
def _file_lock(lock_path: Path):
|
def _file_lock(lock_path: Path, shared: bool = False):
|
||||||
"""Hold an exclusive lock on lock_path, waiting for it if taken.
|
"""Hold a lock on lock_path, waiting for it if taken.
|
||||||
|
|
||||||
Several sessions refresh the same data directories: without it, two
|
Several sessions refresh the same data directories: without it, two
|
||||||
swaps of one tree interleave and the second lands inside the first.
|
swaps of one tree interleave and the second lands inside the first.
|
||||||
@@ -65,14 +65,24 @@ def _file_lock(lock_path: Path):
|
|||||||
yield
|
yield
|
||||||
return
|
return
|
||||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
with open(lock_path, "w") as handle:
|
with open(lock_path, "a") as handle:
|
||||||
fcntl.flock(handle, fcntl.LOCK_EX)
|
fcntl.flock(handle, fcntl.LOCK_SH if shared else fcntl.LOCK_EX)
|
||||||
try:
|
try:
|
||||||
yield
|
yield
|
||||||
finally:
|
finally:
|
||||||
fcntl.flock(handle, fcntl.LOCK_UN)
|
fcntl.flock(handle, fcntl.LOCK_UN)
|
||||||
|
|
||||||
|
|
||||||
|
def cache_lock(cache_dir: str | Path, shared: bool = False):
|
||||||
|
"""The lock a refresh holds while it swaps cache_dir.
|
||||||
|
|
||||||
|
A reader takes it shared: a pack walking data/sdlpal while another run
|
||||||
|
swapped the tree shipped part of it, or nothing.
|
||||||
|
"""
|
||||||
|
cache_dir = Path(cache_dir)
|
||||||
|
return _file_lock(cache_dir.with_name(f".{cache_dir.name}.lock"), shared=shared)
|
||||||
|
|
||||||
|
|
||||||
def _load_versions(versions_path: str = VERSIONS_FILE) -> dict[str, dict]:
|
def _load_versions(versions_path: str = VERSIONS_FILE) -> dict[str, dict]:
|
||||||
path = Path(versions_path)
|
path = Path(versions_path)
|
||||||
if not path.exists():
|
if not path.exists():
|
||||||
@@ -381,8 +391,7 @@ def refresh_entry(
|
|||||||
"""
|
"""
|
||||||
if dry_run:
|
if dry_run:
|
||||||
return _refresh_entry(key, entry, force, dry_run, versions_path)
|
return _refresh_entry(key, entry, force, dry_run, versions_path)
|
||||||
cache_dir = Path(entry["local_cache"])
|
with cache_lock(entry["local_cache"]):
|
||||||
with _file_lock(cache_dir.with_name(f".{cache_dir.name}.lock")):
|
|
||||||
return _refresh_entry(key, entry, force, dry_run, versions_path)
|
return _refresh_entry(key, entry, force, dry_run, versions_path)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -115,5 +115,28 @@ class RefreshConcurrency(unittest.TestCase):
|
|||||||
self.assertEqual(sorted(rdd._load_versions(self.versions)), sorted(keys))
|
self.assertEqual(sorted(rdd._load_versions(self.versions)), sorted(keys))
|
||||||
|
|
||||||
|
|
||||||
|
class PackWalkHoldsTheCache(unittest.TestCase):
|
||||||
|
def test_a_refresh_waits_for_the_walk(self):
|
||||||
|
import fcntl
|
||||||
|
|
||||||
|
from generate_pack import _data_directory_members
|
||||||
|
|
||||||
|
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as tmp:
|
||||||
|
cache = Path(tmp) / "data" / "sdlpal"
|
||||||
|
cache.mkdir(parents=True)
|
||||||
|
(cache / "a.mkf").write_bytes(b"a")
|
||||||
|
(cache / "b.mkf").write_bytes(b"b")
|
||||||
|
systems = {"s": {"data_directories": [{"ref": "sdlpal", "destination": "sdlpal"}]}}
|
||||||
|
registry = {"sdlpal": {"local_cache": str(cache)}}
|
||||||
|
walk = _data_directory_members(systems, registry, "p", "", False, set(), set(), set())
|
||||||
|
next(walk)
|
||||||
|
with open(cache.with_name(".sdlpal.lock"), "a") as handle:
|
||||||
|
with self.assertRaises(BlockingIOError):
|
||||||
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||||
|
list(walk)
|
||||||
|
with open(cache.with_name(".sdlpal.lock"), "a") as handle:
|
||||||
|
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||||
|
fcntl.flock(handle, fcntl.LOCK_UN)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
Reference in new issue
Block a user