diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a092b61..aa6d1cb 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -36,3 +36,9 @@ jobs: docker run --rm --read-only --tmpfs /tmp \ -e BLIMMP_CACHE_DIR=/tmp/blimmp-cache \ blimmp:ci python /app/tests/smoke_test.py + + # Nextflow runs one BLIMMP task per genome, so several processes extract + # the module graphs at once. That raced and killed half the tasks on a + # 9-genome run. + - name: Concurrent extraction + run: docker run --rm blimmp:ci python /app/tests/concurrency_test.py diff --git a/BLIMMP_Scripts/module_detection.py b/BLIMMP_Scripts/module_detection.py index e28cafa..f423075 100644 --- a/BLIMMP_Scripts/module_detection.py +++ b/BLIMMP_Scripts/module_detection.py @@ -2255,21 +2255,45 @@ def _extract_module_graphs(zip_path, destination) -> None: The archive was built on macOS, so it carries __MACOSX metadata and wraps everything in a redundant top-level folder. Both are flattened away. The source zip is deliberately left in place so extraction stays repeatable. + + Extraction happens in a private staging directory that is then renamed into + position, because several BLIMMP processes on one machine share a cache + directory and will race here. Extracting straight into `destination` let + them tear up each other's files: one process would rmtree __MACOSX or + rmdir the nested folder while another was still reading from it, and the + OSError that followed looked to the caller like an unwritable destination. + A rename onto a missing or empty directory is atomic on POSIX, so the first + process to finish wins and the rest adopt its copy. """ print(f"Extracting {os.path.basename(str(zip_path))} to {destination} ...") destination = Path(destination) - with zipfile.ZipFile(str(zip_path), "r") as z: - z.extractall(str(destination)) - - macosx_path = destination / "__MACOSX" - if macosx_path.is_dir(): - shutil.rmtree(str(macosx_path)) - - nested = destination / destination.name - if nested.is_dir(): - for item in os.listdir(str(nested)): - shutil.move(str(nested / item), str(destination / item)) - nested.rmdir() + parent = destination.parent + parent.mkdir(parents=True, exist_ok=True) + + staging = Path(tempfile.mkdtemp(prefix=f".{destination.name}.", dir=str(parent))) + try: + with zipfile.ZipFile(str(zip_path), "r") as z: + z.extractall(str(staging)) + + macosx_path = staging / "__MACOSX" + if macosx_path.is_dir(): + shutil.rmtree(str(macosx_path)) + + nested = staging / destination.name + if nested.is_dir(): + for item in os.listdir(str(nested)): + shutil.move(str(nested / item), str(staging / item)) + nested.rmdir() + + try: + os.replace(str(staging), str(destination)) + except OSError: + # Another process finished first and its directory is non-empty, so + # the rename is refused. Its graphs are as good as ours. + if not _graphs_present(destination): + raise + finally: + shutil.rmtree(str(staging), ignore_errors=True) def validate_paths(paths: "Paths") -> None: diff --git a/tests/concurrency_test.py b/tests/concurrency_test.py new file mode 100644 index 0000000..450e7fc --- /dev/null +++ b/tests/concurrency_test.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +"""Several BLIMMP processes must be able to extract the module graphs at once. + +Nextflow dispatches one BLIMMP task per genome, so a run with N genomes puts N +processes on a node at the same time. When the graphs are not already +extracted, every one of them extracts into the same directory. + +Extracting straight into that shared directory made them destroy each other's +work: one process removed __MACOSX or flattened the nested folder while another +was still reading from it, and the resulting OSError looked to the caller like +an unwritable destination. On a 9-genome run, 4 tasks died with a FATAL +"failed to extract" before any analysis ran. A single-process test cannot see +this, which is how it shipped. +""" + +import os +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +WORKERS = 8 +EXPECTED_GRAPHS = 340 + +CHILD = """ +import sys +from BLIMMP_Scripts.module_detection import ensure_module_graphs +print(ensure_module_graphs(sys.argv[1])) +""" + + +def main() -> int: + import BLIMMP_Scripts + + packaged = Path(BLIMMP_Scripts.__file__).parent / "Graph_Dependencies" + zip_name = "KEGG_Graphs_Generated_March26.zip" + if not (packaged / zip_name).is_file(): + print(f"FAIL: {zip_name} is not in the installed package at {packaged}") + return 1 + + scratch = Path(tempfile.mkdtemp(prefix="blimmp-concurrency-")) + try: + # A Graph_Dependencies directory holding only the archive, so every + # worker has to extract rather than finding graphs already in place. + graph_dir = scratch / "Graph_Dependencies" + graph_dir.mkdir() + shutil.copy2(packaged / zip_name, graph_dir) + + procs = [ + subprocess.Popen( + [sys.executable, "-c", CHILD, str(graph_dir)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for _ in range(WORKERS) + ] + results = [(p.wait(), *p.communicate()) for p in procs] + + failed = [(i, err or out) for i, (rc, out, err) in enumerate(results) if rc != 0] + if failed: + print(f"FAIL: {len(failed)}/{WORKERS} workers exited non-zero") + for i, msg in failed[:3]: + last = msg.strip().splitlines()[-1] if msg.strip() else "(no output)" + print(f" worker {i}: {last}") + return 1 + + extracted = graph_dir / "KEGG_Graphs_Generated_March26" + graphs = list(extracted.glob("module_*_nodes.json")) + if len(graphs) != EXPECTED_GRAPHS: + print(f"FAIL: expected {EXPECTED_GRAPHS} module graphs, found {len(graphs)}") + return 1 + + # Every worker must agree on where the graphs ended up. The extracting + # worker also prints a progress line, so read the path off the last one. + returned = {out.strip().splitlines()[-1] for _, out, _ in results if out.strip()} + if returned != {str(extracted)}: + print(f"FAIL: workers disagreed on the graph directory: {sorted(returned)}") + return 1 + + # A crashed or abandoned extraction leaves its staging directory behind. + leftovers = [p.name for p in graph_dir.iterdir() if p.name.startswith(".")] + if leftovers: + print(f"FAIL: staging directories left behind: {leftovers}") + return 1 + + print(f"PASS: {WORKERS} concurrent workers, {len(graphs)} graphs, no leftovers") + return 0 + finally: + shutil.rmtree(scratch, ignore_errors=True) + + +if __name__ == "__main__": + sys.exit(main())