Merge remote-tracking branch 'origin/main' into perf/523-image-memory-2

# Conflicts:
#	TESTING.md
This commit is contained in:
Vito Sansevero
2026-10-31 16:09:18 -07:00
13 changed files with 442 additions and 68 deletions
+2
View File
@@ -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.
+13 -2
View File
@@ -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.
+1 -1
View File
@@ -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"]
+1 -1
View File
@@ -1,5 +1,5 @@
# Runtime dependencies
watchdog>=6.0.0,<7
watchdog>=6.0.0
Pillow>=12.3.0
aiohttp>=3.14.5
+1 -1
View File
@@ -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
+60
View File
@@ -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 = {
+78 -1
View File
@@ -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):
+5 -2
View File
@@ -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")
}
+27
View File
@@ -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()}
+107 -44
View File
@@ -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)
+12
View File
@@ -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 = {
+48 -5
View File
@@ -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]:
+87 -11
View File
@@ -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