nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(thumbnails): keep going past the per-batch cap until every preview is made

Browse Source 
The missing, rebuild and Convert outdated jobs stopped after
MAX_THUMBNAIL_FILES (10,000) files and said "N more on the next run", but
nothing ever started that next run. The cap now bounds one batch. After
each batch the job plans again, leaving out every file it already worked
on, and continues until nothing is left, the job is cancelled, or a batch
makes no progress. That last case stops on files that fail every time
instead of looping over them.

Activity progress counts against the overall total, and the final line and
the legacy stream's complete frame report every batch. Admission, worker
threads, cancellation and the #316 rules are unchanged. The Thumbnail
maintenance dialog now counts all the work instead of saying "more on the
next run".
This commit is contained in:
Vito Sansevero
2026-10-24 16:57:03 -07:00
parent af0482a12d
commit 726b7b70da
9 changed files with 318 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/api/images.py
+25 -8
View File
@@ -1991,12 +1991,15 @@ class ImageRoutesMixin:
mode: str = "missing",
stop=None,
settings=None,
exclude=None,
) -> dict:
"""Plan every gallery root with one shared per-run generation cap.
"""Plan every gallery root with one shared per-batch generation cap.
See ``_plan_thumbnails``; ``fresh_files`` keeps at most
MAX_FRESH_LISTED files across the roots. ``stop()`` true ends the walk
at the next file (a cancelled job); the partial plan is returned.
*exclude* (a set of source paths) leaves files a job already worked
on out of a later batch's plan.
"""
settings = writer_settings(settings)
combined = {key: 0 for key in _PLAN_COUNTS}
@@ -2006,7 +2009,12 @@ class ImageRoutesMixin:
break
root = Path(root)
plan = self._plan_thumbnails(
root, root / "thumbnails", mode=mode, stop=stop, settings=settings
root,
root / "thumbnails",
mode=mode,
stop=stop,
settings=settings,
exclude=exclude,
)
available = max(0, MAX_THUMBNAIL_FILES - len(combined["targets"]))
combined["targets"].extend(plan["targets"][:available])
@@ -2054,7 +2062,13 @@ class ImageRoutesMixin:
)
def _plan_thumbnails(
self, output_path, thumbnails_dir, mode="missing", stop=None, settings=None
self,
output_path,
thumbnails_dir,
mode="missing",
stop=None,
settings=None,
exclude=None,
):
"""Decide which media files need a preview (blocking walk).
@@ -2071,9 +2085,10 @@ class ImageRoutesMixin:
``fresh_files`` lists the first MAX_FRESH_LISTED fresh files. The walk
is bounded like the gallery listing (MAX_SCAN_DEPTH, MAX_GALLERY_FILES,
no symlinks), each preview folder is listed once, and at most
MAX_THUMBNAIL_FILES targets are queued per run; ``remaining`` counts
the targets beyond that cap. ``stop()`` true ends the walk before the
next file.
MAX_THUMBNAIL_FILES targets are queued per batch (a job plans again
after each batch); ``remaining`` counts the targets beyond that cap.
Sources in *exclude* (worked on by an earlier batch) are left out.
``stop()`` true ends the walk before the next file.
"""
settings = writer_settings(settings)
output_path = Path(output_path)
@@ -2084,6 +2099,8 @@ class ImageRoutesMixin:
for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
if stop is not None and stop():
break
if exclude and src in exclude:
continue
rel_no_ext = src.relative_to(output_path).with_suffix("")
stem = f"{rel_no_ext.name}{thumbnail_settings.THUMB_SUFFIX}"
dst = thumbnails_dir / rel_no_ext.parent / (stem + settings.extension)
@@ -2109,8 +2126,8 @@ class ImageRoutesMixin:
plan["targets"].append((src, dst, is_video))
if plan["remaining"]:
self.logger.warning(
f"Thumbnail run capped at {MAX_THUMBNAIL_FILES} new files; "
f"{plan['remaining']} more will be generated on the next run"
f"Thumbnail batch capped at {MAX_THUMBNAIL_FILES} files; "
f"{plan['remaining']} more are left for the next batch"
)
return plan
py/api/thumbnail_convert.py
+61 -24
View File
@@ -167,21 +167,14 @@ class ThumbnailConvertMixin:
job.log_line(
f"Conversion started · {folders} · to {options.settings.describe()}"
)
plan = await self._run_in_executor(
self._plan_gallery_thumbnails,
output_dirs,
"convert",
stop=lambda: job.cancelled,
settings=options.settings,
)
plan = await self._plan_conversion(job, options, output_dirs)
if job.cancelled:
return self._end_conversion(job, 0, CANCELLED_BEFORE_CONVERTING)
targets = plan["targets"]
if not targets:
if not plan["targets"]:
job.update(progress=100, status_line=NOTHING_OUTDATED)
job.log_line(f"Conversion finished · {NOTHING_OUTDATED.lower()}")
return "finished"
totals = await self._convert_targets(job, options, targets, output_dirs)
totals, left = await self._convert_in_batches(job, options, plan, output_dirs)
self.invalidate_gallery_cache()
if totals["stopped"]:
job.update(status_line=FORMAT_CHANGED.format(converted=totals["converted"]))
@@ -196,7 +189,7 @@ class ThumbnailConvertMixin:
totals["converted"],
f"Cancelled · {_plural(totals['converted'], 'preview')} converted",
)
line = _finished_line(totals, plan["remaining"])
line = _finished_line(totals, left)
job.update(progress=100, status_line=line)
job.log_line(f"Conversion finished · {line}")
self.logger.info(f"Thumbnail conversion: {line}")
@@ -210,18 +203,59 @@ class ThumbnailConvertMixin:
)
return "cancelled"
async def _convert_targets(self, job, options, targets, output_dirs):
"""Convert *targets* in worker batches of preview groups, recording
each file."""
roots = _scan_roots(output_dirs)
total = len(targets)
groups = _group_targets(targets)
job.update(total=total, status_line=_scan_status_line(0, total, 0, 0))
async def _plan_conversion(self, job, options, output_dirs, exclude=None):
"""One batch's plan (at most MAX_THUMBNAIL_FILES targets), off-loop."""
return await self._run_in_executor(
self._plan_gallery_thumbnails,
output_dirs,
"convert",
stop=lambda: job.cancelled,
settings=options.settings,
exclude=exclude,
)
async def _convert_in_batches(self, job, options, plan, output_dirs):
"""Convert ``plan``, then plan again after each batch until nothing
is left, the job is cancelled or stopped, or a batch converted
nothing (files that fail are never planned again). Returns
``(totals across batches, files left)``."""
totals = {"converted": 0, "failed": 0, "before": 0, "after": 0}
totals["stopped"] = False
start = time.monotonic()
run = {
"index": 0,
"batches": 0,
"total": len(plan["targets"]) + plan["remaining"],
}
run["start"] = time.monotonic()
job.update(
total=run["total"], status_line=_scan_status_line(0, run["total"], 0, 0)
)
attempted = set()
while True:
before = totals["converted"]
await self._convert_targets(
job, options, plan["targets"], output_dirs, totals, run
)
attempted.update(src for src, _dst, _video in plan["targets"])
stop = job.cancelled or totals["stopped"]
if stop or not plan["remaining"] or totals["converted"] == before:
return totals, plan["remaining"]
plan = await self._plan_conversion(
job, options, output_dirs, exclude=attempted
)
if job.cancelled:
return totals, 0
run["total"] = run["index"] + len(plan["targets"]) + plan["remaining"]
job.update(total=run["total"])
if not plan["targets"]:
return totals, plan["remaining"]
async def _convert_targets(self, job, options, targets, output_dirs, totals, run):
"""Convert one batch's *targets* in worker batches of preview groups,
recording each file against the job's overall ``run`` counters."""
roots = _scan_roots(output_dirs)
groups = _group_targets(targets)
batch_size = _thumbnail_batch_size(_worker_threads())
index = batches = 0
for batch_start in range(0, len(groups), batch_size):
if job.cancelled or totals["stopped"]:
break
@@ -235,17 +269,20 @@ class ThumbnailConvertMixin:
totals["stopped"] = True
if result["action"] in ("cancelled", "stopped"):
continue
index += 1
run["index"] += 1
index, total = run["index"], max(run["total"], run["index"])
self._record_conversion(job, src, result, roots, totals)
job.update(processed=index, progress=index * 100 // total)
batches += 1
run["batches"] += 1
job.update(
status_line=_scan_status_line(
index, total, batches, time.monotonic() - start
run["index"],
run["total"],
run["batches"],
time.monotonic() - run["start"],
)
)
await asyncio.sleep(0) # let other requests run between batches
return totals
@staticmethod
def _record_conversion(job, src, result, roots, totals):
py/api/thumbnail_jobs.py
+74 -30
View File
@@ -253,42 +253,84 @@ class ThumbnailJobsMixin:
{"phase": "scanning", "message": "Scanning for images and videos..."},
)
)
plan = await self._run_in_executor(
self._plan_gallery_thumbnails,
output_dirs,
options.mode,
stop=lambda: job.cancelled,
settings=options.settings,
)
plan = await self._plan_thumbnail_batch(job, options, output_dirs)
if job.cancelled: # M2: stopped during the walk, before any start frame
return self._end_cancelled(job, 0, CANCELLED_BEFORE_WRITING)
roots = _scan_roots(output_dirs)
self._record_fresh(job, plan, roots)
stats, start = await self._generate_thumbnail_targets(job, options, plan, roots)
stats, start, left, attempted = await self._generate_in_batches(
job, options, plan, roots, output_dirs
)
self.invalidate_gallery_cache()
generated = _plural(stats["generated"], "preview")
if job.cancelled:
return self._end_cancelled(
job, stats["generated"], f"Cancelled · {generated} written"
)
overall = {"targets": attempted, "skipped": plan["skipped"], "remaining": left}
payload = _thumbnail_complete_payload(
len(plan["targets"]) + plan["skipped"] + plan["remaining"],
len(attempted) + plan["skipped"] + left,
stats,
start,
remaining=plan["remaining"],
processed=len(plan["targets"]),
remaining=left,
processed=len(attempted),
)
job.update(progress=100, status_line=_finished_line(plan, stats))
if plan["remaining"]:
job.log_line(
f"{plan['remaining']:,} more previews are left for the next run"
)
job.update(progress=100, status_line=_finished_line(overall, stats))
if left:
job.log_line(f"{left:,} more previews are left for the next run")
errors = _plural(len(stats["errors"]), "error")
job.log_line(f"Thumbnails finished · {generated} written · {errors}")
self.logger.info(payload["message"])
job.publish(_frame("complete", payload))
return "finished"
async def _plan_thumbnail_batch(self, job, options, output_dirs, exclude=None):
"""One batch's plan (at most MAX_THUMBNAIL_FILES targets), off-loop."""
return await self._run_in_executor(
self._plan_gallery_thumbnails,
output_dirs,
options.mode,
stop=lambda: job.cancelled,
settings=options.settings,
exclude=exclude,
)
async def _generate_in_batches(self, job, options, plan, roots, output_dirs):
"""Work through ``plan``, then plan again after each batch until
nothing is left, the job is cancelled, or a batch made no progress
(every file failed: they are not planned again, and the rest waits
for the next run). Files worked on are never planned twice. Returns
``(stats, start, files left, files worked on)``; progress counts
against the overall total."""
run = {
"index": 0,
"batches": 0,
"total": len(plan["targets"]) + plan["remaining"],
}
stats = _new_thumbnail_stats()
stats["skipped"] = plan["skipped"]
start = time.monotonic()
self._publish_thumbnail_start(job, plan, run["total"])
attempted = set()
while True:
before = stats["generated"] + stats["skipped"]
await self._generate_thumbnail_targets(
job, options, plan, roots, stats, run, start
)
attempted.update(src for src, _dst, _video in plan["targets"])
progressed = stats["generated"] + stats["skipped"] > before
if job.cancelled or not plan["remaining"] or not progressed:
return stats, start, plan["remaining"], attempted
plan = await self._plan_thumbnail_batch(
job, options, output_dirs, exclude=attempted
)
if job.cancelled:
return stats, start, 0, attempted
run["total"] = run["index"] + len(plan["targets"]) + plan["remaining"]
job.update(total=run["total"])
if not plan["targets"]:
return stats, start, plan["remaining"], attempted
@staticmethod
def _end_cancelled(job, generated, status_line):
"""Close a cancelled job: its status line, log line and error frame."""
@@ -316,10 +358,10 @@ class ThumbnailJobsMixin:
if plan["skipped"]:
job.log_line(f"= {plan['skipped']:,} previews already up to date")
async def _generate_thumbnail_targets(self, job, options, plan, roots):
"""Generate ``plan``'s targets in worker batches, publishing each file."""
@staticmethod
def _publish_thumbnail_start(job, plan, total):
"""The legacy stream's start frame, with the job's overall total."""
targets = plan["targets"]
total = len(targets)
video_count = sum(1 for _, _, is_video in targets if is_video)
message = (
f"Found {total - video_count} images and {video_count} videos to process"
@@ -335,7 +377,7 @@ class ThumbnailJobsMixin:
{
"total_images": total,
"skipped": plan["skipped"],
"remaining": plan["remaining"],
"remaining": 0,
"phase": "processing",
"image_count": total - video_count,
"video_count": video_count,
@@ -343,17 +385,19 @@ class ThumbnailJobsMixin:
},
)
)
stats = _new_thumbnail_stats()
stats["skipped"] = plan["skipped"]
start = time.monotonic()
async def _generate_thumbnail_targets(
self, job, options, plan, roots, stats, run, start
):
"""Generate one batch (``plan``'s targets) in worker batches,
publishing each file against the job's overall ``run`` counters."""
targets = plan["targets"]
batch_size = _thumbnail_batch_size(_worker_threads())
generate = {
"rebuild": options.rebuild,
"stop": lambda: job.cancelled,
}
index = 0
batches = 0
for batch_start in range(0, total, batch_size):
for batch_start in range(0, len(targets), batch_size):
if job.cancelled:
break
batch = targets[batch_start : batch_start + batch_size]
@@ -363,7 +407,8 @@ class ThumbnailJobsMixin:
for (src, _dst, _video), result in zip(batch, results):
if result["action"] == "cancelled":
continue
index += 1
run["index"] += 1
index, total = run["index"], max(run["total"], run["index"])
_record_thumbnail_result(stats, result)
self._record_thumbnail_outcome(job, src, result, roots)
job.update(processed=index, progress=index * 100 // total)
@@ -380,14 +425,13 @@ class ThumbnailJobsMixin:
_thumbnail_progress_payload(index, total, stats, result, start),
)
)
batches += 1
run["batches"] += 1
job.update(
status_line=_scan_status_line(
index, total, batches, time.monotonic() - start
run["index"], run["total"], run["batches"], time.monotonic() - start
)
)
await asyncio.sleep(0) # let other requests run between batches
return stats, start
@staticmethod
def _record_thumbnail_outcome(job, src, result, roots):