Add BST_REQUIRES_PREVIOUS_SOURCE_STAGE option

This is an element option that allows sources to be staged more
seperately where possible rather than on a per element option.

Part of #982

6 files changed, 104 insertions, 36 deletions

diff --git a/buildstream/_sourcecache.py b/buildstream/_sourcecache.py
index 36f75d040..d00015128 100644
--- a/buildstream/_sourcecache.py
+++ b/buildstream/_sourcecache.py
@@ -118,12 +118,17 @@ class SourceCache(BaseCache):
         ref = source._get_source_name()
 
         # Use tmpdir for now
+        vdir = CasBasedDirectory(self.cas)
+        for previous_source in previous_sources:
+            vdir.import_files(self.export(previous_source))
+
         with utils._tempdir(dir=self.context.tmpdir, prefix='staging-temp') as tmpdir:
-            for previous_source in previous_sources:
-                previous_source._stage(tmpdir)
+            if not vdir.is_empty():
+                vdir.export_files(tmpdir)
             source._stage(tmpdir)
+            vdir.import_files(tmpdir, can_link=True)
 
-            self.cas.commit([ref], tmpdir)
+        self.cas.set_ref(ref, vdir._get_digest())
 
     # export()
     #
diff --git a/buildstream/element.py b/buildstream/element.py
index 30bbc141c..7f68af262 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -107,6 +107,7 @@ from ._artifact import Artifact
 
 from .storage.directory import Directory
 from .storage._filebaseddirectory import FileBasedDirectory
+from .storage._casbaseddirectory import CasBasedDirectory
 from .storage.directory import VirtualDirectoryError
 
 
@@ -233,6 +234,10 @@ class Element(Plugin):
         self.__artifact = None                  # Artifact class for direct artifact composite interaction
         self.__strict_artifact = None           # Artifact for strict cache key
 
+        # the index of the last source in this element that requires previous
+        # sources for staging
+        self.__last_source_requires_previous_ix = None
+
         self.__batch_prepare_assemble = False         # Whether batching across prepare()/assemble() is configured
         self.__batch_prepare_assemble_flags = 0       # Sandbox flags for batching across prepare()/assemble()
         self.__batch_prepare_assemble_collect = None  # Collect dir for batching across prepare()/assemble()
@@ -1552,12 +1557,22 @@ class Element(Plugin):
 
                 if self.__sources:
 
-                    sourcecache = self._get_context().sourcecache
+                    sourcecache = context.sourcecache
+                    # find last required source
+                    last_required_previous_ix = self.__last_source_requires_previous()
+                    import_dir = CasBasedDirectory(context.get_cascache())
+
                     try:
-                        import_dir = sourcecache.export(self.__sources[-1])
+                        for source in self.__sources[last_required_previous_ix:]:
+                            source_dir = sourcecache.export(source)
+                            import_dir.import_files(source_dir)
                     except SourceCacheError as e:
                         raise ElementError("Error trying to export source for {}: {}"
                                            .format(self.name, e))
+                    except VirtualDirectoryError as e:
+                        raise ElementError("Error trying to import sources together for {}: {}"
+                                           .format(self.name, e),
+                                           reason="import-source-files-fail")
 
             with utils._deterministic_umask():
                 vdirectory.import_files(import_dir)
@@ -1938,11 +1953,8 @@ class Element(Plugin):
     def _source_push(self):
         # try and push sources if we've got them
         if self.__sourcecache.has_push_remotes(plugin=self) and self._source_cached():
-            sources = list(self.sources())
-            if sources:
-                source_pushed = self.__sourcecache.push(sources[-1])
-
-                if not source_pushed:
+            for source in self.sources():
+                if not self.__sourcecache.push(source):
                     return False
 
         # Notify successful upload
@@ -2212,24 +2224,27 @@ class Element(Plugin):
     def _fetch(self, fetch_original=False):
         previous_sources = []
         sources = self.__sources
+        fetch_needed = False
         if sources and not fetch_original:
-            source = sources[-1]
-            if self.__sourcecache.contains(source):
-                return
+            for source in self.__sources:
+                if self.__sourcecache.contains(source):
+                    continue
 
-            # try and fetch from source cache
-            if source._get_consistency() < Consistency.CACHED and \
-                    self.__sourcecache.has_fetch_remotes() and \
-                    not self.__sourcecache.contains(source):
-                if self.__sourcecache.pull(source):
-                    return
+                # try and fetch from source cache
+                if source._get_consistency() < Consistency.CACHED and \
+                        self.__sourcecache.has_fetch_remotes():
+                    if self.__sourcecache.pull(source):
+                        continue
+
+                fetch_needed = True
 
         # We need to fetch original sources
-        for source in self.sources():
-            source_consistency = source._get_consistency()
-            if source_consistency != Consistency.CACHED:
-                source._fetch(previous_sources)
-            previous_sources.append(source)
+        if fetch_needed or fetch_original:
+            for source in self.sources():
+                source_consistency = source._get_consistency()
+                if source_consistency != Consistency.CACHED:
+                    source._fetch(previous_sources)
+                previous_sources.append(source)
 
         self.__cache_sources()
 
@@ -2284,12 +2299,22 @@ class Element(Plugin):
     # Check if sources are cached, generating the source key if it hasn't been
     def _source_cached(self):
         if self.__sources:
-            last_source = self.__sources[-1]
-            if not last_source._key:
-                last_source._generate_key(self.__sources[:-1])
-            return self._get_context().sourcecache.contains(last_source)
-        else:
-            return True
+            sourcecache = self._get_context().sourcecache
+
+            # Go through sources we'll cache generating keys
+            for ix, source in enumerate(self.__sources):
+                if not source._key:
+                    if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                        source._generate_key(self.__sources[:ix])
+                    else:
+                        source._generate_key([])
+
+            # Check all sources are in source cache
+            for source in self.__sources:
+                if not sourcecache.contains(source):
+                    return False
+
+        return True
 
     def _should_fetch(self, fetch_original=False):
         """ return bool of if we need to run the fetch stage for this element
@@ -2936,9 +2961,32 @@ class Element(Plugin):
     # Caches the sources into the local CAS
     #
     def __cache_sources(self):
-        sources = self.__sources
-        if sources and not self._source_cached():
-            sources[-1]._cache(sources[:-1])
+        if self.__sources and not self._source_cached():
+            last_requires_previous = 0
+            # commit all other sources by themselves
+            for ix, source in enumerate(self.__sources):
+                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                    self.__sourcecache.commit(source, self.__sources[last_requires_previous:ix])
+                    last_requires_previous = ix
+                else:
+                    self.__sourcecache.commit(source, [])
+
+    # __last_source_requires_previous
+    #
+    # This is the last source that requires previous sources to be cached.
+    # Sources listed after this will be cached separately.
+    #
+    # Returns:
+    #    (int): index of last source that requires previous sources
+    #
+    def __last_source_requires_previous(self):
+        if self.__last_source_requires_previous_ix is None:
+            last_requires_previous = 0
+            for ix, source in enumerate(self.__sources):
+                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+                    last_requires_previous = ix
+            self.__last_source_requires_previous_ix = last_requires_previous
+        return self.__last_source_requires_previous_ix
 
     # __update_state_recursively()
     #
diff --git a/buildstream/plugins/sources/patch.py b/buildstream/plugins/sources/patch.py
index 8e833b411..e42868264 100644
--- a/buildstream/plugins/sources/patch.py
+++ b/buildstream/plugins/sources/patch.py
@@ -52,6 +52,8 @@ from buildstream import utils
 class PatchSource(Source):
     # pylint: disable=attribute-defined-outside-init
 
+    BST_REQUIRES_PREVIOUS_SOURCES_STAGE = True
+
     def configure(self, node):
         self.path = self.node_get_project_path(node, 'path',
                                                check_is_file=True)
diff --git a/buildstream/source.py b/buildstream/source.py
index b1346eb1b..dd1dbd40a 100644
--- a/buildstream/source.py
+++ b/buildstream/source.py
@@ -286,6 +286,17 @@ class Source(Plugin):
     *Since: 1.4*
     """
 
+    BST_REQUIRES_PREVIOUS_SOURCES_STAGE = False
+    """Whether access to previous sources is required during cache
+
+    When set to True:
+      * All sources listed before current source in the given element will be
+        staged with the source when it's cached.
+      * This source can not be the first source for an element.
+
+    *Since: 1.4*
+    """
+
     def __init__(self, context, project, meta, *, alias_override=None, unique_id=None):
         provenance = _yaml.node_get_provenance(meta.config)
         super().__init__("{}-{}".format(meta.element_name, meta.element_index),
@@ -1029,8 +1040,9 @@ class Source(Plugin):
     def _generate_key(self, previous_sources):
         keys = [self._get_unique_key(True)]
 
-        for previous_source in previous_sources:
-            keys.append(previous_source._get_unique_key(True))
+        if self.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
+            for previous_source in previous_sources:
+                keys.append(previous_source._get_unique_key(True))
 
         self.__key = generate_key(keys)
 
diff --git a/tests/sources/local.py b/tests/sources/local.py
index 28ed8f5fc..f568fee78 100644
--- a/tests/sources/local.py
+++ b/tests/sources/local.py
@@ -139,7 +139,7 @@ def test_stage_file_exists(cli, datafiles):
     # Build, checkout
     result = cli.run(project=project, args=['build', 'target.bst'])
     result.assert_main_error(ErrorDomain.STREAM, None)
-    result.assert_task_error(ErrorDomain.SOURCE, 'ensure-stage-dir-fail')
+    result.assert_task_error(ErrorDomain.ELEMENT, "import-source-files-fail")
 
 
 @pytest.mark.datafiles(os.path.join(DATA_DIR, 'directory'))
diff --git a/tests/sources/previous_source_access/plugins/sources/foo_transform.py b/tests/sources/previous_source_access/plugins/sources/foo_transform.py
index 820946454..bec4f9913 100644
--- a/tests/sources/previous_source_access/plugins/sources/foo_transform.py
+++ b/tests/sources/previous_source_access/plugins/sources/foo_transform.py
@@ -18,6 +18,7 @@ class FooTransformSource(Source):
     # We need access to previous both at track time and fetch time
     BST_REQUIRES_PREVIOUS_SOURCES_TRACK = True
     BST_REQUIRES_PREVIOUS_SOURCES_FETCH = True
+    BST_REQUIRES_PREVIOUS_SOURCES_CACHE = True
 
     @property
     def mirror(self):