From b264dac2e8d1bea9ff7ef7cd6c61d3c19145dc98 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 31 Oct 2026 15:58:25 -0700 Subject: [PATCH] fix(images): start no image thread after the teardown; join retired job pools (#523) --- TESTING.md | 8 +++--- py/api/thumbnail_auto.py | 41 ++++++++++++++++++++--------- tests/test_image_pools.py | 51 ++++++++++++++++++++++++++++++++++++ tests/test_utils_parallel.py | 19 ++++++++++++++ utils/parallel.py | 34 +++++++++++++++--------- 5 files changed, 124 insertions(+), 29 deletions(-) diff --git a/TESTING.md b/TESTING.md index 8fb1b06c..2af0a009 100644 --- a/TESTING.md +++ b/TESTING.md @@ -466,11 +466,13 @@ malloc arena at its high-water mark, so image buffers spread over the default executor's threads stay resident. Interactive image work (view previews behind their gate, workspace card facts, both inspector routes) runs on the API's fixed `pm-image` pool (`IMAGE_POOL_SIZE` 3, `_run_image`, `_map_image`; card facts already -cached answer without a trip). Jobs (scan metadata and file facts, thumbnail and +cached answer without a trip; while the gate's 3 preview writes hold every thread, a +card or inspector read waits for one of them, by design of the merged pool; after the +package teardown the pool is not recreated and late calls use the default executor). Jobs (scan metadata and file facts, thumbnail and convert jobs, the legacy gallery generation, capture previews) run through `map_parallel` on one long-lived `pm-worker` pool of Worker threads, reused between -jobs; a new Worker threads value replaces it, and a call made from a job thread runs -inline. After a scan, thumbnail or convert job, or the legacy gallery generation, +jobs; a new Worker threads value replaces it (the teardown also joins a replaced pool +still finishing), and a call made from a job thread runs inline. After a scan, thumbnail or convert job, or the legacy gallery generation, `release_free_memory` (`utils/native_memory.py`) calls glibc's `malloc_trim(0)` once; it does nothing on Windows, macOS and musl and never raises. No allocator setting changes. `tests/test_image_pools.py` asserts the thread of each read and write and the diff --git a/py/api/thumbnail_auto.py b/py/api/thumbnail_auto.py index 8bd5c6d0..4dbb5954 100644 --- a/py/api/thumbnail_auto.py +++ b/py/api/thumbnail_auto.py @@ -62,7 +62,7 @@ import functools import io import os import threading -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import CancelledError, ThreadPoolExecutor from pathlib import Path from aiohttp import web @@ -721,9 +721,14 @@ class AutoThumbnailMixin: def _image_pool(self): """The interactive image pool (#523): IMAGE_POOL_SIZE threads for view previews (behind their gate), card facts and inspector reads, so - image buffers never spread over the default executor's threads.""" + image buffers never spread over the default executor's threads. + + None after the package teardown (``close_auto_thumbnails(final=True)``): + a late request then runs on the default executor (``run_in_executor`` + with None) and starts no thread the teardown no longer stops. + """ with AutoThumbnailMixin._auto_thumbnailer_lock: - if self._image_pool_obj is None: + if self._image_pool_obj is None and not self._auto_thumbnails_shut: self._image_pool_obj = ThreadPoolExecutor( IMAGE_POOL_SIZE, thread_name_prefix=POOL_PREFIX ) @@ -732,35 +737,45 @@ class AutoThumbnailMixin: async def _run_image(self, func, *args, **kwargs): """``_run_in_executor`` for interactive image work: the image pool. - A pool closed in between (a test reset, the teardown) refuses the - call; it then runs on the default executor as before. + After the teardown, or with the pool shut down in between (a test + reset), the call runs on the default executor as before. """ call = functools.partial(func, *args, **kwargs) loop = asyncio.get_running_loop() try: future = loop.run_in_executor(self._image_pool(), call) - except RuntimeError: + except RuntimeError: # submit refused: the pool was shut down future = loop.run_in_executor(None, call) return await future def _map_image(self, func, items): """``func`` over *items* on the image pool, in order (blocking). - Called from an image pool thread, or with the pool closed, the items - run on the calling thread. + Called from an image pool thread, after the teardown, or with the + pool shut down in between (its queued items are cancelled), the + items it did not run run on the calling thread. """ items = list(items) - if not items or _on_image_thread(): + pool = None if not items or _on_image_thread() else self._image_pool() + if pool is None: return [func(item) for item in items] futures = [] try: - pool = self._image_pool() for item in items: futures.append(pool.submit(func, item)) except RuntimeError: - pass # closed in between: the rest run here - done = [future.result() for future in futures] - return done + [func(item) for item in items[len(futures) :]] + pass # shut down in between: the rest run here + results = [] + for index, item in enumerate(items): + future = futures[index] if index < len(futures) else None + if future is None or future.cancelled(): + results.append(func(item)) + continue + try: + results.append(future.result()) + except CancelledError: # cancelled while this caller waited + results.append(func(item)) + return results def _image_pool_threads(self): pool = self._image_pool_obj diff --git a/tests/test_image_pools.py b/tests/test_image_pools.py index df7d1dac..38384a5e 100644 --- a/tests/test_image_pools.py +++ b/tests/test_image_pools.py @@ -132,6 +132,57 @@ class TestInteractiveImagePool(ImagePoolCase): ) self.assertEqual(names, [here, here]) + async def test_after_the_final_close_no_image_thread_starts(self): + await asyncio.get_running_loop().run_in_executor( + None, lambda: self.api.close_auto_thumbnails(final=True) + ) + self.assertIsNone(self.api._image_pool()) + here = threading.current_thread().name + names = self.api._map_image(lambda _: threading.current_thread().name, [1]) + self.assertEqual(names, [here]) + name = await self.api._run_image(lambda: threading.current_thread().name) + self.assertFalse(name.startswith(IMAGE_THREAD), name) + self.assertEqual(self.api._image_pool_threads(), []) + + def test_map_image_runs_items_a_shutdown_cancelled_here(self): + entered = threading.Barrier(auto.IMAGE_POOL_SIZE + 1) + release = threading.Event() + self.addCleanup(release.set) + pool = self.api._image_pool() + for _ in range(auto.IMAGE_POOL_SIZE): # occupy every pool thread + pool.submit(lambda: (entered.wait(WAIT), release.wait(WAIT))) + entered.wait(WAIT) + here = threading.current_thread().name + results, submitted = {}, threading.Event() + + class Counting: + calls = 0 + + def submit(self, *args): + future = pool.submit(*args) + Counting.calls += 1 + if Counting.calls == 2: + submitted.set() + return future + + def call(): + with patch.object(self.api, "_image_pool", return_value=Counting()): + results["names"] = self.api._map_image( + lambda _: threading.current_thread().name, [1, 2] + ) + + caller = threading.Thread(target=call, name="map-image-caller") + caller.start() + # The caller's two items are queued behind the busy threads; the + # shutdown cancels them and the caller runs them itself. + self.assertTrue(submitted.wait(WAIT)) + pool.shutdown(wait=False, cancel_futures=True) + caller.join(WAIT) + release.set() + self.assertFalse(caller.is_alive()) + self.assertEqual(results["names"], ["map-image-caller"] * 2) + self.assertNotEqual(here, "map-image-caller") + async def test_run_image_after_a_close_uses_the_default_executor(self): pool = self.api._image_pool() pool.shutdown(wait=True) diff --git a/tests/test_utils_parallel.py b/tests/test_utils_parallel.py index 75452e9c..21fc717c 100644 --- a/tests/test_utils_parallel.py +++ b/tests/test_utils_parallel.py @@ -64,6 +64,25 @@ class TestMapParallel(unittest.TestCase): self.assertFalse(thread.is_alive(), thread.name) self.assertLessEqual(len(_job_threads()), 3) + def test_shutdown_also_joins_a_pool_a_resize_retired(self): + entered, release = threading.Event(), threading.Event() + + def hold(_): + entered.set() + release.wait(WAIT) + + caller = threading.Thread(target=map_parallel, args=(hold, [1], 1)) + caller.start() + self.addCleanup(caller.join, WAIT) + self.addCleanup(release.set) + self.assertTrue(entered.wait(WAIT)) + map_parallel(lambda n: n, [1, 2], 2) # a new size retires the busy pool + self.assertFalse(parallel.shutdown_job_pool(0.05)) + release.set() + caller.join(WAIT) + self.assertTrue(parallel.shutdown_job_pool(WAIT)) + self.assertFalse(parallel.job_pool_started()) + def test_an_item_run_from_a_job_thread_runs_inline(self): """A nested map never waits for the threads it is running on.""" diff --git a/utils/parallel.py b/utils/parallel.py index 59e46e63..99094224 100644 --- a/utils/parallel.py +++ b/utils/parallel.py @@ -44,6 +44,7 @@ _noted = set() _noted_lock = threading.Lock() _pool_lock = threading.Lock() _pool = None # (workers, ThreadPoolExecutor) +_retired = [] # pools replaced by a new size, finishing what they hold def refused_at_shutdown(error): @@ -63,8 +64,8 @@ def note_refusal(logger, what): def job_pool_started(): - """Whether the long-lived job pool exists (the teardown stops it).""" - return _pool is not None + """Whether a job pool exists (the teardown stops it).""" + return _pool is not None or bool(_retired) def _job_pool(workers): @@ -81,29 +82,36 @@ def _job_pool(workers): current = _pool[1] if retired is not None: retired[1].shutdown(wait=False) # it still runs what it holds + with _pool_lock: + _retired[:] = [p for p in _retired if _alive(p)] + [retired[1]] return current def shutdown_job_pool(timeout=None): - """Stop the job pool: queued items are cancelled (their callers run - them on their own thread), running ones are joined up to ``timeout`` - seconds. True when no job thread is left running.""" + """Stop the job pool (and any pool a resize retired): queued items are + cancelled (their callers run them on their own thread), running ones are + joined up to ``timeout`` seconds. True when no job thread is left.""" global _pool with _pool_lock: - retired, _pool = _pool, None - if retired is None: - return True - executor = retired[1] - # After shutdown: it takes the lock ``submit`` holds while it starts and - # records a thread, so every thread is listed and no new one starts. - executor.shutdown(wait=False, cancel_futures=True) - threads = list(getattr(executor, "_threads", ())) + current, _pool = _pool, None + executors = _retired[:] + ([] if current is None else [current[1]]) + _retired.clear() + threads = [] + for executor in executors: + # After shutdown: it takes the lock ``submit`` holds while it starts + # and records a thread, so every thread is listed and none starts. + executor.shutdown(wait=False, cancel_futures=True) + threads.extend(getattr(executor, "_threads", ())) deadline = None if timeout is None else time.monotonic() + timeout for thread in threads: thread.join(None if deadline is None else max(0.0, deadline - time.monotonic())) return not any(thread.is_alive() for thread in threads) +def _alive(executor): + return any(t.is_alive() for t in getattr(executor, "_threads", ())) + + def _forget(executor): """Drop *executor* as the shared pool (a later call starts a new one).""" global _pool