From 93bf24f47dd7a82fe82c7f526aa502748d94ac80 Mon Sep 17 00:00:00 2001 From: Vito Date: Sat, 31 Oct 2026 21:31:55 +0000 Subject: [PATCH] fix(monitor): keep watchdog threads from holding the exit (#561) (#569) * fix(monitor): keep watchdog threads from holding the exit (#561) watchdog 7.0.0 on Windows starts a plain threading.Thread per watch for its ReadDirectoryChangesW reader (DirectoryChangeReader.start). Created from ComfyUI's main thread it is not a daemon, and the interpreter joins every non-daemon thread before it runs any atexit callback, so the package teardown that would stop the observer never ran and the process never exited. Start the observer from a short-lived daemon thread, so every thread it creates inherits the daemon flag. Run observer.stop(), which joins the emitters without a limit, on a daemon thread and wait for it only until the stop deadline; a later stop waits for the same call. Lift the watchdog<7 cap. Closes #561 * test(exit): exempt watchdog 7's process-wide inotify reader (#561) watchdog 7 on Linux reads every inotify watch through one InotifyFD daemon thread per process that no observer stop ends, by design. The exit and package tests counted every watchdog BaseThread as the package's, so the singleton failed them once the cap was lifted. Exclude it, and require instead that no non-daemon thread is left at exit. --- TESTING.md | 15 ++- pyproject.toml | 2 +- requirements-dev.txt | 2 +- requirements.txt | 2 +- tests/test_image_monitor_singleton.py | 79 +++++++++++++- tests/test_init_package.py | 7 +- tests/test_package_exit.py | 151 ++++++++++++++++++-------- utils/image_monitor.py | 98 +++++++++++++++-- 8 files changed, 293 insertions(+), 63 deletions(-) diff --git a/TESTING.md b/TESTING.md index 1f07393f..df7c6695 100644 --- a/TESTING.md +++ b/TESTING.md @@ -345,8 +345,10 @@ and the time left. The steps, in order: monitor's callbacks (Needs review recorder, preview queue, Images listing listener). 2. `ImageMonitor.stop_monitoring` works against one 5 s deadline. Closing the - image handler joins the image worker; watchdog's `observer.stop()` (not - bounded here) and the observer join get what is left, which can be 0 s. + image handler joins the image worker; watchdog's `observer.stop()` and the + observer join get what is left, which can be 0 s. `stop()` joins watchdog's + emitters without a limit, so it runs on a daemon thread and is waited for + only until the deadline; a later stop waits for that same call (#561). 3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s) and kills ffmpeg. 4. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s). @@ -361,6 +363,15 @@ stored, with the parts not reached False, and the `atexit` callback is removed: the later steps are skipped, no summary line is logged, and a later call returns that result and runs nothing. +The observer is started from a short-lived daemon thread, so every thread it +creates is a daemon (a thread inherits that flag from the thread that creates +it). watchdog 7 on Windows starts a plain `threading.Thread` per watch for its +`ReadDirectoryChangesW` reader; created from ComfyUI's main thread it would not +be a daemon, and the interpreter waits for every such thread before it runs any +`atexit` callback, so the teardown would never start (#561). +`tests/test_package_exit.py` runs a child whose observer does both (a plain +reader thread, a `stop()` that never returns) and requires its exit. + ComfyUI never runs aiohttp's `on_cleanup` hooks, because it never calls `runner.cleanup()`. So `atexit` is the live host's path. It runs on Ctrl+C (`main.py` catches `KeyboardInterrupt` and returns) and on a normal exit. diff --git a/pyproject.toml b/pyproject.toml index 260aa1aa..7a065ad3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ description = "A powerful ComfyUI custom node that extends the standard text enc version = "3.4.1" license = {file = "LICENSE"} requires-python = ">=3.12" -dependencies = ["watchdog>=2.1.0,<7", "Pillow>=8.0.0"] +dependencies = ["watchdog>=2.1.0", "Pillow>=8.0.0"] [build-system] requires = ["setuptools>=84.2.1"] diff --git a/requirements-dev.txt b/requirements-dev.txt index aa8fbc5a..6db215e2 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,5 +1,5 @@ # Runtime dependencies -watchdog>=6.0.0,<7 +watchdog>=6.0.0 Pillow>=12.3.0 aiohttp>=3.14.5 diff --git a/requirements.txt b/requirements.txt index 89160017..550aff47 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ # Runtime dependencies for PromptManager. # aiohttp, numpy and torch come from ComfyUI itself and are not listed here. -watchdog>=2.1.0,<7 # Output directory monitoring +watchdog>=2.1.0 # Output directory monitoring Pillow>=8.0.0 # Image metadata extraction and thumbnails diff --git a/tests/test_image_monitor_singleton.py b/tests/test_image_monitor_singleton.py index 658404c7..aec62710 100644 --- a/tests/test_image_monitor_singleton.py +++ b/tests/test_image_monitor_singleton.py @@ -392,7 +392,10 @@ class TestStartStop(unittest.TestCase): self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs, ): self.assertFalse(self.monitor.stop_monitoring(timeout=5.0)) - self.assertEqual(fake.calls[-1], ("join", 0.0)) + # The stop runs on its own thread and gets 0 s too: it may not have + # returned yet, and then nothing is joined at all. + joins = [call for call in fake.calls if call[0] == "join"] + self.assertTrue(all(call == ("join", 0.0) for call in joins), fake.calls) self.assertIn( "watchdog observer still running after 0.0 s", "\n".join(logs.output) ) @@ -401,6 +404,80 @@ class TestStartStop(unittest.TestCase): def test_stop_monitoring_without_an_observer_returns_true(self): self.assertTrue(self.monitor.stop_monitoring()) + def test_threads_the_observer_starts_are_daemons(self): + # watchdog 7 on Windows starts a plain threading.Thread per watch in + # observer.start(); a non-daemon one holds the interpreter's exit + # before any atexit callback runs (#561). + fake = ReaderObserver() + self.addCleanup(fake.release.set) + self._start_with(fake) + self.assertIsNotNone(fake.reader) + self.assertTrue(fake.reader.daemon) + self.assertFalse(threading.current_thread().daemon) + + def test_an_observer_that_fails_to_start_on_its_thread_is_reported(self): + fake = FakeObserver() + fake.start = MagicMock(side_effect=OSError("handle refused")) + with self.assertLogs("prompt_manager.image_monitor", "ERROR") as logs: + with patch.object(im_mod, "Observer", return_value=fake): + self.monitor.start_monitoring([self.tmp.name]) + self.assertIn("handle refused", "\n".join(logs.output)) + self.assertFalse(self.monitor.is_monitoring) + + def test_an_observer_stop_that_never_returns_is_bounded(self): + # watchdog's observer.stop() joins its emitters (and, in 7.x on + # Windows, their reader threads) without a limit (#561). + fake = StuckStopObserver() + self.addCleanup(fake.release.set) + self._start_with(fake) + with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs: + self.assertFalse(self.monitor.stop_monitoring(timeout=0.05)) + self.assertIn("watchdog observer still running", "\n".join(logs.output)) + self.assertTrue(fake.entered.wait(30)) + self.assertFalse(self.monitor.is_monitoring) + # A second stop waits for the stop already running; it never calls + # observer.stop() a second time. + fake.release.set() + self.assertTrue(self.monitor.stop_monitoring()) + self.assertEqual(fake.stop_calls, 1) + self.assertFalse(self.monitor._unfinished) + + +class ReaderObserver(FakeObserver): + """A FakeObserver whose start() starts a plain thread, as watchdog 7's + Windows emitter does for its ReadDirectoryChangesW reader.""" + + def __init__(self): + super().__init__() + self.release = threading.Event() + self.reader = None + + def start(self): + self.reader = threading.Thread(target=self.release.wait) + self.reader.start() + super().start() + + def stop(self): + self.release.set() + self.reader.join(30) + super().stop() + + +class StuckStopObserver(FakeObserver): + """A FakeObserver whose stop() does not return until ``release``.""" + + def __init__(self): + super().__init__() + self.entered = threading.Event() + self.release = threading.Event() + self.stop_calls = 0 + + def stop(self): + self.stop_calls += 1 + self.entered.set() + self.release.wait(30) + super().stop() + class FakeEvent: def __init__(self, path, is_directory=False): diff --git a/tests/test_init_package.py b/tests/test_init_package.py index 87646e36..42df3b58 100644 --- a/tests/test_init_package.py +++ b/tests/test_init_package.py @@ -596,13 +596,16 @@ PACKAGE_THREAD_NAMES = ("PromptManagerPromptCleanup", "PromptManagerImageWorker" def package_threads(): """Live threads this design owns: the tracker, the image worker, and the - watchdog observer and emitters (which have no name of their own).""" + watchdog observer and emitters (which have no name of their own). + watchdog 7's InotifyFD, one daemon reader per process that no observer + stop ends, is not the package's.""" from watchdog.utils import BaseThread return { thread for thread in threading.enumerate() - if thread.name in PACKAGE_THREAD_NAMES or isinstance(thread, BaseThread) + if thread.name in PACKAGE_THREAD_NAMES + or (isinstance(thread, BaseThread) and type(thread).__name__ != "InotifyFD") } diff --git a/tests/test_package_exit.py b/tests/test_package_exit.py index 3b8c5f7e..724e79a0 100644 --- a/tests/test_package_exit.py +++ b/tests/test_package_exit.py @@ -31,22 +31,65 @@ CHILD = textwrap.dedent(""" import threading root, tmp = sys.argv[1], sys.argv[2] + stuck = sys.argv[3:] == ["stuck-observer"] sys.path.insert(0, root) state = {} + # watchdog 7's Windows observer at its worst (#561): start() starts a + # plain reader thread, and stop() waits for it, which never ends. + class StuckObserver: + + def __init__(self, *args, **kwargs): + self._never = threading.Event() + self._reader = None + + def schedule(self, *args, **kwargs): + pass + + def start(self): + self._reader = threading.Thread( + target=self._never.wait, name="StuckReader" + ) + self._reader.start() + + def stop(self): + self._reader.join() + + def join(self, timeout=None): + self._reader.join(timeout) + + def is_alive(self): + return self._reader.is_alive() + + if stuck: + import watchdog.observers + + watchdog.observers.Observer = StuckObserver + def report(): from watchdog.utils import BaseThread package = state.get("package") result = getattr(package, "_background_work_result", None) # The image worker, the tracker, and watchdog's observer and emitters. + # watchdog 7's InotifyFD is one daemon reader per process that no + # observer stop ends, by design; it is not the package's. alive = sorted( t.name for t in threading.enumerate() if t.name in ("PromptManagerImageWorker", "PromptManagerPromptCleanup") - or isinstance(t, BaseThread) + or (isinstance(t, BaseThread) and type(t).__name__ != "InotifyFD") + ) + # Threads that would hold the exit before any atexit callback (#561). + holding = sorted( + t.name + for t in threading.enumerate() + if not t.daemon and t is not threading.main_thread() + ) + print( + "PM_EXIT_REPORT " + + json.dumps({"result": result, "alive": alive, "holding": holding}) ) - print("PM_EXIT_REPORT " + json.dumps({"result": result, "alive": alive})) sys.stdout.flush() # First registered, so it runs last: after the package's teardown and @@ -64,6 +107,10 @@ CHILD = textwrap.dedent(""" fake_server_module(queue_task_done=True), ) state["package"] = package + if stuck: + # The stop gives up after this long instead of the full 5 s. + monitor_module = sys.modules[package.__name__ + ".utils.image_monitor"] + monitor_module.OBSERVER_JOIN_SECONDS = 0.5 monitor = package._global_image_monitor handler = monitor.handler if monitor is not None else None if handler is None: @@ -78,47 +125,52 @@ CHILD = textwrap.dedent(""" """) +def run_child(testcase, *args): + """Run CHILD with ``args``; returns (completed, report, detail).""" + with tempfile.TemporaryDirectory() as tmp: + os.makedirs(os.path.join(tmp, "output")) + script = os.path.join(tmp, "child.py") + with open(script, "w", encoding="utf-8") as handle: + handle.write(CHILD) + env = dict(os.environ) + env.pop("PROMPT_MANAGER_CONFIG_PATH", None) + # The child's library and log file live in its own folder, never + # in the checkout's logs/ (which this runner may hold open) or a + # data folder the caller set. + env["PROMPT_MANAGER_DATA_DIR"] = tmp + env["PYTHONIOENCODING"] = "utf-8" + try: + completed = subprocess.run( + [sys.executable, script, ROOT, tmp, *args], + cwd=ROOT, + env=env, + capture_output=True, + timeout=CHILD_CEILING_SECONDS, + encoding="utf-8", + errors="replace", + ) + except subprocess.TimeoutExpired as hung: + testcase.fail( + f"the child did not exit within {CHILD_CEILING_SECONDS} s\n" + f"stdout:\n{hung.stdout}\nstderr:\n{hung.stderr}" + ) + # Leaving the block removes the folder: on Windows that fails + # if the child left a handle open, which it cannot once it exited, + # so the check here is that the child exited at all. + detail = f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" + testcase.assertEqual(completed.returncode, 0, detail) + reports = [ + line[len(REPORT_PREFIX) :] + for line in completed.stdout.splitlines() + if line.startswith(REPORT_PREFIX) + ] + testcase.assertEqual(len(reports), 1, detail) + return completed, json.loads(reports[0]), detail + + class TestPackageExit(unittest.TestCase): def test_teardown_runs_at_a_normal_exit_with_the_worker_alive(self): - with tempfile.TemporaryDirectory() as tmp: - os.makedirs(os.path.join(tmp, "output")) - script = os.path.join(tmp, "child.py") - with open(script, "w", encoding="utf-8") as handle: - handle.write(CHILD) - env = dict(os.environ) - env.pop("PROMPT_MANAGER_CONFIG_PATH", None) - # The child's library and log file live in its own folder, never - # in the checkout's logs/ (which this runner may hold open) or a - # data folder the caller set. - env["PROMPT_MANAGER_DATA_DIR"] = tmp - env["PYTHONIOENCODING"] = "utf-8" - try: - completed = subprocess.run( - [sys.executable, script, ROOT, tmp], - cwd=ROOT, - env=env, - capture_output=True, - timeout=CHILD_CEILING_SECONDS, - encoding="utf-8", - errors="replace", - ) - except subprocess.TimeoutExpired as hung: - self.fail( - f"the child did not exit within {CHILD_CEILING_SECONDS} s\n" - f"stdout:\n{hung.stdout}\nstderr:\n{hung.stderr}" - ) - # Leaving the block removes the folder: on Windows that fails - # if the child left a handle open, which it cannot once it exited, - # so the check here is that the child exited at all. - detail = f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" - self.assertEqual(completed.returncode, 0, detail) - reports = [ - line[len(REPORT_PREFIX) :] - for line in completed.stdout.splitlines() - if line.startswith(REPORT_PREFIX) - ] - self.assertEqual(len(reports), 1, detail) - report = json.loads(reports[0]) + completed, report, detail = run_child(self) # "thumbnails" is True here because no preview worker started; its # close result is tested in test_thumbnail_auto and test_init_package. self.assertEqual( @@ -127,6 +179,7 @@ class TestPackageExit(unittest.TestCase): detail, ) self.assertEqual(report["alive"], [], detail) + self.assertEqual(report["holding"], [], detail) # The teardown's own warnings. Import-time warnings (the API's # output-folder search finds no ComfyUI here) are not about exit. for text in ( @@ -137,6 +190,16 @@ class TestPackageExit(unittest.TestCase): ): self.assertNotIn(text, completed.stderr, detail) - -if __name__ == "__main__": - unittest.main() + def test_an_observer_that_never_stops_does_not_hold_the_exit(self): + # Before #561 the reader thread kept the interpreter waiting before + # any atexit callback ran, so the teardown never started. + completed, report, detail = run_child(self, "stuck-observer") + self.assertEqual( + report["result"], + {"monitor": False, "thumbnails": True, "tracker": True}, + detail, + ) + # The stuck reader is still alive, but as a daemon it holds nothing. + self.assertEqual(report["holding"], [], detail) + self.assertIn("watchdog observer still running", completed.stderr, detail) + self.assertIn("Background work not stopped in time", completed.stderr, detail) diff --git a/utils/image_monitor.py b/utils/image_monitor.py index 452bf104..eb4c030a 100644 --- a/utils/image_monitor.py +++ b/utils/image_monitor.py @@ -96,6 +96,65 @@ WORKER_JOIN_SECONDS = 5.0 OBSERVER_JOIN_SECONDS = 5.0 +def _start_daemon_observer(observer) -> None: + """``observer.start()`` on a short-lived daemon thread (#561). + + A thread is a daemon when the thread that creates it is one. watchdog + 7 on Windows starts a plain ``threading.Thread`` per watch inside + ``start()``; created from ComfyUI's main thread it is not a daemon, and + the interpreter waits for every such thread before it runs any + ``atexit`` callback, so the teardown that would stop it never runs. + Started from here, every thread the observer creates is a daemon. + Waits for the start without a limit (start-up, not exit); re-raises + what ``start()`` raised. + """ + raised: List[BaseException] = [] + + def start() -> None: + try: + observer.start() + except BaseException as e: # re-raised on the caller's thread + raised.append(e) + + starter = threading.Thread( + target=start, name="PromptManagerObserverStart", daemon=True + ) + starter.start() + starter.join() + if raised: + raise raised[0] + + +class _ObserverStop: + """One ``observer.stop()`` running on a daemon thread (#561). + + watchdog's ``stop()`` joins its emitters (and, in 7.x on Windows, their + reader threads) without a limit. Run here, a stop that never returns + holds neither the caller past its deadline nor the interpreter's exit. + """ + + def __init__(self, observer) -> None: + self.error: Optional[Exception] = None + self._thread = threading.Thread( + target=self._run, + args=(observer,), + name="PromptManagerObserverStop", + daemon=True, + ) + self._thread.start() + + def _run(self, observer) -> None: + try: + observer.stop() + except Exception as e: # reported by the stop that waits for it + self.error = e + + def wait(self, timeout: float) -> bool: + """True once ``stop()`` has returned or raised.""" + self._thread.join(timeout) + return not self._thread.is_alive() + + class ImageGenerationHandler(FileSystemEventHandler): """Filesystem event handler for detecting new image generation. @@ -1170,6 +1229,9 @@ class ImageMonitor: # (observer, handler) pairs a stop left running: the next stop waits # for them again instead of reporting True. self._unfinished: List[Tuple[Any, Any]] = [] + # Observer stops still running on their thread, by observer: a later + # stop waits for the same call instead of starting another (#561). + self._observer_stops: Dict[Any, _ObserverStop] = {} self.logger = get_logger("prompt_manager.image_monitor") self.review_recorder: Optional[Callable[[str], str]] = None self.thumbnail_queue: Optional[Callable[[str], str]] = None @@ -1273,7 +1335,7 @@ class ImageMonitor: self.logger.warning("No directory could be watched") return try: - observer.start() + _start_daemon_observer(observer) except Exception as e: self.logger.error(f"Image monitoring could not start: {e}") return @@ -1303,12 +1365,12 @@ class ImageMonitor: Closes the handler first (its worker is joined and its dispatch returns at once), then stops the watchdog observer and joins it. The - worker and the observer share the one deadline: the observer's join - gets what the worker left, possibly 0 s. - watchdog's ``observer.stop()`` joins its emitters without a limit; - it counts against the deadline but is not bounded here. A part an - earlier call left running is waited for again. Safe to call when - monitoring never started and more than once. + worker and the observer share the one deadline: the observer's stop + and join get what the worker left, possibly 0 s. watchdog's + ``observer.stop()`` joins its emitters without a limit, so it runs on + a daemon thread and is waited for only until the deadline (#561). A + part an earlier call left running is waited for again. Safe to call + when monitoring never started and more than once. Returns: True when every worker and observer it stopped has ended. @@ -1349,13 +1411,27 @@ class ImageMonitor: return False def _stop_observer(self, observer, timeout: float) -> bool: - """``observer.stop()`` (safe to repeat), then a join of ``timeout`` - seconds; a join that times out is logged with the time it got.""" + """``observer.stop()`` on its own daemon thread, then a join, both + within ``timeout`` seconds; a stop or join still running then is + logged with the time it got. A stop still running from an earlier + call is waited for, never started again.""" if observer is None: return True + deadline = time.monotonic() + timeout + stop = self._observer_stops.get(observer) + if stop is None: + stop = self._observer_stops[observer] = _ObserverStop(observer) + if not stop.wait(timeout): + self.logger.warning( + f"watchdog observer still running after {timeout:.1f} s" + ) + return False + del self._observer_stops[observer] + if stop.error is not None: + self.logger.warning(f"Error while stopping image monitoring: {stop.error}") + return False try: - observer.stop() - observer.join(timeout) + observer.join(max(0.0, deadline - time.monotonic())) except Exception as e: self.logger.warning(f"Error while stopping image monitoring: {e}") return False