diff --git a/py/api/thumbnail_auto.py b/py/api/thumbnail_auto.py index 6e7871ff..65f393c3 100644 --- a/py/api/thumbnail_auto.py +++ b/py/api/thumbnail_auto.py @@ -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 ──────────────────────────────────────────────────── diff --git a/tests/test_thumbnail_auto.py b/tests/test_thumbnail_auto.py index 8f2cc23f..821b45e3 100644 --- a/tests/test_thumbnail_auto.py +++ b/tests/test_thumbnail_auto.py @@ -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"},