Merge pull request #15 from JaneliaSciComp/watchdog-exit-0.9.1

Stop the watchdog at exit (0.9.1)

Author: davidackerman

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 0.9.1 (2026-10-26)
+
+A fix for exit. No placement change, no unit table change, and no default changes.
+
+- The watchdog thread stops at exit, before the interpreter finalizes. On Python up to
+  3.13 a process could abort once all its work was done ("terminate called without an
+  active exception", exit 134) when finalization began while the thread was inside a
+  tensorstore call; an LSF job that did this looked failed. Now an `atexit` handler,
+  which runs once the program's other threads are joined, ends the thread and waits
+  for it, and no new one starts after that.
+
 ## 0.9.0 (2026-10-23)
 
 A job script can size each worker's threads. No placement change, no unit table change, and

docs/spec.md

@@ -692,7 +692,15 @@ this document and that one differ, fix one of them.
    copy and a local read of a key that does not exist; no store is touched, about 0.3 ms
    of CPU). A stranded task then runs within about a second whoever waits on it: the
    package, `vol.ts` directly, or `await`, at any concurrency. The retry stays the outer
-   net. Open question 14 says when the watchdog goes.
+   net. The thread stops at exit (0.9.1): an `atexit` handler, which Python runs once the
+   program's other threads are joined and before it finalizes, ends it and waits for it
+   (at most the longest one kick takes), and no new one starts after that. Python up to
+   3.13 ends a daemon thread that takes the GIL back during finalization with
+   `pthread_exit`, and inside a tensorstore call that aborts the process with "terminate
+   called without an active exception" and exit 134 after all the work is done (once in
+   six of mesh-n-bone's Linux CI runs on Python 3.12, 2026-10-26; every time when a
+   process exits while the thread kicks flat out). Open question 14 says when the
+   watchdog goes.
 
 7. **Contexts.** Without `context`, every open in the process shares `shared_context(0)`:
    tensorstore's cache is off and the levels of one pyramid share one context. Its
@@ -1089,7 +1097,8 @@ through a pickle, and a create whose first attempt lands after its timeout adopt
 array under `exists="error"`. 0.9: `WORLDARRAY_THREADS` ahead of every other variable,
 blank as unset and a bad value refused; `file_io_floor` sizing file I/O under the budget,
 ignored with a thread count, and carried through a pickle; a handle pickled on the driver
-sized in a spawned worker by that worker's `WORLDARRAY_THREADS`.
+sized in a spawned worker by that worker's `WORLDARRAY_THREADS`. 0.9.1: a process whose
+finalization takes a while, exiting while the watchdog kicks flat out, exits cleanly.
 Opt-in (`WORLDARRAY_STRESS=1`, about a minute,
 `tests/test_lost_wakeup.py`): worker processes aim tasks at the moment a one-thread pool's
 worker exits idle, so a plain context strands some for good (13 of about 7,000 on

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "worldarray"
-version = "0.9.0"
+version = "0.9.1"
 description = "Read zarr, N5 and Neuroglancer precomputed volumes in world coordinates through tensorstore: correct voxel placement, pyramid levels, padded region reads, resampling. numpy + tensorstore only."
 readme = "README.md"
 license = { file = "LICENSE" }

src/worldarray/__init__.py

@@ -28,7 +28,7 @@
     parse_chunk_key,
 )
 
-__version__ = "0.9.0"
+__version__ = "0.9.1"
 
 __all__ = [
     "ArrayMeta",

src/worldarray/context.py

@@ -21,11 +21,11 @@
 
 from __future__ import annotations
 
+import atexit
 import json
 import os
 import tempfile
 import threading
-import time
 import weakref
 from typing import Any
 
@@ -50,20 +50,27 @@
 # `gh api repos/google/tensorstore/compare/ed51718085...vX.Y.Z` says "ahead" or "identical"
 # for a release that has it, and `WORLDARRAY_STRESS=1 pytest tests/test_lost_wakeup.py -s`
 # then strands nothing on a plain context.
+#
+# The thread stops at exit, before the interpreter finalizes. Python up to 3.13 ends a
+# daemon thread that takes the GIL back during finalization with pthread_exit, and inside
+# a tensorstore call that aborts the process: "terminate called without an active
+# exception", exit 134, once all the work is done (mesh-n-bone's CI, 2026-10-26).
 KICK_EVERY = 1.0
+KICK_TIMEOUT = 10.0
 _KICK_DIR = tempfile.gettempdir()
 _KICK_KEY = "worldarray-kick-no-such-key"
 _watched: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary()
 _watch_lock = threading.Lock()
 _watchdog: threading.Thread | None = None
+_stop = threading.Event()
 
 
 def _kickers(context: ts.Context) -> tuple:
     """An in-memory array and a local kvstore on ``context``: reading the one copies on its
     decoding pool, reading a missing key from the other runs on its file I/O pool."""
     array = ts.open({"driver": "array", "array": [0], "dtype": "uint8"}, context=context)
     kv = ts.KvStore.open({"driver": "file", "path": _KICK_DIR + "/"}, context=context)
-    return array.result(timeout=10), kv.result(timeout=10)
+    return array.result(timeout=KICK_TIMEOUT), kv.result(timeout=KICK_TIMEOUT)
 
 
 def kick(context: ts.Context) -> list:
@@ -86,38 +93,50 @@ def kick(context: ts.Context) -> list:
 
 def watch(context: ts.Context | None) -> ts.Context | None:
     """Kick ``context``'s pools every ``KICK_EVERY`` seconds for as long as it lives, starting
-    the watchdog thread on first use. Returns ``context``."""
+    the watchdog thread on first use (not once it has stopped at exit). Returns
+    ``context``."""
     global _watchdog
     if context is None:
         return None
     with _watch_lock:
         if context not in _watched:
             _watched[context] = None
-        if _watchdog is None or not _watchdog.is_alive():
+        if not _stop.is_set() and (_watchdog is None or not _watchdog.is_alive()):
             _watchdog = threading.Thread(target=_watch, name="worldarray-kick", daemon=True)
             _watchdog.start()
     return context
 
 
 def _watch() -> None:
     held: list = []
-    while True:
-        time.sleep(KICK_EVERY)
+    while not _stop.wait(KICK_EVERY):
         with _watch_lock:
             contexts = list(_watched.keys())
         held = [f for f in held if not f.done()]
         for context in contexts:
+            if _stop.is_set():
+                break
             held += kick(context)
         del contexts
 
 
+def _stop_watchdog() -> None:
+    """End every watchdog thread and wait for it, at exit: atexit runs once the program's
+    other threads are joined, so they are kicked to the end, and before finalization."""
+    _stop.set()
+    for thread in threading.enumerate():
+        if thread.name == "worldarray-kick" and thread is not threading.current_thread():
+            thread.join(2 * KICK_TIMEOUT)  # the longest one kick takes: two opens
+
+
 def _forget_watchdog() -> None:
     global _watchdog
     _watchdog = None  # a forked child has no such thread; the next watch() starts one
 
 
 if hasattr(os, "register_at_fork"):
     os.register_at_fork(after_in_child=_forget_watchdog)
+atexit.register(_stop_watchdog)
 
 
 THREADS_VAR = "WORLDARRAY_THREADS"

tests/test_access.py

@@ -8,6 +8,9 @@
 import logging
 import os
 import socket
+import subprocess
+import sys
+import textwrap
 import threading
 import time
 import weakref
@@ -205,6 +208,32 @@ def test_every_context_the_package_touches_is_kicked(monkeypatch, tmp_path):
     assert len(wa.context._watched) == n - 1  # a context that died is no longer watched
 
 
+def test_a_process_exits_cleanly_while_the_watchdog_kicks(tmp_path):
+    # Python up to 3.13 ends a daemon thread that takes the GIL back during finalization
+    # with pthread_exit, which aborts inside a tensorstore call; the watchdog stops at exit
+    script = tmp_path / "exit.py"
+    script.write_text(
+        textwrap.dedent("""
+            import time
+            import tensorstore as ts
+            from worldarray import context
+            context.KICK_EVERY = 0.0  # kick flat out
+            limits = [{"cache_pool": {"total_bytes_limit": i}} for i in range(20)]
+            contexts = [context.watch(ts.Context(spec)) for spec in limits]
+            class SlowExit:  # finalization takes a while, with the GIL free
+                def __del__(self):
+                    time.sleep(0.5)
+            slow = SlowExit()
+            time.sleep(0.2)
+        """)
+    )
+    done = subprocess.run(
+        [sys.executable, str(script)], capture_output=True, text=True, timeout=120
+    )
+    assert done.returncode == 0, done.stderr
+    assert "terminate" not in done.stderr
+
+
 class _StalledView:
     """A view whose first ``stalls`` reads never finish (a future nothing resolves, as a
     read stuck on NFS looks from Python); later reads return ``0, 1, 2, 3``."""