[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.
This commit is contained in:
J. Nick Koston
2026-04-26 09:36:13 -05:00
parent e87e78c544
commit 2d9d1eabfa
4 changed files with 140 additions and 23 deletions
+19 -11
View File
@@ -19,7 +19,7 @@ from esphome.const import (
) )
from esphome.core import CORE, ID, HexInt from esphome.core import CORE, ID, HexInt
from esphome.cpp_generator import MockObj 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 from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -63,13 +63,21 @@ def _compute_local_file_path(value: ConfigType) -> Path:
return base_dir / key return base_dir / key
def _download_web_file(value: ConfigType) -> ConfigType: def _download_all_web_files(config: list[ConfigType]) -> list[ConfigType]:
url = value[CONF_URL] """Validate that all web-sourced files are cached, fetching missing/changed
path = _compute_local_file_path(value) ones in parallel before per-item validators read them off disk.
"""
download_content(url, path) items: list[tuple[str, Path]] = []
_LOGGER.debug("download_web_file: path=%s", path) for file_config in config:
return value 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: 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, 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( CONFIG_SCHEMA = cv.All(
cv.only_on_esp32, cv.only_on_esp32,
cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA), cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
_download_all_web_files,
_validate_supported_local_file, _validate_supported_local_file,
) )
@@ -32,7 +32,7 @@ from esphome.const import (
CONF_URL, CONF_URL,
) )
from esphome.core import CORE, HexInt 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__) _LOGGER = logging.getLogger(__name__)
@@ -92,13 +92,21 @@ def _compute_local_file_path(value: dict) -> Path:
return base_dir / key return base_dir / key
def _download_web_file(value): def _download_all_web_files(config):
url = value[CONF_URL] """Validate that all web-sourced files are cached, fetching missing/changed
path = _compute_local_file_path(value) ones in parallel before per-item validators read them off disk.
"""
download_content(url, path) items: list[tuple[str, Path]] = []
_LOGGER.debug("download_web_file: path=%s", path) for file_config in config:
return value 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 = { _PURPOSE_MAP = {
@@ -229,11 +237,10 @@ LOCAL_SCHEMA = cv.Schema(
} }
) )
WEB_SCHEMA = cv.All( WEB_SCHEMA = cv.Schema(
{ {
cv.Required(CONF_URL): cv.url, cv.Required(CONF_URL): cv.url,
}, }
_download_web_file,
) )
@@ -285,7 +292,10 @@ CONFIG_SCHEMA = cv.All(
), ),
# Remove before 2026.10.0 # Remove before 2026.10.0
cv.Optional(CONF_CODEC_SUPPORT_ENABLED): cv.Any(cv.boolean, cv.string), 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.Optional(CONF_TASK_STACK_IN_PSRAM): cv.All(
cv.boolean, cv.requires_component(psram.DOMAIN) cv.boolean, cv.requires_component(psram.DOMAIN)
), ),
+33
View File
@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Iterable
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime from datetime import UTC, datetime
import logging import logging
from pathlib import Path from pathlib import Path
@@ -116,3 +118,34 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
data = req.content data = req.content
path.write_bytes(data) path.write_bytes(data)
return 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))
+66
View File
@@ -282,3 +282,69 @@ def test_download_content_skip_external_update_downloads_when_missing(
assert result == new_content assert result == new_content
assert test_file.read_bytes() == 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)