Time the hang watchdog from the kernel, so disarming it cannot wedge the process (#661)

src/underworld3/mpi.py

@@ -26,10 +26,12 @@
 import faulthandler as _faulthandler
 import os as _os
 import secrets as _secrets
+import signal as _signal
 import sys as _sys
 import io as _io
 import threading as _threading
 import time as _time
+import warnings as _warnings
 from contextlib import contextmanager as _contextmanager
 
 # Pre-import EVERYTHING the watchdog reporter thread can touch. The
@@ -306,19 +308,38 @@ def wrapper(*args, **kwargs):
 # 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.
+# ranks -- the interpreter lock is held for the duration.
 #
-# 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.
+# So the timing is done by the KERNEL and the dump by a SIGNAL handler:
+# `signal.setitimer` raises SIGALRM whether or not the interpreter can run,
+# and faulthandler's handler for it dumps on whichever thread receives the
+# signal. Measured the same way, 8 of 8 expected dumps on every rank blocked
+# in a 4 s `allreduce`, and 300 rounds of allreduce/barrier/bcast under a
+# 100 Hz timer completed with zero collective errors, so the signal does not
+# disturb the traffic it is watching.
+#
+# This deliberately does NOT use `faulthandler.dump_traceback_later`, which
+# was the mechanism until #661. That runs a C thread walking every other
+# thread's LIVE frames, and both arming and disarming it wait on that thread
+# while holding the interpreter lock -- so `unwatch()` could wedge or crash
+# the process it exists to diagnose. Nothing here runs concurrently with the
+# interpreter, and disarming is a syscall that cannot wait on anything.
+#
+# The price is that faulthandler 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()
 
+# POSIX only. Without it the watchdog still reports through its Python timer,
+# which covers every hang that leaves the interpreter lock free but not a rank
+# blocked inside MPI. Underworld runs on Linux and macOS, so this is a
+# statement about what degrades rather than a platform we support.
+_INTERVAL_TIMER_AVAILABLE = hasattr(_signal, "setitimer") and hasattr(_signal, "SIGALRM")
+
 
 def _stack_dump():
     """Every thread's Python stack, main thread first.
@@ -395,6 +416,46 @@ def __init__(self, seconds, stream, abort):
             flush=True,
         )
 
+        # The signal side is set up ONCE, here, rather than on every arm().
+        # `checkpoint()` re-arms and is meant to be cheap enough to leave in a
+        # production loop, and `signal.signal` refuses to run anywhere but the
+        # main thread -- so doing this per checkpoint would make a checkpoint
+        # from a worker thread raise, with `abort` on, which is CI's setting.
+        self._previous_sigalrm = None
+        self._handler_installed = False
+        if _INTERVAL_TIMER_AVAILABLE:
+            pending, _interval = _signal.getitimer(_signal.ITIMER_REAL)
+            if pending > 0.0:
+                _warnings.warn(
+                    f"the hang watchdog takes over ITIMER_REAL, and one was "
+                    f"already armed with {pending:.3g} s to run. That timer is "
+                    f"now cancelled. There is one interval timer per process, "
+                    f"so the watchdog and any other user of signal.alarm() or "
+                    f"setitimer(ITIMER_REAL) cannot both have it.",
+                    RuntimeWarning, stacklevel=3,
+                )
+            self._install_signal_handler()
+
+    def _install_signal_handler(self):
+        """Take SIGALRM, remembering what was there.
+
+        Separate from ``__init__`` because a watchdog can be disarmed and armed
+        again -- ``watching`` cancels the outer one and restores it on exit. A
+        resumed watchdog that moved the clock without re-registering would let
+        the next SIGALRM reach the disposition underneath, which for ``abort``
+        is SIG_DFL: the timer would silently kill the process instead of
+        dumping.
+        """
+        self._previous_sigalrm = _signal.getsignal(_signal.SIGALRM)
+        if self.abort:
+            # Dump, then let SIGALRM's default action terminate us. The handler
+            # chains, so the disposition underneath has to be the default one
+            # rather than whatever was there before.
+            _signal.signal(_signal.SIGALRM, _signal.SIG_DFL)
+        _faulthandler.register(_signal.SIGALRM, file=self.stream,
+                               all_threads=True, chain=self.abort)
+        self._handler_installed = 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.
@@ -408,31 +469,36 @@ def arm(self, label=None, resume=False):
             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
-        )
+        # The mechanism: a kernel interval timer, and faulthandler's SIGNAL
+        # handler to dump when it fires. Re-arming resets the countdown, so a
+        # job that keeps checking in never reaches it, and the timer's repeat
+        # interval keeps it dumping once stuck: two identical stacks a minute
+        # apart say "stuck", one says "slow".
+        #
+        # NOT dump_traceback_later. That runs a C thread which walks every
+        # other thread's LIVE frames, and disarming it waits on that thread
+        # while holding the interpreter lock -- so unwatch() could wedge the
+        # process it exists to diagnose, or crash it (#661, and the test_0054
+        # deadlock triangle before it). Nothing here runs concurrently with
+        # the interpreter: the kernel raises SIGALRM, the handler runs on
+        # whichever thread receives it, and disarming is a syscall that cannot
+        # wait on anything.
+        if _INTERVAL_TIMER_AVAILABLE:
+            # Normally only the clock -- one syscall, safe from any thread,
+            # because the handler went in when the watchdog was built. A
+            # RESUMED watchdog has had its handler removed by the cancel that
+            # suspended it, so it goes back first.
+            if not self._handler_installed:
+                self._install_signal_handler()
+            _signal.setitimer(_signal.ITIMER_REAL, self.seconds, self.seconds)
         self._rearm_timer()
 
     def _rearm_timer(self):
         # Secondary, and only for hangs that leave the interpreter lock free.
-        # It adds the checkpoint label, which faulthandler cannot know about.
-        #
-        # The REPORTER re-arms through this method ALONE, never through
-        # arm(): dump_traceback_later() internally cancels the running C
-        # watchdog thread and waits on its lock, and when that thread is
-        # mid-dump — walking frames the main thread is churning (an
-        # import in progress) — the wait never returns. Measured as the
-        # test_0054 deadlock triangle (native `sample`): the C thread
-        # pinned in dump_traceback, the reporter cond-waiting inside
-        # cancel_dump_traceback_later, the main thread starved in the
-        # import machinery — 12 of 15 runs frozen at a 0.2 s watchdog.
-        # faulthandler was armed with repeat=True; it needs no re-arm
-        # from the reporter. Checkpoints (watch()) still go through
-        # arm(), where resetting the countdown is the point and the main
-        # thread is in ordinary running state.
+        # It adds the checkpoint label, which faulthandler cannot know about,
+        # and it carries `abort` -- the signal handler cannot exit the process
+        # for us. A rank blocked inside MPI never reaches this; the signal
+        # dump above is what covers that case.
         if self.timer is not None:
             self.timer.cancel()
         self.timer = _threading.Timer(self.seconds, self.report)
@@ -441,7 +507,19 @@ def _rearm_timer(self):
 
     def cancel(self):
         self.cancelled = True
-        _faulthandler.cancel_dump_traceback_later()
+        if _INTERVAL_TIMER_AVAILABLE:
+            _signal.setitimer(_signal.ITIMER_REAL, 0.0, 0.0)
+            _faulthandler.unregister(_signal.SIGALRM)
+            self._handler_installed = False
+            # Leave the process's signal state as it was found. unregister()
+            # restores whatever was installed when register() ran, which for
+            # `abort` is the SIG_DFL we put there ourselves, so the caller's
+            # own handler has to be put back explicitly. signal.signal only
+            # runs on the main thread; a disarm from elsewhere leaves the
+            # handler in place, which is inert once the timer is off.
+            if (self._previous_sigalrm is not None
+                    and _threading.current_thread() is _threading.main_thread()):
+                _signal.signal(_signal.SIGALRM, self._previous_sigalrm)
         if self.timer is not None:
             self.timer.cancel()
             self.timer = None

src/underworld3/utilities/hang_report.py

@@ -26,8 +26,15 @@
 import re
 import sys
 
-#: Start of one dump. faulthandler writes this before each set of stacks.
+#: Start of one dump, when it came from ``dump_traceback_later``, which prints
+#: this header before each set of stacks. Dumps taken through faulthandler's
+#: SIGNAL handler -- how the watchdog works since #661 -- carry no header, so
+#: it cannot be the only thing that separates one dump from the next.
 _TIMEOUT = re.compile(r"^Timeout \(")
+#: Start of the stack of the thread that took the dump. faulthandler prints
+#: exactly one of these per dump, ahead of the other threads, so it separates
+#: dumps in a headerless file and in a headed one alike.
+_CURRENT_THREAD = re.compile(r"^Current thread 0x[0-9a-fA-F]+")
 #: Start of one thread's stack within a dump.
 _THREAD = re.compile(r"^(?:Current thread|Thread) 0x[0-9a-fA-F]+")
 #: A single frame.
@@ -70,6 +77,13 @@ def close_dump():
         if _TIMEOUT.match(raw):
             close_dump()
             continue
+        # A headerless dump starts at its "Current thread" line. Closing the
+        # previous dump here is what keeps repeated signal dumps from merging
+        # into one; a "Timeout (" header immediately before simply closes an
+        # already-empty dump, so headed files parse exactly as they did.
+        if _CURRENT_THREAD.match(raw):
+            close_dump()
+            continue
         if _THREAD.match(raw):
             close_thread()
             continue

tests/test_0053_hang_watchdog.py

@@ -204,3 +204,78 @@ def allgather(self, _value):
     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
+
+
+def test_a_pre_existing_sigalrm_handler_is_put_back(report):
+    """The watchdog borrows process-wide signal state; it must give it back.
+
+    With ``abort`` it installs SIG_DFL underneath faulthandler so the dump can
+    chain to it and terminate. Without this, arming the watchdog anywhere in a
+    program would silently discard a handler the program had installed for its
+    own reasons, and only the next SIGALRM would reveal it.
+    """
+    import signal
+
+    def mine(signum, frame):
+        pass
+
+    stream, _read_back = report
+    signal.signal(signal.SIGALRM, mine)
+    try:
+        uw.mpi.watch(seconds=30, stream=stream, abort=True)
+        uw.mpi.unwatch()
+        assert signal.getsignal(signal.SIGALRM) is mine, (
+            "unwatch() left its own SIGALRM disposition behind"
+        )
+    finally:
+        signal.signal(signal.SIGALRM, signal.SIG_DFL)
+
+
+def test_checkpoint_works_off_the_main_thread_with_abort(report):
+    """`checkpoint` is documented as safe to leave in production code.
+
+    ``signal.signal`` refuses to run anywhere but the main thread, so doing the
+    signal setup per checkpoint made a checkpoint from a worker thread raise
+    ValueError -- and only with ``abort`` on, which is CI's setting. The setup
+    belongs in one place, at arm time.
+    """
+    import threading
+
+    stream, _read_back = report
+    uw.mpi.watch(seconds=30, stream=stream, abort=True)
+    outcome = {}
+
+    def worker():
+        try:
+            uw.mpi.checkpoint("from a worker thread")
+            outcome["result"] = "ok"
+        except Exception as exc:                     # noqa: BLE001 - reported below
+            outcome["result"] = f"{type(exc).__name__}: {exc}"
+
+    thread = threading.Thread(target=worker)
+    thread.start()
+    thread.join()
+    uw.mpi.unwatch()
+
+    assert outcome["result"] == "ok", (
+        f"checkpoint() from a worker thread raised: {outcome['result']}"
+    )
+
+
+def test_taking_over_the_interval_timer_is_announced(report):
+    """There is one ITIMER_REAL per process and the watchdog needs it.
+
+    It cannot be shared, so the honest behaviour is to take it and say so
+    rather than cancel someone's timer in silence.
+    """
+    import signal
+
+    stream, _read_back = report
+    signal.setitimer(signal.ITIMER_REAL, 3600.0, 0.0)
+    try:
+        with pytest.warns(RuntimeWarning, match="ITIMER_REAL"):
+            uw.mpi.watch(seconds=30, stream=stream)
+        uw.mpi.unwatch()
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0.0, 0.0)
+        signal.signal(signal.SIGALRM, signal.SIG_DFL)

tests/test_0054_hang_report.py

@@ -172,11 +172,16 @@ def _wait_for(condition, what, cap=600.0, poll=0.25):
 
 
 def _dump_count(dumps, rank):
-    """How many times this rank has dumped so far."""
+    """How many times this rank has dumped so far.
+
+    Counted by the "Current thread" line rather than the "Timeout (" header:
+    the watchdog dumps through faulthandler's signal handler (#661), which
+    writes no header, and there is exactly one such line per dump either way.
+    """
     path = dumps / f"rank{rank:04d}.log"
     if not path.exists():
         return 0
-    return path.read_text(errors="replace").count("Timeout (")
+    return path.read_text(errors="replace").count("Current thread ")
 
 
 def _run_until_the_evidence_exists(argv, ranks, environment, dumps, ready,