diff --git a/src/buildstream/_artifactcache.py b/src/buildstream/_artifactcache.py index c8328f109..d1e4b392a 100644 --- a/src/buildstream/_artifactcache.py +++ b/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 diff --git a/src/buildstream/_scheduler/queues/buildqueue.py b/src/buildstream/_scheduler/queues/buildqueue.py index 4b3fadab6..9d499d4f6 100644 --- a/src/buildstream/_scheduler/queues/buildqueue.py +++ b/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): diff --git a/src/buildstream/_scheduler/queues/pullqueue.py b/src/buildstream/_scheduler/queues/pullqueue.py index 1bb3ed55c..4e3dc6e3e 100644 --- a/src/buildstream/_scheduler/queues/pullqueue.py +++ b/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: diff --git a/src/buildstream/_scheduler/queues/queue.py b/src/buildstream/_scheduler/queues/queue.py index 108786955..30fc64f82 100644 --- a/src/buildstream/_scheduler/queues/queue.py +++ b/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) diff --git a/src/buildstream/element.py b/src/buildstream/element.py index 626bd386d..e0b0723a4 100644 --- a/src/buildstream/element.py +++ b/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. diff --git a/tests/scheduler/__init__.py b/tests/scheduler/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/scheduler/deduplication.py b/tests/scheduler/deduplication.py new file mode 100644 index 000000000..1fb9b2a05 --- /dev/null +++ b/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"} diff --git a/tests/scheduler/deduplication/failure/elements/a.bst b/tests/scheduler/deduplication/failure/elements/a.bst new file mode 100644 index 000000000..4d9c562ef --- /dev/null +++ b/tests/scheduler/deduplication/failure/elements/a.bst @@ -0,0 +1 @@ +kind: fail diff --git a/tests/scheduler/deduplication/failure/elements/b.bst b/tests/scheduler/deduplication/failure/elements/b.bst new file mode 100644 index 000000000..4d9c562ef --- /dev/null +++ b/tests/scheduler/deduplication/failure/elements/b.bst @@ -0,0 +1 @@ +kind: fail diff --git a/tests/scheduler/deduplication/failure/plugins/fail.py b/tests/scheduler/deduplication/failure/plugins/fail.py new file mode 100644 index 000000000..90f3b319b --- /dev/null +++ b/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 diff --git a/tests/scheduler/deduplication/failure/project.conf b/tests/scheduler/deduplication/failure/project.conf new file mode 100644 index 000000000..15100c6cb --- /dev/null +++ b/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 diff --git a/tests/scheduler/deduplication/multi-stage/elements/a.bst b/tests/scheduler/deduplication/multi-stage/elements/a.bst new file mode 100644 index 000000000..ddabcca93 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/elements/a.bst @@ -0,0 +1,4 @@ +kind: stack + +depends: +- subproject-a.bst:hello.bst diff --git a/tests/scheduler/deduplication/multi-stage/elements/b.bst b/tests/scheduler/deduplication/multi-stage/elements/b.bst new file mode 100644 index 000000000..310454cbc --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/elements/b.bst @@ -0,0 +1,4 @@ +kind: stack + +depends: +- subproject-b.bst:hello.bst diff --git a/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst b/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst new file mode 100644 index 000000000..6664eeec6 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst @@ -0,0 +1,5 @@ +kind: junction + +sources: +- kind: local + path: subproject diff --git a/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst b/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst new file mode 100644 index 000000000..6664eeec6 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst @@ -0,0 +1,5 @@ +kind: junction + +sources: +- kind: local + path: subproject diff --git a/tests/scheduler/deduplication/multi-stage/project.conf b/tests/scheduler/deduplication/multi-stage/project.conf new file mode 100644 index 000000000..6224762b9 --- /dev/null +++ b/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 diff --git a/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst b/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst new file mode 100644 index 000000000..003b5bf39 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst @@ -0,0 +1,4 @@ +kind: import +sources: +- kind: local + path: files/hello.txt diff --git a/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt b/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt new file mode 100644 index 000000000..b6fc4c620 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt @@ -0,0 +1 @@ +hello \ No newline at end of file diff --git a/tests/scheduler/deduplication/multi-stage/subproject/project.conf b/tests/scheduler/deduplication/multi-stage/subproject/project.conf new file mode 100644 index 000000000..c6e13a045 --- /dev/null +++ b/tests/scheduler/deduplication/multi-stage/subproject/project.conf @@ -0,0 +1,3 @@ +name: deduplication-multi-stage-subproject +min-version: 2.0 +element-path: elements diff --git a/tests/scheduler/deduplication/simple/elements/a.bst b/tests/scheduler/deduplication/simple/elements/a.bst new file mode 100644 index 000000000..003b5bf39 --- /dev/null +++ b/tests/scheduler/deduplication/simple/elements/a.bst @@ -0,0 +1,4 @@ +kind: import +sources: +- kind: local + path: files/hello.txt diff --git a/tests/scheduler/deduplication/simple/elements/b.bst b/tests/scheduler/deduplication/simple/elements/b.bst new file mode 100644 index 000000000..003b5bf39 --- /dev/null +++ b/tests/scheduler/deduplication/simple/elements/b.bst @@ -0,0 +1,4 @@ +kind: import +sources: +- kind: local + path: files/hello.txt diff --git a/tests/scheduler/deduplication/simple/files/hello.txt b/tests/scheduler/deduplication/simple/files/hello.txt new file mode 100644 index 000000000..b6fc4c620 --- /dev/null +++ b/tests/scheduler/deduplication/simple/files/hello.txt @@ -0,0 +1 @@ +hello \ No newline at end of file diff --git a/tests/scheduler/deduplication/simple/project.conf b/tests/scheduler/deduplication/simple/project.conf new file mode 100644 index 000000000..83365e94e --- /dev/null +++ b/tests/scheduler/deduplication/simple/project.conf @@ -0,0 +1,6 @@ +name: deduplication-simple +min-version: 2.0 +element-path: elements + +variables: + build-root: /buildstream/shared