From 39462303eb9f079603d6cf4bfca4b3d73223911e Mon Sep 17 00:00:00 2001 From: lmoresi Date: Sat, 22 Aug 2026 13:55:53 +1000 Subject: [PATCH 1/3] Diagnose a hung parallel job: per-rank watchdog and a roll-call report A collective entered by only some ranks leaves N-1 ranks stopped in the same frame and one somewhere else. That signature exists only ACROSS the set, so no rank can see it, and the job sits there until the queue kills it with nothing in the output to say which rank went the other way. Two pieces. `uw.mpi.watch` arms a per-rank watchdog: nothing collective, since by the time anything is stuck the ranks have already split and a probe needing all of them can only make it worse. Every rank times itself and dumps alone. `uw.mpi.ranks_agree` is the positive audit, comparing a label so it also catches ranks that arrive together but by different routes, and naming them. `underworld3.utilities.hang_report` does the comparison afterwards, which is the part that turns four stack traces into an answer: UW_HANG_WATCHDOG=120 mpirun -n 4 python myrun.py python -m underworld3.utilities.hang_report uw-hang-dumps ranks [0, 2, 3] stopped at model.py:13 in reduce_the_count rank [1] stopped at model.py:20 in => 3 of 4 ranks are waiting together at model.py:13, and ranks [1] are somewhere else. That frame is where the collective is; ranks [1] are where the bug is. Arming comes from the environment rather than a call in the script, so it is in force during import and mesh construction -- a rank that diverges before reaching a `watch()` call reports nothing. Three measurements shaped this. A Python `threading.Timer` does NOT run while the main thread is inside MPI: at np=4 against a 4 s block in `comm.allreduce`, a re-arming 0.5 s timer fired zero times on the blocked ranks while faulthandler produced all 7 dumps, so reporting has to be in C and the destination has to be a file descriptor, not a buffer. The main thread is identified by its outermost frame, not its depth, after a telemetry thread twelve frames into `requests` was reported as a rank's position. And a rank is placed by the modal position over its recent dumps, not its last one, which had put one of four waiting ranks in a group of its own. Tested by subprocess rather than in-process. faulthandler's `dump_traceback_ later` is a single global slot that pytest's own plugin already owns, and a rank inside a test framework is one process looking at itself -- the one thing this analysis cannot be done from. The end-to-end test runs a real four-rank job that really hangs, lets mpirun kill it, and checks the verdict. Underworld development team with AI support from Claude Code --- src/underworld3/__init__.py | 1 + src/underworld3/mpi.py | 398 +++++++++++++++++++++++ src/underworld3/utilities/hang_report.py | 273 ++++++++++++++++ tests/test_0053_hang_watchdog.py | 206 ++++++++++++ tests/test_0054_hang_report.py | 200 ++++++++++++ 5 files changed, 1078 insertions(+) create mode 100644 src/underworld3/utilities/hang_report.py create mode 100644 tests/test_0053_hang_watchdog.py create mode 100644 tests/test_0054_hang_report.py diff --git a/src/underworld3/__init__.py b/src/underworld3/__init__.py index 450e23680..8da83e480 100644 --- a/src/underworld3/__init__.py +++ b/src/underworld3/__init__.py @@ -182,6 +182,7 @@ def view(): # Needed everywhere import underworld3.mpi from .mpi import pprint, selective_ranks, collective_operation, CollectiveOperationError +from .mpi import watch, unwatch, watching, checkpoint, ranks_agree from ._var_types import * from .utilities._petsc_tools import * from .utilities._nb_tools import * diff --git a/src/underworld3/mpi.py b/src/underworld3/mpi.py index 5f3cdf152..3e1e903d3 100644 --- a/src/underworld3/mpi.py +++ b/src/underworld3/mpi.py @@ -22,10 +22,14 @@ """ from mpi4py import MPI as _MPI +import atexit as _atexit +import faulthandler as _faulthandler import os as _os import secrets as _secrets import sys as _sys import io as _io +import threading as _threading +import time as _time from contextlib import contextmanager as _contextmanager @@ -254,6 +258,12 @@ def wrapper(*args, **kwargs): ) raise CollectiveOperationError(error_msg) + # Entering a declared collective is exactly the progress the watchdog + # wants to hear about, and it gives the report a label without the + # caller having to place one. Costs a single test when disarmed. + if _watchdog is not None: + _watchdog.arm(f"entering {func.__name__}() [collective]") + return func(*args, **kwargs) wrapper.__name__ = func.__name__ @@ -262,6 +272,332 @@ def wrapper(*args, **kwargs): return wrapper +# --------------------------------------------------------------------------- +# Hang watchdog +# --------------------------------------------------------------------------- +# +# When ranks diverge over a collective, the ones that arrived block inside MPI +# and the job sits there until the queue kills it. Nothing in the output says +# which rank went the other way, or from where. +# +# The detector for that must NOT be collective. By the time anything is stuck +# the ranks have already split, so a probe that needs all of them can only make +# it worse -- it either blocks alongside the others or it becomes one more +# collective for the divergent rank to miss. So this is a plain per-rank timer. +# Every rank arms one, every rank reports on its own, and the diagnosis is the +# comparison between reports: three ranks in `allreduce`, one somewhere else. +# +# The reporting has to survive the main thread being inside MPI, and a Python +# thread does not: measured at np=4 against a 4 s block in `comm.allreduce`, +# a re-arming `threading.Timer` set to 0.5 s fired ZERO times on the blocked +# ranks -- the interpreter lock is held for the duration -- while +# `faulthandler.dump_traceback_later` produced all 7 expected dumps. It is +# written in C for exactly this case and does not need the lock. +# +# So faulthandler is the mechanism, and the price is that it writes to a file +# DESCRIPTOR: the destination must be a real file or stream, never a buffer. +# The Python timer is kept alongside it, because it is the only one that can +# print the checkpoint LABEL, and it does run for the many hangs that are not +# inside MPI -- a spin in Python, a stuck read, a solve that releases the lock. + +_watchdog = None +_watchdog_lock = _threading.Lock() + + +def _stack_dump(): + """Every thread's Python stack, main thread first. + + Formatted here rather than through :mod:`faulthandler` because that writes + to a file descriptor, so it cannot report into a pipe, a string buffer or + anything a test can read back. ``sys._current_frames`` needs the + interpreter lock, which mpi4py releases around blocking calls -- so a rank + sitting in ``allreduce`` does still report. If some extension ever holds + the lock through a block, nothing running in Python can report on it. + """ + import traceback + + main = _threading.main_thread().ident + frames = _sys._current_frames() + names = {t.ident: t.name for t in _threading.enumerate()} + + out = [] + for ident in sorted(frames, key=lambda i: (i != main, i)): + who = "MainThread (this is the one that is stuck)" if ident == main \ + else names.get(ident, "unknown") + out.append(f"\n --- thread {ident}: {who} ---") + out.extend( + " " + line.rstrip() + for line in traceback.format_stack(frames[ident]) + ) + return "\n".join(out) + + +class _Watchdog: + """A per-rank timer that reports where this rank is when it stops moving.""" + + def __init__(self, seconds, stream, abort): + self.seconds = float(seconds) + self.stream = stream + self.abort = bool(abort) + self.timer = None + # Set by cancel(). A report already running when the watchdog is + # disarmed must not re-arm behind it: the caller is about to close the + # stream, and faulthandler holds the DESCRIPTOR, so the next dump + # lands in whatever inherits it. + self.cancelled = False + self.label = "watch() -- no checkpoint reached yet" + self.since = _time.monotonic() + self.reports = 0 + + try: + stream.fileno() + except Exception as not_a_file: + raise ValueError( + "the watchdog writes through faulthandler, which needs a real " + "file descriptor, so `stream` cannot be a StringIO or other " + "in-memory buffer. Pass sys.stderr, or a file you opened. " + "This is not a detail to work around: a Python-level fallback " + "cannot report a rank blocked inside MPI, which is the case " + "the watchdog exists for." + ) from not_a_file + + print( + f"=== UW watchdog armed: rank {rank} of {size}, " + f"pid {_os.getpid()}, limit {self.seconds:g} s ===", + file=stream, + flush=True, + ) + + def arm(self, label=None, resume=False): + # `resume` is the deliberate re-arm of a watchdog that was cancelled on + # purpose -- restoring an outer one after a nested `watching` block. + # Without it the `cancelled` latch, which exists to stop an in-flight + # report re-arming behind unwatch(), would also silence the restore. + if resume: + self.cancelled = False + if self.cancelled: + return + if label is not None: + self.label = label + self.since = _time.monotonic() + + # The mechanism. Re-arming resets the countdown, so a job that keeps + # checking in never reaches it. `repeat` keeps it dumping once stuck: + # two identical stacks a minute apart say "stuck", one says "slow". + _faulthandler.dump_traceback_later( + self.seconds, repeat=True, file=self.stream, exit=self.abort + ) + + # Secondary, and only for hangs that leave the interpreter lock free. + # It adds the checkpoint label, which faulthandler cannot know about. + if self.timer is not None: + self.timer.cancel() + self.timer = _threading.Timer(self.seconds, self.report) + self.timer.daemon = True + self.timer.start() + + def cancel(self): + self.cancelled = True + _faulthandler.cancel_dump_traceback_later() + if self.timer is not None: + self.timer.cancel() + self.timer = None + + def report(self): + """The labelled report, when this thread can get the interpreter lock. + + Silent on a rank blocked inside MPI -- faulthandler covers that one. + """ + self.reports += 1 + stalled = _time.monotonic() - self.since + print( + f"\n=== UW HANG WATCHDOG: rank {rank} of {size}, pid {_os.getpid()}" + f" ===\n" + f" no progress for {stalled:.1f} s (limit {self.seconds:g} s)\n" + f" last checkpoint: {self.label}\n" + f" Compare this stack with the other ranks': the one in a " + f"different\n" + f" place is the rank that missed the collective.\n" + f"{_stack_dump()}", + file=self.stream, + flush=True, + ) + if not self.abort: + self.arm() # a no-op once cancel() has run + + +def watch(seconds=300, stream=None, abort=False): + """Arm a per-rank watchdog that reports where this rank is when it hangs. + + Call it once, high up -- at the top of a script, or in a test fixture. + Thereafter :func:`checkpoint` re-arms it, and every function carrying + :func:`collective_operation` re-arms it automatically, so a normally + progressing job never fires. + + Nothing here is collective, which is the point: a rank that has missed a + collective is exactly the rank that cannot take part in a probe. Each rank + times itself and reports alone. + + Parameters + ---------- + seconds : float + How long without a checkpoint counts as stuck. Set it well above the + slowest legitimate phase -- a coarse solve or a mesh generation can + take minutes and is not a hang. + stream : file, optional + Where the report goes. Defaults to ``sys.stderr``. It must be a real + file or stream --- an in-memory buffer is refused, because the dump + goes through :mod:`faulthandler` and that writes to a file descriptor. + Under ``mpirun`` the ranks interleave, so for anything but a small job + give each rank its own file, or use ``mpirun --output-filename``. + abort : bool + Exit the process straight after dumping. Useful in CI, where the + alternative is the job sitting until the wall clock kills it with no + output at all. Off by default: it kills the job. + + Returns + ------- + float + The timeout in force. + + Warnings + -------- + Call :func:`unwatch` before closing *stream*. faulthandler holds the file + DESCRIPTOR, not the Python object, so an armed watchdog over a closed file + writes into whatever that descriptor is reused for next. It is the default + ``sys.stderr`` that makes this harmless most of the time; a log file you + open yourself needs the disarm. + + Example + ------- + >>> uw.mpi.watch(seconds=120) # doctest: +SKIP + >>> for step in range(100): # doctest: +SKIP + ... uw.mpi.checkpoint(f"step {step}") + ... stokes.solve() + """ + global _watchdog + if stream is None: + stream = _sys.stderr + with _watchdog_lock: + if _watchdog is not None: + _watchdog.cancel() + _watchdog = _Watchdog(seconds, stream, abort) + _watchdog.arm() + return float(seconds) + + +def unwatch(): + """Disarm the watchdog. Safe to call when it was never armed.""" + global _watchdog + with _watchdog_lock: + if _watchdog is not None: + _watchdog.cancel() + _watchdog = None + + +def checkpoint(label=None): + """Tell the watchdog this rank is still moving, and where it is. + + A local timer reset -- no communication, and a single test when the + watchdog is not armed, so it is safe to leave in production code. + + The *label* is what makes the eventual report readable: it is printed + alongside the stack, so "rank 3 last saw `step 12 / before adapt`" while + the others saw `step 13` localises the divergence to one iteration. + """ + if _watchdog is not None: + _watchdog.arm(label) + + +@_contextmanager +def watching(seconds=300, stream=None, abort=False): + """:func:`watch` for the duration of a block, then restore what was there. + + >>> with uw.mpi.watching(seconds=60): # doctest: +SKIP + ... mesh.adapt(metric, max_levels=2) + """ + global _watchdog + with _watchdog_lock: + previous = _watchdog + if previous is not None: + previous.cancel() + _watchdog = None + watch(seconds=seconds, stream=stream, abort=abort) + try: + yield + finally: + unwatch() + with _watchdog_lock: + _watchdog = previous + if previous is not None: + previous.arm(resume=True) + + +@collective_operation +def ranks_agree(label, verbose=False): + """Check that every rank reached this point by the same route. COLLECTIVE. + + The watchdog is a post-mortem: it tells you where a job stopped. This is + the positive audit -- put it after a phase you suspect and it either passes + or names the ranks that took a different path. + + It compares *label* across ranks, so it catches divergence the barrier + alone cannot: ranks that all arrive, but from different branches. Give it a + label that varies with the path taken, not a constant. + + Being collective, it can itself hang if a rank never arrives -- that case + is the watchdog's, and the two are meant to be used together. + + Parameters + ---------- + label : str + What this rank believes it just did. Compared verbatim. + verbose : bool + Print the agreed label from rank 0 on success. + + Raises + ------ + CollectiveOperationError + If the labels differ, with the rank-to-label table. + + Example + ------- + >>> uw.mpi.ranks_agree(f"after adapt, {mesh.dm.getNumCells()} cells") + ... # doctest: +SKIP + """ + label = str(label) + checkpoint(f"ranks_agree({label!r})") + seen = comm.allgather(label) + + if len(set(seen)) > 1: + groups = {} + for r, entry in enumerate(seen): + groups.setdefault(entry, []).append(r) + table = "\n".join( + f" ranks {ranks_here}: {entry!r}" + for entry, ranks_here in sorted(groups.items(), + key=lambda kv: kv[1][0]) + ) + raise CollectiveOperationError( + f"\n{'=' * 70}\n" + f"RANKS DISAGREE AT A CHECKPOINT\n" + f"{'=' * 70}\n\n" + f"{len(groups)} different labels across {size} ranks:\n\n" + f"{table}\n\n" + f"The ranks took different paths to get here. Whatever collective " + f"comes\nnext will be reached by some of them and not the others.\n" + f"Look for a branch on rank-local data -- an array's size, an " + f"emptiness\ntest, a `None` -- between the last agreed point and " + f"this one, and\nreduce the predicate before branching on it.\n" + f"{'=' * 70}\n" + ) + + if verbose and rank == 0: + print(f"ranks_agree: all {size} ranks at {label!r}", flush=True) + + return label + + def pprint(*args, proc=0, prefix=None, clean_display=True, flush=False, **kwargs): """ Parallel-safe print that works as a drop-in replacement for print(). @@ -404,3 +740,65 @@ def __exit__(self, *args): dest = rank + 1 if dest < comm.size: comm.send(None, dest=rank + 1, tag=333) + + +def _watch_from_environment(): + """Arm the watchdog from ``UW_HANG_WATCHDOG``, at import. + + Arming here rather than from the user's script is the point: a rank that + dies, or diverges, before reaching a ``watch()`` call reports nothing, and + "before the script got going" covers mesh construction and most of the + import graph. This runs as ``underworld3.mpi`` is imported, which is about + as early as anything can. + + ``UW_HANG_WATCHDOG`` + Seconds of silence that count as stuck. Unset or 0 disables. + ``UW_HANG_WATCHDOG_DIR`` + Directory for the per-rank dumps. Default ``uw-hang-dumps``. + ``UW_HANG_WATCHDOG_ABORT`` + Anything but empty or ``0``: exit after dumping, rather than dumping + repeatedly. For batch queues, where the alternative is the wall clock. + + Usage:: + + UW_HANG_WATCHDOG=120 mpirun -n 4 python myrun.py + python -m underworld3.utilities.hang_report uw-hang-dumps + """ + setting = _os.environ.get("UW_HANG_WATCHDOG", "").strip() + if not setting: + return None + try: + seconds = float(setting) + except ValueError: + print( + f"UW_HANG_WATCHDOG={setting!r} is not a number of seconds; " + "the hang watchdog is NOT armed.", + file=_sys.stderr, flush=True, + ) + return None + if seconds <= 0: + return None + + directory = _os.environ.get("UW_HANG_WATCHDOG_DIR", "uw-hang-dumps") + # Idempotent on every rank, so no collective is needed to make it -- and a + # collective here would be the very thing the watchdog exists to diagnose. + _os.makedirs(directory, exist_ok=True) + path = _os.path.join(directory, f"rank{rank:04d}.log") + + # Left open for the life of the process on purpose: faulthandler holds the + # descriptor, so closing it would point the dump at whatever reuses it. + stream = open(path, "w", buffering=1) + # Disarm on the way out. Interpreter shutdown waits on the faulthandler + # thread, and a watchdog still counting down while the process tears itself + # down keeps the job alive long after the model has finished. + _atexit.register(unwatch) + watch( + seconds=seconds, + stream=stream, + abort=_os.environ.get("UW_HANG_WATCHDOG_ABORT", "").strip() not in ("", "0"), + ) + return path + + +#: Path this rank will dump to, or None when the environment did not ask. +environment_dump_path = _watch_from_environment() diff --git a/src/underworld3/utilities/hang_report.py b/src/underworld3/utilities/hang_report.py new file mode 100644 index 000000000..dfcd8fc8c --- /dev/null +++ b/src/underworld3/utilities/hang_report.py @@ -0,0 +1,273 @@ +"""Read a directory of per-rank hang dumps and say which rank went its own way. + +A collective entered by only some ranks leaves a distinctive trace: N-1 ranks +stopped in the same frame, and one somewhere else. No single rank can see that +--- it exists only across the set --- so the watchdog writes one dump per rank +and the comparison happens here, afterwards. + +That is also why this is a script rather than a test. The analysis needs every +rank's file at once, and a rank inside a test framework is one process looking +at itself. + +Usage:: + + UW_HANG_WATCHDOG=120 mpirun -n 4 python myrun.py + python -m underworld3.utilities.hang_report uw-hang-dumps + +The dumps come from :mod:`faulthandler`, so a frame is ``File "x.py", line N in +func``. It names the CALLING line, not the C function it called: a rank blocked +in ``comm.allreduce`` shows the Python line that made the call, which is the +localisation you want anyway. +""" + +import argparse +import collections +import os +import re +import sys + +#: Start of one dump. faulthandler writes this before each set of stacks. +_TIMEOUT = re.compile(r"^Timeout \(") +#: Start of one thread's stack within a dump. +_THREAD = re.compile(r"^(?:Current thread|Thread) 0x[0-9a-fA-F]+") +#: A single frame. +_FRAME = re.compile(r'^\s+File "(?P.*)", line (?P\d+) in (?P.*)$') + +#: Frames from these belong to the machinery, not to the model. A thread whose +#: own top frame is one of these is a helper --- the watchdog's timer, an I/O +#: thread --- and is not where the rank is stuck. +_MACHINERY = ("threading.py", "concurrent/futures", "selectors.py") + + +class Frame(collections.namedtuple("Frame", "file line func")): + """One stack frame, rendered short enough to compare by eye.""" + + def __str__(self): + return f"{os.path.basename(self.file)}:{self.line} in {self.func}" + + @property + def where(self): + """Identity for grouping: the same source line on any rank.""" + return (os.path.basename(self.file), self.line, self.func) + + +def parse_dump_file(text): + """Every dump in one rank's file, newest last, as lists of thread stacks.""" + dumps, threads, frames = [], [], [] + + def close_thread(): + if frames: + threads.append(list(frames)) + frames.clear() + + def close_dump(): + close_thread() + if threads: + dumps.append(list(threads)) + threads.clear() + + for raw in text.splitlines(): + if _TIMEOUT.match(raw): + close_dump() + continue + if _THREAD.match(raw): + close_thread() + continue + found = _FRAME.match(raw) + if found: + frames.append( + Frame(found["file"], int(found["line"]), found["func"].strip()) + ) + close_dump() + return dumps + + +def main_thread_stack(threads): + """The stack that is the model, not a helper thread. + + faulthandler does not label the main thread when it dumps from a helper, so + it is picked by its OUTERMOST frame --- the last line, since frames print + most-recent-first. A spawned thread bottoms out in ``threading.py``'s + ``_bootstrap``; the main thread bottoms out in the script. + + Depth is the wrong test, and measurably so: a rank blocked two frames into + ``allreduce`` while a telemetry thread sat twelve frames deep inside + ``requests`` was reported as being in the HTTP call. + """ + candidates = [ + stack for stack in threads + if stack and not any(part in stack[-1].file for part in _MACHINERY) + ] + if not candidates: + return None + # A script's main thread ends in ``; under runpy or pytest it does + # not, so that is a preference and not a requirement. + rooted = [stack for stack in candidates if stack[-1].func == ""] + return (rooted or candidates)[0] + + +def read_directory(directory): + """``{rank: [stack, ...]}`` for every ``rank*.log`` found, oldest first.""" + states = {} + for name in sorted(os.listdir(directory)): + found = re.fullmatch(r"rank(\d+)\.log", name) + if not found: + continue + with open(os.path.join(directory, name), errors="replace") as handle: + text = handle.read() + stacks = [ + stack for stack in + (main_thread_stack(threads) for threads in parse_dump_file(text)) + if stack + ] + states[int(found[1])] = stacks + return states + + +#: Dumps to consider when deciding where a rank has settled. +RECENT = 5 + + +def settled_position(stacks, recent=RECENT): + """Where a rank has come to rest, from its last few dumps. + + Not simply the final dump. A rank stuck in a collective shows the same + position every period, so the mode over recent samples is the stable + signal; the last sample alone can catch a rank mid-transition and put it in + a group of its own. Measured: one rank of four fell out of the waiting + group that way, and the report named two culprits instead of one. + + Ties go to the most recent, so a rank that really did move on is not held + at an older position. + """ + window = stacks[-recent:] + counts = collections.Counter(stack[0].where for stack in window) + best = max(counts.items(), key=lambda kv: kv[1])[1] + for stack in reversed(window): + if counts[stack[0].where] == best: + return stack + return window[-1] + + +def roll_call(states): + """Group ranks by where they last were. Returns (groups, ranks_never_stuck). + + ``groups`` maps a position to the ranks stopped there, largest first. A + rank with no dump at all never stopped, which is itself informative: it is + the one that kept going. + """ + stuck, moving = {}, [] + for who, stacks in sorted(states.items()): + if not stacks: + moving.append(who) + continue + stuck[who] = settled_position(stacks) + + groups = collections.defaultdict(list) + for who, stack in stuck.items(): + groups[stack[0].where].append(who) + + ordered = sorted( + ((where, sorted(ranks), stuck[ranks[0]]) + for where, ranks in groups.items()), + key=lambda item: (-len(item[1]), item[1][0]), + ) + return ordered, sorted(moving) + + +def format_report(states, depth=6): + """The verdict, as text.""" + if not states: + return ("No rank*.log files found. Was UW_HANG_WATCHDOG set, and is " + "this the directory it wrote to?") + + groups, moving = roll_call(states) + total = len(states) + out = [f"Hang report: {total} rank(s)", ""] + + if not groups: + out.append("No rank ever stopped long enough to dump. Either nothing " + "hung, or the timeout is longer than the run.") + return "\n".join(out) + + for where, ranks, stack in groups: + head = f"ranks {ranks}" if len(ranks) > 1 else f"rank {ranks}" + out.append(f"{head} stopped at {stack[0]}") + for frame in stack[1:depth]: + out.append(f" via {frame}") + if len(stack) > depth: + out.append(f" ... {len(stack) - depth} more frame(s)") + out.append("") + + if moving: + out.append(f"ranks {moving} never stopped -- still making progress.") + out.append("") + + out.append(_verdict(groups, moving, total)) + return "\n".join(out) + + +def _verdict(groups, moving, total): + """The one sentence worth reading.""" + stopped = sum(len(ranks) for _where, ranks, _stack in groups) + + if moving and stopped: + where, ranks, _stack = groups[0] + return ( + f"=> {stopped} of {total} ranks are waiting at " + f"{where[0]}:{where[1]}, and ranks {moving} went past it.\n" + f" That frame is where the collective is; ranks {moving} are " + f"where the bug is.\n" + f" Look for a branch on rank-local data before it, and reduce " + f"the predicate\n before branching on it." + ) + + if len(groups) > 1: + where, ranks, _stack = groups[0] + others = sorted(r for _w, group, _s in groups[1:] for r in group) + return ( + f"=> {len(ranks)} of {total} ranks are waiting together at " + f"{where[0]}:{where[1]}, and ranks {others}\n" + f" are somewhere else. That frame is where the collective is; " + f"ranks {others} are\n where the bug is. Look for a branch on " + f"rank-local data before it, and reduce\n the predicate before " + f"branching on it." + ) + + if stopped == total: + where = groups[0][0] + return ( + f"=> all {total} ranks are stopped at {where[0]}:{where[1]}, " + f"together.\n" + f" Not a missed collective -- they agree. Something outside the " + f"job is not\n completing, or this phase is simply slower than " + f"the timeout." + ) + + return f"=> {stopped} of {total} ranks stopped; no rank ran on." + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="python -m underworld3.utilities.hang_report", + description="Compare per-rank hang dumps and name the divergent rank.", + ) + parser.add_argument( + "directory", nargs="?", default="uw-hang-dumps", + help="directory of rankNNNN.log files (default: uw-hang-dumps)", + ) + parser.add_argument( + "--depth", type=int, default=6, + help="frames of context to show per group (default: 6)", + ) + args = parser.parse_args(argv) + + if not os.path.isdir(args.directory): + parser.error(f"no such directory: {args.directory}") + + print(format_report(read_directory(args.directory), depth=args.depth)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_0053_hang_watchdog.py b/tests/test_0053_hang_watchdog.py new file mode 100644 index 000000000..792fb8def --- /dev/null +++ b/tests/test_0053_hang_watchdog.py @@ -0,0 +1,206 @@ +"""Serial behaviour of the hang watchdog and the rank-agreement audit. + +The parallel half --- a rank genuinely blocked in a collective still filing a +report --- is in ``tests/parallel/test_0778_hang_watchdog_mpi.py``, because it +needs more than one rank to block against. + +Every test here writes to a real file rather than a buffer, and that is not +incidental. The dump goes through :mod:`faulthandler`, which writes to a file +descriptor. Measured at np=4 against a 4 s block in ``comm.allreduce``: a +re-arming ``threading.Timer`` at 0.5 s fired zero times on the blocked ranks, +while faulthandler produced all 7 expected dumps. A buffer-backed watchdog +would pass every test in this file and still report nothing in the case it +exists for, which is what :func:`test_a_buffer_is_refused` guards. +""" + +import io +import time + +import pytest + +import underworld3 as uw + +pytestmark = [pytest.mark.level_1, pytest.mark.tier_a] + +FIRED = "UW HANG WATCHDOG" + + +@pytest.fixture(autouse=True) +def always_disarm(): + """A leaked timer would fire in the middle of some later test.""" + yield + uw.mpi.unwatch() + + +@pytest.fixture +def report(tmp_path): + """A real file to watch into, and a reader for whatever landed in it. + + Disarms BEFORE closing. faulthandler holds the descriptor, not the Python + object, so a watchdog left armed over a closed file writes into whatever + that descriptor is reused for -- which in a captured pytest run is the + capture pipe, and the session then wedges. The autouse fixture above tears + down after this one, which is too late. + """ + path = tmp_path / "watchdog.log" + handle = path.open("w") + try: + yield handle, (lambda: path.read_text()) + finally: + uw.mpi.unwatch() + handle.close() + + +def test_a_buffer_is_refused(): + """An in-memory stream must be rejected, not quietly half-work. + + faulthandler needs a descriptor. Accepting a buffer would leave the + watchdog silent exactly when a rank is blocked inside MPI. + """ + with pytest.raises(ValueError, match="file descriptor"): + uw.mpi.watch(seconds=5, stream=io.StringIO()) + + +def test_watchdog_reports_when_progress_stops(report): + """The whole point: no checkpoint for `seconds`, and it says so.""" + stream, read_back = report + uw.mpi.watch(seconds=0.3, stream=stream) + uw.mpi.checkpoint("the last thing this rank did") + time.sleep(0.9) + uw.mpi.unwatch() + + text = read_back() + assert FIRED in text + assert f"rank {uw.mpi.rank} of {uw.mpi.size}" in text + assert "the last thing this rank did" in text + + +def test_report_carries_the_main_thread_stack(report): + """A dump of the watchdog's own frames would diagnose nothing. + + The stack must name this test function, which runs on the main thread. + """ + stream, read_back = report + uw.mpi.watch(seconds=0.3, stream=stream) + time.sleep(0.9) + uw.mpi.unwatch() + + text = read_back() + assert "test_report_carries_the_main_thread_stack" in text, ( + f"the dump did not include the main thread:\n{text}" + ) + + +def test_progress_never_fires(report): + """Negative control. A checkpoint inside the window keeps it quiet. + + Without this, the first test proves only that a timer can print something. + """ + stream, read_back = report + uw.mpi.watch(seconds=0.6, stream=stream) + for step in range(8): + time.sleep(0.05) + uw.mpi.checkpoint(f"step {step}") + uw.mpi.unwatch() + + assert FIRED not in read_back(), ( + f"fired despite steady progress:\n{read_back()}" + ) + + +def test_repeats_rather_than_reporting_once(report): + """One dump reads as "slow"; the same dump again reads as "stuck".""" + stream, read_back = report + uw.mpi.watch(seconds=0.3, stream=stream) + time.sleep(1.4) + uw.mpi.unwatch() + + assert read_back().count(FIRED) >= 2 + + +def test_unwatch_disarms(report): + stream, read_back = report + uw.mpi.watch(seconds=0.3, stream=stream) + uw.mpi.unwatch() + time.sleep(0.8) + assert FIRED not in read_back() + + +def test_checkpoint_without_a_watchdog_is_a_no_op(): + """It is meant to be left in production code, so it must cost nothing.""" + uw.mpi.unwatch() + uw.mpi.checkpoint("nobody is listening") + + +def test_watching_restores_the_previous_watchdog(tmp_path): + """A nested block must not silently disarm the outer one on exit.""" + outer_path = tmp_path / "outer.log" + inner_path = tmp_path / "inner.log" + + with outer_path.open("w") as outer, inner_path.open("w") as inner: + try: + uw.mpi.watch(seconds=0.4, stream=outer) + with uw.mpi.watching(seconds=0.3, stream=inner): + time.sleep(0.9) + assert FIRED in inner_path.read_text() + + # The outer watchdog is armed again, and still works. + time.sleep(1.1) + finally: + # Disarm before the files close -- see the `report` fixture. + uw.mpi.unwatch() + + assert FIRED in outer_path.read_text(), ( + "the outer watchdog was left disarmed by the nested block" + ) + + +def test_declared_collectives_check_in_automatically(report): + """`@collective_operation` re-arms the timer and labels the report. + + This is what makes the watchdog usable without seeding `checkpoint` calls + through the library by hand. + """ + @uw.mpi.collective_operation + def a_collective_thing(): + return 42 + + stream, read_back = report + uw.mpi.watch(seconds=0.3, stream=stream) + assert a_collective_thing() == 42 + time.sleep(0.9) + uw.mpi.unwatch() + + assert "a_collective_thing() [collective]" in read_back() + + +def test_ranks_agree_passes_when_they_do(): + assert uw.mpi.ranks_agree("same label") == "same label" + + +def test_ranks_agree_reports_the_split(monkeypatch): + """Serially every rank is this one, so the disagreement is injected. + + The failure has to be provoked somehow, or the test asserts only that + agreement is possible --- which is the case the check is not for. The + parallel suite exercises the real thing. + """ + class ThreeRanksThatSplit: + # mpi4py's Intracomm attributes are read-only, so the communicator + # itself is replaced rather than one of its methods. + size = 3 + + def allgather(self, _value): + return ["took branch A", "took branch B", "took branch A"] + + monkeypatch.setattr(uw.mpi, "comm", ThreeRanksThatSplit()) + monkeypatch.setattr(uw.mpi, "size", 3) + + with pytest.raises(uw.mpi.CollectiveOperationError) as excinfo: + uw.mpi.ranks_agree("took branch A") + + message = str(excinfo.value) + assert "RANKS DISAGREE" in message + assert "took branch A" in message and "took branch B" in message + # The table must name WHICH ranks, or it does not localise anything. + assert "[0, 2]" in message and "[1]" in message diff --git a/tests/test_0054_hang_report.py b/tests/test_0054_hang_report.py new file mode 100644 index 000000000..acff8d02d --- /dev/null +++ b/tests/test_0054_hang_report.py @@ -0,0 +1,200 @@ +"""The hang analyser: parsing, the roll call, and one end-to-end run. + +pytest never arms the watchdog here. It is process-global state --- faulthandler +has one ``dump_traceback_later`` slot and pytest's own plugin already owns it --- +and a rank inside a test framework is one process looking at itself, which is +the one thing the analysis cannot be done from. So the end-to-end test launches +a real ``mpirun`` job as a subprocess and checks the verdict it produces. + +Everything above that is ordinary text processing and is tested directly. +""" + +import os +import shutil +import subprocess +import sys +import textwrap + +import pytest + +from underworld3.utilities import hang_report + +pytestmark = [pytest.mark.level_1, pytest.mark.tier_a] + + +def dump(*thread_stacks): + """One faulthandler dump, in the exact layout it writes.""" + out = ["Timeout (0:00:01.000000)!"] + for index, frames in enumerate(thread_stacks): + out.append(f"Thread 0x{index:016x} (most recent call first):") + out.extend(f' File "{f}", line {n} in {fn}' for f, n, fn in frames) + out.append("") + return "\n".join(out) + + +def stack(*frames): + return [hang_report.Frame(*f) for f in frames] + + +WATCHDOG_THREAD = stack( + ("/usr/lib/python3.12/threading.py", 359, "wait"), + ("/usr/lib/python3.12/threading.py", 1431, "run"), + ("/usr/lib/python3.12/threading.py", 1032, "_bootstrap"), +) +BLOCKED_IN_COLLECTIVE = stack( + ("/uw/discretisation/discretisation_mesh.py", 6954, "get_max_radius"), + ("/uw/discretisation/discretisation_mesh.py", 6140, "_classify_points_in_domain"), + ("/uw/swarm.py", 3786, "migrate"), + ("/home/run/model.py", 41, ""), +) +RAN_AHEAD = stack( + ("/uw/swarm.py", 3801, "migrate"), + ("/home/run/model.py", 41, ""), +) + + +def test_parses_every_dump_in_a_file(): + """`repeat` means many dumps per file; the last one is the live state.""" + text = dump(WATCHDOG_THREAD, BLOCKED_IN_COLLECTIVE) * 3 + dumps = hang_report.parse_dump_file(text) + assert len(dumps) == 3 + assert all(len(threads) == 2 for threads in dumps) + + +def test_main_thread_is_chosen_by_its_root_not_its_depth(): + """Regression: a deep helper thread must not be mistaken for the model. + + Measured on a real run --- rank 0 sat two frames into `allreduce` while a + telemetry thread was twelve frames deep inside `requests`, and picking the + longest stack reported the rank as being in an HTTP call. + """ + deep_helper = stack( + ("/usr/lib/python3.12/http/client.py", 1390, "getresponse"), + ("/site-packages/requests/sessions.py", 579, "request"), + ("/site-packages/urllib3/connectionpool.py", 715, "urlopen"), + ("/usr/lib/python3.12/threading.py", 1032, "_bootstrap"), + ) + assert len(deep_helper) > len(RAN_AHEAD) + + chosen = hang_report.main_thread_stack([deep_helper, RAN_AHEAD]) + assert chosen == RAN_AHEAD, ( + "the deeper helper thread was reported as the main thread" + ) + + +def test_roll_call_separates_the_majority_from_the_odd_one_out(): + states = { + 0: [BLOCKED_IN_COLLECTIVE], + 1: [RAN_AHEAD], + 2: [BLOCKED_IN_COLLECTIVE], + 3: [BLOCKED_IN_COLLECTIVE], + } + groups, moving = hang_report.roll_call(states) + + assert not moving + assert [ranks for _where, ranks, _stack in groups] == [[0, 2, 3], [1]] + + +def test_a_rank_with_no_dump_counts_as_still_moving(): + """Silence is the commonest signature of the rank that caused it. + + A rank that branched away and kept working never stops checking in, so it + never dumps. The report has to say so rather than ignore it. + """ + states = {0: [BLOCKED_IN_COLLECTIVE], 1: [], 2: [BLOCKED_IN_COLLECTIVE]} + groups, moving = hang_report.roll_call(states) + + assert moving == [1] + assert [ranks for _where, ranks, _stack in groups] == [[0, 2]] + + text = hang_report.format_report(states) + assert "never stopped" in text + assert "ranks [1] are where the bug is" in text.replace("\n", " ") + + +def test_all_ranks_together_is_not_reported_as_a_missed_collective(): + """Everyone in the same place is a slow phase, not divergence. + + Saying "missed collective" here would send the reader hunting a bug that + is not there. + """ + states = {r: [BLOCKED_IN_COLLECTIVE] for r in range(4)} + text = hang_report.format_report(states) + assert "together" in text + assert "Not a missed collective" in text + + +def test_empty_directory_says_so_rather_than_passing_quietly(tmp_path): + assert "No rank*.log files found" in hang_report.format_report( + hang_report.read_directory(tmp_path) + ) + + +DIVERGENT = """ +import time +import underworld3 as uw + +def classify(): + '''Stands in for library code deciding how much work this rank has.''' + return [] if uw.mpi.rank == 1 else [1, 2, 3] + +def reduce_the_count(undecided): + '''The un-noticed collective, behind a rank-local guard.''' + return uw.mpi.comm.allreduce(len(undecided)) + +undecided = classify() +if undecided: + reduce_the_count(undecided) +else: + time.sleep(3600) +""" + + +@pytest.mark.timeout(300) +@pytest.mark.skipif(shutil.which("mpirun") is None, reason="needs mpirun") +def test_end_to_end_names_the_divergent_rank(tmp_path): + """A real four-rank job that really hangs, killed, then analysed. + + This is the workflow the tool is for: the job does NOT recover, mpirun + times it out, and the dumps are all that is left. An earlier version had + the divergent rank join late so the job exited cleanly, which tested a + situation nobody is ever in and made termination the flaky part. + + Ranks 0, 2 and 3 block in the collective; rank 1 branches around it and + sleeps. Both groups dump, so the roll call has to separate them. + """ + script = tmp_path / "divergent.py" + script.write_text(textwrap.dedent(DIVERGENT)) + dumps = tmp_path / "uw-hang-dumps" + + environment = dict( + os.environ, + UW_HANG_WATCHDOG="1.0", + UW_HANG_WATCHDOG_DIR=str(dumps), + UW_NO_USAGE_METRICS="1", + ) + # A non-zero exit is EXPECTED -- mpirun kills a job that never finishes. + subprocess.run( + ["mpirun", "--timeout", "25", "-n", "4", sys.executable, "-u", str(script)], + env=environment, capture_output=True, text=True, timeout=120, + ) + + states = hang_report.read_directory(dumps) + assert len(states) == 4, f"expected four dump files, got {sorted(states)}" + + groups, moving = hang_report.roll_call(states) + biggest_where, biggest_ranks, _stack = groups[0] + report = hang_report.format_report(states) + + assert biggest_ranks == [0, 2, 3], ( + f"the waiting ranks were {biggest_ranks}, not [0, 2, 3]:\n{report}" + ) + assert biggest_where[2] == "reduce_the_count", ( + f"the majority was located at {biggest_where}, not at the collective" + ) + odd_ones_out = sorted(r for _w, ranks, _s in groups[1:] for r in ranks) + moving + assert odd_ones_out == [1], ( + f"rank 1 took the branch; the report blamed {odd_ones_out}:\n{report}" + ) + # The verdict has to point at the branch, not merely list stacks. + assert "where the bug is" in report From 9db1ecb7cb01365dcdb5f824092627d79ff82880 Mon Sep 17 00:00:00 2001 From: lmoresi Date: Sat, 22 Aug 2026 16:23:55 +1000 Subject: [PATCH 2/3] Oversubscribe the four-rank diagnostic job, and say so when it does not start A CI runner has two cores, so OpenMPI declines to launch four ranks and the job produced no dumps at all -- surfacing as a FileNotFoundError on the dump directory rather than as the reason. The ranks are blocked or asleep for the whole test, so the cores are not the constraint. Underworld development team with AI support from Claude Code --- tests/test_0054_hang_report.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/tests/test_0054_hang_report.py b/tests/test_0054_hang_report.py index acff8d02d..dff2ea277 100644 --- a/tests/test_0054_hang_report.py +++ b/tests/test_0054_hang_report.py @@ -173,12 +173,24 @@ def test_end_to_end_names_the_divergent_rank(tmp_path): UW_HANG_WATCHDOG_DIR=str(dumps), UW_NO_USAGE_METRICS="1", ) - # A non-zero exit is EXPECTED -- mpirun kills a job that never finishes. - subprocess.run( - ["mpirun", "--timeout", "25", "-n", "4", sys.executable, "-u", str(script)], + # `--oversubscribe` because a CI runner has two cores and this wants four + # ranks. They spend the whole test blocked or asleep, so the cores are not + # the constraint -- but without it OpenMPI declines to start and the job + # produces no dumps at all. + # + # A non-zero exit is EXPECTED: mpirun kills a job that never finishes. + finished = subprocess.run( + ["mpirun", "--oversubscribe", "--timeout", "25", "-n", "4", + sys.executable, "-u", str(script)], env=environment, capture_output=True, text=True, timeout=120, ) + if not dumps.is_dir(): + pytest.fail( + "the job wrote no dumps at all, so it never reached " + f"`import underworld3`:\n{finished.stderr[-2000:]}" + ) + states = hang_report.read_directory(dumps) assert len(states) == 4, f"expected four dump files, got {sorted(states)}" From afa2c8074cac343c2c23a56a65544bc6212d6319 Mon Sep 17 00:00:00 2001 From: lmoresi Date: Sat, 22 Aug 2026 17:38:29 +1000 Subject: [PATCH 3/3] Launch the diagnostic job portably: MPICH rejects OpenMPI's flags CI runs MPICH, which oversubscribes by default and rejects --oversubscribe outright, taking the whole mpirun invocation down with it -- so the job never started and wrote no dumps. --timeout is OpenMPI-only for the same reason. The MPI family is now detected for the oversubscribe flag, and the time limit is enforced from Python, killing the process group so no rank is orphaned holding a dump file open. The job under test is meant never to finish, so that timeout is its normal exit path rather than an error. Underworld development team with AI support from Claude Code --- tests/test_0054_hang_report.py | 59 +++++++++++++++++++++++++++------- 1 file changed, 48 insertions(+), 11 deletions(-) diff --git a/tests/test_0054_hang_report.py b/tests/test_0054_hang_report.py index dff2ea277..66bd625fd 100644 --- a/tests/test_0054_hang_report.py +++ b/tests/test_0054_hang_report.py @@ -11,6 +11,7 @@ import os import shutil +import signal import subprocess import sys import textwrap @@ -130,6 +131,49 @@ def test_empty_directory_says_so_rather_than_passing_quietly(tmp_path): ) +def _launcher(ranks): + """An `mpirun` command line that works on both MPI families. + + The launch flags are NOT portable. OpenMPI refuses to start more ranks than + cores without `--oversubscribe`; MPICH oversubscribes by default and rejects + the flag outright, taking the whole command down with it. `--timeout` is + likewise OpenMPI-only. So the family is detected, and the time limit is + enforced from Python instead of by the launcher. + """ + command = ["mpirun", "-n", str(ranks)] + try: + banner = subprocess.run(["mpirun", "--version"], capture_output=True, + text=True, timeout=30) + flavour = (banner.stdout + banner.stderr) + except (OSError, subprocess.SubprocessError): + flavour = "" + if "Open MPI" in flavour or "OpenRTE" in flavour: + # The ranks are blocked or asleep throughout, so cores are not the + # constraint -- but OpenMPI counts slots, not activity. + command.insert(1, "--oversubscribe") + return command + + +def _run_until_it_hangs(argv, ranks, seconds, environment): + """Launch under MPI, kill the whole job after `seconds`, return stderr. + + The job under test is MEANT never to finish, so the time limit is the + normal exit path rather than an error. `start_new_session` puts the ranks in + their own process group so the kill reaches all of them -- terminating only + `mpirun` can leave orphaned ranks holding the dump files open. + """ + process = subprocess.Popen( + _launcher(ranks) + argv, env=environment, text=True, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True, + ) + try: + _out, err = process.communicate(timeout=seconds) + except subprocess.TimeoutExpired: + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + _out, err = process.communicate() + return err or "" + + DIVERGENT = """ import time import underworld3 as uw @@ -173,22 +217,15 @@ def test_end_to_end_names_the_divergent_rank(tmp_path): UW_HANG_WATCHDOG_DIR=str(dumps), UW_NO_USAGE_METRICS="1", ) - # `--oversubscribe` because a CI runner has two cores and this wants four - # ranks. They spend the whole test blocked or asleep, so the cores are not - # the constraint -- but without it OpenMPI declines to start and the job - # produces no dumps at all. - # - # A non-zero exit is EXPECTED: mpirun kills a job that never finishes. - finished = subprocess.run( - ["mpirun", "--oversubscribe", "--timeout", "25", "-n", "4", - sys.executable, "-u", str(script)], - env=environment, capture_output=True, text=True, timeout=120, + stderr = _run_until_it_hangs( + [sys.executable, "-u", str(script)], ranks=4, seconds=25, + environment=environment, ) if not dumps.is_dir(): pytest.fail( "the job wrote no dumps at all, so it never reached " - f"`import underworld3`:\n{finished.stderr[-2000:]}" + f"`import underworld3`:\n{stderr[-2000:]}" ) states = hang_report.read_directory(dumps)