github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-11 15:27:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify: one custom-name rule in _uri_jobs, ordering owned by the pool

Browse Source
This commit is contained in:
J. Nick Koston
2026-08-31 16:08:57 -04:00
parent 6877513195
commit 328e83ad64
2 changed files with 102 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/platformio/prefetch.py
+30 -54
View File
@@ -55,11 +55,9 @@ def _preserved_sys_path() -> Iterator[None]:
sys.path[:] = saved
# Concurrent registry resolutions / HEAD probes (each is network-bound)
_RESOLVE_WORKERS = 8
# Floor cap for network-bound clones in the pre-install pool
_CLONE_WORKERS = 8
# One cap for concurrent network-bound work: registry resolutions,
# HEAD probes, and the pre-install pool's clone floor
_NETWORK_WORKERS = 8
# A hung child must not block the build; downloads resume on the next run
_PREFETCH_TIMEOUT = 20 * 60
@@ -344,7 +342,7 @@ def _registry_jobs(
if not pending:
return [], 0, []
# Serial resolutions (registry GET + mirror HEAD each) dominate
with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(pending))) as ex:
with ThreadPoolExecutor(max_workers=min(_NETWORK_WORKERS, len(pending))) as ex:
results = list(ex.map(_resolve, pending))
jobs: list[tuple[str, int, Any]] = []
installable: list[tuple[str, Any]] = []
@@ -381,56 +379,31 @@ def _registry_jobs(
_VCS_URI_PREFIXES = ("git+", "hg+", "svn+", "git://", "hg://", "svn://")
def _is_vcs_spec_uri(url: str) -> bool:
def _is_vcs_spec_uri(url: str | None) -> bool:
"""Whether pio's ``install_from_uri`` would clone this URI rather than
copy or download it (PackageSpec normalizes git URLs to ``git+``).
Positive match, with a .git path as the backstop; an unrecognized
scheme is skipped here and left to pio run."""
if url.startswith(("file://", "symlink://", "http://", "https://")):
Positive match, with a .git path as the backstop."""
if not url or url.startswith(("file://", "symlink://", "http://", "https://")):
return False
if url.startswith(_VCS_URI_PREFIXES) or url.split("#", 1)[0].endswith(".git"):
return True
_LOGGER.debug("Unrecognized package URI scheme, leaving it to pio run: %s", url)
return False
return url.startswith(_VCS_URI_PREFIXES) or url.split("#", 1)[0].endswith(".git")
def _spec_name(spec: Any, url: str) -> str:
"""The spec's name; the URL fallbacks are defensive only
(PackageSpec derives a name from the URI itself). Never empty:
an empty name would coalesce distinct to_install keys."""
return spec.name or url.split("#", 1)[0].rstrip("/").rsplit("/", 1)[-1] or url
# (name, spec) from wave 1, (name, spec, compatibility) from dep waves
_Entry = tuple[str, Any] | tuple[str, Any, Any]
def _entry_is_vcs(entry: tuple[str, Any] | tuple[str, Any, Any]) -> bool:
def _entry_is_vcs(entry: _Entry) -> bool:
"""Whether this (name, spec[, compatibility]) pre-install entry is
cloned rather than unpacked."""
url = entry[1].uri
return bool(url and _is_vcs_spec_uri(url))
return _is_vcs_spec_uri(entry[1].uri)
def _clones_first(entries: Iterable[tuple[str, Any]]) -> list[tuple[str, Any]]:
def _clones_first(entries: Iterable[_Entry]) -> list[_Entry]:
"""Clones first: they wait on the network, so they must not queue
behind CPU-bound archive extractions in the pre-install pool."""
return sorted(entries, key=lambda entry: not _entry_is_vcs(entry))
def _installable_in_parallel(
entries: list[tuple[str, Any]], curated: bool
) -> list[tuple[str, Any]]:
"""Drop derived-name clones from uncurated (lib_deps) groups: their
destination dir comes from the cloned manifest, so two entries whose
manifests share a name could race one directory in the pool. The
platform's tool manifests are curated, and pio run installs anything
dropped here serially."""
if curated:
return entries
return [
entry
for entry in entries
if not _entry_is_vcs(entry) or entry[1].has_custom_name()
]
def _uri_jobs(
manager: Any, specs: list[Any], seen: set[str]
) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]:
@@ -451,15 +424,22 @@ def _uri_jobs(
continue
is_vcs = _is_vcs_spec_uri(url)
if not is_vcs and not url.startswith(("http://", "https://")):
if not url.startswith(("file://", "symlink://")):
_LOGGER.debug(
"Unrecognized package URI, leaving it to pio run: %s", url
)
continue # file/symlink specs are copied in place by pio run
if manager.get_package(spec):
continue
name = _spec_name(spec, url)
name = spec.name
if is_vcs:
# No download stage: the pre-install itself clones it.
# Derived-name clones from uncurated groups are dropped at
# the group site by _installable_in_parallel
installable.append((name, spec))
# No download stage: the pre-install itself clones it. Same
# rule as the cached-archive branch below: only a custom name
# is the destination dir. Platform tool specs (owner, name,
# requirements=url) always parse as custom-named, so the
# curated batch stays parallel; pio run installs the rest
if spec.has_custom_name():
installable.append((name, spec))
continue
# PlatformIO downloads URL specs with no checksum
dl_path = Path(manager.compute_download_path(url, ""))
@@ -500,7 +480,7 @@ def _uri_jobs(
if not candidates:
return [], 0, installable
with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex:
with ThreadPoolExecutor(max_workers=min(_NETWORK_WORKERS, len(candidates))) as ex:
sizes = list(ex.map(_head_size, [url for _, url, _, _ in candidates]))
jobs: list[tuple[str, int, Any]] = []
failed = 0
@@ -648,10 +628,6 @@ def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any:
return run
# (name, spec) from wave 1, (name, spec, compatibility) from dep waves
_Entry = tuple[str, Any] | tuple[str, Any, Any]
def _dependency_entries(
manager: Any, entries: list[_Entry], seen_names: set[str]
) -> list[_Entry]:
@@ -766,13 +742,14 @@ def _preinstall(
would hang, not fail). Waves skip dependencies; the installed
manifests feed the next wave. Any failure falls back to pio run.
"""
entries = _clones_first(entries)
clones = sum(1 for entry in entries if _entry_is_vcs(entry))
# Clones are network-bound: let them run wide even on small-core
# runners. Capped, since every worker builds a sibling manager and
# may run a postinstall script; the tail of a mixed wave runs its
# (largely I/O-bound) extractions at the same width
workers = min(
max(get_usable_cpu_count(), min(clones, _CLONE_WORKERS)), len(entries)
max(get_usable_cpu_count(), min(clones, _NETWORK_WORKERS)), len(entries)
)
# One manager per worker (_install mutates instance state); built
# serially because construction rewires the shared manager logger
@@ -946,7 +923,7 @@ def _prefetch(build_dir: Path, env: str) -> None:
jobs += batch_jobs
unresolved += failed
entries += installable
if entries := _installable_in_parallel(entries, curated=is_platform):
if entries:
groups.append(_Group(mgr, entries, is_platform))
sentinel = build_dir / _SENTINEL_NAME
@@ -986,9 +963,8 @@ def _prefetch(build_dir: Path, env: str) -> None:
if name not in failed_names
}
if to_install:
ordered = _clones_first(to_install.values())
try:
_preinstall(mgr, ordered)
_preinstall(mgr, list(to_install.values()))
if is_platform:
platform_packages_installed = True
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
tests/unit_tests/test_platformio_prefetch.py
+72 -129
View File
@@ -597,7 +597,7 @@ def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
m,
[
_FakeSpec(uri="https://x/big.zip", name="big", custom_name=True),
_FakeSpec(uri="git+https://x/repo.git", name="repo"),
_FakeSpec(uri="git+https://x/repo.git", name="repo", custom_name=True),
_FakeSpec(name="registry"),
],
set(),
@@ -614,20 +614,20 @@ def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
"""VCS specs never probe the network here (there is no archive); an
uninstalled one is handed to the pre-install, an installed one and
file/symlink specs are skipped."""
"""VCS specs never probe the network here (there is no archive); a
custom-named uninstalled one is handed to the pre-install, while a
derived-name one, an installed one, and file/symlink specs are left
to pio run (a derived name is not the destination dir)."""
m = _fake_manager(tmp_path)
with patch("esphome.net_retry.http_request") as mock_head:
jobs, failed, installable = pf._uri_jobs(
m,
[
_FakeSpec(uri="git+https://x/tool.git#1.0", name="tool"),
_FakeSpec(uri="hg+https://x/old", name="mercurial"),
# Name falls back to the URL basename, fragment excluded
_FakeSpec(uri="git+https://x/noname#v2", name=None),
# A trailing slash must not derive an empty (colliding) name
_FakeSpec(uri="git+https://x/trail/", name=None),
_FakeSpec(
uri="git+https://x/tool.git#1.0", name="tool", custom_name=True
),
_FakeSpec(uri="hg+https://x/old", name="mercurial", custom_name=True),
_FakeSpec(uri="git+https://x/derived.git", name="derived"),
_FakeSpec(uri="file:///local/dir", name="local"),
_FakeSpec(uri="symlink:///local/dir", name="link"),
],
@@ -635,7 +635,7 @@ def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
)
mock_head.assert_not_called()
assert (jobs, failed) == ([], 0)
assert [n for n, _ in installable] == ["tool", "mercurial", "noname", "trail"]
assert [n for n, _ in installable] == ["tool", "mercurial"]
# Positive classification: an unknown scheme is left to pio run
with patch("esphome.net_retry.http_request") as mock_head:
assert pf._uri_jobs(
@@ -646,7 +646,13 @@ def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
m.get_package.return_value = object() # already installed: warm and silent
with patch("esphome.net_retry.http_request"):
assert pf._uri_jobs(
m, [_FakeSpec(uri="git+https://x/tool.git#1.0", name="tool")], set()
m,
[
_FakeSpec(
uri="git+https://x/tool.git#1.0", name="tool", custom_name=True
)
],
set(),
) == ([], 0, [])
@@ -1469,59 +1475,6 @@ def test_prefetch_installs_cached_archives_without_downloads(
assert not (tmp_path / pf._SENTINEL_NAME).exists()
def test_prefetch_orders_clones_first_in_preinstall(tmp_path: Path) -> None:
"""The wiring, not just the helper: _prefetch hands _preinstall the
clones-first ordering of each group's entries."""
_write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
fake_platform = MagicMock()
fake_platform.packages = {}
config = _fake_config(tmp_path, {"platform": "fake/p@1"})
modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
archive = ("zip", _FakeSpec(uri="https://x/a.zip", name="zip", custom_name=True))
clone = ("repo", _FakeSpec(uri="git+https://x/repo.git", name="repo"))
with (
patch.dict("sys.modules", modules),
patch.object(
pf, "_registry_jobs", side_effect=[([], 0, [archive]), ([], 0, [])]
),
patch.object(pf, "_uri_jobs", side_effect=[([], 0, [clone]), ([], 0, [])]),
patch.object(pf, "_preinstall") as mock_install,
):
pf._prefetch(tmp_path, "testenv")
assert mock_install.call_count == 1
assert mock_install.call_args[0][1] == [clone, archive]
def test_prefetch_drops_derived_name_lib_clones(tmp_path: Path) -> None:
"""A lib_deps clone with a URI-derived name stays with pio run: its
destination dir comes from the cloned manifest, so two entries could
race one directory. Custom-named lib clones and the curated platform
batch keep the parallel pre-install."""
_write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
fake_platform = MagicMock()
fake_platform.packages = {}
config = _fake_config(tmp_path, {"platform": "fake/p@1"})
modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
tool = ("tool", _FakeSpec(uri="git+https://x/tool.git", name="tool"))
derived = ("lib", _FakeSpec(uri="git+https://x/lib.git", name="lib"))
custom = (
"mylib",
_FakeSpec(uri="git+https://x/mylib.git", name="mylib", custom_name=True),
)
with (
patch.dict("sys.modules", modules),
patch.object(pf, "_registry_jobs", return_value=([], 0, [])),
patch.object(
pf,
"_uri_jobs",
side_effect=[([], 0, [tool]), ([], 0, [derived, custom])],
),
patch.object(pf, "_preinstall") as mock_install,
):
pf._prefetch(tmp_path, "testenv")
assert [c.args[1] for c in mock_install.call_args_list] == [[tool], [custom]]
@pytest.mark.parametrize(
("platform_group", "lib_group", "expected"),
[
@@ -1704,12 +1657,9 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
assert installed == ["noise-c"]
def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None:
"""Each worker thread gets its own pre-built manager and installs
genuinely overlap (the barrier deadlocks a serial pool). The worker
count is pinned so a 1-CPU host cannot serialize the pool."""
barrier = threading.Barrier(2, timeout=5)
used: set = set()
def _wave_manager(tmp_path, on_install):
"""A minimal pio-manager stand-in for _preinstall pool tests;
``on_install(manager, spec)`` observes each _install call."""
class _WaveManager:
package_dir = str(tmp_path)
@@ -1740,70 +1690,59 @@ def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None:
return None
def _install(self, spec, skip_dependencies, compatibility=None) -> None:
used.add(id(self))
barrier.wait()
on_install(self, spec)
seed = _WaveManager(str(tmp_path))
with patch.object(pf, "get_usable_cpu_count", return_value=2):
pf._preinstall(
seed,
[
("a@1", _FakeSpec(name="a")),
("b@1", _FakeSpec(name="b")),
],
)
assert len(used) == 2
assert id(seed) not in used
return _WaveManager
def test_preinstall_clone_floor_widens_small_core_pool(tmp_path: Path) -> None:
"""Network-bound clones run wide even on a 1-CPU host: the barrier
deadlocks unless all four clone entries get concurrent workers."""
barrier = threading.Barrier(4, timeout=5)
used: set = set()
class _WaveManager:
package_dir = str(tmp_path)
compatibility = None
def __init__(self, package_dir, **kwargs) -> None:
assert package_dir == str(tmp_path)
def lock(self) -> None:
pass
def unlock(self) -> None:
pass
def memcache_reset(self) -> None:
pass
def get_tmp_dir(self) -> str:
return str(tmp_path)
def get_download_dir(self) -> str:
return str(tmp_path)
def get_package(self, spec):
return None
def get_pkg_dependencies(self, pkg):
return None
def _install(self, spec, skip_dependencies, compatibility=None) -> None:
used.add(id(self))
barrier.wait()
seed = _WaveManager(str(tmp_path))
with patch.object(pf, "get_usable_cpu_count", return_value=1):
pf._preinstall(
seed,
@pytest.mark.parametrize(
("cpu_count", "entries"),
[
(2, [("a@1", _FakeSpec(name="a")), ("b@1", _FakeSpec(name="b"))]),
# The clone floor: network-bound clones run wide on a 1-CPU host
(
1,
[
(f"r{i}", _FakeSpec(uri=f"git+https://x/r{i}.git", name=f"r{i}"))
for i in range(4)
],
),
],
ids=("cpu-sized", "clone-floor"),
)
def test_preinstall_pool_width(tmp_path: Path, cpu_count: int, entries: list) -> None:
"""Each worker gets its own pre-built manager and installs genuinely
overlap: the barrier deadlocks unless every entry runs concurrently."""
barrier = threading.Barrier(len(entries), timeout=5)
used: set = set()
def on_install(mgr, spec) -> None:
used.add(id(mgr))
barrier.wait()
cls = _wave_manager(tmp_path, on_install)
seed = cls(str(tmp_path))
with patch.object(pf, "get_usable_cpu_count", return_value=cpu_count):
pf._preinstall(seed, entries)
assert len(used) == len(entries)
assert id(seed) not in used
def test_preinstall_orders_clones_before_extractions(tmp_path: Path) -> None:
"""The pool owner sorts its own wave: with one worker, the clone
installs before the archive regardless of caller order."""
order: list[str] = []
cls = _wave_manager(tmp_path, lambda mgr, spec: order.append(spec.name))
seed = cls(str(tmp_path))
with patch.object(pf, "get_usable_cpu_count", return_value=1):
pf._preinstall(
seed,
[
("zip", _FakeSpec(uri="https://x/a.zip", name="zip")),
("repo", _FakeSpec(uri="git+https://x/repo.git", name="repo")),
],
)
assert len(used) == 4
assert order == ["repo", "zip"]
def test_sibling_manager_and_sigterm() -> None:
@@ -1957,6 +1896,10 @@ def test_platformio_private_api_contract() -> None:
# _is_vcs_spec_uri relies on bare .git URLs normalizing to git+, on
# both parse paths (raw string, and requirements= for platform tools)
assert PackageSpec("https://github.com/x/y.git#v1").uri.startswith("git+")
assert PackageSpec(
platform_tool = PackageSpec(
owner="o", name="tool-x", requirements="https://github.com/x/y.git"
).uri.startswith("git+")
)
assert platform_tool.uri.startswith("git+")
# A URL requirement re-parses as name=url, marking the name custom;
# this is what keeps platform tool clones in the parallel pre-install
assert platform_tool.has_custom_name()