scheduler: deduplicate build and pull of elements with same cache key

src/buildstream/_artifactcache.py

@@ -238,6 +238,28 @@ def link_key(self, element, oldkey, newkey):
 
         utils.safe_link(os.path.join(self._basedir, oldref), os.path.join(self._basedir, newref))
 
+    # link_element_key():
+    #
+    # Add a reference to an existing artifact for another element
+    # with the same cache key.
+    #
+    # Args:
+    #     element (Element): The Element whose artifact is to be linked
+    #     other (Element): The Element to add the reference for
+    #     key (str): The cache key of the artifact
+    #
+    def link_element_key(self, element, other, key):
+        ref = element.get_artifact_name(key)
+        otherref = other.get_artifact_name(key)
+
+        if ref == otherref:
+            # The two refs are identical, nothing to do
+            return
+
+        dest = os.path.join(self._basedir, otherref)
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        utils.safe_link(os.path.join(self._basedir, ref), dest)
+
     # check_remotes_for_element()
     #
     # Check if the element is available in any of the remotes

src/buildstream/_scheduler/queues/buildqueue.py

@@ -38,7 +38,7 @@ def status(self, element):
         if not element._buildable():
             return QueueStatus.PENDING
 
-        return QueueStatus.READY
+        return self._deduplicate(element)
 
     def done(self, job, element, result, status):
 

src/buildstream/_scheduler/queues/pullqueue.py

@@ -34,11 +34,11 @@ def get_process_func(self):
         return PullQueue._pull_or_skip
 
     def status(self, element):
-        if element._pull_pending():
-            return QueueStatus.READY
-        else:
+        if not element._pull_pending():
             return QueueStatus.SKIP
 
+        return self._deduplicate(element)
+
     def done(self, _, element, result, status):
 
         if status is JobStatus.FAIL:

src/buildstream/_scheduler/queues/queue.py

@@ -32,7 +32,7 @@
 from ...types import FastEnum
 
 if TYPE_CHECKING:
-    from typing import List, Optional
+    from typing import Dict, List, Optional
 
 
 # Queue status for a given element
@@ -48,6 +48,21 @@ class QueueStatus(FastEnum):
     # The element is ready for processing in this queue.
     READY = 3
 
+    # Another element with the same strong cache key is being
+    # processed in this queue, see Queue._deduplicate().
+    WAIT = 4
+
+
+# _KeyGroup()
+#
+# Element instances sharing a strong cache key in a Queue
+#
+class _KeyGroup:
+    def __init__(self):
+        self.owner = None  # The element currently being processed
+        self.cached_element = None  # An element which successfully cached the artifact
+        self.waiters = []  # Elements waiting for the owner to complete
+
 
 # Queue()
 #
@@ -75,6 +90,7 @@ def __init__(self, scheduler, *, imperative=False):
         self._done_queue = deque()  # Processed / Skipped elements
         self._max_retries = 0
         self._queued_elements = 0  # Number of elements queued
+        self._key_groups = {}  # type: Dict[str, _KeyGroup]
 
         self._required_element_check = False  # Whether we should check that elements are required before enqueuing
 
@@ -366,6 +382,70 @@ def _job_done(self, job, element, status, result):
                 self._task_group.add_processed_task()
             else:
                 self._task_group.add_failed_task(element._get_full_name())
+        finally:
+            self._release_key_group(element, status == JobStatus.OK)
+
+    # _deduplicate()
+    #
+    # Ensures that only one of the Element instances sharing a strong
+    # cache key is processed at a time, e.g. the same element loaded
+    # through multiple junctions. Other instances wait for it, then
+    # either adopt its artifact or take over if it failed.
+    #
+    # This is called by Queue implementations from status() for
+    # elements which are otherwise ready to be processed.
+    #
+    # Args:
+    #    element (Element): The element which is ready
+    #
+    # Returns:
+    #    (QueueStatus): The status of the element
+    #
+    def _deduplicate(self, element):
+        key = element._get_cache_key()
+        if key is None:
+            return QueueStatus.READY
+
+        group = self._key_groups.setdefault(key, _KeyGroup())
+        if group.cached_element is not None:
+            element._adopt_artifact(group.cached_element)
+            return QueueStatus.SKIP
+        elif group.owner is not None and group.owner is not element:
+            return QueueStatus.WAIT
+
+        group.owner = element
+        return QueueStatus.READY
+
+    # _release_key_group()
+    #
+    # Re-enqueues elements waiting for the given element, see _deduplicate().
+    #
+    # Args:
+    #    element (Element): The element which completed
+    #    success (bool): Whether the job completed successfully
+    #
+    def _release_key_group(self, element, success):
+        group = self._key_groups.get(element._get_cache_key())
+        if group is None:
+            return
+
+        try:
+            if success and group.cached_element is None and element._cached_success():
+                group.cached_element = element
+
+            if group.owner is element:
+                group.owner = None
+                waiters, group.waiters = group.waiters, []
+                for waiter in waiters:
+                    self._enqueue_element(waiter)
+        except Exception:  # pylint: disable=broad-except
+            # Report unhandled exceptions, the job completion must not be interrupted
+            self._message(
+                element,
+                MessageType.BUG,
+                "Unhandled exception while releasing waiting elements",
+                detail=traceback.format_exc(),
+            )
 
     # Convenience wrapper for Queue implementations to send
     # a message for the element they are processing
@@ -406,6 +486,9 @@ def _enqueue_element(self, element):
             # Push elements which are ready to be processed immediately into the queue
             heapq.heappush(self._ready_queue, (element._depth, self._queued_elements, element))
             self._queued_elements += 1
+        elif status == QueueStatus.WAIT:
+            # Element is re-enqueued by _release_key_group()
+            self._key_groups[element._get_cache_key()].waiters.append(element)
         else:
             # Register a queue specific callback for pending elements
             self.register_pending_element(element)

src/buildstream/element.py

@@ -1661,6 +1661,23 @@ def _assemble_done(self, successful):
             workspace.last_build = key
             self._get_context().get_workspaces().save_config()
 
+    # _adopt_artifact():
+    #
+    # Adopt the cached artifact of another element with the same strong
+    # cache key instead of building or pulling it again.
+    #
+    # Args:
+    #    element (Element): The element which cached the artifact
+    #
+    def _adopt_artifact(self, element):
+        self.__artifacts.link_element_key(element, self, self._get_cache_key())
+
+        if self.__assemble_scheduled:
+            self._assemble_done(True)
+        else:
+            self._load_artifact(pull=False)
+            self._load_artifact_done()
+
     # _assemble():
     #
     # Internal method for running the entire build phase.

tests/scheduler/deduplication.py

@@ -0,0 +1,100 @@
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+# Pylint doesn't play well with fixtures and dependency injection from pytest
+# pylint: disable=redefined-outer-name
+
+import os
+import shutil
+import pytest
+
+from buildstream._testing import cli  # pylint: disable=unused-import
+from buildstream.exceptions import ErrorDomain
+
+from tests.testutils import create_artifact_share
+
+# Project directory
+DATA_DIR = os.path.join(
+    os.path.dirname(os.path.realpath(__file__)),
+    "deduplication",
+)
+
+
+# Elements sharing a strong cache key are built only once
+@pytest.mark.datafiles(DATA_DIR)
+@pytest.mark.parametrize("strict", [True, False], ids=["strict", "non-strict"])
+def test_build(cli, datafiles, strict):
+    project = os.path.join(str(datafiles), "simple")
+    cli.configure({"projects": {"deduplication-simple": {"strict": strict}}})
+
+    assert cli.get_element_key(project, "a.bst") == cli.get_element_key(project, "b.bst")
+
+    result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+    result.assert_success()
+    assert len(result.get_built_elements()) == 1
+
+    states = cli.get_element_states(project, ["a.bst", "b.bst"])
+    assert states == {"a.bst": "cached", "b.bst": "cached"}
+
+
+# The same element loaded through two junctions is built only once
+@pytest.mark.datafiles(DATA_DIR)
+def test_build_junctions(cli, datafiles):
+    project = os.path.join(str(datafiles), "multi-stage")
+
+    result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+    result.assert_success()
+
+    built = [element for element in result.get_built_elements() if element.endswith(":hello.bst")]
+    assert len(built) == 1
+
+    states = cli.get_element_states(project, ["subproject-a.bst:hello.bst", "subproject-b.bst:hello.bst"])
+    assert set(states.values()) == {"cached"}
+
+
+# When the build of an element fails, the elements waiting for it are built
+@pytest.mark.datafiles(DATA_DIR)
+def test_build_failure(cli, datafiles):
+    project = os.path.join(str(datafiles), "failure")
+
+    result = cli.run(project=project, args=["--on-error", "continue", "build", "a.bst", "b.bst"])
+    result.assert_main_error(ErrorDomain.STREAM, None)
+
+    assert sorted(result.get_start_order("build")) == ["a.bst", "b.bst"]
+
+    states = cli.get_element_states(project, ["a.bst", "b.bst"])
+    assert states == {"a.bst": "failed", "b.bst": "failed"}
+
+
+# Elements sharing a strong cache key are pulled only once
+@pytest.mark.datafiles(DATA_DIR)
+def test_pull(cli, tmpdir, datafiles):
+    project = os.path.join(str(datafiles), "simple")
+
+    with create_artifact_share(os.path.join(str(tmpdir), "artifactshare")) as share:
+        cli.configure({"artifacts": {"servers": [{"url": share.repo, "push": True}]}})
+
+        result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+        result.assert_success()
+
+        # Wipe the local cache so that the artifacts must be pulled
+        shutil.rmtree(os.path.join(cli.directory, "cas"))
+        shutil.rmtree(os.path.join(cli.directory, "artifacts"))
+
+        result = cli.run(project=project, args=["artifact", "pull", "a.bst", "b.bst"])
+        result.assert_success()
+        assert len(result.get_pulled_elements()) == 1
+
+        states = cli.get_element_states(project, ["a.bst", "b.bst"])
+        assert states == {"a.bst": "cached", "b.bst": "cached"}

tests/scheduler/deduplication/failure/elements/a.bst

@@ -0,0 +1 @@
+kind: fail

tests/scheduler/deduplication/failure/elements/b.bst

@@ -0,0 +1 @@
+kind: fail

tests/scheduler/deduplication/failure/plugins/fail.py

@@ -0,0 +1,28 @@
+from buildstream import Element, ElementError
+
+
+class Fail(Element):
+    BST_MIN_VERSION = "2.0"
+
+    def configure(self, node):
+        pass
+
+    def preflight(self):
+        pass
+
+    def get_unique_key(self):
+        return {}
+
+    def configure_sandbox(self, sandbox):
+        pass
+
+    def stage(self, sandbox):
+        pass
+
+    def assemble(self, sandbox):
+        raise ElementError("Failing on purpose", reason="fail")
+
+
+# Plugin entry point
+def setup():
+    return Fail

tests/scheduler/deduplication/failure/project.conf

@@ -0,0 +1,12 @@
+name: deduplication-failure
+min-version: 2.0
+element-path: elements
+
+variables:
+  build-root: /buildstream/shared
+
+plugins:
+- origin: local
+  path: plugins
+  elements:
+  - fail

tests/scheduler/deduplication/multi-stage/elements/a.bst

@@ -0,0 +1,4 @@
+kind: stack
+
+depends:
+- subproject-a.bst:hello.bst

tests/scheduler/deduplication/multi-stage/elements/b.bst

@@ -0,0 +1,4 @@
+kind: stack
+
+depends:
+- subproject-b.bst:hello.bst

tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst

@@ -0,0 +1,5 @@
+kind: junction
+
+sources:
+- kind: local
+  path: subproject

tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst

@@ -0,0 +1,5 @@
+kind: junction
+
+sources:
+- kind: local
+  path: subproject

tests/scheduler/deduplication/multi-stage/project.conf

@@ -0,0 +1,8 @@
+name: deduplication-multi-stage
+min-version: 2.0
+element-path: elements
+
+junctions:
+  internal:
+  - subproject-a.bst
+  - subproject-b.bst

tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst

@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt

tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt

@@ -0,0 +1 @@
+hello

tests/scheduler/deduplication/multi-stage/subproject/project.conf

@@ -0,0 +1,3 @@
+name: deduplication-multi-stage-subproject
+min-version: 2.0
+element-path: elements

tests/scheduler/deduplication/simple/elements/a.bst

@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt

tests/scheduler/deduplication/simple/elements/b.bst

@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt

tests/scheduler/deduplication/simple/files/hello.txt

@@ -0,0 +1 @@
+hello