From 38f69c68089abe0a97d1b7cff653f2c1cd426097 Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Sat, 18 Jul 2026 17:28:09 -0500 Subject: [PATCH 1/6] Don't kill a healthy daemon that is slow to open its port laue_orchestrator waited on lsu.wait_for_port(), which only watches the socket and knows nothing about the daemon process. That failed in both directions: - A daemon that is merely slow to start was killed. Startup reads a multi-GB orientation database and then initialises a CUDA context; on a cold page cache or a contended GPU that can exceed the fixed 180 s budget, and the run aborted with "Daemon did not open port in time" despite nothing being wrong. Observed on a 7.2 GB / 100 M-orientation database: both phases reached "Initializing CUDA..." and were SIGKILLed at the 3-minute mark, losing the launch. The identical scan started cleanly on retry. - A daemon that died instantly (bad params, missing file, GPU error) still held the orchestrator for the full 180 s before reporting anything, and the reported message blamed a timeout rather than the actual exit. Replace it with _wait_for_daemon_port(), which polls the port and the process together: fail immediately with the daemon's exit code when it is genuinely dead, keep waiting while it is alive, and log progress every 30 s so a slow start is visible rather than silent. The timeout is retained as a backstop for an alive-but-wedged daemon and raised 180 s -> 900 s, with the error pointing at --port-timeout. Verified: dead daemon returns in 2.0 s with its exit code (was: 900 s); slow-but-alive daemon is waited for and succeeds; alive-but-silent daemon still honours the cap. --- scripts/laue_orchestrator.py | 71 +++++++++++++++++++++++++++++++++--- 1 file changed, 66 insertions(+), 5 deletions(-) diff --git a/scripts/laue_orchestrator.py b/scripts/laue_orchestrator.py index 717d6ed..196da8e 100755 --- a/scripts/laue_orchestrator.py +++ b/scripts/laue_orchestrator.py @@ -84,6 +84,67 @@ def _find_daemon_binary() -> str: ) +def _wait_for_daemon_port( + proc: subprocess.Popen, + port: int, + timeout: float, + poll_interval: float = 2.0, + progress_every: float = 30.0, +) -> bool: + """Wait for the daemon to open *port*, watching the process while we wait. + + ``lsu.wait_for_port()`` is process-blind, which fails in both directions: + + * A daemon that dies immediately (bad params, missing file, GPU error) still + holds the orchestrator for the whole timeout before it reports anything. + * A daemon that is merely *slow* gets killed. Startup reads a multi-GB + orientation database and then initialises a CUDA context; on a cold page + cache, a busy GPU, or a loaded machine that legitimately exceeds a short + fixed budget, and the run aborts with "Daemon did not open port in time" + even though nothing is wrong. + + Polling ``proc`` as well as the port fixes both: we fail fast with the exit + code when the daemon is genuinely dead, and we keep waiting (logging + progress) as long as it is alive. + + Returns True once the port is open, False if the daemon died or the timeout + elapsed while it was still running. + """ + t0 = time.time() + last_note = t0 + while True: + if lsu.is_port_open("127.0.0.1", port): + logger.info(f"Port {port} ready ({time.time() - t0:.1f}s)") + return True + + rc = proc.poll() + if rc is not None: + logger.error( + f"Daemon exited (code {rc}) after {time.time() - t0:.1f}s " + f"without opening port {port}. Check daemon log." + ) + return False + + now = time.time() + if now - t0 >= timeout: + logger.error( + f"Daemon is still running but has not opened port {port} after " + f"{timeout:.0f}s; giving up. If startup is legitimately this slow " + f"(very large orientation database, contended GPU), raise " + f"--port-timeout." + ) + return False + + if now - last_note >= progress_every: + logger.info( + f" still waiting for port {port} " + f"({now - t0:.0f}s elapsed, daemon alive)..." + ) + last_note = now + + time.sleep(poll_interval) + + def _terminate_process(proc: subprocess.Popen, name: str, timeout: float = 10.0) -> None: """Send SIGTERM, wait, then SIGKILL if necessary.""" if proc.poll() is not None: @@ -162,7 +223,7 @@ def run_pipeline( ncpus: int = 1, output_dir: str = "", port: int = lsu.LAUE_STREAM_PORT, - port_timeout: float = 180.0, + port_timeout: float = 900.0, flush_time: float = 5.0, min_unique: int = 2, write_indexfile: bool = True, @@ -185,7 +246,8 @@ def run_pipeline( ncpus: Number of CPUs (passed to daemon). output_dir: Output directory (auto-generated if empty). port: Daemon TCP port. - port_timeout: Max seconds to wait for daemon port. + port_timeout: Max seconds to wait for the daemon port while the + daemon is still alive (a dead daemon aborts immediately). flush_time: Seconds to wait after server finishes before killing daemon. min_unique: Minimum unique spots for orientation filtering. """ @@ -308,8 +370,7 @@ def run_pipeline( # --- 3. Wait for daemon port --- logger.info(f"Waiting for port {port}...") - if not lsu.wait_for_port("127.0.0.1", port, timeout=port_timeout): - logger.error("Daemon did not open port in time. Check daemon log.") + if not _wait_for_daemon_port(daemon_proc, port, port_timeout): _terminate_process(daemon_proc, "daemon") daemon_logf.close() _print_log_tail(daemon_log) @@ -570,7 +631,7 @@ def main() -> None: help=f"Daemon TCP port (default: {lsu.LAUE_STREAM_PORT})" ) parser.add_argument( - "--port-timeout", type=float, default=180.0, + "--port-timeout", type=float, default=900.0, help="Max seconds to wait for daemon port (default: 180)" ) parser.add_argument( From ec64ad4fc39ecd99f4cf426ba701002629a44675 Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Sat, 18 Jul 2026 18:14:32 -0500 Subject: [PATCH 2/6] run_laue.sh: make the documented WATCH="" actually batch a folder The CONFIG block says 'set WATCH="" to batch an existing folder', but the default was applied with ${WATCH:-"--watch"}, and the ":-" form substitutes the default when the variable is unset *or empty*. Setting WATCH="" therefore still produced --watch, so a run over an already-complete folder silently stayed in watch mode: it indexed every frame, then parked forever waiting for more, and never ran post-processing or wrote any output.h5 until someone noticed and touched STOP_LAUE by hand. Use ${WATCH-"--watch"} so an explicitly empty WATCH means batch mode while an unset WATCH still defaults to watching. Verified: WATCH="" -> "", unset -> "--watch". This matters for unattended batch processing, where nothing is watching to notice that a finished scan is sitting idle. --- scripts/pipeline/run_laue.sh | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/scripts/pipeline/run_laue.sh b/scripts/pipeline/run_laue.sh index e4d7f41..89b09c1 100755 --- a/scripts/pipeline/run_laue.sh +++ b/scripts/pipeline/run_laue.sh @@ -26,7 +26,10 @@ BETA_CONFIG=${BETA_CONFIG:-"$WORK/params/params_beta.txt"} # set BETA_CONFIG="" ALPHA_GPU=${ALPHA_GPU:-0}; ALPHA_PORT=${ALPHA_PORT:-60517} BETA_GPU=${BETA_GPU:-1}; BETA_PORT=${BETA_PORT:-60518} NCPUS=${NCPUS:-32} # CPU cores for the refinement stage -WATCH=${WATCH:-"--watch"} # set WATCH="" to batch an existing folder +# NB: "-" not ":-" — ${WATCH:-...} would substitute the default for an *empty* +# WATCH too, so the documented WATCH="" (batch an existing folder) never took +# effect and every run silently stayed in watch mode. +WATCH=${WATCH-"--watch"} # set WATCH="" to batch an existing folder # ----------------------------------------------------------------------------- FOLDER=${1:?usage: run_laue.sh DATA_FOLDER [H5_LOCATION]} From d6c81079b16fd411b4c12dbbdcefa3c060fb4077 Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Sat, 18 Jul 2026 21:43:16 -0500 Subject: [PATCH 3/6] Wait for the daemon to finish writing, not merely to have started MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The flush wait broke as soon as solutions.txt existed and was non-empty, slept 2 s, and SIGTERMed the daemon. But solutions.txt is appended to as each frame is fitted, so its existence means the FIRST frame landed, not the last. The daemon is routinely behind — it logs "receive queue was full N times (backpressure from processing)" — so terminating it two seconds after the first result silently discards everything still queued. Observed on a 6561-frame batch run: the last 31 frames were missing, contiguously, with no error in any log. The scan reported success. Watch-mode runs had been masking this, because a finished scan sits idle until someone touches STOP_LAUE, which happens to give the daemon minutes-to-hours to drain; in batch mode (WATCH="") there is no such pause and the tail is lost every time. Wait for solutions.txt to stop growing instead: that is the daemon actually draining. Quiescence window is max(flush_time, 10 s), with an hour-long backstop and a warning that explicitly says results may be truncated if it trips. Verified against a synthetic writer: the loop does not break while the file is still growing, and captures every byte written. --- scripts/laue_orchestrator.py | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/scripts/laue_orchestrator.py b/scripts/laue_orchestrator.py index 196da8e..a199aaa 100755 --- a/scripts/laue_orchestrator.py +++ b/scripts/laue_orchestrator.py @@ -430,19 +430,35 @@ def run_pipeline( logger.info(f"Image server exited (code {server_proc.returncode}). " f"Waiting for daemon to write results...") - # Poll for solutions.txt (mirrors integrator_batch_process.py pattern) - flush_deadline = time.time() + flush_time + 60 + # Wait for the daemon to finish WRITING, not merely to have started. + # + # solutions.txt is appended to as each frame is fitted, so "the file exists and + # is non-empty" means the *first* frame landed, not the last. Breaking there and + # SIGTERMing the daemon two seconds later silently discards everything still in + # its queue — and the daemon is routinely behind (it logs "receive queue was full + # N times (backpressure from processing)"). Observed: a 6561-frame batch run lost + # the last 31 frames, contiguously, with no error anywhere. + # + # Instead, wait until the file stops growing: that is the daemon actually draining. + quiet_needed = max(flush_time, 10.0) + flush_deadline = time.time() + flush_time + 3600 + last_size, last_change = -1, time.time() while time.time() < flush_deadline: - if os.path.isfile(solutions_file) and os.path.getsize(solutions_file) > 0: - logger.info(f"solutions.txt detected ({os.path.getsize(solutions_file)} bytes)") - time.sleep(2.0) # extra wait for file to be fully flushed - break if daemon_proc.poll() is not None: logger.info("Daemon exited on its own.") break + size = os.path.getsize(solutions_file) if os.path.isfile(solutions_file) else 0 + now = time.time() + if size != last_size: + last_size, last_change = size, now + elif size > 0 and (now - last_change) >= quiet_needed: + logger.info(f"solutions.txt quiescent at {size} bytes after " + f"{quiet_needed:.0f}s without growth — daemon has drained") + break time.sleep(1.0) else: - logger.warning("Timed out waiting for daemon to write solutions.txt") + logger.warning("Timed out waiting for the daemon to finish writing " + "solutions.txt; results may be truncated") # --- 7. Terminate daemon --- _terminate_process(daemon_proc, "daemon") From edca6504ea4c7022db4e137e304092e8212da4fd Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Mon, 20 Jul 2026 19:23:02 -0500 Subject: [PATCH 4/6] Guard frame_mapping with a lock: a JSON dump race silently truncated scans frame_mapping is written by BOTH the sender thread and the consumer thread, while the consumer also serialises it to JSON every save_interval frames. Nothing guarded it, so json.dump() iterated the dict while the other thread inserted into it and raised: RuntimeError: dictionary changed size during iteration laue_image_server.py _consumer_thread -> save_frame_mapping laue_stream_utils.py json.dump(mapping, f, indent=1) That exception killed the consumer thread. The server then shut down through its normal path and reported Done: 15955 sent, 0 skipped, 9599.3s total on a 40,401-frame scan -- a silent 61% truncation presented as success, with post-processing then run over the partial result. Nothing in the orchestrator or the logs flagged it as a failure. The window scales with frame count and with backpressure, which is why it went unnoticed on 6.5k-frame scans and appeared on a 40k-frame one that logged 15,933 queue-full events. Fixes: - add mapping_lock; take it around every frame_mapping mutation in both threads - periodic save takes a snapshot under the lock and writes the file outside it, and can no longer kill the consumer: a failed periodic save is now a warning, since the final save after the producer finishes is the authoritative one - final save also snapshots under the lock Verified with a threaded reproducer: unguarded, dumps raise RuntimeError while a writer mutates concurrently; with the lock, 400/400 dumps succeed. --- scripts/laue_image_server.py | 40 +++++++++++++++++++++++++++--------- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/scripts/laue_image_server.py b/scripts/laue_image_server.py index 0043023..e740fad 100755 --- a/scripts/laue_image_server.py +++ b/scripts/laue_image_server.py @@ -232,11 +232,18 @@ def _settled(path): send_error: list = [] # shared error flag frame_mapping: dict = {} + # frame_mapping is written by BOTH the sender thread and the consumer thread, and + # serialised to JSON by the consumer every save_interval frames. Without a lock, + # json.dump() iterates the dict while the other thread inserts into it and raises + # "RuntimeError: dictionary changed size during iteration". That killed the consumer + # thread mid-run, after which the server reported a normal "Done: N sent, 0 skipped" + # having silently processed only part of the scan (observed: 15,955 of 40,401). + mapping_lock = threading.Lock() sent_count_lock = threading.Lock() counters = {"sent": 0, "skip": 0} def _sender_thread(sock, send_q, send_error, frame_mapping, counters, - labels_h5f): + labels_h5f, mapping_lock): """Drain the queue and send frames over TCP.""" while True: item = send_q.get() @@ -265,7 +272,8 @@ def _sender_thread(sock, send_q, send_error, frame_mapping, counters, with sent_count_lock: counters["skip"] += 1 send_error.append(e) - frame_mapping[str(image_num)] = mapping_entry + with mapping_lock: + frame_mapping[str(image_num)] = mapping_entry send_q.task_done() # Open labels HDF5 for writing @@ -278,7 +286,8 @@ def _sender_thread(sock, send_q, send_error, frame_mapping, counters, sender = threading.Thread( target=_sender_thread, - args=(sock, send_q, send_error, frame_mapping, counters, labels_h5f), + args=(sock, send_q, send_error, frame_mapping, counters, labels_h5f, + mapping_lock), daemon=True, ) sender.start() @@ -368,10 +377,11 @@ def _consumer_thread(): reason = result["error"] if reason != "no_spots": logger.error(f" Frame {fr_idx} of {h5bn}: {reason}") - frame_mapping[str(img_num)] = { - "file": h5bn, "frame": fr_idx, - "skipped": True, "reason": reason, - } + with mapping_lock: + frame_mapping[str(img_num)] = { + "file": h5bn, "frame": fr_idx, + "skipped": True, "reason": reason, + } skip_count += 1 else: mapping_entry = { @@ -403,7 +413,15 @@ def _consumer_thread(): f"({rate:.1f} img/s, ETA {remaining:.0f}s)" ) if processed > 0 and processed % save_interval == 0: - lsu.save_frame_mapping(frame_mapping, mapping_file) + with mapping_lock: + snapshot = dict(frame_mapping) + try: + lsu.save_frame_mapping(snapshot, mapping_file) + except Exception as e: + # a periodic save is only for crash resilience; the final + # save after the producer finishes is the authoritative one. + # Never let it take down the consumer thread. + logger.warning(f" periodic mapping save failed: {e}") # Signal sender thread: no more frames send_q.put(None) @@ -424,8 +442,10 @@ def _consumer_thread(): send_q.put(None) # sentinel sender.join(timeout=60) - # Final mapping save - lsu.save_frame_mapping(frame_mapping, mapping_file) + # Final mapping save (authoritative) + with mapping_lock: + snapshot = dict(frame_mapping) + lsu.save_frame_mapping(snapshot, mapping_file) logger.info(f"Final mapping saved to {mapping_file}") # Close socket From abdcbc3b0c3b7de78fd659f93a6d2e87c7b01da6 Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Mon, 20 Jul 2026 19:23:57 -0500 Subject: [PATCH 5/6] run_laue.sh: pin CUDA_DEVICE_ORDER so GPU N means the card nvidia-smi calls N The launcher set CUDA_VISIBLE_DEVICES=$gpu without CUDA_DEVICE_ORDER. CUDA then enumerates by its default FASTEST_FIRST ordering, which need not match nvidia-smi indices on a host with mixed GPUs. Observed on a shared machine with two H200s (nvidia-smi 0,1) and two RTX PRO 6000s (2,3): a run configured for GPU 0/1 placed its daemon on physical GPU 2 -- a card already running another user's multi-day job -- while the intended H200s sat completely idle. Nothing in the logs indicated the substitution. Pin CUDA_DEVICE_ORDER=PCI_BUS_ID at launch so the configured index always selects the intended card. This matters most on shared systems, where the failure mode is quietly competing for someone else's GPU. --- scripts/pipeline/run_laue.sh | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/scripts/pipeline/run_laue.sh b/scripts/pipeline/run_laue.sh index 89b09c1..e5258f4 100755 --- a/scripts/pipeline/run_laue.sh +++ b/scripts/pipeline/run_laue.sh @@ -43,7 +43,13 @@ launch() { # launch PHASE CONFIG GPU PORT [ -z "$config" ] && return 0 [ -f "$config" ] || { echo "ERROR: config not found: $config" >&2; exit 1; } local out="$WORK/results/${phase}_$TS" - CUDA_VISIBLE_DEVICES=$gpu setsid nohup "$PY" "$SCRIPTS/laue_orchestrator.py" \ + # CUDA_DEVICE_ORDER=PCI_BUS_ID is required for $gpu to mean the same card that + # nvidia-smi calls $gpu. CUDA defaults to FASTEST_FIRST, so on a mixed-GPU host + # "GPU 0" can land on a completely different physical card -- observed on a shared + # machine where the daemon landed on another user's GPU while the intended cards + # sat idle. + CUDA_DEVICE_ORDER=PCI_BUS_ID CUDA_VISIBLE_DEVICES=$gpu \ + setsid nohup "$PY" "$SCRIPTS/laue_orchestrator.py" \ --config "$config" --folder "$FOLDER" --h5-location "$H5LOC" \ --ncpus "$NCPUS" --port "$port" $WATCH --output-dir "$out" \ > "$WORK/results/${phase}_$TS.launch.log" 2>&1 < /dev/null & From 291ecc9a8c357b46da4fa425108233d32b7eacd8 Mon Sep 17 00:00:00 2001 From: Hemant Sharma Date: Thu, 23 Jul 2026 11:07:17 -0500 Subject: [PATCH 6/6] Add regression tests for the five streaming-pipeline fixes Locks in all five fixes in this PR. Ten tests, no GPU required: port wait fails fast on a dead daemon (was: waited the full timeout); waits while alive; honours the cap when the port never opens WATCH="" expands to batch mode; unset still defaults to --watch drain orchestrator waits for solutions.txt to go quiescent, and warns rather than silently proceeding if it gives up race frame_mapping mutations and the dump are lock-guarded (source contract); the exact "changed size during iteration" RuntimeError is reproduced deterministically; the snapshot-under-lock pattern is shown immune under concurrency GPU order run_laue.sh sets CUDA_DEVICE_ORDER on the same line as CUDA_VISIBLE_DEVICES The two reachable functions (port wait) are exercised directly; the inline logic (drain, race) is pinned with source contracts that fail if the construct is reverted, plus a deterministic behavioural test for the race. --- scripts/tests/test_streaming_regressions.py | 233 ++++++++++++++++++++ 1 file changed, 233 insertions(+) create mode 100644 scripts/tests/test_streaming_regressions.py diff --git a/scripts/tests/test_streaming_regressions.py b/scripts/tests/test_streaming_regressions.py new file mode 100644 index 0000000..74185bd --- /dev/null +++ b/scripts/tests/test_streaming_regressions.py @@ -0,0 +1,233 @@ +"""Regression tests for five silent-failure bugs in the streaming pipeline. + +All five were found while running ~150,000 Laue frames through the pipeline across +four machines during the July 2026 34-ID-E beam time. Four share a signature: the +failure reported success, so partial or misplaced work looked like a clean run. Each +test below locks in one fix; the module docstring of each fix and PR #5 have the full +story. + +Where the fixed logic is a reachable function it is exercised directly; where it is +inline in a long driver it is pinned with a source contract (which still fails if the +construct is reverted) plus, for the concurrency bug, a behavioural reproduction of the +race the fix removes. + +Run: + cd ~/opt/LaueMatching && python -m pytest scripts/tests/test_streaming_regressions.py -v +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import threading +import time +from pathlib import Path + +import pytest + +# macOS CI sometimes links two OpenMP runtimes via numpy/scipy; harmless here. +os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE") + +_SCRIPTS = Path(__file__).resolve().parent.parent +if str(_SCRIPTS) not in sys.path: + sys.path.insert(0, str(_SCRIPTS)) + +_RUN_LAUE = _SCRIPTS / "pipeline" / "run_laue.sh" +_ORCH_SRC = (_SCRIPTS / "laue_orchestrator.py").read_text() +_SERVER_SRC = (_SCRIPTS / "laue_image_server.py").read_text() + + +def _free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +# --------------------------------------------------------------------------- +# Bug 1: the port wait killed a healthy-but-slow daemon and waited the full +# timeout on a dead one. _wait_for_daemon_port polls the process too. +# --------------------------------------------------------------------------- + +class _FakeAlive: + def poll(self): + return None + + +def test_port_wait_fails_fast_when_daemon_dies(): + """A daemon that exits without opening the port must be reported at once, + with its exit code — not after the whole timeout (old behaviour: 180 s+).""" + import laue_orchestrator as lo + + proc = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(3)"]) + t0 = time.time() + ok = lo._wait_for_daemon_port(proc, _free_port(), timeout=900.0, poll_interval=0.2) + dt = time.time() - t0 + assert ok is False + assert dt < 15, f"took {dt:.1f}s; should fail fast on a dead daemon" + + +def test_port_wait_keeps_waiting_while_daemon_is_alive(): + """A slow-but-alive daemon must be waited for, not killed at a fixed budget.""" + import laue_orchestrator as lo + + port = _free_port() + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + + def open_late(): + time.sleep(4) + srv.bind(("127.0.0.1", port)) + srv.listen(1) + + threading.Thread(target=open_late, daemon=True).start() + alive = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + t0 = time.time() + ok = lo._wait_for_daemon_port(alive, port, timeout=30.0, + poll_interval=0.2, progress_every=2.0) + dt = time.time() - t0 + assert ok is True + assert dt >= 3.5, "must not return before the port actually opens" + finally: + alive.kill() + srv.close() + + +def test_port_wait_honours_cap_when_never_opens(): + """An alive daemon that never opens the port must still give up at the cap.""" + import laue_orchestrator as lo + + hang = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + t0 = time.time() + ok = lo._wait_for_daemon_port(hang, _free_port(), timeout=3.0, + poll_interval=0.2, progress_every=1.0) + dt = time.time() - t0 + assert ok is False + assert 2.5 <= dt < 12 + finally: + hang.kill() + + +# --------------------------------------------------------------------------- +# Bug 2: run_laue.sh ignored WATCH="". ${WATCH:-default} substitutes for an +# empty value too, so the documented batch mode never took effect. +# --------------------------------------------------------------------------- + +def _watch_expansion(env_value): + """Return what run_laue.sh's WATCH expansion yields for a given WATCH env.""" + line = [l for l in _RUN_LAUE.read_text().splitlines() + if l.strip().startswith("WATCH=")][0].split("#")[0].strip() + # e.g. WATCH=${WATCH-"--watch"} + script = f'{("WATCH="+chr(39)+env_value+chr(39)+"; ") if env_value is not None else ""}{line}; printf "%s" "$WATCH"' + out = subprocess.run(["bash", "-c", script], capture_output=True, text=True) + return out.stdout + + +def test_watch_empty_means_batch(): + """WATCH="" must yield an empty flag (batch mode), not --watch.""" + assert _watch_expansion("") == "", ( + "WATCH='' still expands to --watch; batch mode silently stays in watch mode" + ) + + +def test_watch_unset_still_defaults_to_watch(): + """An unset WATCH must still default to --watch (live mode).""" + assert _watch_expansion(None) == "--watch" + + +# --------------------------------------------------------------------------- +# Bug 3: the daemon was terminated as soon as solutions.txt EXISTED, before it +# finished writing, losing the tail of the scan. The fix waits for the +# file to stop growing. (Logic is inline in run_pipeline; pin by contract.) +# --------------------------------------------------------------------------- + +def test_drain_waits_for_quiescence_not_mere_existence(): + """The flush wait must key on the file no longer growing, not on it existing.""" + # the old, buggy predicate broke as soon as the file was non-empty + assert 'os.path.getsize(solutions_file) > 0:' not in _ORCH_SRC or \ + "quiescent" in _ORCH_SRC, ( + "orchestrator appears to break on solutions.txt existence again; " + "it must wait for the file to stop growing" + ) + assert "quiescent" in _ORCH_SRC, "drain-quiescence logic missing" + # and it must warn (not silently proceed) if it gives up while still growing + assert "may be truncated" in _ORCH_SRC + + +# --------------------------------------------------------------------------- +# Bug 4: frame_mapping was mutated by two threads while a third serialised it, +# raising "dictionary changed size during iteration", which killed the +# consumer and left the server reporting a truncated scan as success. +# --------------------------------------------------------------------------- + +def test_frame_mapping_mutations_are_lock_guarded(): + """Every frame_mapping write, and the JSON dump, must hold the lock.""" + assert "mapping_lock = threading.Lock()" in _SERVER_SRC, "mapping_lock missing" + # the dump must operate on a snapshot taken under the lock, never the live dict + assert "with mapping_lock:" in _SERVER_SRC + assert "snapshot = dict(frame_mapping)" in _SERVER_SRC, ( + "JSON dump must use a snapshot taken under the lock, not the live dict" + ) + + +def test_mutating_a_dict_during_iteration_raises(): + """The exact failure mode the fix removes, made deterministic: mutating a dict + while it is being iterated (which is what json.dump does) raises RuntimeError.""" + # json.dump(mapping, f, indent=1) iterates the dict's items; a concurrent insert + # from the sender thread mid-iteration is exactly this RuntimeError. Reproduced + # deterministically here without threads. + d = {i: i for i in range(100)} + with pytest.raises(RuntimeError, match="changed size during iteration"): + for _ in d: + d[len(d)] = 1 + + +def test_snapshot_under_lock_is_immune(): + """The fix pattern — copy the dict under the lock, dump the copy — is safe even + while the live dict keeps changing. This is deterministic: it must always pass.""" + live, lock, stop = {}, threading.Lock(), threading.Event() + + def churn(): + i = 0 + while not stop.is_set() and i < 500_000: + with lock: + live[str(i)] = {"frame": i} + i += 1 + + th = threading.Thread(target=churn, daemon=True) + th.start() + try: + for _ in range(500): + with lock: + snapshot = dict(live) # exactly what the consumer thread does + json.dumps(snapshot) # never touches the live, mutating dict + finally: + stop.set() + th.join(timeout=2) + + +# --------------------------------------------------------------------------- +# Bug 5: CUDA_VISIBLE_DEVICES without CUDA_DEVICE_ORDER selects by CUDA's +# FASTEST_FIRST ordering, which need not match nvidia-smi indices. +# --------------------------------------------------------------------------- + +def test_run_laue_pins_cuda_device_order(): + """The launch line must set CUDA_DEVICE_ORDER=PCI_BUS_ID so the configured GPU + index selects the card nvidia-smi calls that index.""" + src = _RUN_LAUE.read_text() + assert "CUDA_DEVICE_ORDER=PCI_BUS_ID" in src, ( + "run_laue.sh selects a GPU without pinning CUDA_DEVICE_ORDER; on a mixed-GPU " + "host the job can land on the wrong card" + ) + # the two env vars must share a line so the ordering applies to the selection + # (setsid/the command may continue on the next physical line via a trailing backslash) + dev = [l for l in src.splitlines() if "CUDA_VISIBLE_DEVICES=" in l] + assert dev and "CUDA_DEVICE_ORDER=PCI_BUS_ID" in dev[0], ( + "CUDA_DEVICE_ORDER must sit on the same line as CUDA_VISIBLE_DEVICES" + )