Merge branch 'juerg/dynamic-plan' into 'master'

Restore dynamic build plan feature

See merge request BuildStream/buildstream!1926

5 files changed, 139 insertions, 14 deletions

diff --git a/src/buildstream/_loader/loader.py b/src/buildstream/_loader/loader.py
index 3cfaf8be6..37e00ad4b 100644
--- a/src/buildstream/_loader/loader.py
+++ b/src/buildstream/_loader/loader.py
@@ -292,6 +292,9 @@ class Loader:
             # Optimization for junctions with a single local source
             basedir = sources[0]._get_local_path()
         else:
+            # Stage sources
+            element._set_required()
+
             # Note: We use _KeyStrength.WEAK here because junctions
             # cannot have dependencies, therefore the keys are
             # equivalent.
diff --git a/src/buildstream/_scheduler/queues/queue.py b/src/buildstream/_scheduler/queues/queue.py
index 295161ed2..986ac6c0a 100644
--- a/src/buildstream/_scheduler/queues/queue.py
+++ b/src/buildstream/_scheduler/queues/queue.py
@@ -181,7 +181,10 @@ class Queue:
 
         # Obtain immediate element status
         for elt in elts:
-            self._enqueue_element(elt)
+            if self._required_element_check and not elt._is_required():
+                elt._set_required_callback(self._enqueue_element)
+            else:
+                self._enqueue_element(elt)
 
     # dequeue()
     #
@@ -238,6 +241,13 @@ class Queue:
             for element in ready
         ]
 
+    # set_required_element_check()
+    #
+    # This ensures that, for the first non-track queue, we must check
+    # whether elements are required before enqueuing them
+    def set_required_element_check(self):
+        self._required_element_check = True
+
     # any_failed_elements()
     #
     # Returns whether any elements in this queue have failed their jobs
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 48390ba14..dc91db6b7 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -371,8 +371,8 @@ class Stream:
 
         self._scheduler.clear_queues()
         track_queue = TrackQueue(self._scheduler)
-        self._add_queue(track_queue)
-        self._enqueue_plan(elements)
+        self._add_queue(track_queue, track=True)
+        self._enqueue_plan(elements, queue=track_queue)
         self._run()
 
     # pull()
@@ -1286,10 +1286,10 @@ class Stream:
             # others are requested dynamically as needed.
             # This avoids pulling, fetching, or building unneeded build-only dependencies.
             for element in elements:
-                element._schedule_assembly_when_necessary()
+                element._set_required()
         else:
             for element in selected:
-                element._schedule_assembly_when_necessary()
+                element._set_required()
 
         return selected
 
@@ -1308,7 +1308,11 @@ class Stream:
     # Args:
     #    queue (Queue): Queue to add to the pipeline
     #
-    def _add_queue(self, queue):
+    def _add_queue(self, queue, *, track=False):
+        if not track and not self.queues:
+            # First non-track queue
+            queue.set_required_element_check()
+
         self.queues.append(queue)
 
     # _enqueue_plan()
diff --git a/src/buildstream/element.py b/src/buildstream/element.py
index e3d4ffc68..bd702b112 100644
--- a/src/buildstream/element.py
+++ b/src/buildstream/element.py
@@ -253,6 +253,7 @@ class Element(Plugin):
         self.__splits = None  # Resolved regex objects for computing split domains
         self.__whitelist_regex = None  # Resolved regex object to check if file is allowed to overlap
         self.__tainted = None  # Whether the artifact is tainted and should not be shared
+        self.__required = False  # Whether the artifact is required in the current session
         self.__artifact_files_required = False  # Whether artifact files are required in the local cache
         self.__build_result = None  # The result of assembling this Element (success, description, detail)
         self._build_log_path = None  # The path of the build log for this Element
@@ -271,6 +272,7 @@ class Element(Plugin):
         self.__batch_prepare_assemble_collect = None  # type: Optional[str]
 
         # Callbacks
+        self.__required_callback = None  # Callback to Queues
         self.__can_query_cache_callback = None  # Callback to PullQueue/FetchQueue
         self.__buildable_callback = None  # Callback to BuildQueue
 
@@ -1142,7 +1144,7 @@ class Element(Plugin):
     # - _update_artifact_state()
     #   - Computes the state of the element's artifact using the
     #     cache key.
-    # - _schedule_assembly_when_necessary()
+    # - __schedule_assembly_when_necessary()
     #   - Schedules assembly of an element, iff its current state
     #     allows/necessitates it
     # - __update_cache_key_non_strict()
@@ -1384,6 +1386,41 @@ class Element(Plugin):
         # Ensure deterministic owners of sources at build time
         vdirectory.set_deterministic_user()
 
+    # _set_required():
+    #
+    # Mark this element and its runtime dependencies as required.
+    # This unblocks pull/fetch/build.
+    #
+    def _set_required(self):
+        if self.__required:
+            # Already done
+            return
+
+        self.__required = True
+
+        # Request artifacts of runtime dependencies
+        for dep in self.dependencies(Scope.RUN, recurse=False):
+            dep._set_required()
+
+        # When an element becomes required, it must be assembled for
+        # the current pipeline. `__schedule_assembly_when_necessary()`
+        # will abort if some other state prevents it from being built,
+        # and changes to such states will cause re-scheduling, so this
+        # is safe.
+        self.__schedule_assembly_when_necessary()
+
+        # Callback to the Queue
+        if self.__required_callback is not None:
+            self.__required_callback(self)
+            self.__required_callback = None
+
+    # _is_required():
+    #
+    # Returns whether this element has been marked as required.
+    #
+    def _is_required(self):
+        return self.__required
+
     # _set_artifact_files_required():
     #
     # Mark artifact files for this element and its runtime dependencies as
@@ -1422,6 +1459,9 @@ class Element(Plugin):
             # We're not processing
             not processing
             and
+            # We're required for the current build
+            self._is_required()
+            and
             # We have figured out the state of our artifact
             self.__artifact
             and
@@ -1429,12 +1469,12 @@ class Element(Plugin):
             not self._cached()
         )
 
-    # _schedule_assembly_when_necessary():
+    # __schedule_assembly_when_necessary():
     #
     # This is called in the main process before the element is assembled
     # in a subprocess.
     #
-    def _schedule_assembly_when_necessary(self):
+    def __schedule_assembly_when_necessary(self):
         # FIXME: We could reduce the number of function calls a bit by
         # factoring this out of this method (and checking whether we
         # should schedule at the calling end).
@@ -1448,7 +1488,7 @@ class Element(Plugin):
 
         # Requests artifacts of build dependencies
         for dep in self.dependencies(Scope.BUILD, recurse=False):
-            dep._schedule_assembly_when_necessary()
+            dep._set_required()
 
         # Once we schedule an element for assembly, we know that our
         # build dependencies have strong cache keys, so we can update
@@ -1708,7 +1748,7 @@ class Element(Plugin):
 
         # We may not have actually pulled an artifact - the pull may
         # have failed. We might therefore need to schedule assembly.
-        self._schedule_assembly_when_necessary()
+        self.__schedule_assembly_when_necessary()
         # If we've finished pulling, an artifact might now exist
         # locally, so we might need to update a non-strict strong
         # cache key.
@@ -2113,6 +2153,22 @@ class Element(Plugin):
             return not self._has_all_sources_cached()
         return not self._has_all_sources_in_source_cache()
 
+    # _set_required_callback()
+    #
+    #
+    # Notify the pull/fetch/build queue that the element is potentially
+    # ready to be processed.
+    #
+    # _Set the _required_callback - the _required_callback is invoked when an
+    # element is marked as required. This informs us that the element needs to
+    # either be pulled or fetched + built.
+    #
+    # Args:
+    #    callback (callable) - The callback function
+    #
+    def _set_required_callback(self, callback):
+        self.__required_callback = callback
+
     # _set_can_query_cache_callback()
     #
     # Notify the pull/fetch queue that the element is potentially
@@ -2221,6 +2277,11 @@ class Element(Plugin):
         assert "_Element__buildable_callback" in state
         state["_Element__buildable_callback"] = None
 
+        # This callback is not even read in the child process, so delete it.
+        # If this assumption is invalidated, we will get an attribute error to
+        # let us know, and we will need to update accordingly.
+        del state["_Element__required_callback"]
+
         return self.__meta_kind, state
 
     def _walk_artifact_files(self):
@@ -3016,7 +3077,7 @@ class Element(Plugin):
     # to this element.
     #
     # If the state changes, this will subsequently call
-    # `self._schedule_assembly_when_necessary()` to schedule assembly if it becomes
+    # `self.__schedule_assembly_when_necessary()` to schedule assembly if it becomes
     # possible.
     #
     # Element.__update_cache_keys() must be called before this to have
@@ -3032,7 +3093,7 @@ class Element(Plugin):
         if not context.get_strict() and not self.__artifact:
             # We've calculated the weak_key, so instantiate artifact instance member
             self.__artifact = Artifact(self, context, weak_key=self.__weak_cache_key)
-            self._schedule_assembly_when_necessary()
+            self.__schedule_assembly_when_necessary()
 
         if not self.__strict_cache_key:
             return
@@ -3044,7 +3105,7 @@ class Element(Plugin):
 
             if context.get_strict():
                 self.__artifact = self.__strict_artifact
-                self._schedule_assembly_when_necessary()
+                self.__schedule_assembly_when_necessary()
             else:
                 self.__update_cache_key_non_strict()
 
diff --git a/tests/frontend/pull.py b/tests/frontend/pull.py
index 2f8d7a5ba..6a5a41977 100644
--- a/tests/frontend/pull.py
+++ b/tests/frontend/pull.py
@@ -542,3 +542,50 @@ def test_pull_artifact(cli, tmpdir, datafiles):
 
         # And assert that it's again in the local cache, without having built
         assert os.path.exists(os.path.join(local_cache, "artifacts", "refs", artifact_ref))
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_dynamic_build_plan(cli, tmpdir, datafiles):
+    project = str(datafiles)
+    target = "checkout-deps.bst"
+    build_dep = "import-dev.bst"
+    runtime_dep = "import-bin.bst"
+    all_elements = [target, build_dep, runtime_dep]
+
+    with create_artifact_share(os.path.join(str(tmpdir), "artifactshare")) as share:
+
+        # First build the target element and push to the remote.
+        cli.configure({"artifacts": {"url": share.repo, "push": True}})
+        result = cli.run(project=project, args=["build", target])
+        result.assert_success()
+
+        # Assert that everything is now cached in the remote.
+        for element_name in all_elements:
+            assert_shared(cli, share, project, element_name)
+
+        # Now we've pushed, delete the user's local artifact cache directory
+        casdir = os.path.join(cli.directory, "cas")
+        shutil.rmtree(casdir)
+        artifactdir = os.path.join(cli.directory, "artifacts")
+        shutil.rmtree(artifactdir)
+
+        # Assert that nothing is cached locally anymore
+        states = cli.get_element_states(project, all_elements)
+        assert not any(states[e] == "cached" for e in all_elements)
+
+        # Now try to rebuild target
+        result = cli.run(project=project, args=["build", target])
+        result.assert_success()
+
+        # Assert that target and runtime dependency were pulled
+        # but build dependency was not pulled as it wasn't needed
+        # (dynamic build plan).
+        assert target in result.get_pulled_elements()
+        assert runtime_dep in result.get_pulled_elements()
+        assert build_dep not in result.get_pulled_elements()
+
+        # And assert that the pulled elements are again in the local cache
+        states = cli.get_element_states(project, all_elements)
+        assert states[target] == "cached"
+        assert states[runtime_dep] == "cached"
+        assert states[build_dep] != "cached"