From 2d9d1eabfa1ea45e1b52fbb3fc97893eeab45da4 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 26 Apr 2026 09:36:13 -0500 Subject: [PATCH] [core] Download external_files in parallel Each component that uses external_files (audio_file, speaker media_player, ...) currently calls download_content once per file inline inside a per-item config validator. With ~24 cached audio files in a Home Assistant Voice PE config, that means ~24 sequential HEAD round-trips, even when every response is a 304. This adds download_content_many(items, ...) which fans the per-file checks out across a ThreadPoolExecutor (capped at 16 workers so configs with hundreds of files don't open hundreds of sockets), then refactors audio_file and speaker.media_player to collect URLs at the list level and call the batch helper once instead of downloading inside each per-item validator. Wall time for the validation phase drops from sum(latency) to roughly max(latency) when the cache is warm. --- esphome/components/audio_file/__init__.py | 30 +++++---- .../speaker/media_player/__init__.py | 34 ++++++---- esphome/external_files.py | 33 ++++++++++ tests/unit_tests/test_external_files.py | 66 +++++++++++++++++++ 4 files changed, 140 insertions(+), 23 deletions(-) diff --git a/esphome/components/audio_file/__init__.py b/esphome/components/audio_file/__init__.py index bb1ce257db..4db8f11432 100644 --- a/esphome/components/audio_file/__init__.py +++ b/esphome/components/audio_file/__init__.py @@ -19,7 +19,7 @@ from esphome.const import ( ) from esphome.core import CORE, ID, HexInt from esphome.cpp_generator import MockObj -from esphome.external_files import download_content +from esphome.external_files import download_content_many from esphome.types import ConfigType _LOGGER = logging.getLogger(__name__) @@ -63,13 +63,21 @@ def _compute_local_file_path(value: ConfigType) -> Path: return base_dir / key -def _download_web_file(value: ConfigType) -> ConfigType: - url = value[CONF_URL] - path = _compute_local_file_path(value) - - download_content(url, path) - _LOGGER.debug("download_web_file: path=%s", path) - return value +def _download_all_web_files(config: list[ConfigType]) -> list[ConfigType]: + """Validate that all web-sourced files are cached, fetching missing/changed + ones in parallel before per-item validators read them off disk. + """ + items: list[tuple[str, Path]] = [] + for file_config in config: + conf_file = file_config.get(CONF_FILE, {}) + if conf_file.get(CONF_TYPE) != TYPE_WEB: + continue + url = conf_file[CONF_URL] + path = _compute_local_file_path(conf_file) + items.append((url, path)) + _LOGGER.debug("download_web_file: path=%s", path) + download_content_many(items) + return config def _file_schema(value: ConfigType | str) -> ConfigType: @@ -142,11 +150,10 @@ LOCAL_SCHEMA = cv.Schema( } ) -WEB_SCHEMA = cv.All( +WEB_SCHEMA = cv.Schema( { cv.Required(CONF_URL): cv.url, - }, - _download_web_file, + } ) @@ -209,6 +216,7 @@ def _validate_supported_local_file(config: list[ConfigType]) -> list[ConfigType] CONFIG_SCHEMA = cv.All( cv.only_on_esp32, cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA), + _download_all_web_files, _validate_supported_local_file, ) diff --git a/esphome/components/speaker/media_player/__init__.py b/esphome/components/speaker/media_player/__init__.py index abfd599808..7c4b366947 100644 --- a/esphome/components/speaker/media_player/__init__.py +++ b/esphome/components/speaker/media_player/__init__.py @@ -32,7 +32,7 @@ from esphome.const import ( CONF_URL, ) from esphome.core import CORE, HexInt -from esphome.external_files import download_content +from esphome.external_files import download_content_many _LOGGER = logging.getLogger(__name__) @@ -92,13 +92,21 @@ def _compute_local_file_path(value: dict) -> Path: return base_dir / key -def _download_web_file(value): - url = value[CONF_URL] - path = _compute_local_file_path(value) - - download_content(url, path) - _LOGGER.debug("download_web_file: path=%s", path) - return value +def _download_all_web_files(config): + """Validate that all web-sourced files are cached, fetching missing/changed + ones in parallel before per-item validators read them off disk. + """ + items: list[tuple[str, Path]] = [] + for file_config in config: + conf_file = file_config.get(CONF_FILE, {}) + if conf_file.get(CONF_TYPE) != TYPE_WEB: + continue + url = conf_file[CONF_URL] + path = _compute_local_file_path(conf_file) + items.append((url, path)) + _LOGGER.debug("download_web_file: path=%s", path) + download_content_many(items) + return config _PURPOSE_MAP = { @@ -229,11 +237,10 @@ LOCAL_SCHEMA = cv.Schema( } ) -WEB_SCHEMA = cv.All( +WEB_SCHEMA = cv.Schema( { cv.Required(CONF_URL): cv.url, - }, - _download_web_file, + } ) @@ -285,7 +292,10 @@ CONFIG_SCHEMA = cv.All( ), # Remove before 2026.10.0 cv.Optional(CONF_CODEC_SUPPORT_ENABLED): cv.Any(cv.boolean, cv.string), - cv.Optional(CONF_FILES): cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA), + cv.Optional(CONF_FILES): cv.All( + cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA), + _download_all_web_files, + ), cv.Optional(CONF_TASK_STACK_IN_PSRAM): cv.All( cv.boolean, cv.requires_component(psram.DOMAIN) ), diff --git a/esphome/external_files.py b/esphome/external_files.py index b6f6149ebb..0cf6ab7006 100644 --- a/esphome/external_files.py +++ b/esphome/external_files.py @@ -1,5 +1,7 @@ from __future__ import annotations +from collections.abc import Iterable +from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime import logging from pathlib import Path @@ -116,3 +118,34 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by data = req.content path.write_bytes(data) return data + + +# Cap concurrent connections so a config with hundreds of remote files doesn't +# open hundreds of sockets at once. 16 is wide enough that wall time is +# dominated by the slowest single request for normal configs (a couple dozen +# files), and tight enough to be polite to the upstream host. +DEFAULT_DOWNLOAD_WORKERS = 16 + + +def download_content_many( + items: Iterable[tuple[str, Path]], + timeout: int = NETWORK_TIMEOUT, + max_workers: int = DEFAULT_DOWNLOAD_WORKERS, +) -> None: + """Run `download_content` for each (url, path) pair concurrently. + + Wall time drops from `sum(latency)` to roughly `max(latency)` for cached + files where the HEAD round-trip dominates. The first exception raised by + any worker is propagated; remaining workers complete before this returns. + """ + items = list(items) + if not items: + return + if len(items) == 1: + url, path = items[0] + download_content(url, path, timeout) + return + workers = min(max_workers, len(items)) + with ThreadPoolExecutor(max_workers=workers) as ex: + # list() forces iteration so exceptions surface here, not silently. + list(ex.map(lambda item: download_content(item[0], item[1], timeout), items)) diff --git a/tests/unit_tests/test_external_files.py b/tests/unit_tests/test_external_files.py index 4b0826db04..182e424adb 100644 --- a/tests/unit_tests/test_external_files.py +++ b/tests/unit_tests/test_external_files.py @@ -282,3 +282,69 @@ def test_download_content_skip_external_update_downloads_when_missing( assert result == new_content assert test_file.read_bytes() == new_content + + +@patch("esphome.external_files.download_content") +def test_download_content_many_empty_is_noop( + mock_download: MagicMock, setup_core: Path +) -> None: + """Empty input shouldn't spin up a thread pool or call download_content.""" + external_files.download_content_many([]) + mock_download.assert_not_called() + + +@patch("esphome.external_files.download_content") +def test_download_content_many_single_item_avoids_pool( + mock_download: MagicMock, setup_core: Path +) -> None: + """A single item should be downloaded inline (no thread pool overhead).""" + item = ("https://example.com/file.txt", setup_core / "f.txt") + external_files.download_content_many([item]) + mock_download.assert_called_once_with( + item[0], item[1], external_files.NETWORK_TIMEOUT + ) + + +@patch("esphome.external_files.download_content") +def test_download_content_many_runs_in_parallel( + mock_download: MagicMock, setup_core: Path +) -> None: + """Multiple items should run concurrently — total wall time ≈ max latency.""" + import threading + + barrier = threading.Barrier(3) + + def slow_download(url: str, path: Path, timeout: int) -> bytes: + # If calls were serial this would deadlock (third caller never arrives + # while the first is blocked at the barrier). + barrier.wait(timeout=2.0) + return b"" + + mock_download.side_effect = slow_download + items = [ + ("https://example.com/a", setup_core / "a"), + ("https://example.com/b", setup_core / "b"), + ("https://example.com/c", setup_core / "c"), + ] + external_files.download_content_many(items, max_workers=4) + assert mock_download.call_count == 3 + + +@patch("esphome.external_files.download_content") +def test_download_content_many_propagates_errors( + mock_download: MagicMock, setup_core: Path +) -> None: + """An exception from any worker must propagate out of download_content_many.""" + + def fake_download(url: str, path: Path, timeout: int) -> bytes: + if url.endswith("bad"): + raise Invalid(f"could not download {url}") + return b"" + + mock_download.side_effect = fake_download + items = [ + ("https://example.com/ok", setup_core / "ok"), + ("https://example.com/bad", setup_core / "bad"), + ] + with pytest.raises(Invalid, match="could not download"): + external_files.download_content_many(items)