This is an automated email from the ASF dual-hosted git repository. striker pushed a commit to branch striker/speculative-actions in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit df29a87fc37d1928d3c71d490b109211e451eddf Author: Sander Striker <[email protected]> AuthorDate: Sat Mar 21 00:03:57 2026 +0100 speculative-actions: Global instantiated_actions for cross-element resolution Replace per-element adapted_digests/action_outputs state with a global _instantiated_actions dict (base_action_hash -> adapted_action_digest) shared across all elements during priming. This fixes cross-element ACTION overlay resolution when dependency elements are adapted but not rebuilt (e.g. intermediate files like generated headers or .o files produced by a dependency's subactions). Key changes: - Generator: fix overlay sort order to SOURCE > ACTION > ARTIFACT. ACTION overlays resolve intermediate files (.o, generated headers) not present in artifacts, so they should be tried before ARTIFACT. - Instantiator: replace action_outputs parameter with global instantiated_actions dict. Add step-0 already-instantiated check. Add resolved_cache parameter to avoid re-resolving overlays across passes when an SA is deferred. - Priming queue: global _instantiated_actions dict and _primed_elements set as class-level shared state. Unresolvable ACTION overlays are removed from the in-memory SA proto when their source_element has finished priming. Cache deserialized spec_actions proto on the element so mutations and resolution caches survive across passes. Add dep-primed callback to trigger incremental priming when a dependency finishes priming (earlier than dep-cached). - Element: add _set_build_dep_primed_callback and _notify_build_deps_primed for the dep-primed notification mechanism. - Architecture doc: add 6 example scenarios, expand ReferencedSAs future optimization, describe global instantiated_actions approach. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> --- doc/source/arch_speculative_actions.rst | 291 +++++++++++++++++++-- .../queues/speculativecacheprimingqueue.py | 219 ++++++++++------ src/buildstream/_speculative_actions/generator.py | 23 +- .../_speculative_actions/instantiator.py | 105 +++++--- src/buildstream/element.py | 33 ++- .../test_pipeline_integration.py | 61 +++-- 6 files changed, 565 insertions(+), 167 deletions(-) diff --git a/doc/source/arch_speculative_actions.rst b/doc/source/arch_speculative_actions.rst index 2f5b672a6..c713f69a2 100644 --- a/doc/source/arch_speculative_actions.rst +++ b/doc/source/arch_speculative_actions.rst @@ -28,10 +28,11 @@ action cache instead of being executed from scratch. Overview -------- -A typical rebuild scenario: a developer modifies a leaf library. Every -downstream element needs rebuilding because its dependency changed. But -the downstream elements' own source code hasn't changed — only the -dependency artifacts are different. Speculative actions exploit this by: +A typical rebuild scenario: a developer modifies a low-level library +(e.g. a base SDK element). Every downstream element needs rebuilding +because its dependency changed. But the downstream elements' own source +code hasn't changed — only the dependency artifacts are different. +Speculative actions exploit this by: 1. **Recording** subactions from the previous build (via recc through the ``remote-apis-socket``) @@ -96,7 +97,7 @@ element with subaction digests: contains an intermediate file produced by a dependency's subaction (not in the artifact — those are ARTIFACT overlays), a cross-element ACTION overlay is created with ``source_element`` set to the - dependency name. + dependency name. Subsequently, a copy of the SpeculativeAction that is referenced by the ACTION overlay, is added to the list of speculative actions with its element field set to its originating element. We then evaluate that SA in the same way and copy in further SA's as needed, setting element to the dependency if it isn't set. We only need to walk the list of SAs of the dependency, which by definition is complete. This approach makes the SA list self-sufficient, at the cost o [...] 4. Stores the ``SpeculativeActions`` proto on the artifact, which is saved under both the strong and weak cache keys. @@ -123,13 +124,17 @@ Overlay Fallback Resolution When the same file digest appears in both a dependency's source tree and its artifact (e.g. a header file), both SOURCE and ARTIFACT overlays are generated. At instantiation time, they are tried in -priority order: SOURCE first, then ARTIFACT, then ACTION. - -This enables parallelism: if a dependency is rebuilding, its SOURCE -overlay can resolve as soon as the dependency's sources are fetched -(before its full build completes), while the ARTIFACT overlay serves -as a fallback if the sources are not available (dependency not -rebuilding this invocation — its artifact is already cached). +priority order: **SOURCE first, then ACTION, then ARTIFACT**. + +- **SOURCE** overlays enable the earliest resolution — as soon as + the element's sources are fetched, before any build completes. +- **ACTION** overlays resolve intermediate files (e.g. ``.o`` files) + that are produced by prior subactions but not present in artifacts. + They are tried before ARTIFACT because they provide a more direct + resolution path for intermediate files. +- **ARTIFACT** overlays serve as a fallback when sources are not + available (dependency not rebuilding this invocation — its artifact + is already cached). Overlay data availability at priming time: @@ -139,8 +144,8 @@ Overlay data availability at priming time: - If a referenced element **is rebuilding**: its old artifact is invalidated (new strong key), so ARTIFACT resolution returns None. SOURCE resolution may succeed if the Fetch queue has already run. - If neither resolves, the subaction is deferred until the dependency - completes. + If neither resolves, the subaction is deferred until the dependency's + sources become available or its artifact is cached. Action Instantiation @@ -149,20 +154,27 @@ Action Instantiation The ``SpeculativeActionInstantiator`` adapts stored actions for the current dependency versions: +0. **Already-instantiated check**: if the base action's hash is found + in the global ``instantiated_actions`` dict, returns the previously + adapted digest immediately (avoids redundant work when multiple + elements reference the same dependency subaction) 1. Fetches the base action from CAS 2. Resolves each overlay with fallback (first resolved wins per target - digest): + digest), in priority order **SOURCE > ACTION > ARTIFACT**: - **SOURCE** overlays: finds the current file digest in the element's source tree by path + - **ACTION** overlays: looks up the producing subaction's adapted + digest in the global ``instantiated_actions`` dict, then fetches + the ``ActionResult`` from the action cache to find the output + file's current digest. If the producing action was never + instantiated, the overlay is dropped gracefully. - **ARTIFACT** overlays: finds the current file digest in the dependency's artifact tree by path - - **ACTION** overlays: finds the current output file digest from the - producing subaction's ``ActionResult`` by path — looked up in - ``action_outputs`` (intra-element) or via the action cache - (cross-element) -3. Builds a digest replacement map (old hash → new digest) +3. Builds a digest replacement map (old hash → new digest), skipping + when old hash == new digest. If the replacement map is empty, the + SA is marked as done. 4. Recursively traverses the action's input tree, replacing file digests 5. Stores the modified action in CAS 6. If no digests changed, returns the base action digest (already cached) @@ -194,20 +206,31 @@ PENDING: 2. **Background priming**: pre-fetches CAS blobs, instantiates subactions whose overlays are resolvable from already-cached deps, submits them fire-and-forget (reads first stream response to - confirm acceptance, then drops the stream) + confirm acceptance, then drops the stream). Each instantiated + action is recorded in the global ``instantiated_actions`` dict. 3. **Per-dep callback**: as each dependency becomes cached, the callback triggers incremental priming — newly resolvable ARTIFACT and ACTION overlays are resolved and submitted 4. **Final pass** (element becomes buildable): all dependencies are built, all ``ActionResults`` are in the action cache. Remaining - ACTION overlays are resolved using adapted digests from earlier - submissions. Remaining subactions are submitted fire-and-forget. + ACTION overlays are resolved via the global ``instantiated_actions`` + dict. Remaining subactions are submitted fire-and-forget. 5. Element proceeds to BuildQueue with all actions primed Unchanged actions (instantiated digest equals base digest) skip submission — they are already in the action cache from the previous build. +**Global instantiated_actions**: A shared dict +(``base_action_hash → adapted_action_digest``) accessible to all +elements during priming. When element A instantiates a subaction, +the mapping is immediately visible to element B's priming. This +enables cross-element ACTION overlay resolution — element B can look +up element A's adapted subaction digest to find intermediate files +(e.g. generated headers) that aren't in artifacts. The dict is +protected by a threading lock for write access; reads are safe under +the GIL. + **Build Queue**: Builds elements as usual. When recc runs a compile or link command, it checks the action cache first. If priming succeeded, the adapted action is already cached → **action cache hit**. @@ -217,6 +240,171 @@ the build queue. Generates overlays from newly recorded subactions and stores them for future priming. +Example Scenarios +----------------- + +The following scenarios illustrate how speculative actions behave across +different dependency change patterns. In each case, "unchanged" means +the element's own sources did not change (its weak key is stable), so +its stored SA is available for priming. + + +Single dependency change +~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + base (sources CHANGED) → liba (unchanged) → app (unchanged) + +The most common CI scenario: a low-level element is modified, all +downstream elements need rebuilding. + +- **base**: weak key changed → no SA available → builds from scratch. +- **liba**: weak key unchanged → SA available. + + - SOURCE overlays (liba's own ``.c`` files): resolve immediately. + - ARTIFACT overlays (base's headers): deferred until base builds. + - When base finishes → per-dep callback fires → ARTIFACT overlays + resolve → liba's compile actions are submitted fire-and-forget. + - ACTION overlays (intra-element, e.g. ``ar`` consuming ``.o`` files + from compile): resolve sequentially from ``instantiated_actions`` + once the compile actions complete. + +- **app**: same pattern — waits for liba, then resolves and submits. + +Result: every downstream element gets full cache hits on all subactions. + + +Cross-element intermediate files +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + codegen (unchanged) → liba (unchanged) + | | + generates gen.h compiles with gen.h + +codegen's build produces ``gen.h`` as a subaction output. liba's +compile subaction uses ``gen.h`` as an input, tracked by a cross-element +ACTION overlay. + +- **codegen**: weak key unchanged → SA available → primed. Its compile + action is recorded in ``instantiated_actions``. +- **liba**: ACTION overlay for ``gen.h`` looks up codegen's subaction + in ``instantiated_actions`` → found → fetches ``ActionResult`` from + AC → resolves ``gen.h``'s adapted digest → submitted. + +Result: the global ``instantiated_actions`` dict enables cross-element +resolution of intermediate files. Without the global dict (per-element +state only), liba would not see codegen's adapted digest. + + +Intra-element action chains (compile → archive) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + base (sources CHANGED) → liba (unchanged) + compile: base.h + liba.c → liba.o + archive: ar rcs libliba.a liba.o + +liba's archive action depends on ``.o`` files produced by liba's own +compile actions. These ``.o`` files are intermediate — they are not +installed in artifacts. + +- **liba**: priming processes subactions in order. + + 1. Compile action: ARTIFACT overlay for ``base.h`` deferred until + base builds. When base finishes → resolves → submitted → + recorded in ``instantiated_actions``. + 2. Archive action: ACTION overlay for ``liba.o`` looks up compile's + hash in ``instantiated_actions`` → found → fetches + ``ActionResult`` → resolves ``liba.o`` → submitted. + +Result: the full compile → archive chain fires as soon as base completes. +Downstream elements that depend on ``libliba.a`` via ARTIFACT overlays +resolve once liba's artifact is cached. + + +Changed element breaks the action chain +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + codegen (sources CHANGED) → liba (unchanged) → app (unchanged) + | | + generates gen.h compiles with gen.h + +When codegen's sources change, its weak key changes, so its SA is +unavailable and codegen builds from scratch. Its subactions are never +primed and do not appear in ``instantiated_actions``. + +- **liba**: ACTION overlay for ``gen.h`` references codegen's subaction + → ``instantiated_actions.get(...)`` returns None → **overlay dropped**. + If ``gen.h`` is also installed in codegen's artifact, an ARTIFACT + overlay exists as fallback → resolves once codegen finishes building. + If ``gen.h`` is truly intermediate (not in the artifact), that + specific compile action cannot be adapted and falls back to full + execution during liba's build. + +- **liba finishes priming**: its adapted actions are recorded in + ``instantiated_actions``. From this point, all downstream elements + (app, etc.) can resolve ACTION overlays referencing liba's subactions. + +Result: one level of delay (codegen must build before the chain +resumes), but the chain propagates from liba onward. See +:ref:`referenced_speculative_actions` for a future optimization that +would eliminate this delay. + + +Multiple source changes in a chain +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + base (sources CHANGED) → liba (sources CHANGED) → app (unchanged) + +Both base and liba have changed sources, so both have changed weak keys +and no SAs available. Both build from scratch. + +- **app**: weak key unchanged → SA available. + + - ARTIFACT overlays for liba: deferred until liba builds → resolve. + - ACTION overlays for liba's subactions: liba was never primed → + ``instantiated_actions`` has no entries for liba's subactions → + **overlays dropped**. App's compile actions that depend on liba's + intermediate files (e.g. ``.o`` files not in the artifact) cannot + be adapted. + +Result: app gets cache hits for subactions that only depend on +artifacts (the common case), but misses on subactions that depend on +intermediate files from liba. This is a graceful degradation — those +subactions execute normally during app's build. + + +Dependency adapted but sources unchanged +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:: + + base (deps CHANGED, not sources) → liba (unchanged) → app (unchanged) + +base's own sources didn't change (weak key stable), but one of base's +dependencies changed, so base has a different strong key. + +- **base**: weak key unchanged → SA available → primed. Adapted + actions recorded in ``instantiated_actions``. +- **liba**: ACTION overlays for base's subactions → found in + ``instantiated_actions`` (populated by base's priming) → resolve. +- **app**: similarly resolves via ``instantiated_actions``. + +Result: the global dict ensures that base's adapted digests propagate +to all downstream elements, even though base's artifact hasn't changed +content-wise. Without the global dict, liba would fail to look up +base's adapted digests. + + + Scaling Considerations ---------------------- @@ -256,14 +444,59 @@ build configuration, so this only happens when the element itself changed — in which case the SA is correctly invalidated. +.. _referenced_speculative_actions: + Future Optimizations -------------------- -1. **Topological prioritization**: Prime elements in build order (leaves - first) to maximize the chance priming completes before building starts. +1. **ReferencedSpeculativeActions**: Store + ``repeated ReferencedSpeculativeActions`` on the SA proto — pointers + (``element_name``, ``sa_digest``) to dependency elements' SAs. This + enables a downstream element to instantiate a dependency's SA even + when the dependency's weak key changed (its sources changed). + + Consider this scenario:: + + codegen (sources CHANGED) → liba (unchanged) + | | + generates gen.h compiles with gen.h + + Currently, codegen's weak key changes, so its SA is unavailable. + liba's ACTION overlay for ``gen.h`` is dropped because codegen was + never primed and ``instantiated_actions`` has no entry for codegen's + subaction. liba must wait for codegen to build before the ARTIFACT + fallback (if ``gen.h`` is installed) or full execution (if ``gen.h`` + is truly intermediate) can proceed. + + With ReferencedSAs, liba's artifact would store a reference to + codegen's SA from the previous build. During priming, liba could + retrieve codegen's SA, instantiate codegen's subactions (adapting + them with codegen's new sources), and populate + ``instantiated_actions`` with codegen's adapted digests. The ACTION + overlay for ``gen.h`` would then resolve immediately, eliminating + the one-level delay. + + The benefit is most pronounced when a low-level element with + generated intermediate files (headers, protocol buffer outputs, + code-generated sources) changes frequently and has many downstream + dependents. The cost is additional complexity in SA storage and + retrieval, plus the overhead of instantiating dependency SAs during + priming. Whether this trade-off is worthwhile depends on real-world + profiling of rebuild patterns. + +2. **Topological prioritization**: Prime elements in build order + (dependencies first) to maximize the chance priming completes before + building starts. + +3. **Selective priming**: Skip cheap actions (fast link steps), prioritize + expensive ones (long compilations). Only skip when it doesn't break + SA chains. + +4. **Batch FetchTree**: Collect all input root digests and fetch in + parallel or in a single batch. -2. **Selective priming**: Skip cheap actions (fast link steps), prioritize - expensive ones (long compilations). +5. **Storage**: Store SAs more efficiently so that they can be pulled + down efficiently. -3. **Batch FetchTree**: Collect all input root digests and fetch in - parallel or in a single batch. +6. **Generation**: Find a way to make the output tree to input tree + matching more efficient. diff --git a/src/buildstream/_scheduler/queues/speculativecacheprimingqueue.py b/src/buildstream/_scheduler/queues/speculativecacheprimingqueue.py index 107b3e2a9..2cca3c6ca 100644 --- a/src/buildstream/_scheduler/queues/speculativecacheprimingqueue.py +++ b/src/buildstream/_scheduler/queues/speculativecacheprimingqueue.py @@ -31,8 +31,16 @@ element is released to the BuildQueue. By then, most adapted actions are already in the action cache — recc gets cache hits. Elements without stored SpeculativeActions skip this queue entirely. + +Cross-element ACTION overlay resolution uses a global +``_instantiated_actions`` dict (base_action_hash → adapted_action_digest) +shared across all elements. When element A's priming instantiates a +subaction, the mapping is immediately visible to element B's priming, +enabling cross-element intermediate file resolution. """ +import threading + # Local imports from . import Queue, QueueStatus from ..jobs import JobStatus @@ -45,22 +53,40 @@ class SpeculativeCachePrimingQueue(Queue): complete_name = "Cache primed" resources = [ResourceType.UPLOAD] + # Global shared state: maps base_action_hash -> adapted_action_digest + # Populated by all elements during priming, enabling cross-element + # ACTION overlay resolution. + _instantiated_actions = {} + _instantiated_actions_lock = threading.Lock() + + # Elements whose priming has completed (all passes done). + # Used to determine if an ACTION overlay's producing element has + # finished priming — if so and the action isn't in _instantiated_actions, + # the overlay can be permanently dropped from the SA proto. + _primed_elements = set() + def get_process_func(self): # Runs when element is READY (buildable) — final priming pass return SpeculativeCachePrimingQueue._final_prime_pass def status(self, element): if element._cached(): + # Already cached — no priming needed. Record as primed so + # downstream elements know this element's actions won't appear + # in _instantiated_actions. + SpeculativeCachePrimingQueue._primed_elements.add(element.name) return QueueStatus.SKIP weak_key = element._get_weak_cache_key() if not weak_key: + SpeculativeCachePrimingQueue._primed_elements.add(element.name) return QueueStatus.SKIP context = element._get_context() artifactcache = context.artifactcache spec_actions = artifactcache.lookup_speculative_actions_by_weak_key(element, weak_key) if not spec_actions or not spec_actions.actions: + SpeculativeCachePrimingQueue._primed_elements.add(element.name) return QueueStatus.SKIP # Has SAs. If not buildable, enter PENDING — background @@ -75,6 +101,11 @@ class SpeculativeCachePrimingQueue(Queue): # Register per-dep callback for incremental overlay resolution element._set_build_dep_cached_callback(self._on_dep_cached) + # Register per-dep callback for when a dependency finishes + # priming — its adapted actions are now in _instantiated_actions + # and downstream elements can resolve cross-element ACTION overlays + element._set_build_dep_primed_callback(self._on_dep_primed) + # Also register buildable callback so we get re-enqueued # when the element becomes fully buildable element._set_buildable_callback(self._enqueue_element) @@ -100,6 +131,17 @@ class SpeculativeCachePrimingQueue(Queue): self._launch_incremental_prime, element, dep ) + def _on_dep_primed(self, element, dep): + """Called each time a build dependency finishes priming. + + The dep's adapted action digests are now in _instantiated_actions. + Launches incremental priming to resolve cross-element ACTION + overlays that reference the dep's subactions. + """ + self._scheduler.loop.call_soon( + self._launch_incremental_prime, element, dep + ) + def _launch_incremental_prime(self, element, dep): self._scheduler.loop.run_in_executor( None, SpeculativeCachePrimingQueue._incremental_prime, element, dep @@ -116,11 +158,22 @@ class SpeculativeCachePrimingQueue(Queue): else: element.info(f"Primed {primed}/{total} actions") - # Clear priming state and per-dep callback + # Record element as primed so other elements can determine + # whether ACTION overlay producers have finished. + with SpeculativeCachePrimingQueue._instantiated_actions_lock: + SpeculativeCachePrimingQueue._primed_elements.add(element.name) + + # Notify reverse build deps that this element finished priming — + # its adapted actions are now in _instantiated_actions and + # downstream elements can resolve cross-element ACTION overlays. + element._notify_build_deps_primed() + + # Clear priming state and per-dep callbacks element._set_build_dep_cached_callback(None) + element._set_build_dep_primed_callback(None) element.__priming_submitted = None - element.__priming_action_outputs = None - element.__priming_adapted_digests = None + element.__priming_spec_actions = None + element.__priming_resolved = None # ----------------------------------------------------------------- # Background priming (runs in thread pool while element is PENDING) @@ -151,23 +204,43 @@ class SpeculativeCachePrimingQueue(Queue): Iterates over all subactions, skipping already-submitted ones. For each remaining subaction, attempts to resolve all overlays. If resolvable, instantiates and submits fire-and-forget. + + ACTION overlay resolution uses the global _instantiated_actions + dict. For each ACTION overlay: + - If the producing action is in _instantiated_actions → check + AC for the ActionResult; if not yet available, defer the SA + - If the producing action is NOT in _instantiated_actions AND + its source_element has finished priming → the producing + action will never appear; remove the overlay from the proto + - If the producing action is NOT in _instantiated_actions AND + its source_element has NOT finished priming → skip for now, + it may appear on a later pass + + Dropped overlays are removed directly from the in-memory SA + proto (which is discarded after the build). """ from ..._speculative_actions.instantiator import SpeculativeActionInstantiator from ..._protos.buildstream.v2 import speculative_actions_pb2 + from ..._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 context = element._get_context() cas = context.get_cascache() artifactcache = context.artifactcache - weak_key = element._get_weak_cache_key() - spec_actions = artifactcache.lookup_speculative_actions_by_weak_key(element, weak_key) - if not spec_actions or not spec_actions.actions: - return - - # Recover or initialize state + # Use the cached spec_actions proto if available (mutations and + # _resolved_cache attributes must survive across passes). Only + # deserialize from CAS on the first pass. + spec_actions = getattr(element, "_SpeculativeCachePrimingQueue__priming_spec_actions", None) + if spec_actions is None: + weak_key = element._get_weak_cache_key() + spec_actions = artifactcache.lookup_speculative_actions_by_weak_key(element, weak_key) + if not spec_actions or not spec_actions.actions: + return + + # Recover or initialize per-element state submitted = getattr(element, "_SpeculativeCachePrimingQueue__priming_submitted", None) or set() - action_outputs = getattr(element, "_SpeculativeCachePrimingQueue__priming_action_outputs", None) or {} - adapted_digests = getattr(element, "_SpeculativeCachePrimingQueue__priming_adapted_digests", None) or {} + # Per-SA resolution caches: {base_action_hash -> {target_hash -> new_digest}} + resolved_caches = getattr(element, "_SpeculativeCachePrimingQueue__priming_resolved", None) or {} # Pre-fetch CAS blobs only on first pass if not submitted: @@ -191,46 +264,78 @@ class SpeculativeCachePrimingQueue(Queue): ac_service = casd.get_ac_service() instantiator = SpeculativeActionInstantiator(cas, artifactcache, ac_service=ac_service) + # References to global state (reads are GIL-safe) + instantiated_actions = SpeculativeCachePrimingQueue._instantiated_actions + primed_elements = SpeculativeCachePrimingQueue._primed_elements + for spec_action in spec_actions.actions: base_hash = spec_action.base_action_digest.hash if base_hash in submitted: continue - # Check overlay resolvability + # Check ACTION overlay resolvability against global state, + # removing overlays that will never resolve. resolvable = True - for overlay in spec_action.overlays: - if overlay.type == speculative_actions_pb2.SpeculativeActions.Overlay.ACTION: - key = (overlay.source_action_digest.hash, overlay.source_path) - if key not in action_outputs and ac_service: - # The AC stores results under the adapted digest - # (what was actually executed), but overlays reference - # the base digest. Look up with adapted, store under base. - base_key_hash = overlay.source_action_digest.hash - lookup_digest = adapted_digests.get( - base_key_hash, - overlay.source_action_digest, - ) - SpeculativeCachePrimingQueue._fetch_action_outputs_keyed( - ac_service, lookup_digest, base_key_hash, - action_outputs, - ) - if key not in action_outputs: - resolvable = False - break + to_remove = [] + for i, overlay in enumerate(spec_action.overlays): + if overlay.type != speculative_actions_pb2.SpeculativeActions.Overlay.ACTION: + continue + + source_hash = overlay.source_action_digest.hash + adapted = instantiated_actions.get(source_hash) + + if adapted is not None: + # Producing action was instantiated — check if + # result is in AC + if ac_service: + try: + request = remote_execution_pb2.GetActionResultRequest( + action_digest=adapted, + ) + action_result = ac_service.GetActionResult(request) + if not action_result: + # Submitted but not yet complete — defer + resolvable = False + break + except Exception: + resolvable = False + break + else: + # Not in instantiated_actions — check if the + # producing element has finished priming + source_elem = overlay.source_element or element.name + if source_elem in primed_elements: + # Element finished priming without instantiating + # this action — it will never appear. Mark for + # removal from the proto. + to_remove.append(i) + # else: source element not yet primed, skip for now + + # Remove dropped overlays from the proto (reverse order to + # preserve indices) + for i in reversed(to_remove): + del spec_action.overlays[i] if not resolvable: continue try: + # Get or create per-SA resolution cache + sa_cache = resolved_caches.setdefault(base_hash, {}) action_digest = instantiator.instantiate_action( spec_action, element, element_lookup, - action_outputs=action_outputs, + instantiated_actions=instantiated_actions, + resolved_cache=sa_cache, ) if not action_digest: continue + # Record in global state (write-locked) + with SpeculativeCachePrimingQueue._instantiated_actions_lock: + instantiated_actions[base_hash] = action_digest + # Skip unchanged actions (already in AC from previous build) if action_digest.hash == base_hash: submitted.add(base_hash) @@ -244,16 +349,15 @@ class SpeculativeCachePrimingQueue(Queue): f"(base {base_hash[:8]})" ) submitted.add(base_hash) - adapted_digests[base_hash] = action_digest except Exception as e: element.warn(f"Failed to prime action: {e}") continue - # Store state for next pass + # Store per-element state for next pass element.__priming_submitted = submitted - element.__priming_action_outputs = action_outputs - element.__priming_adapted_digests = adapted_digests + element.__priming_spec_actions = spec_actions + element.__priming_resolved = resolved_caches # ----------------------------------------------------------------- # Final priming pass (runs as a job when element becomes READY) @@ -266,18 +370,16 @@ class SpeculativeCachePrimingQueue(Queue): All deps are built, so all ActionResults are in AC. Resolve any remaining ACTION overlays and submit. """ - # Run the same logic — it will pick up where background left off + # Run the same logic — it will pick up where background left off. + # By now all deps are built, so _primed_elements contains all + # producing elements. Any ACTION overlay whose source_element + # is in _primed_elements but whose action is not in + # _instantiated_actions will be removed from the proto. SpeculativeCachePrimingQueue._do_prime_pass(element) # Count results submitted = getattr(element, "_SpeculativeCachePrimingQueue__priming_submitted", None) or set() - - from ..._protos.buildstream.v2 import speculative_actions_pb2 - - context = element._get_context() - artifactcache = context.artifactcache - weak_key = element._get_weak_cache_key() - spec_actions = artifactcache.lookup_speculative_actions_by_weak_key(element, weak_key) + spec_actions = getattr(element, "_SpeculativeCachePrimingQueue__priming_spec_actions", None) if not spec_actions: return (0, 0, 0) @@ -320,35 +422,6 @@ class SpeculativeCachePrimingQueue(Queue): except Exception: pass - @staticmethod - def _fetch_action_outputs(ac_service, action_digest, action_outputs): - """Fetch ActionResult from action cache and record output file digests.""" - SpeculativeCachePrimingQueue._fetch_action_outputs_keyed( - ac_service, action_digest, action_digest.hash, action_outputs - ) - - @staticmethod - def _fetch_action_outputs_keyed(ac_service, action_digest, key_hash, action_outputs): - """Fetch ActionResult and store outputs keyed by a specified hash. - - When resolving ACTION overlays, the overlay references the base - action digest but the AC stores the result under the adapted - digest. This method allows looking up with one digest but - storing results under a different key hash. - """ - try: - from ..._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 - - request = remote_execution_pb2.GetActionResultRequest( - action_digest=action_digest, - ) - action_result = ac_service.GetActionResult(request) - if action_result: - for output_file in action_result.output_files: - action_outputs[(key_hash, output_file.path)] = output_file.digest - except Exception: - pass - @staticmethod def _submit_action_async(exec_service, action_digest, element): """Submit an Execute request fire-and-forget style. diff --git a/src/buildstream/_speculative_actions/generator.py b/src/buildstream/_speculative_actions/generator.py index caf6c9ad3..427e1cbde 100644 --- a/src/buildstream/_speculative_actions/generator.py +++ b/src/buildstream/_speculative_actions/generator.py @@ -22,7 +22,7 @@ Generates SpeculativeActions and artifact overlays after element builds. This module is responsible for: 1. Extracting subaction digests from ActionResult 2. Traversing action input trees to find all file digests -3. Resolving digests to their source elements (SOURCE > ARTIFACT > ACTION priority) +3. Resolving digests to their source elements (SOURCE > ACTION > ARTIFACT priority) 4. Creating overlays for each digest 5. Generating artifact_overlays for the element's output files 6. Tracking inter-subaction output dependencies via ACTION overlays @@ -56,7 +56,7 @@ class SpeculativeActionsGenerator: self._artifactcache = artifactcache # Cache for digest.hash -> list of (element, path, type) lookups # Multiple entries per digest enable fallback resolution: - # SOURCE overlays are tried first, then ARTIFACT, then ACTION. + # SOURCE overlays are tried first, then ACTION, then ARTIFACT. self._digest_cache: Dict[str, list] = {} def generate_speculative_actions(self, element, subaction_digests, dependencies): @@ -128,7 +128,16 @@ class SpeculativeActionsGenerator: overlay.target_digest.size_bytes = digest_size spec_action.overlays.append(overlay) + # Sort overlays: SOURCE > ACTION > ARTIFACT + # This ensures the instantiator tries SOURCE first, then + # ACTION (intermediate files), then ARTIFACT as fallback. if spec_action: + type_priority = { + speculative_actions_pb2.SpeculativeActions.Overlay.SOURCE: 0, + speculative_actions_pb2.SpeculativeActions.Overlay.ACTION: 1, + speculative_actions_pb2.SpeculativeActions.Overlay.ARTIFACT: 2, + } + spec_action.overlays.sort(key=lambda o: type_priority.get(o.type, 99)) spec_actions.actions.append(spec_action) # Fetch this subaction's ActionResult and record its outputs @@ -209,7 +218,7 @@ class SpeculativeActionsGenerator: Build a cache mapping file digests to their source elements. Multiple entries per digest are stored to enable fallback - resolution at instantiation time (SOURCE > ARTIFACT > ACTION). + resolution at instantiation time (SOURCE > ACTION > ARTIFACT). Args: element: The element being processed @@ -355,7 +364,7 @@ class SpeculativeActionsGenerator: input_digests = self._extract_digests_from_action(action) # Resolve each digest to overlays (may produce multiple per digest - # for fallback resolution: SOURCE > ARTIFACT) + # for fallback resolution: SOURCE > ACTION > ARTIFACT) for digest in input_digests: overlays = self._resolve_digest_to_overlays(digest, element) spec_action.overlays.extend(overlays) @@ -448,10 +457,12 @@ class SpeculativeActionsGenerator: overlays.append(overlay) - # Sort: SOURCE first, then ARTIFACT — instantiator tries in order + # Sort: SOURCE first, then ARTIFACT — instantiator tries in order. + # ACTION overlays are added separately in generate_speculative_actions() + # and the final sort there establishes SOURCE > ACTION > ARTIFACT. type_priority = { speculative_actions_pb2.SpeculativeActions.Overlay.SOURCE: 0, - speculative_actions_pb2.SpeculativeActions.Overlay.ARTIFACT: 1, + speculative_actions_pb2.SpeculativeActions.Overlay.ARTIFACT: 2, } overlays.sort(key=lambda o: type_priority.get(o.type, 99)) diff --git a/src/buildstream/_speculative_actions/instantiator.py b/src/buildstream/_speculative_actions/instantiator.py index abd5b9bbd..4ec0412ed 100644 --- a/src/buildstream/_speculative_actions/instantiator.py +++ b/src/buildstream/_speculative_actions/instantiator.py @@ -20,10 +20,11 @@ SpeculativeActionInstantiator Instantiates SpeculativeActions by applying overlays. This module is responsible for: -1. Fetching base actions from CAS -2. Applying SOURCE and ARTIFACT overlays -3. Replacing file digests in action input trees -4. Storing modified actions back to CAS +1. Checking if the action is already instantiated (via global instantiated_actions) +2. Fetching base actions from CAS +3. Applying SOURCE, ACTION, and ARTIFACT overlays (in that priority order) +4. Replacing file digests in action input trees +5. Storing modified actions back to CAS """ @@ -51,20 +52,34 @@ class SpeculativeActionInstantiator: self._artifactcache = artifactcache self._ac_service = ac_service - def instantiate_action(self, spec_action, element, element_lookup, action_outputs=None): + def instantiate_action(self, spec_action, element, element_lookup, + instantiated_actions=None, resolved_cache=None): """ Instantiate a SpeculativeAction by applying overlays. + Previously resolved overlays can be passed in via resolved_cache + to avoid re-resolving overlays that succeeded on a prior pass but + whose SA couldn't be fully instantiated yet (e.g. an ACTION + overlay was deferred). + Args: - spec_action: SpeculativeAction proto + spec_action: SpeculativeAction proto (may be mutated: overlays + removed by the priming queue) element: Element being primed element_lookup: Dict mapping element names to Element objects - action_outputs: Optional dict of (subaction_index_str, output_path) -> new_digest - for resolving ACTION overlays from prior subaction executions + instantiated_actions: Optional dict mapping base_action_hash -> adapted_action_digest + (global across all elements, populated by the priming queue) + resolved_cache: Optional dict of {target_digest_hash -> new_digest} + from prior passes, updated in-place with new resolutions Returns: Digest of instantiated action, or None if overlays cannot be applied """ + # Step 0: Check if already instantiated (e.g. by another element's priming) + base_hash = spec_action.base_action_digest.hash + if instantiated_actions is not None and base_hash in instantiated_actions: + return instantiated_actions[base_hash] + # Fetch the base action base_action = self._cas.fetch_action(spec_action.base_action_digest) if not base_action: @@ -80,9 +95,12 @@ class SpeculativeActionInstantiator: action = remote_execution_pb2.Action() action.CopyFrom(base_action) - # Track if we made any modifications - modified = False - digest_replacements = {} # old_hash -> new_digest + # Seed digest_replacements from the resolution cache (if provided). + # This avoids re-resolving overlays that succeeded on a prior + # pass but whose SA couldn't be fully instantiated yet. + if resolved_cache is None: + resolved_cache = {} + digest_replacements = dict(resolved_cache) skipped_count = 0 applied_count = 0 @@ -91,7 +109,8 @@ class SpeculativeActionInstantiator: # They are stored in priority order (SOURCE first); once a target # is resolved, subsequent overlays for it are skipped. for overlay in spec_action.overlays: - # Skip if this target was already resolved by a higher-priority overlay + # Skip if this target was already resolved (by a higher-priority + # overlay or from the resolution cache) if overlay.target_digest.hash in digest_replacements: continue @@ -101,13 +120,20 @@ class SpeculativeActionInstantiator: skipped_count += 1 continue - replacement = self._resolve_overlay(overlay, element, element_lookup, action_outputs=action_outputs) + replacement = self._resolve_overlay(overlay, element, element_lookup, instantiated_actions=instantiated_actions) if replacement: # replacement is (old_digest, new_digest) digest_replacements[replacement[0].hash] = replacement[1] - if replacement[0].hash != replacement[1].hash: - modified = True - applied_count += 1 + applied_count += 1 + + # Update the resolution cache in-place for the next pass + resolved_cache.update(digest_replacements) + + # Check if any replacements actually change a digest + modified = any( + old_hash != new_digest.hash + for old_hash, new_digest in digest_replacements.items() + ) # Log optimization results if skipped_count > 0: @@ -198,7 +224,7 @@ class SpeculativeActionInstantiator: return False - def _resolve_overlay(self, overlay, element, element_lookup, action_outputs=None): + def _resolve_overlay(self, overlay, element, element_lookup, instantiated_actions=None): """ Resolve an overlay to get current file digest. @@ -206,8 +232,7 @@ class SpeculativeActionInstantiator: overlay: Overlay proto element: Current element element_lookup: Dict mapping element names to Element objects - action_outputs: Optional dict of (subaction_index_str, output_path) -> new_digest - for resolving ACTION overlays + instantiated_actions: Optional dict mapping base_action_hash -> adapted_action_digest Returns: Tuple of (old_digest, new_digest) or None @@ -216,10 +241,10 @@ class SpeculativeActionInstantiator: if overlay.type == speculative_actions_pb2.SpeculativeActions.Overlay.SOURCE: return self._resolve_source_overlay(overlay, element, element_lookup) + elif overlay.type == speculative_actions_pb2.SpeculativeActions.Overlay.ACTION: + return self._resolve_action_overlay(overlay, instantiated_actions) elif overlay.type == speculative_actions_pb2.SpeculativeActions.Overlay.ARTIFACT: return self._resolve_artifact_overlay(overlay, element, element_lookup) - elif overlay.type == speculative_actions_pb2.SpeculativeActions.Overlay.ACTION: - return self._resolve_action_overlay(overlay, action_outputs) return None @@ -320,40 +345,42 @@ class SpeculativeActionInstantiator: return None - def _resolve_action_overlay(self, overlay, action_outputs): + def _resolve_action_overlay(self, overlay, instantiated_actions): """ - Resolve an ACTION overlay using outputs from prior subaction executions. + Resolve an ACTION overlay using the global instantiated_actions map. - For intra-element overlays (source_element == ""), uses the - action_outputs dict populated during sequential priming. + Looks up the producing subaction's adapted digest in + instantiated_actions, then fetches the ActionResult from the + action cache to find the output file's current digest. - For cross-element overlays (source_element set), falls back to - the action cache — the dependency's subaction may have been - executed during the dependency's own priming or build. + Works for both intra-element and cross-element ACTION overlays, + since instantiated_actions is global across all elements. Args: overlay: Overlay proto with type ACTION - action_outputs: Dict of (base_action_digest_hash, output_path) -> new_digest + instantiated_actions: Dict mapping base_action_hash -> adapted_action_digest Returns: Tuple of (old_digest, new_digest) or None """ - key = (overlay.source_action_digest.hash, overlay.source_path) + source_hash = overlay.source_action_digest.hash - # Check action_outputs first (intra-element, populated during priming) - if action_outputs: - new_digest = action_outputs.get(key) - if new_digest: - return (overlay.target_digest, new_digest) + # Step 1: Look up the adapted digest for the producing action + adapted_digest = None + if instantiated_actions: + adapted_digest = instantiated_actions.get(source_hash) + + if adapted_digest is None: + # Producing action was never instantiated — drop this overlay + return None - # For cross-element: look up the producing subaction's ActionResult - # from the action cache directly - if overlay.source_element and self._ac_service: + # Step 2: Fetch ActionResult using the adapted digest from AC + if self._ac_service: try: from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 request = remote_execution_pb2.GetActionResultRequest( - action_digest=overlay.source_action_digest, + action_digest=adapted_digest, ) action_result = self._ac_service.GetActionResult(request) if action_result: diff --git a/src/buildstream/element.py b/src/buildstream/element.py index b9169076e..972709121 100644 --- a/src/buildstream/element.py +++ b/src/buildstream/element.py @@ -300,7 +300,8 @@ class Element(Plugin): self.__required_callback = None # Callback to Queues self.__can_query_cache_callback = None # Callback to PullQueue/FetchQueue self.__buildable_callback = None # Callback to BuildQueue - self.__build_dep_cached_callback = None # Callback to PrimingQueue (per-dep) + self.__build_dep_cached_callback = None # Callback to PrimingQueue (per-dep, on cached) + self.__build_dep_primed_callback = None # Callback to PrimingQueue (per-dep, on primed) self.__resolved_initial_state = False # Whether the initial state of the Element has been resolved @@ -2507,6 +2508,36 @@ class Element(Plugin): def _set_build_dep_cached_callback(self, callback): self.__build_dep_cached_callback = callback + # _set_build_dep_primed_callback() + # + # Set a callback invoked each time a build dependency finishes + # priming (its speculative actions have been instantiated and + # submitted to the action cache). + # + # Unlike _set_build_dep_cached_callback (which fires when a dep + # becomes cached after building), this fires earlier — when a + # dep's priming completes. This enables downstream elements to + # re-evaluate cross-element ACTION overlays sooner, since the + # dep's adapted action digests are now in instantiated_actions. + # + # Args: + # callback (callable) - Called with (element, dep) where dep is + # the just-primed dependency + # + def _set_build_dep_primed_callback(self, callback): + self.__build_dep_primed_callback = callback + + # _notify_build_deps_primed() + # + # Notify reverse build dependencies that this element has finished + # priming. Called by SpeculativeCachePrimingQueue.done() after an + # element's priming completes. + # + def _notify_build_deps_primed(self): + for rdep in self.__reverse_build_deps: + if rdep.__build_dep_primed_callback is not None: + rdep.__build_dep_primed_callback(rdep, self) + # _set_depth() # # Set the depth of the Element. diff --git a/tests/speculative_actions/test_pipeline_integration.py b/tests/speculative_actions/test_pipeline_integration.py index bf1b72e76..afb0c8281 100644 --- a/tests/speculative_actions/test_pipeline_integration.py +++ b/tests/speculative_actions/test_pipeline_integration.py @@ -896,12 +896,13 @@ class TestActionOverlays: for overlay in sub1_sa[0].overlays: assert overlay.type != speculative_actions_pb2.SpeculativeActions.Overlay.ACTION - def test_action_overlay_instantiation_with_action_outputs(self, tmp_path): + def test_action_overlay_instantiation_with_instantiated_actions(self, tmp_path): """ - Instantiate an ACTION overlay using action_outputs from a prior - subaction's execution result. + Instantiate an ACTION overlay using instantiated_actions from a prior + subaction's priming, with the ActionResult in the AC. """ cas = FakeCAS() + ac_service = FakeACService() artifactcache = FakeArtifactCache(cas, str(tmp_path)) # Build an action whose input tree has main.o @@ -911,26 +912,36 @@ class TestActionOverlays: link_action_digest = _build_action(cas, link_input) # Create a SpeculativeAction with an ACTION overlay - # Use a fake compile action digest as the producing action - compile_action_digest = _make_digest(b'fake-compile-action') + # Use a fake compile action digest as the producing action's base + compile_base_digest = _make_digest(b'fake-compile-action-base') + # The adapted digest (what was actually executed after priming) + compile_adapted_digest = _make_digest(b'fake-compile-action-adapted') spec_action = speculative_actions_pb2.SpeculativeActions.SpeculativeAction() spec_action.base_action_digest.CopyFrom(link_action_digest) overlay = spec_action.overlays.add() overlay.type = speculative_actions_pb2.SpeculativeActions.Overlay.ACTION - overlay.source_action_digest.CopyFrom(compile_action_digest) + overlay.source_action_digest.CopyFrom(compile_base_digest) overlay.source_path = "main.o" overlay.target_digest.CopyFrom(old_main_o_digest) - # Simulate: compile subaction executed and produced new main.o + # Simulate: compile subaction was instantiated and executed, + # producing new main.o — result is in AC under adapted digest new_main_o = b'new-object-code' new_main_o_digest = _make_digest(new_main_o) - action_outputs = {(compile_action_digest.hash, "main.o"): new_main_o_digest} + compile_result = remote_execution_pb2.ActionResult() + out = compile_result.output_files.add() + out.path = "main.o" + out.digest.CopyFrom(new_main_o_digest) + ac_service.store_action_result(compile_adapted_digest, compile_result) + + # Global instantiated_actions: base -> adapted + instantiated_actions = {compile_base_digest.hash: compile_adapted_digest} element = FakeElement("app.bst") - instantiator = SpeculativeActionInstantiator(cas, artifactcache) + instantiator = SpeculativeActionInstantiator(cas, artifactcache, ac_service=ac_service) result_digest = instantiator.instantiate_action( spec_action, element, {}, - action_outputs=action_outputs, + instantiated_actions=instantiated_actions, ) assert result_digest is not None @@ -945,7 +956,7 @@ class TestActionOverlays: def test_action_overlay_full_roundtrip(self, tmp_path): """ Full roundtrip: generate ACTION overlays, store, retrieve, - instantiate with action_outputs from sequential priming execution. + instantiate with instantiated_actions from sequential priming execution. Models the compile→link scenario where dep.h changes, causing main.o to change, which should be chained to the link action. @@ -1022,26 +1033,34 @@ class TestActionOverlays: # --- Sequential instantiation (simulating priming queue) --- element_lookup = {"dep.bst": dep_element_v2} - instantiator = SpeculativeActionInstantiator(cas, artifactcache) - action_outputs = {} + instantiator = SpeculativeActionInstantiator(cas, artifactcache, ac_service=ac_service) + instantiated_actions = {} # 1) Instantiate compile action (SOURCE + ARTIFACT overlays) compile_result_digest = instantiator.instantiate_action( retrieved.actions[0], element_v2, element_lookup, - action_outputs=action_outputs, + instantiated_actions=instantiated_actions, ) assert compile_result_digest is not None + # Record in instantiated_actions (as the priming queue would) + instantiated_actions[compile_digest.hash] = compile_result_digest + # Simulate compile execution producing new main.o - # Key by the compile's base_action_digest hash (as the priming queue would) + # Store the result in the AC under the adapted digest main_o_v2 = b'main-object-v2' main_o_v2_digest = _make_digest(main_o_v2) - action_outputs[(compile_digest.hash, "main.o")] = main_o_v2_digest + compile_v2_result = remote_execution_pb2.ActionResult() + out = compile_v2_result.output_files.add() + out.path = "main.o" + out.digest.CopyFrom(main_o_v2_digest) + ac_service.store_action_result(compile_result_digest, compile_v2_result) - # 2) Instantiate link action (ACTION overlay resolves from action_outputs) + # 2) Instantiate link action (ACTION overlay resolves via + # instantiated_actions + AC lookup) link_result_digest = instantiator.instantiate_action( retrieved.actions[1], element_v2, element_lookup, - action_outputs=action_outputs, + instantiated_actions=instantiated_actions, ) assert link_result_digest is not None assert link_result_digest.hash != link_digest.hash @@ -1300,9 +1319,13 @@ class TestCrossElementActionOverlays: instantiator = SpeculativeActionInstantiator( cas, artifactcache, ac_service=ac_service ) + # Cross-element: the dep's codegen action was instantiated and + # its result is in AC. instantiated_actions maps base -> adapted + # (in this case, same digest since we stored under dep_codegen_digest). + instantiated_actions = {dep_codegen_digest.hash: dep_codegen_digest} result_digest = instantiator.instantiate_action( spec_action, element, {}, - action_outputs={}, # Empty — cross-element resolves via AC + instantiated_actions=instantiated_actions, ) assert result_digest is not None
