fix(thumbnails): answer shared flights with the outcome only and restart a failed worker
The history and serve routes share a flight key, so a flight now answers only the ensure outcome whichever route started it; the serve route derives the preview path and, for a fallback, looks its source up again on the view pool. A view joining a history flight no longer gets a 500 or an image answered with the video placeholder. The capture worker clears its running state however its loop ends, and a worker that cannot start leaves the file queued for the next enqueue.
This commit is contained in:
Vito Sansevero
+77
-25
@@ -400,18 +400,44 @@ class AutoThumbnailer:
|
||||
self._queue.append((key, Path(src), Path(dst), bool(is_video)))
|
||||
self._busy = True
|
||||
if self._start_worker and not self._worker_running:
|
||||
self._worker_running = True
|
||||
worker = threading.Thread(
|
||||
target=self._run, name=WORKER_NAME, daemon=True
|
||||
)
|
||||
worker.pm_preview_token = _MODULE_TOKEN
|
||||
worker.pm_preview_close = self.close
|
||||
self._worker = worker
|
||||
worker.start()
|
||||
self._start_worker_locked()
|
||||
self._changed.notify_all()
|
||||
return True
|
||||
|
||||
def _start_worker_locked(self):
|
||||
"""Start the worker (lock held). A start that fails leaves the file
|
||||
queued and the next enqueue tries again."""
|
||||
worker = threading.Thread(target=self._run, name=WORKER_NAME, daemon=True)
|
||||
worker.pm_preview_token = _MODULE_TOKEN
|
||||
worker.pm_preview_close = self.close
|
||||
self._worker_running = True
|
||||
try:
|
||||
worker.start()
|
||||
except RuntimeError:
|
||||
self._worker_running = False
|
||||
self._busy = False
|
||||
self._logger.warning(
|
||||
"The automatic preview worker could not start; the next output "
|
||||
"tries again",
|
||||
exc_info=True,
|
||||
)
|
||||
return
|
||||
self._worker = worker
|
||||
|
||||
def _run(self):
|
||||
try:
|
||||
self._drain()
|
||||
except Exception:
|
||||
self._logger.warning("The automatic preview worker stopped", exc_info=True)
|
||||
finally:
|
||||
# Whatever ended the loop, the next enqueue starts a fresh worker.
|
||||
with self._changed:
|
||||
self._worker_running = False
|
||||
self._busy = False
|
||||
self._batch = 0
|
||||
self._changed.notify_all()
|
||||
|
||||
def _drain(self):
|
||||
while True:
|
||||
with self._changed:
|
||||
if self._closed or not self._queue:
|
||||
@@ -630,29 +656,37 @@ class AutoThumbnailMixin:
|
||||
return video_placeholder_jpeg()
|
||||
|
||||
async def _admit_preview(self, src, dst, is_video):
|
||||
"""Make one preview through the view admission (the history route)."""
|
||||
root_parts = (Path(dst).name,)
|
||||
"""Make one preview through the view admission (the history route).
|
||||
|
||||
Shares the serve route's flight key; every flight answers the
|
||||
``ensure`` outcome (or None, or OVER), whichever route started it.
|
||||
"""
|
||||
return await self._preview_gate().run(
|
||||
flight_key(Path(dst).parent, root_parts),
|
||||
functools.partial(self._auto_thumbnails().ensure, src, dst, is_video),
|
||||
flight_key(Path(dst).parent, (Path(dst).name,)),
|
||||
functools.partial(self._ensure_view, src, dst, is_video),
|
||||
self._preview_pool(),
|
||||
)
|
||||
|
||||
def _ensure_view(self, src, dst, is_video):
|
||||
"""``ensure``, reporting a preview removed under us as busy."""
|
||||
outcome = self._auto_thumbnails().ensure(src, dst, is_video)
|
||||
if outcome == READY and not Path(dst).is_file():
|
||||
return BUSY # removed under us: the fallback this once
|
||||
return outcome
|
||||
|
||||
# ── on first view ─────────────────────────────────────────────────
|
||||
|
||||
def _preview_on_demand(self, root, parts):
|
||||
"""``(outcome, src, dst, is_video)`` for a missing preview under
|
||||
*root*, or None when no media file previews as *parts*."""
|
||||
"""The ``ensure`` outcome for the preview *parts* under *root*
|
||||
(``deferred`` when its folder is linked out of the root), or None
|
||||
when no media file previews as *parts*."""
|
||||
located = _thumbnail_source(root, parts)
|
||||
if located is None:
|
||||
return None
|
||||
src, dst, is_video = located
|
||||
if dst is None:
|
||||
return DEFERRED, src, None, is_video
|
||||
outcome = self._auto_thumbnails().ensure(src, dst, is_video)
|
||||
if outcome == READY and not dst.is_file():
|
||||
outcome = BUSY # removed under us: the fallback this once
|
||||
return outcome, src, dst, is_video
|
||||
return DEFERRED
|
||||
return self._ensure_view(src, dst, is_video)
|
||||
|
||||
def _preview_response(self, outcome, src, dst, is_video):
|
||||
if outcome == READY and dst is not None:
|
||||
@@ -669,7 +703,13 @@ class AutoThumbnailMixin:
|
||||
|
||||
async def _serve_missing_thumbnail(self, output_dirs, filepath):
|
||||
"""The made preview, or its fallback; None when *filepath* names no
|
||||
preview of a media file in *output_dirs*."""
|
||||
preview of a media file in *output_dirs*.
|
||||
|
||||
The flight answers only the outcome (a history flight may be the one
|
||||
joined); the file it previews is derived here: the preview path is
|
||||
the request's own, and a fallback looks its source up again on the
|
||||
view pool (bounded by it, never the default executor).
|
||||
"""
|
||||
if _has_traversal(filepath) or not _is_preview_request(filepath):
|
||||
return None
|
||||
parts = tuple(filepath.split("/"))
|
||||
@@ -677,17 +717,29 @@ class AutoThumbnailMixin:
|
||||
pool = self._preview_pool()
|
||||
for root in output_dirs:
|
||||
root = Path(root)
|
||||
answer = await gate.run(
|
||||
outcome = await gate.run(
|
||||
flight_key(root, parts),
|
||||
functools.partial(self._preview_on_demand, root, parts),
|
||||
pool,
|
||||
)
|
||||
if answer is OVER:
|
||||
if outcome is None:
|
||||
continue
|
||||
if outcome is OVER:
|
||||
# Past the cap: exact names only, no listing, no write.
|
||||
quick = _quick_source(root, parts)
|
||||
answer = None if quick is None else (BUSY, quick[0], None, quick[1])
|
||||
if answer is not None:
|
||||
return self._preview_response(*answer)
|
||||
if quick is None:
|
||||
continue
|
||||
return self._preview_response(BUSY, quick[0], None, quick[1])
|
||||
dst = root.joinpath(*parts)
|
||||
if outcome == READY:
|
||||
return self._file_response(dst)
|
||||
located = await gate.loop.run_in_executor(
|
||||
pool, _thumbnail_source, root, parts
|
||||
)
|
||||
if located is None:
|
||||
continue
|
||||
src, _dst, is_video = located
|
||||
return self._preview_response(outcome, src, None, is_video)
|
||||
return None
|
||||
|
||||
# ── on capture ────────────────────────────────────────────────────
|
||||
|
||||
@@ -727,6 +727,74 @@ class TestShutdown(AutoThumbnailCase):
|
||||
|
||||
|
||||
class TestHistorySharesTheAdmission(AutoThumbnailCase):
|
||||
async def _shared_flight(self, name, is_video, first):
|
||||
"""Start the history route's or the serve route's flight (``first``),
|
||||
let the other one join it, release, and return both answers."""
|
||||
src = self.output_dir / name
|
||||
if is_video:
|
||||
src.write_bytes(b"AV1 stream no decoder here")
|
||||
else:
|
||||
make_png(src, size=(256, 256))
|
||||
url = _entry(self.api, self.output_dir, name)["thumbnail_url"]
|
||||
dst = self.output_dir / "thumbnails" / Path(url.split("?")[0]).name
|
||||
calls = self._count_writes()
|
||||
entered, release, _state = self._gate_writes()
|
||||
|
||||
def history():
|
||||
return asyncio.ensure_future(self.api._admit_preview(src, dst, is_video))
|
||||
|
||||
def view():
|
||||
return asyncio.ensure_future(self.client.get(url))
|
||||
|
||||
leader = history() if first == "history" else view()
|
||||
await self._wait(entered)
|
||||
follower = view() if first == "history" else history()
|
||||
await self._waiting(2)
|
||||
release.set()
|
||||
answers = {first: await leader}
|
||||
answers["view" if first == "history" else "history"] = await follower
|
||||
self.assertEqual(len(calls), 1, "one flight, one write")
|
||||
return answers, dst
|
||||
|
||||
async def test_a_view_joining_a_history_flight_gets_the_thumbnail(self):
|
||||
answers, dst = await self._shared_flight("a.png", False, "history")
|
||||
self.assertEqual(answers["history"], auto.READY)
|
||||
resp = answers["view"]
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(resp.headers["Content-Type"], "image/png")
|
||||
self.assertEqual(await resp.read(), dst.read_bytes())
|
||||
|
||||
async def test_history_joining_a_view_flight_gets_the_outcome(self):
|
||||
answers, dst = await self._shared_flight("b.png", False, "view")
|
||||
self.assertEqual(answers["history"], auto.READY)
|
||||
self.assertEqual(await answers["view"].read(), dst.read_bytes())
|
||||
|
||||
async def test_a_held_shared_flight_answers_each_route_in_its_own_type(self):
|
||||
from py.api.jobs import RESTORE_JOB_KIND
|
||||
|
||||
make_png(self.output_dir / "c.png", size=(64, 64))
|
||||
(self.output_dir / "c.mp4").write_bytes(b"AV1 stream no decoder here")
|
||||
self._hold(RESTORE_JOB_KIND)
|
||||
for name, is_video in (("c.png", False), ("c.mp4", True)):
|
||||
with self.subTest(name=name):
|
||||
src = self.output_dir / name
|
||||
url = _entry(self.api, self.output_dir, name)["thumbnail_url"]
|
||||
dst = self.output_dir / "thumbnails" / Path(url.split("?")[0]).name
|
||||
history = asyncio.ensure_future(
|
||||
self.api._admit_preview(src, dst, is_video)
|
||||
)
|
||||
resp = await self.client.get(url)
|
||||
self.assertEqual(await history, auto.DEFERRED)
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(resp.headers["Cache-Control"], "no-store")
|
||||
body = await resp.read()
|
||||
if is_video:
|
||||
self.assertEqual(resp.headers["Content-Type"], "image/jpeg")
|
||||
self.assertTrue(body.startswith(b"\xff\xd8"))
|
||||
else:
|
||||
self.assertEqual(resp.headers["Content-Type"], "image/png")
|
||||
self.assertEqual(body, src.read_bytes())
|
||||
|
||||
def test_history_and_serve_route_share_one_flight_key(self):
|
||||
root = self.output_dir
|
||||
preview = root / "thumbnails" / "sub" / "a_thumb.png"
|
||||
@@ -828,6 +896,53 @@ class TestThumbnailOnCapture(AutoThumbnailCase):
|
||||
await self._idle()
|
||||
self.assertFalse((self.output_dir / "thumbnails").exists())
|
||||
|
||||
async def test_a_worker_that_raises_never_blocks_the_next_one(self):
|
||||
# A raising config read stops the worker; the next enqueue starts one.
|
||||
thumbnailer = self.api._auto_thumbnails()
|
||||
real_workers = thumbnailer._workers
|
||||
failures = iter([RuntimeError("config unreadable")])
|
||||
|
||||
def workers():
|
||||
failure = next(failures, None)
|
||||
if failure is not None:
|
||||
raise failure
|
||||
return real_workers()
|
||||
|
||||
thumbnailer._workers = workers
|
||||
first = make_png(self.output_dir / "first.png", size=(32, 32))
|
||||
with self.assertLogs("prompt_manager.api", "WARNING"):
|
||||
self.assertEqual(self.api.queue_capture_thumbnail(str(first)), "queued")
|
||||
await self._idle()
|
||||
self.assertFalse(thumbnailer._worker_running)
|
||||
second = make_png(self.output_dir / "second.png", size=(32, 32))
|
||||
self.assertEqual(self.api.queue_capture_thumbnail(str(second)), "queued")
|
||||
await self._idle()
|
||||
for name in ("first", "second"):
|
||||
self.assertTrue(
|
||||
(self.output_dir / "thumbnails" / f"{name}_thumb.png").is_file()
|
||||
)
|
||||
|
||||
async def test_a_worker_that_cannot_start_is_started_by_the_next_enqueue(self):
|
||||
thumbnailer = self.api._auto_thumbnails()
|
||||
first = make_png(self.output_dir / "first.png", size=(32, 32))
|
||||
with (
|
||||
patch.object(
|
||||
threading.Thread,
|
||||
"start",
|
||||
side_effect=RuntimeError("can't start new thread"),
|
||||
),
|
||||
self.assertLogs("prompt_manager.api", "WARNING"),
|
||||
):
|
||||
self.assertEqual(self.api.queue_capture_thumbnail(str(first)), "queued")
|
||||
self.assertFalse(thumbnailer._worker_running)
|
||||
second = make_png(self.output_dir / "second.png", size=(32, 32))
|
||||
self.assertEqual(self.api.queue_capture_thumbnail(str(second)), "queued")
|
||||
await self._idle()
|
||||
for name in ("first", "second"):
|
||||
self.assertTrue(
|
||||
(self.output_dir / "thumbnails" / f"{name}_thumb.png").is_file()
|
||||
)
|
||||
|
||||
def test_queue_is_bounded(self):
|
||||
thumbnailer = auto.AutoThumbnailer(
|
||||
lambda *a: {"action": "generated"},
|
||||
|
||||
Reference in New Issue
Block a user