From d6758377d14a8ab63781a4c035162166d4889a12 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 31 Aug 2026 22:25:58 -0400 Subject: [PATCH] [core] Clone git libraries in parallel in the library prefetch (#18836) --- esphome/git.py | 9 ++ esphome/platformio/library.py | 103 +++++++++++++++----- tests/unit_tests/test_git.py | 19 ++++ tests/unit_tests/test_platformio_library.py | 81 ++++++++++++++- 4 files changed, 184 insertions(+), 28 deletions(-) diff --git a/esphome/git.py b/esphome/git.py index 9815377f51..14145a639b 100644 --- a/esphome/git.py +++ b/esphome/git.py @@ -457,6 +457,15 @@ def _clone_complete_marker_path(repo_dir: Path) -> Path: return repo_dir / ".git" / _CLONE_COMPLETE_MARKER +def has_complete_clone( + url: str, ref: str | None, domain: str, subpath: Path | None = None +) -> bool: + """Lock-free probe for a complete clone; can go stale immediately, so + best-effort decisions only, never a substitute for ``clone_or_update``.""" + repo_dir = _repo_entry_dir(_cache_key(url, ref), domain, subpath) + return _clone_complete_marker_path(repo_dir).is_file() + + def _clear_clone_complete_marker(repo_dir: Path) -> None: """Best-effort removal of the completion marker. diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py index 0402311a9a..3ff60f8aaa 100644 --- a/esphome/platformio/library.py +++ b/esphome/platformio/library.py @@ -13,7 +13,7 @@ regardless of which toolchain consumes the result. """ from collections import deque -from collections.abc import Callable, Iterable +from collections.abc import Callable, Hashable, Iterable from dataclasses import dataclass, field from functools import partial import glob @@ -99,6 +99,17 @@ class Source: ) -> Path: raise NotImplementedError + def prefetch_key(self, dir_suffix: str) -> Hashable | None: + """Prefetch dedup identity; None = not prefetchable. Sources that + could write one cache dir must return equal keys (workers must never + share a dir); a coarser key only skips a prefetch.""" + return None + + def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool: + """Whether a completed fetch exists; only consulted when + ``prefetch_key()`` is not None, True is the safe default.""" + return True + def source_root(self, build_path: Path) -> Path: """Directory holding the library's own files (manifest + sources). @@ -127,6 +138,9 @@ class URLSource(Source): h.update(salt.encode()) return base_dir / h.hexdigest()[:8] / dir_suffix + def prefetch_key(self, dir_suffix: str) -> Hashable | None: + return self.url if self.size else None + def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool: """Whether a completed extraction already exists for this source.""" return ( @@ -177,14 +191,29 @@ class GitSource(Source): self.url = url self.ref = ref - def download( - self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = "" - ) -> Path: + @staticmethod + def _domain(salt: str, namespace: str) -> str: domain = DOMAIN if namespace: domain = f"{domain}/{namespace}" if salt: domain = f"{domain}/{salt}" + return domain + + def prefetch_key(self, dir_suffix: str) -> Hashable | None: + # The clone target dir is hash(url@ref)/ + return (self.url, self.ref, dir_suffix) + + def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool: + """Whether a completed clone already exists for this source.""" + return git.has_complete_clone( + self.url, self.ref, self._domain(salt, namespace), Path(dir_suffix) + ) + + def download( + self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = "" + ) -> Path: + domain = self._domain(salt, namespace) path, _ = git.clone_or_update( url=self.url, ref=self.ref, @@ -988,56 +1017,78 @@ def _fetch_source( ) +def _clone_source( + component: ConvertedLibrary, + salt: str, + namespace: str, + tracker: Callable[[int], None], +) -> None: + # No byte progress from git; one tick so a cancelled batch stops here + tracker(0) + component.source.download( + component.get_sanitized_name(), salt=salt, namespace=namespace + ) + + def _prefetch_wave( wave: list[tuple[str, ConvertedLibrary]], salt: str, namespace: str ) -> None: - """Best-effort parallel download of a wave's registry archives. + """Best-effort parallel fetch of a wave's registry archives and git clones. - The walk's own ``download()`` stays authoritative; duplicate URLs + The walk's own ``download()`` stays authoritative; duplicate sources prefetch once so two threads never share a cache directory. Archives whose size the registry did not report are left to the sequential loop, whose per-file bars don't interleave. A node a sibling in the - same wave supersedes has its archive fetched in vain (knowing better + same wave supersedes has its source fetched in vain (knowing better would need the manifests being downloaded). """ try: - components: list[ConvertedLibrary] = [] - seen: set[str] = set() + archives: list[ConvertedLibrary] = [] + clones: list[ConvertedLibrary] = [] + seen: set[Hashable] = set() for _key, component in wave: source = component.source - if not isinstance(source, URLSource) or not source.size: + name = component.get_sanitized_name() + dedup_key = source.prefetch_key(name) + if dedup_key is None or dedup_key in seen: continue - if source.url in seen: - continue - seen.add(source.url) + seen.add(dedup_key) try: - cached = source.is_cached( - component.get_sanitized_name(), salt=salt, namespace=namespace - ) + cached = source.is_cached(name, salt=salt, namespace=namespace) except OSError as err: # Best-effort, but visibly: a systematic probe failure makes - # every warm build re-download every archive + # every warm build re-fetch every source _LOGGER.warning("Cache probe for %s failed: %s", component.name, err) cached = False if cached: # A warm build must stay silent continue - components.append(component) - if not components: + (archives if isinstance(source, URLSource) else clones).append(component) + if not archives and not clones: return # Single-item waves (a dependency chain discovers one archive per # wave) go through the same runner: one download method, one bar - _LOGGER.info( - "Downloading %d library archive(s): %s", - len(components), - ", ".join(c.name for c in components), - ) + if archives: + _LOGGER.info( + "Downloading %d library archive(s): %s", + len(archives), + ", ".join(c.name for c in archives), + ) + if clones: + _LOGGER.info( + "Cloning %d library repo(s): %s", + len(clones), + ", ".join(c.name for c in clones), + ) failures = run_batch_downloads( "Downloading libraries", [ (c.name, c.source.size, partial(_fetch_source, c, salt, namespace)) - for c in components - ], + for c in archives + ] + # Size 0: clones share the worker pool without skewing the + # byte bar, whose total stays the archive sum + + [(c.name, 0, partial(_clone_source, c, salt, namespace)) for c in clones], ) # The sequential call below retries and raises the real error warn_prefetch_failures( diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py index e296d48a46..0f7e0339c9 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -714,6 +714,25 @@ def test_run_git_command_without_git_dir_raises_error( git.run_git_command(["git", "clone", "https://invalid.url/repo.git"]) +def test_has_complete_clone(tmp_path: Path) -> None: + """The lock-free probe tracks the completion marker, subpath included.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + subpath = Path("lib") + assert not git.has_complete_clone(url, "v1", "test_domain", subpath) + + repo_dir = _compute_repo_dir(url, "v1", "test_domain") / subpath + (repo_dir / ".git").mkdir(parents=True) + # A directory without the marker is an incomplete clone + assert not git.has_complete_clone(url, "v1", "test_domain", subpath) + + _mark_clone_complete(repo_dir) + assert git.has_complete_clone(url, "v1", "test_domain", subpath) + # The ref is part of the cache key + assert not git.has_complete_clone(url, "v2", "test_domain", subpath) + + def test_clone_or_update_with_never_refresh( tmp_path: Path, mock_run_git_command: Mock ) -> None: diff --git a/tests/unit_tests/test_platformio_library.py b/tests/unit_tests/test_platformio_library.py index 0a16b118fc..3bae39b3c1 100644 --- a/tests/unit_tests/test_platformio_library.py +++ b/tests/unit_tests/test_platformio_library.py @@ -638,7 +638,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel( setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: """Registry archives in one wave download concurrently, deduped by URL; - git/local sources and failures are left to the sequential call.""" + local sources and failures are left to the sequential call.""" calls: list[str] = [] def fake_download( @@ -658,7 +658,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel( # into the same cache directory) ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))), ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))), - ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))), + ("l", ConvertedLibrary("l", "*", LocalSource("/some/lib"))), ] lib._prefetch_wave(wave, "", "idf") assert sorted(calls) == [ @@ -670,6 +670,83 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel( assert "Prefetch of c failed (retrying sequentially)" in caplog.text +def test_prefetch_wave_clones_git_sources_in_parallel( + setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Git sources join the same prefetch batch as the archives, deduped by + clone target; a clone failure warns and is left to the sequential call.""" + caplog.set_level("INFO") + calls: list[str] = [] + + def fake_clone(self, dir_suffix, force=False, salt="", namespace=""): + calls.append(f"{self}/{dir_suffix}") + if "boom" in self.url: + raise RuntimeError("boom") + + monkeypatch.setattr(GitSource, "download", fake_clone) + wave = [ + ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))), + ("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))), + # Same url@ref and target dir must clone once + ("g2", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))), + ("h", ConvertedLibrary("h", "*", GitSource("https://x/boom.git", None))), + ] + monkeypatch.setattr( + URLSource, "download", lambda self, dir_suffix, progress=None, **kw: None + ) + lib._prefetch_wave(wave, "", "idf") + assert sorted(calls) == ["https://x/boom.git/h", "https://x/g.git#v1/g"] + assert "Cloning 2 library repo(s): g, h" in caplog.text + assert "Prefetch of h failed (retrying sequentially)" in caplog.text + + +def test_source_base_prefetch_defaults() -> None: + """The base Source is not prefetchable and reports cached (nothing to do).""" + source = Source() + assert source.prefetch_key("x") is None + assert source.is_cached("x") is True + + +def test_prefetch_wave_single_clone_uses_the_batch( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A wave with only git sources still clones through the batch runner.""" + caplog.set_level("INFO") + calls: list[str] = [] + monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: False) + monkeypatch.setattr( + GitSource, + "download", + lambda self, dir_suffix, force=False, salt="", namespace="": calls.append( + self.url + ), + ) + lib._prefetch_wave( + [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))], + "", + "idf", + ) + assert calls == ["https://x/g.git"] + assert "Cloning 1 library repo(s): g" in caplog.text + assert "Downloading" not in caplog.text + + +def test_prefetch_wave_warm_git_cache_is_silent( + setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """An already-complete clone is neither re-fetched nor announced.""" + caplog.set_level("INFO") + monkeypatch.setattr( + GitSource, + "download", + lambda self, dir_suffix, **kw: (_ for _ in ()).throw(AssertionError("cloned")), + ) + monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: True) + wave = [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))] + lib._prefetch_wave(wave, "", "idf") + assert "Cloning" not in caplog.text + + def test_prefetch_wave_unknown_size_left_to_sequential( setup_core, monkeypatch: pytest.MonkeyPatch ) -> None: