From 328e83ad641cb8b8cb45aa3f86e0435957fdf67a Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 31 Aug 2026 16:08:57 -0400 Subject: [PATCH] Simplify: one custom-name rule in _uri_jobs, ordering owned by the pool --- esphome/platformio/prefetch.py | 84 +++----- tests/unit_tests/test_platformio_prefetch.py | 201 +++++++------------ 2 files changed, 102 insertions(+), 183 deletions(-) diff --git a/esphome/platformio/prefetch.py b/esphome/platformio/prefetch.py index ce9e80ad18..465f895507 100644 --- a/esphome/platformio/prefetch.py +++ b/esphome/platformio/prefetch.py @@ -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 diff --git a/tests/unit_tests/test_platformio_prefetch.py b/tests/unit_tests/test_platformio_prefetch.py index f979dde660..98216d5e18 100644 --- a/tests/unit_tests/test_platformio_prefetch.py +++ b/tests/unit_tests/test_platformio_prefetch.py @@ -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()