diff --git a/app/features/tasks/router.py b/app/features/tasks/router.py index 6c086ee5..9b94c095 100644 --- a/app/features/tasks/router.py +++ b/app/features/tasks/router.py @@ -439,12 +439,11 @@ async def task_metadata(request: Request, repo: TasksRepository, config: Config, opts["proxy"] = proxy try: - from httpx_curl_cffi import AsyncCurlTransport, CurlOpt + from httpx_curl_cffi import AsyncCurlTransport opts["transport"] = AsyncCurlTransport( impersonate="chrome", default_headers=True, - curl_options={CurlOpt.FRESH_CONNECT: True}, ) opts.pop("headers", None) except Exception: diff --git a/app/library/config.py b/app/library/config.py index c139ed42..6bca0d7e 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -207,6 +207,9 @@ class Config(metaclass=Singleton): live_premiere_buffer: int = 5 """The buffer time in minutes to add to video duration to wait before starting premiere download.""" + add_items_concurrency: int = 4 + """The number of concurrent add items to be processed at same time.""" + playlist_items_concurrency: int = 4 """The number of concurrent playlist items to be processed at same time.""" @@ -274,6 +277,7 @@ class Config(metaclass=Singleton): "auto_clear_history_days", "default_pagination", "extract_info_concurrency", + "add_items_concurrency", "flaresolverr_max_timeout", "flaresolverr_client_timeout", "flaresolverr_cache_ttl", diff --git a/app/library/downloads/item_adder.py b/app/library/downloads/item_adder.py index af7aeaba..40036a41 100644 --- a/app/library/downloads/item_adder.py +++ b/app/library/downloads/item_adder.py @@ -76,7 +76,11 @@ async def add_item( async def add( - queue: "DownloadQueue", item: "Item", already: set | None = None, entry: dict | None = None + queue: "DownloadQueue", + item: "Item", + already: set | None = None, + entry: dict | None = None, + playlist: bool = False, ) -> dict[str, str]: """ Add an item to the download queue. @@ -86,11 +90,23 @@ async def add( item: Item to be added to the queue already: Set of already downloaded items entry: Entry associated with the item (if already extracted) + playlist: Whether the item is part of a playlist Returns: dict[str, str]: Status dict with "status" and optional "msg" keys """ + if playlist: + # If not done like this, it would lead to deadlocks when adding playlist items. + return await _add_impl(queue=queue, item=item, already=already, entry=entry) + + async with queue.add_sem: + return await _add_impl(queue=queue, item=item, already=already, entry=entry) + + +async def _add_impl( + queue: "DownloadQueue", item: "Item", already: set | None = None, entry: dict | None = None +) -> dict[str, str]: _preset: Preset | None = Presets.get_instance().get(item.preset) if item.has_cli(): diff --git a/app/library/downloads/playlist_processor.py b/app/library/downloads/playlist_processor.py index 2a7006d3..1ea39dcd 100644 --- a/app/library/downloads/playlist_processor.py +++ b/app/library/downloads/playlist_processor.py @@ -98,9 +98,9 @@ async def process_playlist( ): dct = merge_dict(merge_dict({"_type": "video"}, etr), entry) dct.pop("entries", None) - return await queue.add(item=newItem, entry=dct, already=already) + return await queue.add(item=newItem, entry=dct, already=already, playlist=True) - return await queue.add(item=newItem, already=already) + return await queue.add(item=newItem, already=already, playlist=True) finally: if acquired: queue.processors.release() diff --git a/app/library/downloads/queue_manager.py b/app/library/downloads/queue_manager.py index 88a35ce6..c4ab5b11 100644 --- a/app/library/downloads/queue_manager.py +++ b/app/library/downloads/queue_manager.py @@ -43,6 +43,8 @@ class DownloadQueue(metaclass=Singleton): "DataStore for the download queue." self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) "Semaphore to limit the number of concurrent processors." + self.add_sem = asyncio.Semaphore(self.config.add_items_concurrency) + "Semaphore to limit the number of concurrent add item operations." self.pool = PoolManager(queue=self, config=self.config) "Pool manager for coordinating download execution." @@ -221,7 +223,9 @@ class DownloadQueue(metaclass=Singleton): async def on_shutdown(self, _: web.Application): await self.pool.shutdown() - async def add(self, item: Item, already: set | None = None, entry: dict | None = None) -> dict[str, str]: + async def add( + self, item: Item, already: set | None = None, entry: dict | None = None, playlist: bool = False + ) -> dict[str, str]: """ Add an item to the download queue. @@ -229,12 +233,13 @@ class DownloadQueue(metaclass=Singleton): item: Item to be added to the queue already: Set of already downloaded items entry: Entry associated with the item (if already extracted) + playlist: Whether the item is part of a playlist Returns: dict[str, str]: Status dict with "status" and optional "msg" keys """ - return await add_impl(queue=self, item=item, already=already, entry=entry) + return await add_impl(queue=self, item=item, already=already, entry=entry, playlist=playlist) async def cancel(self, ids: list[str]) -> dict[str, str]: """