perf(images): start a stale listing's refresh walk a second after the read (#348)
After the faster walk, a read that found the listing stale still took 0.5-1.0 s against 0.07 s: it was served from the old listing but ran beside the refresh walk it had just started, sharing the GIL. The refresh now starts REFRESH_DELAY (1 s) after that read through a per-root rewalk_handle, so the page's own requests finish first. One pending refresh per root and none while a walk runs; any walk start (a cold read, Rescan, the warm-up) cancels a pending one; on_cleanup cancels them before its first await.
This commit is contained in:
@@ -548,6 +548,11 @@ class PromptManagerAPI(
|
||||
async def cleanup_gallery_listings(self, app=None):
|
||||
"""Stop the background listing walks (aiohttp on_cleanup)."""
|
||||
self._workspace_warm_enabled = False
|
||||
# Before any await, so no delayed refresh can start a walk meanwhile.
|
||||
for record in self._root_listings.values():
|
||||
handle, record.rewalk_handle = record.rewalk_handle, None
|
||||
if handle is not None:
|
||||
handle.cancel()
|
||||
warm, self._workspace_warm_task = self._workspace_warm_task, None
|
||||
if warm is not None:
|
||||
warm.cancel()
|
||||
|
||||
@@ -21,6 +21,11 @@ from typing import Callable, NamedTuple
|
||||
# patch it.
|
||||
REFRESH_AFTER = 30
|
||||
|
||||
# The stale read is served at once; its refresh walk starts this many seconds
|
||||
# later, so the walk does not share the GIL with the page's own requests
|
||||
# (#348 live test: reads during the walk took 2.2 s instead of 0.07 s).
|
||||
REFRESH_DELAY = 1.0
|
||||
|
||||
# A reader with no listing gives up (500) after this many walks in a row were
|
||||
# discarded by invalidations made while they ran.
|
||||
MAX_DISCARDED_WALKS = 3
|
||||
@@ -75,6 +80,8 @@ class RootListing:
|
||||
flight: asyncio.Future | None = None
|
||||
# True while walks keep failing: one warning per streak.
|
||||
failing: bool = False
|
||||
# The delayed refresh not started yet (loop.call_later), or None.
|
||||
rewalk_handle: asyncio.TimerHandle | None = None
|
||||
|
||||
def drop(self) -> None:
|
||||
self.files = None
|
||||
|
||||
+18
-2
@@ -1422,9 +1422,22 @@ class ImageRoutesMixin:
|
||||
)
|
||||
|
||||
def _refresh_if_stale(self, output_path, record):
|
||||
"""Walk a stale listing again, REFRESH_DELAY after this read (on the loop).
|
||||
|
||||
At most one pending refresh per root, none while a walk runs.
|
||||
"""
|
||||
age = _time.monotonic() - (record.walked_at or 0.0)
|
||||
if age >= gallery_listing.REFRESH_AFTER:
|
||||
self._schedule_walk(output_path)
|
||||
if age < gallery_listing.REFRESH_AFTER:
|
||||
return
|
||||
if record.flight is not None or record.rewalk_handle is not None:
|
||||
return
|
||||
record.rewalk_handle = asyncio.get_running_loop().call_later(
|
||||
gallery_listing.REFRESH_DELAY, self._delayed_walk, output_path, record
|
||||
)
|
||||
|
||||
def _delayed_walk(self, output_path, record):
|
||||
record.rewalk_handle = None
|
||||
self._schedule_walk(output_path)
|
||||
|
||||
def _schedule_walk(self, output_path):
|
||||
"""Start a background walk of *output_path* unless one is running."""
|
||||
@@ -1436,6 +1449,9 @@ class ImageRoutesMixin:
|
||||
"""Start the root's one walk; returns its flight future (on the loop)."""
|
||||
loop = asyncio.get_running_loop()
|
||||
self._listing_loop = loop
|
||||
handle, record.rewalk_handle = record.rewalk_handle, None
|
||||
if handle is not None:
|
||||
handle.cancel() # this walk is the refresh it was waiting to start
|
||||
record.walk_seq += 1
|
||||
flight = loop.create_future()
|
||||
record.flight = flight
|
||||
|
||||
@@ -2497,6 +2497,11 @@ class TestSharedListing(ImageAPITestCase):
|
||||
self.scan = _GatedScan(self.api._scan_gallery_files_sync)
|
||||
self.api._scan_gallery_files_sync = self.scan
|
||||
self.addCleanup(self.scan.release_all)
|
||||
# The stale refresh waits REFRESH_DELAY after the read (#348); these
|
||||
# tests start it on the next loop turn and _settle waits for it.
|
||||
delay = patch.object(gallery_listing, "REFRESH_DELAY", 0)
|
||||
delay.start()
|
||||
self.addCleanup(delay.stop)
|
||||
|
||||
async def _json(self, method, path, **kwargs):
|
||||
# Bounded, like every wait in this class: a stuck walk fails the test
|
||||
@@ -2510,6 +2515,12 @@ class TestSharedListing(ImageAPITestCase):
|
||||
def _record(self):
|
||||
return self.api._root_listings[str(self.output_dir)]
|
||||
|
||||
def _pending(self):
|
||||
return any(
|
||||
record.rewalk_handle is not None
|
||||
for record in self.api._root_listings.values()
|
||||
)
|
||||
|
||||
def _warming(self):
|
||||
task = self.api._workspace_warm_task
|
||||
return task is not None and not task.done()
|
||||
@@ -2524,7 +2535,9 @@ class TestSharedListing(ImageAPITestCase):
|
||||
"""Wait for every background walk this API started."""
|
||||
|
||||
async def settle():
|
||||
while self.api._listing_tasks or self._warming():
|
||||
while self.api._listing_tasks or self._warming() or self._pending():
|
||||
if self._pending():
|
||||
await asyncio.sleep(0.01) # a delayed refresh not started yet
|
||||
await asyncio.gather(*list(self.api._listing_tasks))
|
||||
# A finished walk leaves the set in a done callback, which needs
|
||||
# a loop turn; gather over finished tasks does not yield one.
|
||||
@@ -2854,6 +2867,66 @@ class TestSharedListing(ImageAPITestCase):
|
||||
self.assertIsNone(self._record().flight)
|
||||
self.assertEqual(self.scan.calls, 0)
|
||||
|
||||
async def test_a_stale_read_starts_its_refresh_after_a_delay(self):
|
||||
# The read is served from the old listing; the walk starts
|
||||
# REFRESH_DELAY later so it does not share the GIL with the page's
|
||||
# own requests (#348 live test: 2.2 s reads during the walk).
|
||||
make_png(self.output_dir / "a.png")
|
||||
await self._json("GET", self.URL)
|
||||
self._age()
|
||||
with patch.object(gallery_listing, "REFRESH_DELAY", 30):
|
||||
status, data = await self._json("GET", self.URL)
|
||||
await self._json("GET", self.URL) # still one pending refresh
|
||||
record = self._record()
|
||||
handle = record.rewalk_handle
|
||||
self.assertEqual((status, data["total"]), (200, 1))
|
||||
self.assertIsNotNone(handle)
|
||||
self.assertIsNone(record.flight)
|
||||
self.assertEqual(self.scan.calls, 1)
|
||||
self.assertAlmostEqual(
|
||||
handle.when() - asyncio.get_running_loop().time(), 30, delta=2
|
||||
)
|
||||
handle.cancel()
|
||||
record.rewalk_handle = None
|
||||
|
||||
async def test_the_delayed_refresh_walks_once_when_it_fires(self):
|
||||
make_png(self.output_dir / "a.png")
|
||||
await self._json("GET", self.URL)
|
||||
make_png(self.output_dir / "b.png")
|
||||
self._age()
|
||||
await self._json("GET", self.URL)
|
||||
await self._json("GET", self.URL)
|
||||
await self._settle()
|
||||
self.assertEqual(self.scan.calls, 2)
|
||||
self.assertIsNone(self._record().rewalk_handle)
|
||||
_, data = await self._json("GET", self.URL)
|
||||
self.assertEqual(data["total"], 2)
|
||||
|
||||
async def test_a_walk_started_meanwhile_cancels_the_delayed_refresh(self):
|
||||
make_png(self.output_dir / "a.png")
|
||||
await self._json("GET", self.URL)
|
||||
self._age()
|
||||
with patch.object(gallery_listing, "REFRESH_DELAY", 30):
|
||||
await self._json("GET", self.URL)
|
||||
handle = self._record().rewalk_handle
|
||||
status, _ = await self._json("POST", "/prompt_manager/gallery/rescan")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(handle.cancelled())
|
||||
self.assertIsNone(self._record().rewalk_handle)
|
||||
self.assertEqual(self.scan.calls, 2)
|
||||
|
||||
async def test_cleanup_cancels_a_delayed_refresh(self):
|
||||
make_png(self.output_dir / "a.png")
|
||||
await self._json("GET", self.URL)
|
||||
self._age()
|
||||
with patch.object(gallery_listing, "REFRESH_DELAY", 30):
|
||||
await self._json("GET", self.URL)
|
||||
handle = self._record().rewalk_handle
|
||||
await self.api.cleanup_gallery_listings(web.Application())
|
||||
self.assertTrue(handle.cancelled())
|
||||
self.assertIsNone(self._record().rewalk_handle)
|
||||
self.assertEqual(self.scan.calls, 1)
|
||||
|
||||
WORKSPACE_URL = "/prompt_manager/images/output?mode=workspace&limit=1"
|
||||
|
||||
async def _warmed(self):
|
||||
|
||||
@@ -193,7 +193,10 @@ class TestWorkspaceImages(ImageAPITestCase):
|
||||
"_scan_gallery_files_sync",
|
||||
wraps=self.api._scan_gallery_files_sync,
|
||||
) as scan:
|
||||
await self.get() # served at once; one background walk runs
|
||||
with patch.object(gallery_listing, "REFRESH_DELAY", 0):
|
||||
await self.get() # served at once; one refresh walk follows
|
||||
while record.rewalk_handle is not None:
|
||||
await asyncio.sleep(0.01)
|
||||
while self.api._listing_tasks:
|
||||
await asyncio.gather(*list(self.api._listing_tasks))
|
||||
# A finished walk leaves the set in a done callback, which needs
|
||||
|
||||
@@ -174,8 +174,10 @@ class RecordTests(unittest.TestCase):
|
||||
self.assertEqual(failed.kind, FAILED)
|
||||
self.assertIs(failed.error, error)
|
||||
|
||||
def test_refresh_constant(self):
|
||||
def test_refresh_constants(self):
|
||||
self.assertEqual(gallery_listing.REFRESH_AFTER, 30)
|
||||
self.assertEqual(gallery_listing.REFRESH_DELAY, 1.0)
|
||||
self.assertIsNone(RootListing().rewalk_handle)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user