diff --git a/src/underworld3/__init__.py b/src/underworld3/__init__.py index 450e2368..8da83e48 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 5f3cdf15..3e1e903d 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 00000000..dfcd8fc8 --- /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 00000000..792fb8de --- /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 00000000..66bd625f --- /dev/null +++ b/tests/test_0054_hang_report.py @@ -0,0 +1,249 @@ +"""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 signal +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) + ) + + +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 + +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", + ) + 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{stderr[-2000:]}" + ) + + 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