_stream.py: Add internal _resolve_elements()

Replaces Pipeline `resolve_elements()`.

1 files changed, 31 insertions, 1 deletions

diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index a34bbe389..3bbf27971 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -1363,7 +1363,7 @@ class Stream:
 
         # Now move on to loading primary selection.
         #
-        self._pipeline.resolve_elements(self.targets)
+        self._resolve_elements(self.targets)
         selected = self._pipeline.get_selection(self.targets, selection, silent=False)
         selected = self._pipeline.except_elements(self.targets, selected, except_elements)
 
@@ -1379,6 +1379,36 @@ class Stream:
 
         return selected
 
+    # _resolve_elements()
+    #
+    # Resolve element state and cache keys.
+    #
+    # Args:
+    #    targets (list of Element): The list of toplevel element targets
+    #
+    def _resolve_elements(self, targets):
+        with self._context.messenger.simple_task("Resolving cached state", silent_nested=True) as task:
+            # We need to go through the project to access the loader
+            if task:
+                task.set_maximum_progress(self._project.loader.loaded)
+
+            # XXX: Now that Element._update_state() can trigger recursive update_state calls
+            # it is possible that we could get a RecursionError. However, this is unlikely
+            # to happen, even for large projects (tested with the Debian stack). Although,
+            # if it does become a problem we may have to set the recursion limit to a
+            # greater value.
+            for element in self._pipeline.dependencies(targets, _Scope.ALL):
+                # Determine initial element state.
+                element._initialize_state()
+
+                # We may already have Elements which are cached and have their runtimes
+                # cached, if this is the case, we should immediately notify their reverse
+                # dependencies.
+                element._update_ready_for_runtime_and_cached()
+
+                if task:
+                    task.add_current_progress()
+
     # _add_queue()
     #
     # Adds a queue to the stream