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