diff --git a/FAQ.md b/FAQ.md index 5e2ae000..fd034fc6 100644 --- a/FAQ.md +++ b/FAQ.md @@ -57,6 +57,7 @@ or the `environment:` section in `compose.yaml` file. | YTP_TASK_HANDLER_RANDOM_DELAY | The maximum random delay in seconds before starting a task handler. | `60` | | YTP_IGNORE_ARCHIVED_ITEMS | Don't report archived items in the download history. | `false` | | YTP_CHECK_FOR_UPDATES | Whether to check for application updates. | `true` | +| YTP_EXTRACT_INFO_CONCURRENCY | The number of concurrent extract info operations. | `4` | > [!NOTE] > To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_`. diff --git a/app/library/Events.py b/app/library/Events.py index 2f7a4738..05afb8cd 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -9,7 +9,7 @@ from typing import Any from .BackgroundWorker import BackgroundWorker from .Singleton import Singleton -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("library.events") class Events: diff --git a/app/library/config.py b/app/library/config.py index 13b44051..18832e53 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -117,6 +117,9 @@ class Config(metaclass=Singleton): extract_info_timeout: int = 70 """The timeout to use for extracting video information.""" + extract_info_concurrency: int = 4 + """The number of concurrent extract_info calls allowed.""" + db_file: str = "{config_path}{os_sep}ytptube.db" """The path to the database file.""" diff --git a/app/library/downloads/item_adder.py b/app/library/downloads/item_adder.py index 2d7690bc..d0a23473 100644 --- a/app/library/downloads/item_adder.py +++ b/app/library/downloads/item_adder.py @@ -45,7 +45,7 @@ if TYPE_CHECKING: from .queue_manager import DownloadQueue -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.add") async def add_item( @@ -201,23 +201,26 @@ async def add( LOG.error(msg) return {"status": "error", "msg": msg} - LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.") + if entry: + LOG.info(f"[P] Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.") if not entry: - entry: dict | None = await asyncio.wait_for( - fut=asyncio.get_running_loop().run_in_executor( - None, - functools.partial( - extract_info, - config=yt_conf, - url=item.url, - debug=bool(queue.config.ytdlp_debug), - no_archive=False, - follow_redirect=True, + async with queue.extractors: + LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.") + entry: dict | None = await asyncio.wait_for( + fut=asyncio.get_running_loop().run_in_executor( + None, + functools.partial( + extract_info, + config=yt_conf, + url=item.url, + debug=bool(queue.config.ytdlp_debug), + no_archive=False, + follow_redirect=True, + ), ), - ), - timeout=queue.config.extract_info_timeout, - ) + timeout=queue.config.extract_info_timeout, + ) if not entry: LOG.error(f"Unable to extract info for '{item.url}'. Logs: {logs}") diff --git a/app/library/downloads/monitors.py b/app/library/downloads/monitors.py index c83802f0..3938e580 100644 --- a/app/library/downloads/monitors.py +++ b/app/library/downloads/monitors.py @@ -12,7 +12,7 @@ from app.library.Utils import dt_delta, str_to_dt if TYPE_CHECKING: from .queue_manager import DownloadQueue -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.monitors") async def check_for_stale(queue: "DownloadQueue") -> None: diff --git a/app/library/downloads/playlist_processor.py b/app/library/downloads/playlist_processor.py index a1327d77..2a7006d3 100644 --- a/app/library/downloads/playlist_processor.py +++ b/app/library/downloads/playlist_processor.py @@ -13,7 +13,7 @@ if TYPE_CHECKING: from .queue_manager import DownloadQueue -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.playlist") async def process_playlist( diff --git a/app/library/downloads/pool_manager.py b/app/library/downloads/pool_manager.py index 89584fb8..944cfe5e 100644 --- a/app/library/downloads/pool_manager.py +++ b/app/library/downloads/pool_manager.py @@ -15,7 +15,7 @@ if TYPE_CHECKING: from .queue_manager import DownloadQueue -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.pool") class PoolManager: diff --git a/app/library/downloads/queue_manager.py b/app/library/downloads/queue_manager.py index cc6cf8e1..b9d66ad3 100644 --- a/app/library/downloads/queue_manager.py +++ b/app/library/downloads/queue_manager.py @@ -25,7 +25,7 @@ from .pool_manager import PoolManager if TYPE_CHECKING: from app.library.DataStore import StoreType -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.queue") class DownloadQueue(metaclass=Singleton): @@ -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.extractors = asyncio.Semaphore(self.config.extract_info_concurrency) + "Semaphore to limit the number of concurrent extract_info calls." self.pool = PoolManager(queue=self, config=self.config) "Pool manager for coordinating download execution." diff --git a/app/library/downloads/video_processor.py b/app/library/downloads/video_processor.py index 67c7a644..93c16844 100644 --- a/app/library/downloads/video_processor.py +++ b/app/library/downloads/video_processor.py @@ -17,7 +17,7 @@ if TYPE_CHECKING: from .queue_manager import DownloadQueue -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("downloads.video") async def add_video(queue: "DownloadQueue", entry: dict, item: "Item", logs: list[str] | None = None) -> dict[str, str]: diff --git a/app/tests/test_item_adder_extract_concurrency.py b/app/tests/test_item_adder_extract_concurrency.py new file mode 100644 index 00000000..d1211e55 --- /dev/null +++ b/app/tests/test_item_adder_extract_concurrency.py @@ -0,0 +1,66 @@ +import time +import asyncio +import threading +import pytest + +from app.library.config import Config +from app.library.downloads.queue_manager import DownloadQueue +from app.library.ItemDTO import Item + + +@pytest.mark.asyncio +async def test_extract_concurrency_limited(monkeypatch): + """Ensure that concurrent extract_info calls are limited by the configured semaphore.""" + # Configure a low concurrency to make the timing assertions stable + cfg = Config.get_instance() + cfg.extract_info_concurrency = 2 + + # Reset singleton so new DownloadQueue picks up updated config + DownloadQueue._reset_singleton() + queue = DownloadQueue.get_instance() + + sleep_time = 0.18 + + # Thread-safe counters to observe concurrency usage in the executor threads + lock = threading.Lock() + current = {"val": 0} + max_seen = {"max": 0} + + def fake_extract_info(config, url, **kwargs): + # Track concurrent starts + with lock: + current["val"] += 1 + if current["val"] > max_seen["max"]: + max_seen["max"] = current["val"] + + # Blocking call to simulate expensive IO/work executed in executor + time.sleep(sleep_time) + + with lock: + current["val"] -= 1 + + return {"_type": "video", "id": url.split("/")[-1], "title": "t", "url": url} + + monkeypatch.setattr("app.library.Utils.extract_info", fake_extract_info) + + items = [Item(url=f"http://example.com/{i}") for i in range(6)] + + start = time.perf_counter() + tasks = [asyncio.create_task(queue.add(item=item)) for item in items] + + await asyncio.gather(*tasks) + elapsed = time.perf_counter() - start + + # Assert we never exceeded the configured concurrency + assert max_seen["max"] <= cfg.extract_info_concurrency, ( + f"Max concurrent extractions {max_seen['max']} exceeded limit {cfg.extract_info_concurrency}" + ) + + # Sanity timing check (relaxed to avoid flakiness) + rounds = -(-len(items) // cfg.extract_info_concurrency) # ceil division + assert elapsed >= rounds * sleep_time * 0.8, ( + f"Elapsed {elapsed:.2f}s too low; expected at least {rounds * sleep_time * 0.8:.2f}s" + ) + + # Cleanup + DownloadQueue._reset_singleton()