diff --git a/FAQ.md b/FAQ.md index 16d6dcdc..e5fa31f2 100644 --- a/FAQ.md +++ b/FAQ.md @@ -17,7 +17,8 @@ or the `environment:` section in `compose.yaml` file. | YTP_INSTANCE_TITLE | The title of the instance | `empty string` | | YTP_FILE_LOGGING | Whether to log to file | `false` | | YTP_DOWNLOAD_PATH | Path to where the downloads will be saved | `/downloads` | -| YTP_MAX_WORKERS | How many works to use for downloads | `1` | +| YTP_MAX_WORKERS | The maximum number of workers to use for downloading | `20` | +| YTP_MAX_WORKERS_PER_EXTRACTOR | The maximum number of concurrent downloads per extractor | `2` | | YTP_AUTH_USERNAME | Username for basic authentication | `empty string` | | YTP_AUTH_PASSWORD | Password for basic authentication | `empty string` | | YTP_CONSOLE_ENABLED | Whether to enable the console | `false` | @@ -50,6 +51,12 @@ or the `environment:` section in `compose.yaml` file. | YTP_TEMP_DISABLED | Disable temp files handling. | `false` | | YTP_DOWNLOAD_PATH_DEPTH | How many subdirectories to show in auto complete. | `1` | +> [!NOTE] +> To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_`. +> The extractor name must be in uppercase, to know the extractor name, check the log for the specific extractor used for the download. +> The limit should not exceed the `YTP_MAX_WORKERS` value as it will be ignored. + + # Browser extensions & bookmarklets ## Simple bookmarklet diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 0aac9c68..b82e3363 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -2,6 +2,7 @@ import asyncio import functools import glob import logging +import os import time import traceback import uuid @@ -60,6 +61,8 @@ class DownloadQueue(metaclass=Singleton): "Semaphore to limit the number of concurrent downloads." self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) "Semaphore to limit the number of concurrent processors." + self.limits: dict[str, asyncio.Semaphore] = {} + "Per-extractor semaphores to limit concurrent downloads per extractor." self.paused = asyncio.Event() "Event to pause the download queue." self.event = asyncio.Event() @@ -82,6 +85,35 @@ class DownloadQueue(metaclass=Singleton): """ return DownloadQueue() + def _get_limit(self, extractor: str) -> asyncio.Semaphore: + """ + Get or create a semaphore for the given extractor. + + Args: + extractor (str): The extractor name. + + Returns: + asyncio.Semaphore: The semaphore for the extractor. + + """ + if extractor not in self.limits: + env_limit: str | None = os.environ.get(f"YTP_MAX_WORKERS_FOR_{extractor.upper()}") + + # Determine effective limit + if env_limit and env_limit.isdigit() and 1 <= int(env_limit): + limit: int = min(int(env_limit), self.config.max_workers) + else: + if env_limit: + LOG.warning(f"Invalid extractor limit '{env_limit}' for '{extractor}', using default limit.") + limit = self.config.max_workers_per_extractor + + limit = min(limit, self.config.max_workers) + + self.limits[extractor] = asyncio.Semaphore(limit) + LOG.info(f"Created limits container for extractor '{extractor}': {limit}") + + return self.limits[extractor] + def attach(self, _: web.Application) -> None: """ Attach the download queue to the application. @@ -983,36 +1015,67 @@ class DownloadQueue(metaclass=Singleton): """ Create a pool of workers to download the files. """ + adaptive_sleep = 0.2 # Start with base sleep + max_sleep = 5.0 # Maximum sleep to avoid excessive delays + while True: while not self.queue.has_downloads(): LOG.info("Waiting for item to download.") await self.event.wait() self.event.clear() + adaptive_sleep = 0.2 if self.is_paused(): - LOG.info("Download pool is paused.") + LOG.warning("Download pool is paused.") await self.paused.wait() LOG.info("Download pool resumed downloading.") + adaptive_sleep = 0.2 + + items_processed = 0 for _id, entry in list(self.queue.items()): if entry.started() or entry.is_cancelled() or entry.info.auto_start is False: continue + extractor: str = entry.info.get_extractor() or "unknown" + + # Live downloads bypass all limits. if entry.is_live: - task = asyncio.create_task(self._download_live(_id, entry), name=f"download_live_{_id}") + task: asyncio.Task[None] = asyncio.create_task( + self._download_live(_id, entry), name=f"download_live_{extractor}_{_id}" + ) task.add_done_callback(self._handle_task_exception) + items_processed += 1 else: + _limit: asyncio.Semaphore = self._get_limit(extractor) + + # Skip this item in this iteration if no slots are available. + if self.workers.locked() or _limit.locked(): + continue + await self.workers.acquire() + await _limit.acquire() - task = asyncio.create_task(self._download_file(_id, entry), name=f"download_file_{_id}") + task: asyncio.Task[None] = asyncio.create_task( + self._download_file(_id, entry), name=f"download_file_{extractor}_{_id}" + ) - def _release_semaphore(t: asyncio.Task): + def _release(t: asyncio.Task, sem=_limit) -> None: + sem.release() self.workers.release() self._handle_task_exception(t) - task.add_done_callback(_release_semaphore) + task.add_done_callback(_release) + items_processed += 1 - await asyncio.sleep(0.5) + # No items could be processed, back off a bit to avoid busy-waiting. + if 0 == items_processed: + adaptive_sleep: float = min(adaptive_sleep * 1.5, max_sleep) + LOG.info(f"No download slots available. Backing off for {adaptive_sleep:.2f}s before next attempt.") + else: + adaptive_sleep = 0.2 + + await asyncio.sleep(adaptive_sleep) async def _download_live(self, _id: str, entry: Download) -> None: LOG.info(f"Creating temporary worker for entry '{entry.info.name()}'.") diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index b1282f42..12449961 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -167,11 +167,10 @@ class Item: return Item(**data) def get_archive_id(self) -> str | None: - if not self.url: - return None + return get_archive_id(self.url).get("archive_id") if self.url else None - idDict: dict = get_archive_id(self.url) - return idDict.get("archive_id") + def get_extractor(self) -> str | None: + return get_archive_id(self.url).get("ie_key") if self.url else None def get_ytdlp_opts(self) -> YTDLPOpts: params: YTDLPOpts = YTDLPOpts.get_instance() @@ -326,6 +325,15 @@ class ItemDTO: return self.archive_id + def get_extractor(self) -> str | None: + if self.archive_id: + return self.archive_id.split(" ")[0] + + idDict: dict[str, str | None] = get_archive_id(self.url) + self.archive_id = idDict.get("archive_id") + + return idDict.get("ie_key") if self.url else None + def get_ytdlp_opts(self) -> YTDLPOpts: """ Get the yt-dlp options for the item. diff --git a/app/library/Utils.py b/app/library/Utils.py index 65ada868..8411827b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -87,6 +87,8 @@ DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2} T = TypeVar("T") "Generic type variable." +ARCHIVE_IDS_CACHE: dict[str, dict] = {} +"Cache for archive IDs." class StreamingError(Exception): """Raised when an error occurs during streaming.""" @@ -1114,6 +1116,9 @@ def get_archive_id(url: str) -> dict[str, str | None]: "archive_id": None, } + if url in ARCHIVE_IDS_CACHE: + return ARCHIVE_IDS_CACHE[url] + if YTDLP_INFO_CLS is None: YTDLP_INFO_CLS = YTDLP( params={ @@ -1146,6 +1151,7 @@ def get_archive_id(url: str) -> dict[str, str | None]: LOG.exception(e) LOG.error(f"Error getting archive ID: {e}") + ARCHIVE_IDS_CACHE.update({url: idDict}) return idDict diff --git a/app/library/config.py b/app/library/config.py index f5268f65..d3f099ec 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -69,9 +69,12 @@ class Config(metaclass=Singleton): base_path: str = "/" """The base path to use for the application.""" - max_workers: int = 1 + max_workers: int = 20 """The maximum number of workers to use for downloading.""" + max_workers_per_extractor: int = 2 + """The maximum number of concurrent downloads per extractor.""" + streamer_vcodec: str = "" """The video codec to use for streaming. If empty, auto-detect.""" @@ -205,6 +208,7 @@ class Config(metaclass=Singleton): _int_vars: tuple = ( "port", "max_workers", + "max_workers_per_extractor", "extract_info_timeout", "debugpy_port", "playlist_items_concurrency", @@ -240,6 +244,7 @@ class Config(metaclass=Singleton): "remove_files", "ui_update_title", "max_workers", + "max_workers_per_extractor", "default_preset", "instance_title", "console_enabled",