diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index ee0fdb52..f173bc84 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -62,6 +62,7 @@ from common import ( yaml_load, ) import packresolve +from refresh_data_dirs import cache_lock import region as region_mod import slot as slot_mod import slots @@ -345,22 +346,25 @@ def _data_directory_members( dd_prefix = base_dest else: dd_prefix = dd_dest - for root, _dirs, filenames in os.walk(local_path): - for fname in filenames: - src = os.path.join(root, fname) - rel = os.path.relpath(src, local_path) - full = f"{dd_prefix}/{rel}" - if full in seen_destinations or ( - full.lower() in seen_lower and case_insensitive - ): - continue - if _has_path_conflict(full, seen_destinations, seen_parents): - continue - seen_destinations.add(full) - _register_path(full, seen_destinations, seen_parents) - if case_insensitive: - seen_lower.add(full.lower()) - yield src, full + # Held while the caller writes what is yielded: a refresh swapping + # the tree waits until the walk is done. + with cache_lock(local_path, shared=True): + for root, _dirs, filenames in os.walk(local_path): + for fname in filenames: + src = os.path.join(root, fname) + rel = os.path.relpath(src, local_path) + full = f"{dd_prefix}/{rel}" + if full in seen_destinations or ( + full.lower() in seen_lower and case_insensitive + ): + continue + if _has_path_conflict(full, seen_destinations, seen_parents): + continue + seen_destinations.add(full) + _register_path(full, seen_destinations, seen_parents) + if case_insensitive: + seen_lower.add(full.lower()) + yield src, full def _pack_data_directories( diff --git a/scripts/refresh_data_dirs.py b/scripts/refresh_data_dirs.py index 3eaaaadd..50a24411 100644 --- a/scripts/refresh_data_dirs.py +++ b/scripts/refresh_data_dirs.py @@ -52,8 +52,8 @@ def load_registry(registry_path: str = DEFAULT_REGISTRY) -> dict[str, dict]: @contextlib.contextmanager -def _file_lock(lock_path: Path): - """Hold an exclusive lock on lock_path, waiting for it if taken. +def _file_lock(lock_path: Path, shared: bool = False): + """Hold a lock on lock_path, waiting for it if taken. Several sessions refresh the same data directories: without it, two swaps of one tree interleave and the second lands inside the first. @@ -65,14 +65,24 @@ def _file_lock(lock_path: Path): yield return lock_path.parent.mkdir(parents=True, exist_ok=True) - with open(lock_path, "w") as handle: - fcntl.flock(handle, fcntl.LOCK_EX) + with open(lock_path, "a") as handle: + fcntl.flock(handle, fcntl.LOCK_SH if shared else fcntl.LOCK_EX) try: yield finally: 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]: path = Path(versions_path) if not path.exists(): @@ -381,8 +391,7 @@ def refresh_entry( """ if dry_run: return _refresh_entry(key, entry, force, dry_run, versions_path) - cache_dir = Path(entry["local_cache"]) - with _file_lock(cache_dir.with_name(f".{cache_dir.name}.lock")): + with cache_lock(entry["local_cache"]): return _refresh_entry(key, entry, force, dry_run, versions_path) diff --git a/tests/test_refresh_data_dirs.py b/tests/test_refresh_data_dirs.py index 15ce8648..5a6e936f 100644 --- a/tests/test_refresh_data_dirs.py +++ b/tests/test_refresh_data_dirs.py @@ -115,5 +115,28 @@ class RefreshConcurrency(unittest.TestCase): 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__": unittest.main()