Merge branch 'juerg/fetch-subprojects' into 'master'

Improve subproject fetching

See merge request BuildStream/buildstream!1414

9 files changed, 98 insertions, 122 deletions

diff --git a/src/buildstream/_exceptions.py b/src/buildstream/_exceptions.py
index 819f9538c..f57e4b34a 100644
--- a/src/buildstream/_exceptions.py
+++ b/src/buildstream/_exceptions.py
@@ -198,9 +198,6 @@ class LoadErrorReason(Enum):
     # Failure to load a project from a specified junction
     INVALID_JUNCTION = 13
 
-    # Subproject needs to be fetched
-    SUBPROJECT_FETCH_NEEDED = 14
-
     # Subproject has no ref
     SUBPROJECT_INCONSISTENT = 15
 
diff --git a/src/buildstream/_frontend/app.py b/src/buildstream/_frontend/app.py
index a9dd46b34..68cf7ec4d 100644
--- a/src/buildstream/_frontend/app.py
+++ b/src/buildstream/_frontend/app.py
@@ -216,33 +216,12 @@ class App():
         except BstError as e:
             self._error_exit(e, "Error instantiating artifact cache")
 
-        #
-        # Load the Project
-        #
-        try:
-            self.project = Project(directory, self.context, cli_options=self._main_options['option'],
-                                   default_mirror=self._main_options.get('default_mirror'))
-        except LoadError as e:
-
-            # Help users that are new to BuildStream by suggesting 'init'.
-            # We don't want to slow down users that just made a mistake, so
-            # don't stop them with an offer to create a project for them.
-            if e.reason == LoadErrorReason.MISSING_PROJECT_CONF:
-                click.echo("No project found. You can create a new project like so:", err=True)
-                click.echo("", err=True)
-                click.echo("    bst init", err=True)
-
-            self._error_exit(e, "Error loading project")
-
-        except BstError as e:
-            self._error_exit(e, "Error loading project")
-
         # Now that we have a logger and message handler,
         # we can override the global exception hook.
         sys.excepthook = self._global_exception_handler
 
         # Create the stream right away, we'll need to pass it around
-        self.stream = Stream(self.context, self.project, self._session_start,
+        self.stream = Stream(self.context, self._session_start,
                              session_start_callback=self.session_start_cb,
                              interrupt_callback=self._interrupt_handler,
                              ticker_callback=self._tick,
@@ -259,6 +238,30 @@ class App():
         if session_name:
             self._message(MessageType.START, session_name)
 
+        #
+        # Load the Project
+        #
+        try:
+            self.project = Project(directory, self.context, cli_options=self._main_options['option'],
+                                   default_mirror=self._main_options.get('default_mirror'),
+                                   fetch_subprojects=self.stream.fetch_subprojects)
+
+            self.stream.set_project(self.project)
+        except LoadError as e:
+
+            # Help users that are new to BuildStream by suggesting 'init'.
+            # We don't want to slow down users that just made a mistake, so
+            # don't stop them with an offer to create a project for them.
+            if e.reason == LoadErrorReason.MISSING_PROJECT_CONF:
+                click.echo("No project found. You can create a new project like so:", err=True)
+                click.echo("", err=True)
+                click.echo("    bst init", err=True)
+
+            self._error_exit(e, "Error loading project")
+
+        except BstError as e:
+            self._error_exit(e, "Error loading project")
+
         # Run the body of the session here, once everything is loaded
         try:
             yield
diff --git a/src/buildstream/_includes.py b/src/buildstream/_includes.py
index f792b7716..8f507b566 100644
--- a/src/buildstream/_includes.py
+++ b/src/buildstream/_includes.py
@@ -104,7 +104,7 @@ class Includes:
         shortname = include
         if ':' in include:
             junction, include = include.split(':', 1)
-            junction_loader = loader._get_loader(junction, fetch_subprojects=True)
+            junction_loader = loader._get_loader(junction)
             current_loader = junction_loader
         else:
             current_loader = loader
diff --git a/src/buildstream/_loader/loader.py b/src/buildstream/_loader/loader.py
index d52a8a72e..fa3539b22 100644
--- a/src/buildstream/_loader/loader.py
+++ b/src/buildstream/_loader/loader.py
@@ -45,11 +45,12 @@ from .._message import Message, MessageType
 # Args:
 #    context (Context): The Context object
 #    project (Project): The toplevel Project object
+#    fetch_subprojects (callable): A function to fetch subprojects
 #    parent (Loader): A parent Loader object, in the case this is a junctioned Loader
 #
 class Loader():
 
-    def __init__(self, context, project, *, parent=None):
+    def __init__(self, context, project, *, fetch_subprojects, parent=None):
 
         # Ensure we have an absolute path for the base directory
         basedir = project.element_path
@@ -69,6 +70,7 @@ class Loader():
         self._basedir = basedir              # Base project directory
         self._first_pass_options = project.first_pass_config.options  # Project options (OptionPool)
         self._parent = parent                # The parent loader
+        self._fetch_subprojects = fetch_subprojects
 
         self._meta_elements = {}  # Dict of resolved meta elements by name
         self._elements = {}       # Dict of elements
@@ -85,12 +87,11 @@ class Loader():
     #                       this is a bit more expensive due to deep copies
     #    ticker (callable): An optional function for tracking load progress
     #    targets (list of str): Target, element-path relative bst filenames in the project
-    #    fetch_subprojects (bool): Whether to fetch subprojects while loading
     #
     # Raises: LoadError
     #
     # Returns: The toplevel LoadElement
-    def load(self, targets, rewritable=False, ticker=None, fetch_subprojects=False):
+    def load(self, targets, rewritable=False, ticker=None):
 
         for filename in targets:
             if os.path.isabs(filename):
@@ -109,9 +110,8 @@ class Loader():
 
         for target in targets:
             with PROFILER.profile(Topics.LOAD_PROJECT, target):
-                _junction, name, loader = self._parse_name(target, rewritable, ticker,
-                                                           fetch_subprojects=fetch_subprojects)
-                element = loader._load_file(name, rewritable, ticker, fetch_subprojects)
+                _junction, name, loader = self._parse_name(target, rewritable, ticker)
+                element = loader._load_file(name, rewritable, ticker)
                 target_elements.append(element)
 
         #
@@ -255,13 +255,12 @@ class Loader():
     #    filename (str): The element-path relative bst file
     #    rewritable (bool): Whether we should load in round trippable mode
     #    ticker (callable): A callback to report loaded filenames to the frontend
-    #    fetch_subprojects (bool): Whether to fetch subprojects while loading
     #    provenance (Provenance): The location from where the file was referred to, or None
     #
     # Returns:
     #    (LoadElement): A loaded LoadElement
     #
-    def _load_file(self, filename, rewritable, ticker, fetch_subprojects, provenance=None):
+    def _load_file(self, filename, rewritable, ticker, provenance=None):
 
         # Silently ignore already loaded files
         if filename in self._elements:
@@ -290,14 +289,12 @@ class Loader():
                 current_element[2].append(dep.name)
 
                 if dep.junction:
-                    self._load_file(dep.junction, rewritable, ticker,
-                                    fetch_subprojects, dep.provenance)
+                    self._load_file(dep.junction, rewritable, ticker, dep.provenance)
                     loader = self._get_loader(dep.junction,
                                               rewritable=rewritable,
                                               ticker=ticker,
-                                              fetch_subprojects=fetch_subprojects,
                                               provenance=dep.provenance)
-                    dep_element = loader._load_file(dep.name, rewritable, ticker, fetch_subprojects, dep.provenance)
+                    dep_element = loader._load_file(dep.name, rewritable, ticker, dep.provenance)
                 else:
                     dep_element = self._elements.get(dep.name)
 
@@ -553,13 +550,12 @@ class Loader():
     #
     # Args:
     #    filename (str): Junction name
-    #    fetch_subprojects (bool): Whether to fetch subprojects while loading
     #
     # Raises: LoadError
     #
     # Returns: A Loader or None if specified junction does not exist
     def _get_loader(self, filename, *, rewritable=False, ticker=None, level=0,
-                    fetch_subprojects=False, provenance=None):
+                    provenance=None):
 
         provenance_str = ""
         if provenance is not None:
@@ -582,14 +578,13 @@ class Loader():
             # junctions in the parent take precedence over junctions defined
             # in subprojects
             loader = self._parent._get_loader(filename, rewritable=rewritable, ticker=ticker,
-                                              level=level + 1, fetch_subprojects=fetch_subprojects,
-                                              provenance=provenance)
+                                              level=level + 1, provenance=provenance)
             if loader:
                 self._loaders[filename] = loader
                 return loader
 
         try:
-            self._load_file(filename, rewritable, ticker, fetch_subprojects)
+            self._load_file(filename, rewritable, ticker)
         except LoadError as e:
             if e.reason != LoadErrorReason.MISSING_FILE:
                 # other load error
@@ -619,26 +614,18 @@ class Loader():
         # find loader for that project.
         if element.target:
             subproject_loader = self._get_loader(element.target_junction, rewritable=rewritable, ticker=ticker,
-                                                 level=level, fetch_subprojects=fetch_subprojects,
-                                                 provenance=provenance)
+                                                 level=level, provenance=provenance)
             loader = subproject_loader._get_loader(element.target_element, rewritable=rewritable, ticker=ticker,
-                                                   level=level, fetch_subprojects=fetch_subprojects,
-                                                   provenance=provenance)
+                                                   level=level, provenance=provenance)
             self._loaders[filename] = loader
             return loader
 
         # Handle the case where a subproject needs to be fetched
         #
         if element._get_consistency() == Consistency.RESOLVED:
-            if fetch_subprojects:
-                if ticker:
-                    ticker(filename, 'Fetching subproject')
-                element._fetch()
-            else:
-                detail = "Try fetching the project with `bst source fetch {}`".format(filename)
-                raise LoadError(LoadErrorReason.SUBPROJECT_FETCH_NEEDED,
-                                "{}Subproject fetch needed for junction: {}".format(provenance_str, filename),
-                                detail=detail)
+            if ticker:
+                ticker(filename, 'Fetching subproject')
+            self._fetch_subprojects([element])
 
         # Handle the case where a subproject has no ref
         #
@@ -670,7 +657,8 @@ class Loader():
         try:
             from .._project import Project  # pylint: disable=cyclic-import
             project = Project(project_dir, self._context, junction=element,
-                              parent_loader=self, search_for_project=False)
+                              parent_loader=self, search_for_project=False,
+                              fetch_subprojects=self._fetch_subprojects)
         except LoadError as e:
             if e.reason == LoadErrorReason.MISSING_PROJECT_CONF:
                 message = (
@@ -698,14 +686,13 @@ class Loader():
     #   rewritable (bool): Whether the loaded files should be rewritable
     #                      this is a bit more expensive due to deep copies
     #   ticker (callable): An optional function for tracking load progress
-    #   fetch_subprojects (bool): Whether to fetch subprojects while loading
     #
     # Returns:
     #   (tuple): - (str): name of the junction element
     #            - (str): name of the element
     #            - (Loader): loader for sub-project
     #
-    def _parse_name(self, name, rewritable, ticker, fetch_subprojects=False):
+    def _parse_name(self, name, rewritable, ticker):
         # We allow to split only once since deep junctions names are forbidden.
         # Users who want to refer to elements in sub-sub-projects are required
         # to create junctions on the top level project.
@@ -713,9 +700,8 @@ class Loader():
         if len(junction_path) == 1:
             return None, junction_path[-1], self
         else:
-            self._load_file(junction_path[-2], rewritable, ticker, fetch_subprojects)
-            loader = self._get_loader(junction_path[-2], rewritable=rewritable, ticker=ticker,
-                                      fetch_subprojects=fetch_subprojects)
+            self._load_file(junction_path[-2], rewritable, ticker)
+            loader = self._get_loader(junction_path[-2], rewritable=rewritable, ticker=ticker)
             return junction_path[-2], junction_path[-1], loader
 
     # Print a warning message, checks warning_token against project configuration
diff --git a/src/buildstream/_pipeline.py b/src/buildstream/_pipeline.py
index e6ae94cfd..0758cf5ff 100644
--- a/src/buildstream/_pipeline.py
+++ b/src/buildstream/_pipeline.py
@@ -92,25 +92,19 @@ class Pipeline():
     #
     # Args:
     #    target_groups (list of lists): Groups of toplevel targets to load
-    #    fetch_subprojects (bool): Whether we should fetch subprojects as a part of the
-    #                              loading process, if they are not yet locally cached
     #    rewritable (bool): Whether the loaded files should be rewritable
     #                       this is a bit more expensive due to deep copies
     #
     # Returns:
     #    (tuple of lists): A tuple of grouped Element objects corresponding to target_groups
     #
-    def load(self, target_groups, *,
-             fetch_subprojects=True,
-             rewritable=False):
+    def load(self, target_groups, *, rewritable=False):
 
         # First concatenate all the lists for the loader's sake
         targets = list(itertools.chain(*target_groups))
 
         with PROFILER.profile(Topics.LOAD_PIPELINE, "_".join(t.replace(os.sep, "-") for t in targets)):
-            elements = self._project.load_elements(targets,
-                                                   rewritable=rewritable,
-                                                   fetch_subprojects=fetch_subprojects)
+            elements = self._project.load_elements(targets, rewritable=rewritable)
 
             # Now create element groups to match the input target groups
             elt_iter = iter(elements)
diff --git a/src/buildstream/_project.py b/src/buildstream/_project.py
index 1fdc84acb..114d25054 100644
--- a/src/buildstream/_project.py
+++ b/src/buildstream/_project.py
@@ -95,7 +95,7 @@ class Project():
 
     def __init__(self, directory, context, *, junction=None, cli_options=None,
                  default_mirror=None, parent_loader=None,
-                 search_for_project=True):
+                 search_for_project=True, fetch_subprojects=None):
 
         # The project name
         self.name = None
@@ -157,7 +157,7 @@ class Project():
         self._project_includes = None
 
         with PROFILER.profile(Topics.LOAD_PROJECT, self.directory.replace(os.sep, '-')):
-            self._load(parent_loader=parent_loader)
+            self._load(parent_loader=parent_loader, fetch_subprojects=fetch_subprojects)
 
         self._partially_loaded = True
 
@@ -440,18 +440,13 @@ class Project():
     #    targets (list): Target names
     #    rewritable (bool): Whether the loaded files should be rewritable
     #                       this is a bit more expensive due to deep copies
-    #    fetch_subprojects (bool): Whether we should fetch subprojects as a part of the
-    #                              loading process, if they are not yet locally cached
     #
     # Returns:
     #    (list): A list of loaded Element
     #
-    def load_elements(self, targets, *,
-                      rewritable=False, fetch_subprojects=False):
+    def load_elements(self, targets, *, rewritable=False):
         with self._context.timed_activity("Loading elements", silent_nested=True):
-            meta_elements = self.loader.load(targets, rewritable=rewritable,
-                                             ticker=None,
-                                             fetch_subprojects=fetch_subprojects)
+            meta_elements = self.loader.load(targets, rewritable=rewritable, ticker=None)
 
         with self._context.timed_activity("Resolving elements"):
             elements = [
@@ -558,7 +553,7 @@ class Project():
     #
     # Raises: LoadError if there was a problem with the project.conf
     #
-    def _load(self, parent_loader=None):
+    def _load(self, *, parent_loader=None, fetch_subprojects):
 
         # Load builtin default
         projectfile = os.path.join(self.directory, _PROJECT_CONF_FILE)
@@ -613,7 +608,7 @@ class Project():
         self._fatal_warnings = _yaml.node_get(pre_config_node, list, 'fatal-warnings', default_value=[])
 
         self.loader = Loader(self._context, self,
-                             parent=parent_loader)
+                             parent=parent_loader, fetch_subprojects=fetch_subprojects)
 
         self._project_includes = Includes(self.loader, copy_tree=False)
 
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 8097f451d..a7db33bb9 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -49,7 +49,6 @@ from . import Scope, Consistency
 #
 # Args:
 #    context (Context): The Context object
-#    project (Project): The Project object
 #    session_start (datetime): The time when the session started
 #    session_start_callback (callable): A callback to invoke when the session starts
 #    interrupt_callback (callable): A callback to invoke when we get interrupted
@@ -59,7 +58,7 @@ from . import Scope, Consistency
 #
 class Stream():
 
-    def __init__(self, context, project, session_start, *,
+    def __init__(self, context, session_start, *,
                  session_start_callback=None,
                  interrupt_callback=None,
                  ticker_callback=None,
@@ -80,8 +79,8 @@ class Stream():
         self._artifacts = context.artifactcache
         self._sourcecache = context.sourcecache
         self._context = context
-        self._project = project
-        self._pipeline = Pipeline(context, project, self._artifacts)
+        self._project = None
+        self._pipeline = None
         self._scheduler = Scheduler(context, session_start,
                                     interrupt_callback=interrupt_callback,
                                     ticker_callback=ticker_callback,
@@ -98,6 +97,18 @@ class Stream():
         if self._project:
             self._project.cleanup()
 
+    # set_project()
+    #
+    # Set the top-level project.
+    #
+    # Args:
+    #    project (Project): The Project object
+    #
+    def set_project(self, project):
+        assert self._project is None
+        self._project = project
+        self._pipeline = Pipeline(self._context, project, self._artifacts)
+
     # load_selection()
     #
     # An all purpose method for loading a selection of elements, this
@@ -121,7 +132,6 @@ class Stream():
             target_objects, _ = self._load(targets, (),
                                            selection=selection,
                                            except_targets=except_targets,
-                                           fetch_subprojects=False,
                                            use_artifact_config=use_artifact_config,
                                            load_refs=load_refs)
 
@@ -242,7 +252,6 @@ class Stream():
                        use_artifact_config=use_config,
                        artifact_remote_url=remote,
                        use_source_config=True,
-                       fetch_subprojects=True,
                        dynamic_plan=True)
 
         # Remove the tracking elements from the main targets
@@ -323,7 +332,6 @@ class Stream():
                        except_targets=except_targets,
                        track_except_targets=track_except_targets,
                        track_cross_junctions=track_cross_junctions,
-                       fetch_subprojects=True,
                        use_source_config=use_source_config,
                        source_remote_url=remote)
 
@@ -356,8 +364,7 @@ class Stream():
                        selection=selection, track_selection=selection,
                        except_targets=except_targets,
                        track_except_targets=except_targets,
-                       track_cross_junctions=cross_junctions,
-                       fetch_subprojects=True)
+                       track_cross_junctions=cross_junctions)
 
         track_queue = TrackQueue(self._scheduler)
         self._add_queue(track_queue, track=True)
@@ -390,8 +397,7 @@ class Stream():
                                  selection=selection,
                                  ignore_junction_targets=ignore_junction_targets,
                                  use_artifact_config=use_config,
-                                 artifact_remote_url=remote,
-                                 fetch_subprojects=True)
+                                 artifact_remote_url=remote)
 
         if not self._artifacts.has_fetch_remotes():
             raise StreamError("No artifact caches available for pulling artifacts")
@@ -431,8 +437,7 @@ class Stream():
                                  selection=selection,
                                  ignore_junction_targets=ignore_junction_targets,
                                  use_artifact_config=use_config,
-                                 artifact_remote_url=remote,
-                                 fetch_subprojects=True)
+                                 artifact_remote_url=remote)
 
         if not self._artifacts.has_push_remotes():
             raise StreamError("No artifact caches available for pushing artifacts")
@@ -496,9 +501,7 @@ class Stream():
 
         # if pulling we need to ensure dependency artifacts are also pulled
         selection = PipelineSelection.RUN if pull else PipelineSelection.NONE
-        elements, _ = self._load(
-            (target,), (), selection=selection,
-            fetch_subprojects=True, use_artifact_config=True)
+        elements, _ = self._load((target,), (), selection=selection, use_artifact_config=True)
 
         target = elements[-1]
 
@@ -644,8 +647,7 @@ class Stream():
 
         elements, _ = self._load((target,), (),
                                  selection=deps,
-                                 except_targets=except_targets,
-                                 fetch_subprojects=True)
+                                 except_targets=except_targets)
 
         # Assert all sources are cached in the source dir
         if fetch:
@@ -951,6 +953,23 @@ class Stream():
 
         return list(output_elements)
 
+    # fetch_subprojects()
+    #
+    # Fetch subprojects as part of the project and element loading process.
+    #
+    # Args:
+    #    junctions (list of Element): The junctions to fetch
+    #
+    def fetch_subprojects(self, junctions):
+        old_queues = self.queues
+        try:
+            queue = FetchQueue(self._scheduler)
+            queue.enqueue(junctions)
+            self.queues = [queue]
+            self._run()
+        finally:
+            self.queues = old_queues
+
     #############################################################
     #                 Scheduler API forwarding                  #
     #############################################################
@@ -1039,7 +1058,6 @@ class Stream():
     #    use_source_config (bool): Whether to initialize remote source caches with the config
     #    artifact_remote_url (str): A remote url for initializing the artifacts
     #    source_remote_url (str): A remote url for initializing source caches
-    #    fetch_subprojects (bool): Whether to fetch subprojects while loading
     #
     # Returns:
     #    (list of Element): The primary element selection
@@ -1056,7 +1074,6 @@ class Stream():
               use_source_config=False,
               artifact_remote_url=None,
               source_remote_url=None,
-              fetch_subprojects=False,
               dynamic_plan=False,
               load_refs=False):
 
@@ -1075,8 +1092,7 @@ class Stream():
         # Load all target elements
         elements, except_elements, track_elements, track_except_elements = \
             self._pipeline.load([target_elements, except_targets, track_targets, track_except_targets],
-                                rewritable=rewritable,
-                                fetch_subprojects=fetch_subprojects)
+                                rewritable=rewritable)
 
         # Obtain the ArtifactElement objects
         artifacts = [self._project.create_artifact_element(ref) for ref in target_artifacts]
diff --git a/tests/format/junctions.py b/tests/format/junctions.py
index a85308e39..8842bc617 100644
--- a/tests/format/junctions.py
+++ b/tests/format/junctions.py
@@ -333,18 +333,8 @@ def test_git_show(cli, tmpdir, datafiles):
     }
     _yaml.dump(element, os.path.join(project, 'base.bst'))
 
-    # Verify that bst show does not implicitly fetch subproject
-    result = cli.run(project=project, args=['show', 'target.bst'])
-    result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_FETCH_NEEDED)
-
-    # Assert that we have the expected provenance encoded into the error
-    assert "target.bst [line 3 column 2]" in result.stderr
-
-    # Explicitly fetch subproject
-    result = cli.run(project=project, args=['source', 'fetch', 'base.bst'])
-    result.assert_success()
-
-    # Check that bst show succeeds now and the pipeline includes the subproject element
+    # Check that bst show succeeds with implicit subproject fetching and the
+    # pipeline includes the subproject element
     element_list = cli.get_pipeline(project, ['target.bst'])
     assert 'base.bst:target.bst' in element_list
 
diff --git a/tests/frontend/show.py b/tests/frontend/show.py
index 0f6d74c65..4ef97dd84 100644
--- a/tests/frontend/show.py
+++ b/tests/frontend/show.py
@@ -277,15 +277,10 @@ def test_unfetched_junction(cli, tmpdir, datafiles, ref_storage, element_name, w
         ])
         result.assert_success()
 
-    # Assert the correct error when trying to show the pipeline
+    # Assert successful bst show (requires implicit subproject fetching)
     result = cli.run(project=project, silent=True, args=[
         'show', element_name])
-
-    # If a workspace is open, no fetch is needed
-    if workspaced:
-        result.assert_success()
-    else:
-        result.assert_main_error(ErrorDomain.LOAD, LoadErrorReason.SUBPROJECT_FETCH_NEEDED)
+    result.assert_success()
 
 
 @pytest.mark.datafiles(os.path.join(DATA_DIR, 'project'))