diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index 56975f37..cc3967f5 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -3223,46 +3223,48 @@ def _main(held: contextlib.ExitStack) -> None: return data_registry = load_data_dir_registry(args.platforms_dir) - if data_registry and not args.offline and not args.no_refresh_data: - from refresh_data_dirs import load_registry, refresh_all + # The output directory is claimed before the data directory refresh: a + # refresh waits for every run that holds a cache, so a second build + # during a pipeline sat silent for hours instead of being refused. + with _pack_output_lock(args.output_dir): + if data_registry and not args.offline and not args.no_refresh_data: + from refresh_data_dirs import load_registry, refresh_all - registry = load_registry(os.path.join(args.platforms_dir, "_data_dirs.yml")) - results = refresh_all(registry, force=args.refresh_data) - updated = sum(1 for v in results.values() if v) - if updated: - print(f"Refreshed {updated} data director{'ies' if updated > 1 else 'y'}") - failed = sorted(k for k, v in results.items() if v is None) - if failed: - print(f"WARNING: data directory refresh failed: {', '.join(failed)}") - if data_registry: - hold_data_dirs(held, data_registry) + registry = load_registry(os.path.join(args.platforms_dir, "_data_dirs.yml")) + results = refresh_all(registry, force=args.refresh_data) + updated = sum(1 for v in results.values() if v) + if updated: + print(f"Refreshed {updated} data director{'ies' if updated > 1 else 'y'}") + failed = sorted(k for k, v in results.items() if v is None) + if failed: + print(f"WARNING: data directory refresh failed: {', '.join(failed)}") + if data_registry: + hold_data_dirs(held, data_registry) - emu_profiles = load_emulator_profiles(args.emulators_dir) + emu_profiles = load_emulator_profiles(args.emulators_dir) - target_cores_cache: dict[str, set[str] | None] = {} - if args.target: - try: - target_cores_cache, platforms = build_target_cores_cache( - platforms, - args.target, - args.platforms_dir, - is_all=args.all, - ) - except (FileNotFoundError, ValueError) as e: - print(f"ERROR: {e}", file=sys.stderr) - sys.exit(1) + target_cores_cache: dict[str, set[str] | None] = {} + if args.target: + try: + target_cores_cache, platforms = build_target_cores_cache( + platforms, + args.target, + args.platforms_dir, + is_all=args.all, + ) + except (FileNotFoundError, ValueError) as e: + print(f"ERROR: {e}", file=sys.stderr) + sys.exit(1) - groups = group_identical_platforms( - platforms, args.platforms_dir, target_cores_cache if args.target else None - ) + groups = group_identical_platforms( + platforms, args.platforms_dir, target_cores_cache if args.target else None + ) - if args.manifest: - with _pack_output_lock(args.output_dir): + if args.manifest: _run_manifest_mode( args, groups, db, zip_contents, emu_profiles, target_cores_cache ) - else: - with _pack_output_lock(args.output_dir): + else: _run_platform_packs( args, groups, @@ -3273,9 +3275,9 @@ def _main(held: contextlib.ExitStack) -> None: target_cores_cache, system_filter, ) - if args.verify_packs: - with _pack_output_lock(args.output_dir, exclusive=False): - _run_verify_packs(args) + if not args.manifest and args.verify_packs: + with _pack_output_lock(args.output_dir, exclusive=False): + _run_verify_packs(args) # Manifest generation (JSON inventory for install.py) diff --git a/scripts/refresh_data_dirs.py b/scripts/refresh_data_dirs.py index 4c06b995..db8e4a1c 100644 --- a/scripts/refresh_data_dirs.py +++ b/scripts/refresh_data_dirs.py @@ -23,6 +23,7 @@ import tempfile import urllib.error import urllib.request import zipfile +from collections.abc import Iterator from pathlib import Path from artifacts import file_lock as _file_lock from common import platform_arg, require_platform, yaml_load @@ -61,7 +62,34 @@ def cache_lock( 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) + return _announced_lock(cache_dir.with_name(f".{cache_dir.name}.lock"), shared, cache_dir) + + +@contextlib.contextmanager +def _announced_lock(lock_path: Path, shared: bool, cache_dir: Path) -> Iterator[None]: + """file_lock, saying so when it has to wait. + + A pipeline holds every cache for its whole run: a refresh started beside + it waited hours with nothing on screen. + """ + try: + import fcntl + except ImportError: + with _file_lock(lock_path, shared=shared): + yield + return + lock_path.parent.mkdir(parents=True, exist_ok=True) + mode = fcntl.LOCK_SH if shared else fcntl.LOCK_EX + with open(lock_path, "a") as handle: + try: + fcntl.flock(handle, mode | fcntl.LOCK_NB) + except BlockingIOError: + log.warning("waiting for %s: another run holds it", cache_dir) + fcntl.flock(handle, mode) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) def hold_data_dirs(stack: contextlib.ExitStack, registry: dict) -> None: diff --git a/tests/test_artifact_lock.py b/tests/test_artifact_lock.py index 905db767..03f37a77 100644 --- a/tests/test_artifact_lock.py +++ b/tests/test_artifact_lock.py @@ -110,6 +110,36 @@ class PackLockCliTest(unittest.TestCase): self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr) self.assertIn("is in use by another run", proc.stdout) + def test_a_held_output_is_refused_before_the_data_refresh(self): + """The refresh waits for every run holding a data tree, and a + pipeline holds them all for its whole run: probed after the refresh, + a second build sat silent for hours instead of being refused.""" + from refresh_data_dirs import cache_lock # noqa: PLC0415 + + platforms = Path(self.dir) / "platforms" + platforms.mkdir() + for name in ("_registry.yml", "_shared.yml", "misterfpga.yml"): + shutil.copyfile(REPO_ROOT / "platforms" / name, platforms / name) + cache = Path(self.dir) / "cache" / "probe" + cache.mkdir(parents=True) + (platforms / "_data_dirs.yml").write_text( + "data_directories:\n probe:\n source_type: zip\n" + f" source_url: https://example.invalid/probe.zip\n" + f" local_cache: {cache}\n" + ) + output = Path(self.dir) / "out" + with artifact_lock(str(output)), cache_lock(cache): + try: + proc = subprocess.run( + [sys.executable, "scripts/generate_pack.py", "--platform", "misterfpga", + "--platforms-dir", str(platforms), "--output-dir", str(output)], + capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=120, + ) + except subprocess.TimeoutExpired: + self.fail("generate_pack waited on the data tree instead of refusing") + self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr) + self.assertIn("is in use by another run", proc.stdout) + def test_every_writing_mode_refuses_a_locked_output(self): """Two of seven modes took the lock, and five wrote straight into it. diff --git a/tests/test_refresh_data_dirs.py b/tests/test_refresh_data_dirs.py index d737c811..22f31c59 100644 --- a/tests/test_refresh_data_dirs.py +++ b/tests/test_refresh_data_dirs.py @@ -166,6 +166,33 @@ class EmulatorPackHoldsTheCache(unittest.TestCase): self.assertEqual(held, [True]) +class AWaitIsAnnounced(unittest.TestCase): + def test_a_held_tree_is_named_before_the_wait(self): + """A pipeline holds every tree for its run: the refresh beside it + waited with nothing on screen.""" + with tempfile.TemporaryDirectory() as tmp: + cache = Path(tmp) / "probe" + taken = threading.Event() + + def second_run() -> None: + with rdd.cache_lock(cache): + taken.set() + + with self.assertLogs(rdd.log, "WARNING") as logs: + with rdd.cache_lock(cache): + worker = threading.Thread(target=second_run) + worker.start() + for _ in range(100): + if logs.output: + break + threading.Event().wait(0.05) + self.assertFalse(taken.is_set()) + worker.join(10) + self.assertTrue(taken.is_set()) + self.assertIn("waiting for", logs.output[0]) + self.assertIn("probe", logs.output[0]) + + class DryRunReport(unittest.TestCase): """What a dry run says: why each entry would be fetched, and how many."""