diff --git a/README.md b/README.md index e71786e3..cc783519 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,8 @@ Searches your library by text, category, tags and minimum rating, then outputs t ![Prompt Search List node producing three images](assets/readme/node-search.webp) +To get the images of a Search List batch linked to their prompts, wire the Search List output into a Prompt Manager Text node, not straight into CLIP Text Encode. Wired straight in, the images stay unlinked: the search text is never saved as a prompt. + ## Install **Requirements:** ComfyUI and Python 3.12 or newer. `watchdog` and `Pillow` are the only dependencies of their own; the rest comes with ComfyUI. diff --git a/TESTING.md b/TESTING.md index 2af0a009..03229a2b 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), kills ffmpeg and shuts the interactive image pool down. 4. `shutdown_job_pool` (`utils/parallel.py`) cancels the long-lived job pool's @@ -364,6 +366,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_file_prompt.py b/tests/test_file_prompt.py index 8b525659..2ecb2d3d 100644 --- a/tests/test_file_prompt.py +++ b/tests/test_file_prompt.py @@ -564,6 +564,66 @@ class TestUndecidedGraph(unittest.TestCase): self.assertEqual(decision.review, "untraced") +def _search_list(text="fox", side="positive"): + """f22: a PromptSearchList batch (no PromptManager node) feeds CLIPTextEncode.""" + graph = parity.build("f22")["prompt"] + graph["10"]["inputs"]["text"] = text + if side == "negative": + # The search feeds the negative; the positive is typed text. + graph["6"]["inputs"]["text"] = parity.POS + graph["7"]["inputs"]["text"] = ["10", 0] + return graph + + +class TestSearchTextIsNeverAPrompt(unittest.TestCase): + """#548: the PromptSearchList query is a search, never the prompt.""" + + def test_the_search_text_is_never_linked_or_stored(self): + for policy in Policy: + with self.subTest(policy=policy): + decision = _decide( + _graph(_search_list()), policy=policy, lookup=_Lookup() + ) + self.assertEqual(decision.kind, "never") + self.assertIsNone(decision.text) + self.assertEqual(decision.unlinked_status, "unresolved") + self.assertTrue(decision.try_recorded_run) + + def test_a_search_text_from_a_primitive_is_never_linked(self): + graph = _search_list() + graph["10"]["inputs"]["text"] = ["30", 0] + graph["30"] = {"class_type": "PrimitiveString", "inputs": {"value": "fox"}} + for policy in Policy: + with self.subTest(policy=policy): + decision = _decide(_graph(graph), policy=policy, lookup=_Lookup()) + self.assertEqual(decision.kind, "never") + self.assertIsNone(decision.text) + + def test_an_empty_search_text_is_unchanged(self): + """Regression guard (passes on the base): nothing to exclude.""" + for policy in Policy: + with self.subTest(policy=policy): + decision = _decide( + _graph(_search_list("")), policy=policy, lookup=_Lookup() + ) + self.assertEqual((decision.kind, decision.text), ("never", None)) + + def test_without_a_sampler_any_search_list_query_is_excluded(self): + graph = _search_list() + del graph["3"] + self.assertTrue(file_prompt._search_text(graph, " fox ")) + self.assertFalse(file_prompt._search_text(graph, "a fox")) + self.assertFalse(file_prompt._search_text(None, "fox")) + + def test_a_search_on_the_negative_keeps_the_typed_positive(self): + """Regression guard (passes on the base): only the positive's source + is excluded, so the legacy answer from typed text still links.""" + graph = _search_list(side="negative") + decision = _decide(_graph(graph), lookup=_Lookup()) + self.assertEqual((decision.kind, decision.text), ("positive", parity.POS)) + self.assertEqual(decision.source, "legacy-parser") + + class TestStepsWithoutAnUndecidedGraph(unittest.TestCase): def test_step_9_ambiguous_parameters_only(self): meta = { 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 1ac151bd..bf05f17b 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_legacy_parity_gate.py b/tests/test_legacy_parity_gate.py index 4b7a7980..b5cf1d59 100644 --- a/tests/test_legacy_parity_gate.py +++ b/tests/test_legacy_parity_gate.py @@ -108,6 +108,15 @@ LEGACY_WRONG = { "a cozy library with tall shelves", "legacy shows branch 1; the file name names branch 4's saver", ), + # #548, the one shape where nothing is read on purpose: legacy shows a + # PromptSearchList's search text ("fox") as the positive, but that is the + # library query, not the prompt the batch ran. The scan and the monitor + # never store or link it either (utils.file_prompt._search_text). + ("inspection_parity", "f22"): ( + None, + "legacy shows the PromptSearchList search text; each item is decided " + "at run time, so no positive is read (#548)", + ), } @@ -149,6 +158,24 @@ class LegacyParityGateTest(unittest.TestCase): self.assertEqual(text, expected) self.assertNotEqual(text, legacy) + def test_the_search_list_query_is_never_stored_either(self): + """#548: the scan and the monitor depart from legacy for f22 too.""" + from utils.file_prompt import Policy, decide_file_prompt + + graph, workflow = corpora()["inspection_parity"]["f22"] + self.assertEqual(legacy_positive(graph, workflow), "fox") + for policy in Policy: + with self.subTest(policy=policy): + decision = decide_file_prompt( + {"prompt": graph, "workflow": workflow}, + file_name="f22.png", + file_size=None, + in_output=True, + lookup=lambda: None, + policy=policy, + ) + self.assertEqual((decision.kind, decision.text), ("never", None)) + def test_no_graph_leaves_the_legacy_agreement(self): recorded = json.loads(AGREEMENT_PATH.read_text(encoding="utf-8")) current = {name: agreeing(graphs) for name, graphs in corpora().items()} 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/tests/test_pair_scan.py b/tests/test_pair_scan.py index 3f91f41d..aa0c5f1f 100644 --- a/tests/test_pair_scan.py +++ b/tests/test_pair_scan.py @@ -90,6 +90,18 @@ class TestPairScan(AdminAPITestCase): self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0) self.assertEqual(self._texts(), []) + def test_a_prompt_search_list_query_is_never_stored_as_a_prompt(self): + # #548: the search text was stored and every batch image linked to it. + g = sampled_graph() + g["10"] = { + "class_type": "PromptSearchList", + "inputs": {"text": "fox", "category": "", "tags": "", "limit": 50}, + } + g["1"]["inputs"]["text"] = ["10", 0] + self.assertEqual(self._scanned("search.png", g), "unresolved") + self.assertEqual(self._texts(), []) + self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0) + def test_numeric_widgets_are_never_stored_as_a_prompt(self): # The pre-redesign scan stored "2048" from a resolution widget here. g = { diff --git a/utils/file_prompt.py b/utils/file_prompt.py index 0061277f..230c6c99 100644 --- a/utils/file_prompt.py +++ b/utils/file_prompt.py @@ -230,6 +230,9 @@ def _undecided_graph(found: _File) -> Decision: if switched is not None: # The legacy answer is a switched-off node's own text (#90) return Decision(switched, canonical=found.canonical) + if text is not None and _search_text(found.canonical["prompt"], text): + # The query of a Search List batch is a search, never a prompt (#548) + return _never(found, _unlinked(found), try_recorded_run=True) if text is not None and _joined_text(found.canonical["prompt"], text): # The legacy parser joins prepend + text + append: the text a # PromptManager node stores cannot be told from it (#343). @@ -311,6 +314,8 @@ _NOT_TEXT_INPUTS = frozenset( } ) _MAX_SWITCH_VISITS = 5000 +# The batch node whose ``text`` input is a library search query (#548). +SEARCH_LIST_TYPE = "PromptSearchList" def _negative_name(key: Any) -> bool: @@ -328,10 +333,28 @@ def _switched_off_upstream(graph: Any) -> Optional[str]: """ if not isinstance(graph, dict): return None - roots = _positive_roots(graph) - if not roots: + upstream = _positive_upstream(graph) + if upstream is None: ids = _prompt_manager_ids(graph) return _switch_kind(save_setting(graph, node_id) for node_id in ids) + found = [ + node_id + for node_id in upstream + if graph[node_id].get("class_type") in PROMPT_MANAGER_TYPES + ] + return _switch_kind(save_setting(graph, node_id) for node_id in found) + + +def _positive_upstream(graph: dict) -> Optional[set]: + """The ids of the nodes upstream of a sampler's positive, or None when + the graph has no recognisable sampler. + + Follows every input except negative and non-text ones (#90's walk follows + only prompt-like names). + """ + roots = _positive_roots(graph) + if not roots: + return None seen, found = set(), set() stack = [str(link[0]) for link in roots] while stack and len(seen) < _MAX_SWITCH_VISITS: @@ -342,8 +365,7 @@ def _switched_off_upstream(graph: Any) -> Optional[str]: node = graph.get(node_id) if not isinstance(node, dict): continue - if node.get("class_type") in PROMPT_MANAGER_TYPES: - found.add(node_id) + found.add(node_id) inputs = node.get("inputs") for key, value in (inputs.items() if isinstance(inputs, dict) else ()): if key in _NOT_TEXT_INPUTS or _negative_name(key): @@ -351,7 +373,28 @@ def _switched_off_upstream(graph: Any) -> Optional[str]: if isinstance(value, (list, tuple)) and len(value) == 2: if isinstance(value[0], (str, int)) and isinstance(value[1], int): stack.append(str(value[0])) - return _switch_kind(save_setting(graph, node_id) for node_id in found) + return found + + +def _search_text(graph: Any, text: str) -> bool: + """Whether ``text`` is the search query of a PromptSearchList node that + feeds a positive (any such node when no sampler is recognised). + + The legacy parser traces a positive wired from a Search List to the + node's ``text`` input, which is the database search, never the prompt + the batch ran (#548). Each item is counted by the node at run time. + """ + if not isinstance(graph, dict): + return False + upstream = _positive_upstream(graph) + ids = graph if upstream is None else upstream + wanted = text.strip() + return any( + isinstance(graph[node_id], dict) + and graph[node_id].get("class_type") == SEARCH_LIST_TYPE + and resolve_text(graph, node_id) == wanted + for node_id in ids + ) def _off_text(graph: Any, text: str) -> Optional[str]: 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