fix: standalone writers take the repository lock

This commit is contained in:
Abdessamad Derraz committed 2026-11-06 03:42:09 +01:00
1 parent 639204712b
commit 1ed605969d
10 files changed
+152 -30

No files matched your search

+48 -15
View File
@@ -208,11 +208,16 @@ def _strip_timestamps(text: str) -> str:
class ArtifactLockBusy(RuntimeError):
"""Raised when another process already holds the artifact directory."""
# A run that holds an artifact directory for its whole duration names it
# here for the steps it spawns: they would otherwise be refused by it.
# A run that holds artifact directories for its whole duration names them
# here, os.pathsep apart, for the steps it spawns: they would otherwise be
# refused by it. A pipeline holds the repository root and its --output-dir.
HELD_LOCK_ENV = "RETROBIOS_HELD_ARTIFACT_DIR"
def _held_directories() -> list[str]:
return [d for d in os.environ.get(HELD_LOCK_ENV, "").split(os.pathsep) if d]
@contextlib.contextmanager
def artifact_lock(directory: str, exclusive: bool = True) -> Iterator[None]:
"""Serialize access to a shared artifact directory across processes.
@@ -229,7 +234,7 @@ def artifact_lock(directory: str, exclusive: bool = True) -> Iterator[None]:
return
os.makedirs(directory, exist_ok=True)
if os.environ.get(HELD_LOCK_ENV) == os.path.realpath(directory):
if os.path.realpath(directory) in _held_directories():
yield
return
lock_path = os.path.join(directory, ".lock")
@@ -256,16 +261,24 @@ def hold_artifact_lock(directory: str) -> Iterator[None]:
run can purge the packs one step built before the next step checks
them. The children this process spawns inherit the hold.
"""
with artifact_lock(directory):
previous = os.environ.get(HELD_LOCK_ENV)
os.environ[HELD_LOCK_ENV] = os.path.realpath(directory)
try:
yield
finally:
if previous is None:
os.environ.pop(HELD_LOCK_ENV, None)
else:
os.environ[HELD_LOCK_ENV] = previous
with artifact_lock(directory), publish_hold(directory):
yield
@contextlib.contextmanager
def publish_hold(directory: str) -> Iterator[None]:
"""Name a directory this process holds to the steps it spawns."""
previous = os.environ.get(HELD_LOCK_ENV)
os.environ[HELD_LOCK_ENV] = os.pathsep.join(
[*_held_directories(), os.path.realpath(directory)]
)
try:
yield
finally:
if previous is None:
os.environ.pop(HELD_LOCK_ENV, None)
else:
os.environ[HELD_LOCK_ENV] = previous
# What a pipeline run writes at fixed paths, whatever its --output-dir.
@@ -280,20 +293,40 @@ def hold_repository_outputs(root: str = ".") -> Iterator[None]:
on dist/: both rewrote database.json, and the second to reach the
manifest step found install/ held after minutes of work, or, when the
steps did not overlap, rewrote the committed manifests from its own
state. The lock file is the repository root's .lock.
state. The lock file is the repository root's .lock. A pipeline passes
it to its steps with publish_hold.
"""
stack = contextlib.ExitStack()
try:
stack.enter_context(artifact_lock(root))
except ArtifactLockBusy as exc:
raise ArtifactLockBusy(
f"another pipeline run is writing {', '.join(REPOSITORY_OUTPUTS)} "
f"another run is writing {', '.join(REPOSITORY_OUTPUTS)} "
f"(lock: {os.path.join(root, '.lock')}). Wait for it to finish."
) from exc
with stack:
yield
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
def repository_output_lock(path: str) -> contextlib.AbstractContextManager[None]:
"""The repository hold, when *path* is one of its fixed outputs.
Only a pipeline took it: generate_db.py run beside a pipeline replaced
database.json between its verify and its packs, and a manifest run on
install/ made the pipeline's own manifest step fail after minutes of
work. A path elsewhere (a scratch --output) takes nothing.
"""
target = os.path.realpath(path)
for output in REPOSITORY_OUTPUTS:
fixed = os.path.join(_REPO_ROOT, output.rstrip("/"))
if target == fixed or target.startswith(fixed + os.sep):
return hold_repository_outputs(_REPO_ROOT)
return contextlib.nullcontext()
def _build_timestamp(db: dict | None = None) -> str:
"""Timestamp for generated artifacts.
+3 -1
View File
@@ -3547,6 +3547,8 @@ from artifacts import ( # noqa: E402,F401
artifact_lock,
hold_artifact_lock,
hold_repository_outputs,
publish_hold,
repository_output_lock,
_TIMESTAMP_PATTERNS,
_strip_timestamps,
)
@@ -3588,7 +3590,7 @@ def run_cli(main: Callable[[], int | None]) -> None:
try:
code = main()
sys.stdout.flush()
except InputMissing as exc:
except (InputMissing, ArtifactLockBusy) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
sys.exit(1)
except BrokenPipeError:
+8 -1
View File
@@ -26,6 +26,7 @@ from common import (
DEFAULT_PROVENANCE_DIR,
list_registered_platforms,
load_provenance_snapshots,
repository_output_lock,
require_yaml,
set_plain_mode,
source_yaml_paths,
@@ -430,7 +431,11 @@ def main() -> int:
help="Directory with dump-catalog snapshots",
)
args = parser.parse_args()
with repository_output_lock(args.output):
return _generate(args)
def _generate(args: argparse.Namespace) -> int:
bios_dir = Path(args.bios_dir)
if not bios_dir.is_dir():
print(f"Error: BIOS directory '{bios_dir}' not found", file=sys.stderr)
@@ -742,4 +747,6 @@ def _collect_all_aliases(
if __name__ == "__main__":
sys.exit(main() or 0)
from common import run_cli
run_cli(main)
+7
View File
@@ -49,6 +49,7 @@ from common import ( # noqa: E402
load_emulator_profiles,
load_platform_config,
native_layer,
repository_output_lock,
run_cli,
)
from region import normalize_declared # noqa: E402
@@ -661,7 +662,13 @@ def main() -> int:
help="exit 1 when a file would change, naming what differs (implies --quiet)")
parser.add_argument("--quiet", action="store_true", help="print only failures")
args = parser.parse_args()
if args.check:
return _index(args)
with repository_output_lock(str(args.output_dir)):
return _index(args)
def _index(args: argparse.Namespace) -> int:
files, coverage = render(args.output_dir)
if not (args.quiet or args.check):
print_coverage(coverage, files)
+5 -2
View File
@@ -48,6 +48,7 @@ from common import (
profiles_for_systems,
write_text_atomic,
artifact_lock,
repository_output_lock,
ArtifactLockBusy,
build_target_cores_cache,
build_zip_contents_index,
@@ -2809,9 +2810,11 @@ def _run_manifest_mode(
@contextlib.contextmanager
def _pack_output_lock(output_dir: str, exclusive: bool = True) -> Iterator[None]:
"""Hold the output directory for the duration of a pack run."""
"""Hold the output directory for the duration of a pack run, and the
repository's fixed outputs when it is one of them (install/)."""
try:
with artifact_lock(output_dir, exclusive=exclusive):
with artifact_lock(output_dir, exclusive=exclusive), \
repository_output_lock(output_dir) if exclusive else contextlib.nullcontext():
yield
except ArtifactLockBusy as exc:
print(f"ERROR: {exc}", file=sys.stderr)
+5
View File
@@ -30,6 +30,7 @@ from common import (
load_emulator_profiles,
load_platform_config,
load_platform_registry,
repository_output_lock,
resolve_platform_cores,
system_index,
unique_emulator_profiles,
@@ -1034,7 +1035,11 @@ def main() -> None:
help="platform YAML directory (default: %(default)s)",
)
args = parser.parse_args()
with repository_output_lock("README.md"):
_write(args)
def _write(args: argparse.Namespace) -> None:
import contextlib
from common import load_data_dir_registry
+5 -2
View File
@@ -12080,11 +12080,14 @@ def main() -> None:
# files in the sweep, so the second is refused instead.
import contextlib
from common import load_data_dir_registry
from common import load_data_dir_registry, repository_output_lock
from refresh_data_dirs import hold_data_dirs
# The coverage pages resolve against data/, held while they are read.
with artifact_lock(args.docs_dir), contextlib.ExitStack() as held:
# Whatever --docs-dir says, website/src/data and wiki/ are written.
with artifact_lock(args.docs_dir), \
repository_output_lock(str(PUBLISHED_PARTS_FILE)), \
contextlib.ExitStack() as held:
hold_data_dirs(held, load_data_dir_registry(args.platforms_dir))
_generate(args)
except ArtifactLockBusy as exc:
+6 -3
View File
@@ -73,6 +73,7 @@ from common import (
hold_artifact_lock,
hold_repository_outputs,
load_data_dir_registry,
publish_hold,
)
from refresh_data_dirs import hold_data_dirs
@@ -1002,9 +1003,11 @@ def main() -> None:
# rewrote them beside the run on dist/.
held = contextlib.ExitStack()
try:
held.enter_context(
hold_repository_outputs(str(Path(__file__).resolve().parent.parent))
)
root = str(Path(__file__).resolve().parent.parent)
held.enter_context(hold_repository_outputs(root))
# Its steps write database.json, install/ and README.md themselves,
# and take the same lock to do so.
held.enter_context(publish_hold(root))
if not args.skip_packs:
held.enter_context(hold_artifact_lock(args.output_dir))
except ArtifactLockBusy as exc:
+10 -5
View File
@@ -735,13 +735,18 @@ def main() -> int:
if not args.fetch and not any(path.is_dir() for path in cache_dirs):
print(f"Error: no tree cache at {', '.join(map(str, cache_dirs))}", file=sys.stderr)
return 1
from common import load_emulator_profiles, write_if_changed # noqa: PLC0415
from common import ( # noqa: PLC0415
load_emulator_profiles,
repository_output_lock,
write_if_changed,
)
profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
held, unread = build_source_paths(profiles, cache_dirs, args.fetch)
held = keep_unread(held, unread, read_record(args.output))
args.output.parent.mkdir(parents=True, exist_ok=True)
write_if_changed(str(args.output), json.dumps(held, indent=1, sort_keys=True) + "\n")
with repository_output_lock(str(args.output)):
held, unread = build_source_paths(profiles, cache_dirs, args.fetch)
held = keep_unread(held, unread, read_record(args.output))
args.output.parent.mkdir(parents=True, exist_ok=True)
write_if_changed(str(args.output), json.dumps(held, indent=1, sort_keys=True) + "\n")
print(f"{sum(map(len, held.values()))} paths held in {len(held)} trees, "
f"{len(unread)} trees not cached: {args.output}")
return 0
+55 -1
View File
@@ -23,6 +23,7 @@ from common import ( # noqa: E402
artifact_lock,
hold_artifact_lock,
hold_repository_outputs,
publish_hold,
)
@@ -218,7 +219,7 @@ class PipelineHoldsItsFixedOutputs(unittest.TestCase):
capture_output=True, text=True, cwd=str(REPO_ROOT), timeout=120,
)
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertIn("another pipeline run is writing database.json", proc.stderr)
self.assertIn("another run is writing database.json", proc.stderr)
self.assertNotIn("generate database", proc.stdout.lower())
def test_two_holds_of_the_repository_exclude_each_other(self):
@@ -228,6 +229,59 @@ class PipelineHoldsItsFixedOutputs(unittest.TestCase):
pass
class StandaloneWritersHonourTheRepositoryHold(unittest.TestCase):
"""Only pipeline runs took the repository's lock: generate_db.py beside
a pipeline replaced database.json between its verify and its packs, and
a manifest run on install/ failed the pipeline's own manifest step."""
def _beside_a_pipeline(self, argv: list[str]) -> subprocess.CompletedProcess:
env = {k: v for k, v in os.environ.items() if k != "RETROBIOS_HELD_ARTIFACT_DIR"}
return subprocess.run(
[sys.executable, *argv], capture_output=True, text=True,
cwd=str(REPO_ROOT), timeout=300, env=env, check=False,
)
def test_a_writer_of_a_fixed_output_is_refused(self):
with hold_repository_outputs(str(REPO_ROOT)):
for argv in (
["scripts/generate_db.py", "--bios-dir", "bios", "--output", "database.json"],
["scripts/generate_pack.py", "--all", "--manifest", "--offline",
"--output-dir", "install"],
["scripts/generate_games_index.py", "--quiet"],
):
with self.subTest(script=argv[0]):
proc = self._beside_a_pipeline(argv)
self.assertEqual(proc.returncode, 1, proc.stdout[-500:] + proc.stderr[-500:])
self.assertIn("another run is writing database.json", proc.stderr)
self.assertNotIn("Traceback", proc.stderr)
def _child(self, target: str, env: dict) -> subprocess.CompletedProcess:
code = (
"import sys; sys.path.insert(0, 'scripts')\n"
"from common import artifact_lock, repository_output_lock\n"
f"with repository_output_lock({target!r}), artifact_lock({self.scratch!r}):\n"
" pass\n"
)
return subprocess.run([sys.executable, "-c", code], cwd=str(REPO_ROOT), env=env,
capture_output=True, text=True, timeout=60, check=False)
def test_a_step_the_pipeline_spawns_passes(self):
"""A pipeline holds the root and its --output-dir, and its steps
write both: the environment names the two, os.pathsep apart."""
self.scratch = _tempdir(self)
with hold_repository_outputs(str(REPO_ROOT)), publish_hold(str(REPO_ROOT)), \
hold_artifact_lock(self.scratch):
proc = self._child(str(REPO_ROOT / "install" / "games"), dict(os.environ))
self.assertEqual(proc.returncode, 0, proc.stderr)
def test_a_scratch_output_takes_nothing(self):
self.scratch = _tempdir(self)
env = {k: v for k, v in os.environ.items() if k != "RETROBIOS_HELD_ARTIFACT_DIR"}
with hold_repository_outputs(str(REPO_ROOT)):
proc = self._child(os.path.join(self.scratch, "database.json"), env)
self.assertEqual(proc.returncode, 0, proc.stderr)
class HeldLockTest(unittest.TestCase):
"""pipeline.py took the lock step by step: between the pack build and the
integrity check dist/ was free, another run's purge emptied it, and the