WIP: notesbschubert/pipeline-bak

6 files changed, 53 insertions, 30 deletions

diff --git a/buildstream/_pipeline.py b/buildstream/_pipeline.py
index 3dc3019eb..faf4c6fc6 100644
--- a/buildstream/_pipeline.py
+++ b/buildstream/_pipeline.py
@@ -136,9 +136,6 @@ class Pipeline():
                 # Preflight
                 element._preflight()
 
-                # Determine initial element state.
-                element._update_state()
-
     # dependencies()
     #
     # Generator function to iterate over elements and optionally
diff --git a/buildstream/_scheduler/queues/buildqueue.py b/buildstream/_scheduler/queues/buildqueue.py
index 7b3a8e51f..563d0baa0 100644
--- a/buildstream/_scheduler/queues/buildqueue.py
+++ b/buildstream/_scheduler/queues/buildqueue.py
@@ -106,10 +106,6 @@ class BuildQueue(Queue):
         # Inform element in main process that assembly is done
         element._assemble_done()
 
-        # Update the state of all reverse dependencies
-        for reverse_dependency in element.reverse_build_dependencies:
-            reverse_dependency._update_state()
-
         # This has to be done after _assemble_done, such that the
         # element may register its cache key as required
         #
diff --git a/buildstream/_scheduler/queues/fetchqueue.py b/buildstream/_scheduler/queues/fetchqueue.py
index fc11fd1d1..fdf27bd8b 100644
--- a/buildstream/_scheduler/queues/fetchqueue.py
+++ b/buildstream/_scheduler/queues/fetchqueue.py
@@ -44,9 +44,6 @@ class FetchQueue(Queue):
         element._fetch()
 
     def status(self, element):
-        # state of dependencies may have changed, recalculate element state
-        element._update_state()
-
         if not element._is_required():
             # Artifact is not currently required but it may be requested later.
             # Keep it in the queue.
@@ -72,7 +69,7 @@ class FetchQueue(Queue):
         if status == JobStatus.FAIL:
             return
 
-        element._update_state()
-
+        element._fetch_done()
+        
         # Successful fetch, we must be CACHED now
         assert element._get_consistency() == Consistency.CACHED
diff --git a/buildstream/_scheduler/queues/pullqueue.py b/buildstream/_scheduler/queues/pullqueue.py
index dbeb806e5..013ee6489 100644
--- a/buildstream/_scheduler/queues/pullqueue.py
+++ b/buildstream/_scheduler/queues/pullqueue.py
@@ -39,9 +39,6 @@ class PullQueue(Queue):
             raise SkipJob(self.action_name)
 
     def status(self, element):
-        # state of dependencies may have changed, recalculate element state
-        element._update_state()
-
         if not element._is_required():
             # Artifact is not currently required but it may be requested later.
             # Keep it in the queue.
diff --git a/buildstream/_stream.py b/buildstream/_stream.py
index 62eff1d8b..396b18c96 100644
--- a/buildstream/_stream.py
+++ b/buildstream/_stream.py
@@ -1018,17 +1018,6 @@ class Stream():
 
         _, status = self._scheduler.run(self.queues)
 
-        # Force update element states after a run, such that the summary
-        # is more coherent
-        try:
-            for element in self.total_elements:
-                element._update_state()
-        except BstError as e:
-            self._message(MessageType.ERROR, "Error resolving final state", detail=str(e))
-            set_last_task_error(e.domain, e.reason)
-        except Exception as e:   # pylint: disable=broad-except
-            self._message(MessageType.BUG, "Unhandled exception while resolving final state", detail=str(e))
-
         if status == SchedStatus.ERROR:
             raise StreamError()
         elif status == SchedStatus.TERMINATED:
diff --git a/buildstream/element.py b/buildstream/element.py
index a1a9abac1..0a0513c40 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -197,7 +197,7 @@ class Element(Plugin):
 
         self.__runtime_dependencies = []        # Direct runtime dependency Elements
         self.__build_dependencies = []          # Direct build dependency Elements
-        self.reverse_build_dependencies = []    # Direct reverse build dependency Elements
+        self.__reverse_build_dependencies = []    # Direct reverse build dependency Elements
         self.__sources = []                     # List of Sources
         self.__weak_cache_key = None            # Our cached weak cache key
         self.__strict_cache_key = None          # Our cached cache key for strict builds
@@ -228,6 +228,8 @@ class Element(Plugin):
         self.__metadata_workspaced = {}               # Boolean of whether it's workspaced
         self.__metadata_workspaced_dependencies = {}  # List of which dependencies are workspaced
 
+        self.__is_workspaced = None
+
         # Ensure we have loaded this class's defaults
         self.__init_defaults(plugin_conf)
 
@@ -371,6 +373,13 @@ class Element(Plugin):
     #############################################################
     #                       Public Methods                      #
     #############################################################
+    @property
+    def is_workspaced(self):
+        if self.__is_workspaced is None:
+            self.__is_workspaced = self._get_workspace() is not None
+
+        return self.__is_workspaced
+
     def sources(self):
         """A generator function to enumerate the element sources
 
@@ -440,6 +449,27 @@ class Element(Plugin):
         if should_yield and (recurse or recursed) and scope != Scope.BUILD:
             yield self
 
+    def reverse_build_dependencies(self, recurse=False):
+        if not recurse:
+            yield from self.__reverse_build_dependencies
+            return
+
+        # visited = set()
+
+        def recurse_rdeps(element):
+            # if element in visited:
+                # return
+
+            # visited.add(element)
+
+            yield element
+
+            for rdep in element.__reverse_build_dependencies:
+                yield from recurse_rdeps(rdep)
+
+        for rdep in self.__reverse_build_dependencies:
+            yield from recurse_rdeps(rdep)
+
     def search(self, scope, name):
         """Search for a dependency by name
 
@@ -931,7 +961,7 @@ class Element(Plugin):
         for meta_dep in meta.build_dependencies:
             dependency = Element._new_from_meta(meta_dep)
             element.__build_dependencies.append(dependency)
-            dependency.reverse_build_dependencies.append(element)
+            dependency.__reverse_build_dependencies.append(element)
 
         return element
 
@@ -1281,6 +1311,9 @@ class Element(Plugin):
         for source in self.sources():
             source._preflight()
 
+        # Determine initial element state.
+        self._update_state()
+
     # _schedule_tracking():
     #
     # Force an element state to be inconsistent. Any sources appear to be
@@ -1308,6 +1341,9 @@ class Element(Plugin):
 
         self._update_state()
 
+        for rdep in self.reverse_build_dependencies(recurse=True):
+            rdep._update_state()
+
     # _track():
     #
     # Calls track() on the Element sources
@@ -1448,6 +1484,7 @@ class Element(Plugin):
     # This unblocks pull/fetch/build.
     #
     def _set_required(self):
+        # FIXME: this should enqueue stuff in the queue, it should not be here by default
         if self.__required:
             # Already done
             return
@@ -1458,6 +1495,7 @@ class Element(Plugin):
         for dep in self.dependencies(Scope.RUN, recurse=False):
             dep._set_required()
 
+        # FIXME: this should not be done at all here
         self._update_state()
 
     # _is_required():
@@ -1501,10 +1539,16 @@ class Element(Plugin):
 
         self.__assemble_scheduled = False
         self.__assemble_done = True
-
+        # FIXME: only if workspaced
         self._update_state()
 
-        if self._get_workspace() and self._cached_success():
+        if self.is_workspaced:
+
+            # Update the state of all reverse dependencies
+            for reverse_dependency in self.reverse_build_dependencies(recurse=True):
+                reverse_dependency._update_state()
+
+        if self.is_workspaced and self._cached_success():
             assert utils._is_main_process(), \
                 "Attempted to save workspace configuration from child process"
             #
@@ -2037,6 +2081,9 @@ class Element(Plugin):
                 source._fetch(previous_sources)
             previous_sources.append(source)
 
+    def _fetch_done(self):
+        self._update_state()
+
     #############################################################
     #                   Private Local Methods                   #
     #############################################################