diff --git a/app/tests/test_download_queue.py b/app/tests/test_download_queue.py index 66b5a0d..0e6f0a7 100644 --- a/app/tests/test_download_queue.py +++ b/app/tests/test_download_queue.py @@ -330,7 +330,7 @@ async def test_retry_restores_playlist_output_context(dq_env): chapter_template="", ) failed_info.status = "error" - dq.done.put(Download(None, None, None, None, "best", "any", {}, failed_info)) + await dq.done.put(Download(None, None, None, None, "best", "any", {}, failed_info)) def fake_extract(self, extracted_url, *_args, **_kwargs): return { @@ -389,7 +389,7 @@ async def test_retry_keeps_playlist_context_through_url_indirection(dq_env): dq = DownloadQueue(dq_env, notifier) url = "https://example.com/watch?v=1" resolved = "https://example.com/resolved?v=1" - dq.done.put(Download(None, None, None, None, "best", "any", {}, _failed_playlist_item(url))) + await dq.done.put(Download(None, None, None, None, "best", "any", {}, _failed_playlist_item(url))) def fake_extract(self, extracted_url, *_args, **_kwargs): if extracted_url == url: @@ -427,7 +427,7 @@ async def test_retry_reapplies_current_options_gates(dq_env): ytdl_options_presets=["Still There", "Removed Preset"], ytdl_options_overrides={"paths": {"home": "/etc"}}, ) - dq.done.put(Download(None, None, None, None, "best", "any", {}, info)) + await dq.done.put(Download(None, None, None, None, "best", "any", {}, info)) def fake_extract(self, extracted_url, *_args, **_kwargs): return { @@ -457,7 +457,7 @@ async def test_retry_keeps_overrides_while_still_allowed(dq_env): dq = DownloadQueue(dq_env, notifier) url = "https://example.com/watch?v=1" info = _failed_playlist_item(url, ytdl_options_overrides={"writesubtitles": True}) - dq.done.put(Download(None, None, None, None, "best", "any", {}, info)) + await dq.done.put(Download(None, None, None, None, "best", "any", {}, info)) def fake_extract(self, extracted_url, *_args, **_kwargs): return { @@ -481,7 +481,7 @@ async def test_retry_carries_the_sponsorblock_flag(dq_env): notifier = AsyncMock() dq = DownloadQueue(dq_env, notifier) url = "https://example.com/watch?v=1" - dq.done.put( + await dq.done.put( Download(None, None, None, None, "best", "any", {}, _failed_playlist_item(url, sponsorblock=True)) ) @@ -1441,9 +1441,9 @@ async def test_post_download_cleanup_clears_filename_on_error(dq_env): notifier = AsyncMock() dq = DownloadQueue(dq_env, notifier) download = _make_download(dq_env, status="downloading", filename="../tmp/partial.mp4") - dq.queue.put(download) + await dq.queue.put(download) - dq._post_download_cleanup(download) + await dq._post_download_cleanup(download) assert download.info.status == "error" assert download.info.filename is None @@ -1456,9 +1456,9 @@ async def test_post_download_cleanup_keeps_captured_subtitles_on_error(dq_env): dq = DownloadQueue(dq_env, notifier) download = _make_download(dq_env, download_type="captions", status="downloading", filename="en.srt") download.info.subtitle_files = [{"filename": "en.srt", "size": 42}] - dq.queue.put(download) + await dq.queue.put(download) - dq._post_download_cleanup(download) + await dq._post_download_cleanup(download) assert download.info.status == "error" assert download.info.filename == "en.srt" @@ -1478,7 +1478,7 @@ async def test_clear_skips_deletion_outside_download_directory(dq_env): # A crafted/legacy relative filename that escapes DOWNLOAD_DIR via '..'. escaping_filename = os.path.relpath(outside_file, dq_env.DOWNLOAD_DIR) download = _make_download(dq_env, status="finished", filename=escaping_filename) - dq.done.put(download) + await dq.done.put(download) await dq.clear([download.info.url]) diff --git a/app/tests/test_persistent_queue.py b/app/tests/test_persistent_queue.py index 6885315..e8c2fa0 100644 --- a/app/tests/test_persistent_queue.py +++ b/app/tests/test_persistent_queue.py @@ -2,8 +2,11 @@ from __future__ import annotations +import asyncio import json import os +import threading +import time import shelve import sys import tempfile @@ -69,22 +72,22 @@ def _create_legacy_shelf(path: str, *infos: DownloadInfo) -> None: shelf[info.url] = info -class PersistentQueueTests(unittest.TestCase): - def test_put_get_delete_roundtrip(self): +class PersistentQueueTests(unittest.IsolatedAsyncioTestCase): + async def test_put_get_delete_roundtrip(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq = PersistentQueue("queue", path) dl = _FakeDownload(_make_info("http://a.example")) - pq.put(dl) + await pq.put(dl) self.assertTrue(os.path.exists(path + ".json")) self.assertTrue(pq.exists("http://a.example")) self.assertFalse(pq.empty()) got = pq.get("http://a.example") self.assertEqual(got.info.url, "http://a.example") - pq.delete("http://a.example") + await pq.delete("http://a.example") self.assertFalse(pq.exists("http://a.example")) - def test_saved_items_sorted_by_timestamp(self): + async def test_saved_items_sorted_by_timestamp(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq = PersistentQueue("queue", path) @@ -92,16 +95,16 @@ class PersistentQueueTests(unittest.TestCase): b = _FakeDownload(_make_info("http://second.example")) a.info.timestamp = 100 b.info.timestamp = 200 - pq.put(a) - pq.put(b) + await pq.put(a) + await pq.put(b) keys = [k for k, _ in pq.saved_items()] self.assertEqual(keys, ["http://first.example", "http://second.example"]) - def test_load_restores_from_json(self): + async def test_load_restores_from_json(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq1 = PersistentQueue("queue", path) - pq1.put(_FakeDownload(_make_info("http://load.example"))) + await pq1.put(_FakeDownload(_make_info("http://load.example"))) pq2 = PersistentQueue("queue", path) pq2.load() self.assertTrue(pq2.exists("http://load.example")) @@ -115,7 +118,7 @@ class PersistentQueueTests(unittest.TestCase): self.assertTrue(pq.exists("http://legacy.example")) self.assertTrue(os.path.exists(path + ".json")) - def test_queue_persists_only_compact_entry_subset(self): + async def test_queue_persists_only_compact_entry_subset(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq = PersistentQueue("queue", path) @@ -128,7 +131,7 @@ class PersistentQueueTests(unittest.TestCase): "formats": [{"id": "huge"}], "description": "very large payload", } - pq.put(_FakeDownload(info)) + await pq.put(_FakeDownload(info)) with open(path + ".json", encoding="utf-8") as f: payload = json.load(f) @@ -146,7 +149,7 @@ class PersistentQueueTests(unittest.TestCase): self.assertNotIn("formats", record["entry"]) self.assertNotIn("description", record["entry"]) - def test_completed_queue_persists_only_failed_retry_context(self): + async def test_completed_queue_persists_only_failed_retry_context(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "completed") pq = PersistentQueue("completed", path) @@ -161,7 +164,7 @@ class PersistentQueueTests(unittest.TestCase): "formats": [{"id": "huge"}], } info.filename = "done.mp4" - pq.put(_FakeDownload(info)) + await pq.put(_FakeDownload(info)) with open(path + ".json", encoding="utf-8") as f: payload = json.load(f) @@ -180,7 +183,7 @@ class PersistentQueueTests(unittest.TestCase): self.assertEqual(record["filename"], "done.mp4") info.status = "finished" - pq.put(_FakeDownload(info)) + await pq.put(_FakeDownload(info)) with open(path + ".json", encoding="utf-8") as f: payload = json.load(f) self.assertNotIn("entry", payload["items"][0]["info"]) @@ -256,7 +259,7 @@ class PersistentQueueTests(unittest.TestCase): self.assertNotIn("speed", record) self.assertNotIn("eta", record) - def test_put_rollbacks_in_memory_queue_when_state_write_fails(self): + async def test_put_rollbacks_in_memory_queue_when_state_write_fails(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq = PersistentQueue("queue", path) @@ -272,18 +275,18 @@ class PersistentQueueTests(unittest.TestCase): with patch("ytdl.AtomicJsonStore.save", bad_save): with self.assertRaises(OSError): - pq.put(dl) + await pq.put(dl) self.assertFalse(pq.exists("http://rollback.example")) - def test_put_rollbacks_to_previous_download_when_replace_fails(self): + async def test_put_rollbacks_to_previous_download_when_replace_fails(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "queue") pq = PersistentQueue("queue", path) first = _FakeDownload(_make_info("http://same.example")) second = _FakeDownload(_make_info("http://same.example")) second.info.title = "Replaced title" - pq.put(first) + await pq.put(first) orig_save = __import__("state_store").AtomicJsonStore.save @@ -294,10 +297,68 @@ class PersistentQueueTests(unittest.TestCase): with patch("ytdl.AtomicJsonStore.save", bad_save): with self.assertRaises(OSError): - pq.put(second) + await pq.put(second) self.assertEqual(pq.get("http://same.example").info.title, "Title") +class StateWriteOffEventLoopTests(unittest.IsolatedAsyncioTestCase): + """State writes fsync twice; on a slow disk that must not stall the loop. + + Before this, put()/delete() wrote inline, so a queue mutation blocked every + other request the server was serving for as long as the filesystem took. + See issue #980. + """ + + async def test_save_runs_off_the_event_loop_thread(self): + with tempfile.TemporaryDirectory() as tmp: + pq = PersistentQueue("queue", os.path.join(tmp, "queue")) + self.addCleanup(pq.close) + loop_thread = threading.get_ident() + save_threads = [] + + orig_save = __import__("state_store").AtomicJsonStore.save + + def recording_save(store, data): + save_threads.append(threading.get_ident()) + return orig_save(store, data) + + with patch("ytdl.AtomicJsonStore.save", recording_save): + await pq.put(_FakeDownload(_make_info("http://a.example"))) + + self.assertEqual(len(save_threads), 1) + self.assertNotEqual(save_threads[0], loop_thread) + + async def test_a_slow_write_does_not_stall_other_coroutines(self): + with tempfile.TemporaryDirectory() as tmp: + pq = PersistentQueue("queue", os.path.join(tmp, "queue")) + self.addCleanup(pq.close) + orig_save = __import__("state_store").AtomicJsonStore.save + + def slow_save(store, data): + time.sleep(0.3) + return orig_save(store, data) + + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + ticking = asyncio.create_task(ticker()) + try: + with patch("ytdl.AtomicJsonStore.save", slow_save): + await pq.put(_FakeDownload(_make_info("http://a.example"))) + finally: + ticking.cancel() + + # An inline write would have starved the loop for the whole 0.3s and + # left ticks at 0. + self.assertGreater(ticks, 5) + self.assertTrue(pq.exists("http://a.example")) + + if __name__ == "__main__": unittest.main() diff --git a/app/ytdl.py b/app/ytdl.py index 5d28d3f..9d39853 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -1111,6 +1111,19 @@ class PersistentQueue: self.path = f"{path}.json" self.store = AtomicJsonStore(self.path, kind=f"persistent_queue:{name}") self.dict = OrderedDict() + # A state write fsyncs twice (the file and its directory). On a slow or + # contended filesystem that is seconds, and running it inline in an + # async caller blocked the event loop -- every other request stalled + # behind a single queue mutation. One dedicated thread keeps the writes + # off the loop and, being single, keeps them ordered. The default + # executor is not usable for this: extract_info shares it and can hold + # its threads for minutes, which is exactly when state writes happen. + self._store_executor = ThreadPoolExecutor( + max_workers=1, thread_name_prefix=f"state-{name}") + # Guards the mutate-write-rollback section. The write is awaited now, so + # without this two callers could interleave between changing self.dict + # and persisting it, and a rollback could revert the wrong entry. + self._lock = asyncio.Lock() def load(self): for k, v in self.saved_items(): @@ -1151,8 +1164,14 @@ class PersistentQueue: for key, download in self.dict.items() ] - def _save_dict(self): - self.store.save({"items": self._serialize_items()}) + async def _save_dict_async(self): + # Serialize on the event loop -- it is pure CPU and sub-millisecond -- + # and hand the finished payload to the writer thread. Serializing in the + # thread instead would have it walk live DownloadInfo objects while the + # loop mutates them. + payload = {"items": self._serialize_items()} + await asyncio.get_running_loop().run_in_executor( + self._store_executor, self.store.save, payload) def _load_state_items(self): payload = self.store.load() @@ -1193,32 +1212,39 @@ class PersistentQueue: self.store.save({"items": items}) return items - def put(self, value): + async def put(self, value): key = value.info.url - old = self.dict.get(key) - self.dict[key] = value - try: - self._save_dict() - except Exception: - if old is None: - del self.dict[key] - else: - self.dict[key] = old - raise - - def delete(self, key): - if key in self.dict: - old = self.dict[key] - del self.dict[key] + async with self._lock: + old = self.dict.get(key) + self.dict[key] = value try: - self._save_dict() + await self._save_dict_async() except Exception: - self.dict[key] = old + if old is None: + del self.dict[key] + else: + self.dict[key] = old raise + async def delete(self, key): + async with self._lock: + if key in self.dict: + old = self.dict[key] + del self.dict[key] + try: + await self._save_dict_async() + except Exception: + self.dict[key] = old + raise + def empty(self): return not bool(self.dict) + def close(self): + # wait=True so a write already in flight reaches disk before the + # process exits; there is at most one, and it is the newest state. + self._store_executor.shutdown(wait=True) + class DownloadQueue: def __init__(self, config, notifier): self.config = config @@ -1395,8 +1421,8 @@ class DownloadQueue: if not info.error: info.error = str(exc) self._unregister_scheduled(url) - self.queue.delete(url) - self.done.put(download) + await self.queue.delete(url) + await self.done.put(download) await self.notifier.completed(info) else: log.warning( @@ -1430,9 +1456,9 @@ class DownloadQueue: await self.notifier.updated(info) bg_tasks.create_task(self.__start_download(download), name="start_download") - def _schedule_upcoming_download(self, download: Download) -> None: + async def _schedule_upcoming_download(self, download: Download) -> None: download.info.status = 'scheduled' - self.queue.put(download) + await self.queue.put(download) self._register_scheduled(download) def _force_start_scheduled(self, download: Download) -> None: @@ -1451,9 +1477,9 @@ class DownloadQueue: log.info(f"Download {download.info.title} was canceled, skipping start.") return await download.start(self.notifier, self._download_executor) - self._post_download_cleanup(download) + await self._post_download_cleanup(download) - def _post_download_cleanup(self, download): + async def _post_download_cleanup(self, download): if download.info.status != 'finished': if download.tmpfilename and os.path.isfile(download.tmpfilename): try: @@ -1473,11 +1499,11 @@ class DownloadQueue: download.info.size = None download.close() if self.queue.exists(download.info.url): - self.queue.delete(download.info.url) + await self.queue.delete(download.info.url) if download.canceled: bg_tasks.create_task(self.notifier.canceled(download.info.url), name="notify_canceled") else: - self.done.put(download) + await self.done.put(download) bg_tasks.create_task(self.notifier.completed(download.info), name="notify_completed") try: clear_after = int(self.config.CLEAR_COMPLETED_AFTER) @@ -1585,12 +1611,12 @@ class DownloadQueue: ) if auto_start is True: if is_upcoming: - self._schedule_upcoming_download(download) + await self._schedule_upcoming_download(download) else: - self.queue.put(download) + await self.queue.put(download) bg_tasks.create_task(self.__start_download(download), name="start_download") else: - self.pending.put(download) + await self.pending.put(download) await self.notifier.added(dl) def __write_feed_metadata_sync(self, entry, etype, download_type, folder, @@ -1899,7 +1925,7 @@ class DownloadQueue: info.status = 'error' info.msg = msg download = Download(None, None, None, None, quality, format, {}, info) - self.done.put(download) + await self.done.put(download) await self.notifier.completed(info) async def add( @@ -2086,11 +2112,11 @@ class DownloadQueue: for id in ids: if self.pending.exists(id): dl = self.pending.get(id) - self.pending.delete(id) + await self.pending.delete(id) if getattr(dl.info, 'live_status', None) == 'is_upcoming': - self._schedule_upcoming_download(dl) + await self._schedule_upcoming_download(dl) else: - self.queue.put(dl) + await self.queue.put(dl) bg_tasks.create_task(self.__start_download(dl), name="start_download") continue if self.queue.exists(id): @@ -2106,7 +2132,7 @@ class DownloadQueue: # Track URL so playlist add loop won't re-queue it self._canceled_urls.add(id) if self.pending.exists(id): - self.pending.delete(id) + await self.pending.delete(id) await self.notifier.canceled(id) continue if not self.queue.exists(id): @@ -2119,7 +2145,7 @@ class DownloadQueue: dl.cancel() else: dl.canceled = True - self.queue.delete(id) + await self.queue.delete(id) await self.notifier.canceled(id) return {'status': 'ok'} @@ -2157,7 +2183,7 @@ class DownloadQueue: pass except OSError as e: log.warning(f'deleting file "{rel_name}" for download {id} failed with error message {e!r}') - self.done.delete(id) + await self.done.delete(id) await self.notifier.cleared(id) return {'status': 'ok'} @@ -2175,3 +2201,7 @@ class DownloadQueue: if download.started() and download.running(): download.cancel() self._download_executor.shutdown(wait=False, cancel_futures=True) + # Unlike the download executor these are drained, not cancelled: a + # queued write is the newest state and must reach disk before exit. + for queue in (self.queue, self.pending, self.done): + queue.close()