[Notes] [Git][BuildStream/buildstream][bst-1.2] 15 commits: tests/artifa

Tristan Van Berkom pushed to branch bst-1.2 at BuildStream / buildstream

Commits:

80e912e6

by Tristan Van Berkom at 2018-09-10T06:56:11Z

tests/artifactcache/expiry.py: Fix test case expectations.

f81637c5

by Tristan Van Berkom at 2018-09-10T06:56:11Z

_scheduler/queues: Mark build and pull queue as requiring shared access to the CACHE

This is what the whole resource.py thing was created for, the
cleanup job must have exclusive access to the cache, while the pull
and build jobs which result in adding new artifacts, must only require
shared access.

This is a part of #623

f074a8ac

by Tristan Van Berkom at 2018-09-10T06:56:11Z

_scheduler/resources.py: Set CACHE max resources to 0

This means there is no cap for shared resource requests.

Together with the previous commit, this causes the cleanup
job and the pull/build jobs to all require unlimited shared
access to the CACHE resource; while only the cleanup job
requires exclusive access to the resource.

This is a part of #623

d2f5c4f3

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_scheduler/scheduler.py: Make CacheSizeJob() not require the CACHE exclusively

This runs after every pull, and does not need the cache exclusively,
only the cleanup job requires the cache exclusively.

Without this, every time a cache_size job is queued, all pull and
build jobs need to complete before cache_size job can run exclusively,
which is not good.

This is a part of #623

eb9eb6db

by Tristan Van Berkom at 2018-09-10T06:56:12Z

element.py: Adding missing API documenting comment

8293d8da

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_artifactcache: Making public methods public

The artifact cache provides the following public methods for
external callers, but was hiding them away as if they are private.

  o ArtifactCache.add_artifact_size()

  o ArtifactCache.set_cache_size()

Mark these properly public

879d5117

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_scheduler: API fixes; _check_cache_size_real() -> check_cache_size()

Here we have a very private looking _check_cache_size_real() function
which no-one would ever want to call from outside of the _scheduler,
especially given it's `_real()` prefix we should look for another
outward facing API to use.

However this is not private to the scheduler, and is intended to
be called by the `Queue` implementations.

  o Renamed this to check_cache_size()

  o Moved it to the public API section of the Scheduler object

  o Added the missing API documenting comment

  o Also added the missing API documenting comment to the private
    `_run_cleanup()` callback which runs in response to completion
    of the cache size calculation job.

  o Also place the cleanup job logs into a cleanup subdirectory,
    for better symmetry with the cache_size jobs which now have
    their own subdirectory

2c898796

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_artifactcache/artifactcache.py: Removing unused variable.

The ArtifactCache._local variable used to exist in order to
use a special hack to allow absolute paths to a remote artifact
cache, this was all for the purpose of testing.

This has all gone away with the introduction of CAS, leaving behind
a stale variable.

55266475

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_artifactcache/artifactcache.py: Sealing away some the estimated size

Previously, the API contract was to expose the estimated_size variable
on the ArtifactCache instance for all to see, however it is only relevant
to the ArtifactCache abstract class code. Subclasses were informed to
update the estimated_size variable in their calculate_cache_size()
implementation.

To untangle this and hide away the estimated size, this commit
does the following:

  o Introduces ArtifactCache.compute_cache_size() API for external
    callers

  o ArtifactCache.compute_cache_size() calls the abstract method
    for the CasCache subclass to implement

  o ArtifactCache.compute_cache_size() updates the private
    estimated_size variable

  o All direct callers to ArtifactCache.calculate_cache_size(),
    have been updated to use the ArtifactCache.compute_cache_size()
    method instead, which takes care of updating anything local
    to the ArtifactCache abstract class code (the estimated_size)

af0bfaee

by Tristan Van Berkom at 2018-09-10T06:56:12Z

element.py: Remove _get_artifact_size()

There is no justification to hold onto this state here.

Instead, just make `Element._assemble()` return the size of the
artifact it cached, and localize handling of that return value in
the BuildQueue implementation where the value is observed.

222c5034

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_scheduler/jobs/cachesizejob.py: Use Platform.get_platform() public accessor

This was previously poking directly at the Platform._instance.

Also, use the name 'artifacts' to hold the artifact cache to
be consistent with other parts of the codebase, instead of
calling it 'cache'.

67fa96e1

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_scheduler/jobs/cleanupjob.py: Use Platform.get_platform() public accessor

This was previously poking directly at the Platform._instance.

Also, use the name 'artifacts' to hold the artifact cache to
be consistent with other parts of the codebase, instead of
calling it 'cache'.

303240ab

by Tristan Van Berkom at 2018-09-10T06:56:12Z

element.py: Remove _get_artifact_cache() accessor.

The artifact cache is anyway a singleton, the Element itself
has an internal handle on the artifact cache because it is
needed very frequently; but an artifact cache is not element
specific and should not be looked up by surrounding code
on a per element basis.

Updated _scheduler/queues/queue.py, _scheduler/queues/buildqueue.py
and _scheduler/jobs/elementjob.py to get the artifact cache directly
from the Platform

78499b7d

by Tristan Van Berkom at 2018-09-10T06:56:12Z

_artifactcache: There shalt be only one cache size

This does a lot of house cleaning, finally bringing cache
cleanup logic to a level of comprehensibility.

Changes in this commit include:

  o _artifactcache/artifactcache.py: _cache_size, _cache_quota and
    _cache_lower_threshold are now all private variables.

    get_approximate_cache_size() is now get_cache_size()

    Added get_quota_exceeded() for the purpose of safely checking
    if we have exceeded the quota.

    set_cache_size() now asserts that the passed size is not None,
    it is not acceptable to set a None size cache anymore.

  o _artifactcache/cascache.py: No longer set the ArtifactCache
    'cache_size' variable violently in the commit() method.

    Also the calculate_cache_size() method now unconditionally
    calculates the cache size, that is what it's for.

  o _scheduler/jobs/cachesizejob.py & _scheduler/jobs/cleanupjob.py:

    Now check the success status. Don't try to set the cache size
    in the case that the job was terminated.

  o _scheduler/jobs/elementjob.py & _scheduler/queues/queue.py:

    No longer passing around the cache size from child tasks,
    this happens only explicitly, not implicitly for all tasks.

  o _scheduler/queues/buildqueue.py & _scheduler/scheduler.py:

    Use get_quota_exceeded() accessor

This is a part of #623

32d0ce66

by Tristan Van Berkom at 2018-09-10T09:22:57Z

Merge branch 'tristan/fix-cache-exclusivity-1.2' into 'bst-1.2'

Tristan/fix cache exclusivity 1.2

See merge request BuildStream/buildstream!779

12 changed files:

buildstream/_artifactcache/artifactcache.py
buildstream/_artifactcache/cascache.py
buildstream/_scheduler/jobs/cachesizejob.py
buildstream/_scheduler/jobs/cleanupjob.py
buildstream/_scheduler/jobs/elementjob.py
buildstream/_scheduler/queues/buildqueue.py
buildstream/_scheduler/queues/pullqueue.py
buildstream/_scheduler/queues/queue.py
buildstream/_scheduler/resources.py
buildstream/_scheduler/scheduler.py
buildstream/element.py
tests/artifactcache/expiry.py

Changes:

buildstream/_artifactcache/artifactcache.py

@@ -81,19 +81,16 @@ ArtifactCacheSpec.__new__.__defaults__ = (None, None, None)
  class ArtifactCache():
      def __init__(self, context):
          self.context = context
 -        self.required_artifacts = set()
          self.extractdir = os.path.join(context.artifactdir, 'extract')
          self.tmpdir = os.path.join(context.artifactdir, 'tmp')
 -        self.estimated_size = None
+-
          self.global_remote_specs = []
          self.project_remote_specs = {}
 -        self._local = False
 -        self.cache_size = None
 -        self.cache_quota = None
 -        self.cache_lower_threshold = None
 +        self._required_artifacts = set()      # The artifacts required for this session
 +        self._cache_size = None               # The current cache size, sometimes it's an estimate
 +        self._cache_quota = None              # The cache quota
 +        self._cache_lower_threshold = None    # The target cache size for a cleanup
          os.makedirs(self.extractdir, exist_ok=True)
          os.makedirs(self.tmpdir, exist_ok=True)
@@ -212,8 +209,8 @@ class ArtifactCache():
              weak_key = element._get_cache_key(strength=_KeyStrength.WEAK)
              for key in (strong_key, weak_key):
 -                if key and key not in self.required_artifacts:
 -                    self.required_artifacts.add(key)
 +                if key and key not in self._required_artifacts:
 +                    self._required_artifacts.add(key)
                      # We also update the usage times of any artifacts
                      # we will be using, which helps preventing a
@@ -228,10 +225,16 @@ class ArtifactCache():
+     #
      # Clean the artifact cache as much as possible.
+     #
 +    # Returns:
 +    #    (int): The size of the cache after having cleaned up
 +    #
      def clean(self):
          artifacts = self.list_artifacts()
 -        while self.calculate_cache_size() >= self.cache_quota - self.cache_lower_threshold:
 +        # Do a real computation of the cache size once, just in case
 +        self.compute_cache_size()
++
 +        while self.get_cache_size() >= self._cache_lower_threshold:
              try:
                  to_remove = artifacts.pop(0)
              except IndexError:
@@ -245,7 +248,7 @@ class ArtifactCache():
                            "Please increase the cache-quota in {}."
                            .format(self.context.config_origin or default_conf))
 -                if self.calculate_cache_size() > self.cache_quota:
 +                if self.get_quota_exceeded():
                      raise ArtifactError("Cache too full. Aborting.",
                                          detail=detail,
                                          reason="cache-too-full")
@@ -253,46 +256,94 @@ class ArtifactCache():
                      break
              key = to_remove.rpartition('/')[2]
 -            if key not in self.required_artifacts:
 +            if key not in self._required_artifacts:
++
 +                # Remove the actual artifact, if it's not required.
                  size = self.remove(to_remove)
 -                if size:
 -                    self.cache_size -= size
++
 +                # Remove the size from the removed size
 +                self.set_cache_size(self._cache_size - size)
          # This should be O(1) if implemented correctly
 -        return self.calculate_cache_size()
 +        return self.get_cache_size()
 -    # get_approximate_cache_size()
 +    # compute_cache_size()
+     #
 -    # A cheap method that aims to serve as an upper limit on the
 -    # artifact cache size.
 +    # Computes the real artifact cache size by calling
 +    # the abstract calculate_cache_size() method.
+     #
 -    # The cache size reported by this function will normally be larger
 -    # than the real cache size, since it is calculated using the
 -    # pre-commit artifact size, but for very small artifacts in
 -    # certain caches additional overhead could cause this to be
 -    # smaller than, but close to, the actual size.
 +    # Returns:
 +    #    (int): The size of the artifact cache.
 +    #
 +    def compute_cache_size(self):
 +        self._cache_size = self.calculate_cache_size()
++
 +        return self._cache_size
++
 +    # add_artifact_size()
+     #
 -    # Nonetheless, in practice this should be safe to use as an upper
 -    # limit on the cache size.
 +    # Adds the reported size of a newly cached artifact to the
 +    # overall estimated size.
+     #
 -    # If the cache has built-in constant-time size reporting, please
 -    # feel free to override this method with a more accurate
 -    # implementation.
 +    # Args:
 +    #     artifact_size (int): The size to add.
 +    #
 +    def add_artifact_size(self, artifact_size):
 +        cache_size = self.get_cache_size()
 +        cache_size += artifact_size
++
 +        self.set_cache_size(cache_size)
++
 +    # get_cache_size()
 +    #
 +    # Fetches the cached size of the cache, this is sometimes
 +    # an estimate and periodically adjusted to the real size
 +    # when a cache size calculation job runs.
 +    #
 +    # When it is an estimate, the value is either correct, or
 +    # it is greater than the actual cache size.
+     #
      # Returns:
      #     (int) An approximation of the artifact cache size.
+     #
 -    def get_approximate_cache_size(self):
 -        # If we don't currently have an estimate, figure out the real
 -        # cache size.
 -        if self.estimated_size is None:
 +    def get_cache_size(self):
++
 +        # If we don't currently have an estimate, figure out the real cache size.
 +        if self._cache_size is None:
              stored_size = self._read_cache_size()
              if stored_size is not None:
 -                self.estimated_size = stored_size
 +                self._cache_size = stored_size
              else:
 -                self.estimated_size = self.calculate_cache_size()
 +                self.compute_cache_size()
++
 +        return self._cache_size
 -        return self.estimated_size
 +    # set_cache_size()
 +    #
 +    # Forcefully set the overall cache size.
 +    #
 +    # This is used to update the size in the main process after
 +    # having calculated in a cleanup or a cache size calculation job.
 +    #
 +    # Args:
 +    #     cache_size (int): The size to set.
 +    #
 +    def set_cache_size(self, cache_size):
++
 +        assert cache_size is not None
++
 +        self._cache_size = cache_size
 +        self._write_cache_size(self._cache_size)
++
 +    # get_quota_exceeded()
 +    #
 +    # Checks if the current artifact cache size exceeds the quota.
 +    #
 +    # Returns:
 +    #    (bool): True of the quota is exceeded
 +    #
 +    def get_quota_exceeded(self):
 +        return self.get_cache_size() > self._cache_quota
      ################################################
      # Abstract methods for subclasses to implement #
@@ -484,11 +535,8 @@ class ArtifactCache():
+     #
      # Return the real artifact cache size.
+     #
 -    # Implementations should also use this to update estimated_size.
 -    #
      # Returns:
 -    #
 -    # (int) The size of the artifact cache.
 +    #    (int): The size of the artifact cache.
+     #
      def calculate_cache_size(self):
          raise ImplError("Cache '{kind}' does not implement calculate_cache_size()"
@@ -535,39 +583,13 @@ class ArtifactCache():
          with self.context.timed_activity("Initializing remote caches", silent_nested=True):
              self.initialize_remotes(on_failure=remote_failed)
 -    # _add_artifact_size()
 -    #
 -    # Since we cannot keep track of the cache size between threads,
 -    # this method will be called by the main process every time a
 -    # process that added something to the cache finishes.
 -    #
 -    # This will then add the reported size to
 -    # ArtifactCache.estimated_size.
 -    #
 -    def _add_artifact_size(self, artifact_size):
 -        if not self.estimated_size:
 -            self.estimated_size = self.calculate_cache_size()
+-
 -        self.estimated_size += artifact_size
 -        self._write_cache_size(self.estimated_size)
+-
 -    # _set_cache_size()
 -    #
 -    # Similarly to the above method, when we calculate the actual size
 -    # in a child thread, we can't update it. We instead pass the value
 -    # back to the main thread and update it there.
 -    #
 -    def _set_cache_size(self, cache_size):
 -        self.estimated_size = cache_size
+-
 -        # set_cache_size is called in cleanup, where it may set the cache to None
 -        if self.estimated_size is not None:
 -            self._write_cache_size(self.estimated_size)
+-
      # _write_cache_size()
+     #
      # Writes the given size of the artifact to the cache's size file
+     #
 +    # Args:
 +    #    size (int): The size of the artifact cache to record
 +    #
      def _write_cache_size(self, size):
          assert isinstance(size, int)
          size_file_path = os.path.join(self.context.artifactdir, CACHE_SIZE_FILE)
@@ -579,6 +601,9 @@ class ArtifactCache():
      # Reads and returns the size of the artifact cache that's stored in the
      # cache's size file
+     #
 +    # Returns:
 +    #    (int): The size of the artifact cache, as recorded in the file
 +    #
      def _read_cache_size(self):
          size_file_path = os.path.join(self.context.artifactdir, CACHE_SIZE_FILE)
@@ -628,13 +653,13 @@ class ArtifactCache():
          stat = os.statvfs(artifactdir_volume)
          available_space = (stat.f_bsize * stat.f_bavail)
 -        cache_size = self.get_approximate_cache_size()
 +        cache_size = self.get_cache_size()
          # Ensure system has enough storage for the cache_quota
+         #
          # If cache_quota is none, set it to the maximum it could possibly be.
+         #
 -        # Also check that cache_quota is atleast as large as our headroom.
 +        # Also check that cache_quota is at least as large as our headroom.
+         #
          if cache_quota is None:  # Infinity, set to max system storage
              cache_quota = cache_size + available_space
@@ -660,8 +685,8 @@ class ArtifactCache():
          # if we end up writing more than 2G, but hey, this stuff is
          # already really fuzzy.
+         #
 -        self.cache_quota = cache_quota - headroom
 -        self.cache_lower_threshold = self.cache_quota / 2
 +        self._cache_quota = cache_quota - headroom
 +        self._cache_lower_threshold = self._cache_quota / 2
  # _configured_remote_artifact_cache_specs():

buildstream/_artifactcache/cascache.py

@@ -120,8 +120,6 @@ class CASCache(ArtifactCache):
          for ref in refs:
              self.set_ref(ref, tree)
 -        self.cache_size = None
+-
      def diff(self, element, key_a, key_b, *, subdir=None):
          ref_a = self.get_artifact_fullname(element, key_a)
          ref_b = self.get_artifact_fullname(element, key_b)
@@ -488,11 +486,7 @@ class CASCache(ArtifactCache):
              raise ArtifactError("Attempt to access unavailable artifact: {}".format(e)) from e
      def calculate_cache_size(self):
 -        if self.cache_size is None:
 -            self.cache_size = utils._get_dir_size(self.casdir)
 -            self.estimated_size = self.cache_size
+-
 -        return self.cache_size
 +        return utils._get_dir_size(self.casdir)
      # list_artifacts():
+     #

buildstream/_scheduler/jobs/cachesizejob.py

@@ -24,15 +24,19 @@ class CacheSizeJob(Job):
      def __init__(self, *args, complete_cb, **kwargs):
          super().__init__(*args, **kwargs)
          self._complete_cb = complete_cb
 -        self._cache = Platform._instance.artifactcache
++
 +        platform = Platform.get_platform()
 +        self._artifacts = platform.artifactcache
      def child_process(self):
 -        return self._cache.calculate_cache_size()
 +        return self._artifacts.compute_cache_size()
      def parent_complete(self, success, result):
 -        self._cache._set_cache_size(result)
 -        if self._complete_cb:
 -            self._complete_cb(result)
 +        if success:
 +            self._artifacts.set_cache_size(result)
++
 +            if self._complete_cb:
 +                self._complete_cb(result)
      def child_process_data(self):
          return {}

buildstream/_scheduler/jobs/cleanupjob.py

@@ -24,15 +24,19 @@ class CleanupJob(Job):
      def __init__(self, *args, complete_cb, **kwargs):
          super().__init__(*args, **kwargs)
          self._complete_cb = complete_cb
 -        self._cache = Platform._instance.artifactcache
++
 +        platform = Platform.get_platform()
 +        self._artifacts = platform.artifactcache
      def child_process(self):
 -        return self._cache.clean()
 +        return self._artifacts.clean()
      def parent_complete(self, success, result):
 -        self._cache._set_cache_size(result)
 -        if self._complete_cb:
 -            self._complete_cb()
 +        if success:
 +            self._artifacts.set_cache_size(result)
++
 +            if self._complete_cb:
 +                self._complete_cb()
      def child_process_data(self):
          return {}

buildstream/_scheduler/jobs/elementjob.py

@@ -109,13 +109,7 @@ class ElementJob(Job):
          data = {}
          workspace = self._element._get_workspace()
 -        artifact_size = self._element._get_artifact_size()
 -        cache_size = self._element._get_artifact_cache().calculate_cache_size()
+-
          if workspace is not None:
              data['workspace'] = workspace.to_dict()
 -        if artifact_size is not None:
 -            data['artifact_size'] = artifact_size
 -        data['cache_size'] = cache_size
          return data

buildstream/_scheduler/queues/buildqueue.py

@@ -20,6 +20,7 @@
  from . import Queue, QueueStatus
  from ..resources import ResourceType
 +from ..._platform import Platform
  # A queue which assembles elements
@@ -28,11 +29,10 @@ class BuildQueue(Queue):
      action_name = "Build"
      complete_name = "Built"
 -    resources = [ResourceType.PROCESS]
 +    resources = [ResourceType.PROCESS, ResourceType.CACHE]
      def process(self, element):
 -        element._assemble()
 -        return element._get_unique_id()
 +        return element._assemble()
      def status(self, element):
          # state of dependencies may have changed, recalculate element state
@@ -51,18 +51,22 @@ class BuildQueue(Queue):
          return QueueStatus.READY
 -    def _check_cache_size(self, job, element):
 -        if not job.child_data:
 -            return
 +    def _check_cache_size(self, job, element, artifact_size):
 -        artifact_size = job.child_data.get('artifact_size', False)
 +        # After completing a build job, add the artifact size
 +        # as returned from Element._assemble() to the estimated
 +        # artifact cache size
 +        #
 +        platform = Platform.get_platform()
 +        artifacts = platform.artifactcache
 -        if artifact_size:
 -            cache = element._get_artifact_cache()
 -            cache._add_artifact_size(artifact_size)
 +        artifacts.add_artifact_size(artifact_size)
 -            if cache.get_approximate_cache_size() > cache.cache_quota:
 -                self._scheduler._check_cache_size_real()
 +        # If the estimated size outgrows the quota, ask the scheduler
 +        # to queue a job to actually check the real cache size.
 +        #
 +        if artifacts.get_quota_exceeded():
 +            self._scheduler.check_cache_size()
      def done(self, job, element, result, success):
@@ -70,8 +74,8 @@ class BuildQueue(Queue):
              # Inform element in main process that assembly is done
              element._assemble_done()
 -        # This has to be done after _assemble_done, such that the
 -        # element may register its cache key as required
 -        self._check_cache_size(job, element)
 +            # This has to be done after _assemble_done, such that the
 +            # element may register its cache key as required
 +            self._check_cache_size(job, element, result)
          return True

buildstream/_scheduler/queues/pullqueue.py

@@ -29,7 +29,7 @@ class PullQueue(Queue):
      action_name = "Pull"
      complete_name = "Pulled"
 -    resources = [ResourceType.DOWNLOAD]
 +    resources = [ResourceType.DOWNLOAD, ResourceType.CACHE]
      def process(self, element):
          # returns whether an artifact was downloaded or not
@@ -62,7 +62,7 @@ class PullQueue(Queue):
          # Build jobs will check the "approximate" size first. Since we
          # do not get an artifact size from pull jobs, we have to
          # actually check the cache size.
 -        self._scheduler._check_cache_size_real()
 +        self._scheduler.check_cache_size()
          # Element._pull() returns True if it downloaded an artifact,
          # here we want to appear skipped if we did not download.

buildstream/_scheduler/queues/queue.py

@@ -300,8 +300,6 @@ class Queue():
          # Update values that need to be synchronized in the main task
          # before calling any queue implementation
          self._update_workspaces(element, job)
 -        if job.child_data:
 -            element._get_artifact_cache().cache_size = job.child_data.get('cache_size')
          # Give the result of the job to the Queue implementor,
          # and determine if it should be considered as processed

buildstream/_scheduler/resources.py

@@ -8,7 +8,7 @@ class ResourceType():
  class Resources():
      def __init__(self, num_builders, num_fetchers, num_pushers):
          self._max_resources = {
 -            ResourceType.CACHE: 1,
 +            ResourceType.CACHE: 0,
              ResourceType.DOWNLOAD: num_fetchers,
              ResourceType.PROCESS: num_builders,
              ResourceType.UPLOAD: num_pushers

buildstream/_scheduler/scheduler.py

@@ -239,6 +239,25 @@ class Scheduler():
          self._schedule_queue_jobs()
          self._sched()
 +    # check_cache_size():
 +    #
 +    # Queues a cache size calculation job, after the cache
 +    # size is calculated, a cleanup job will be run automatically
 +    # if needed.
 +    #
 +    # FIXME: This should ensure that only one cache size job
 +    #        is ever pending at a given time. If a cache size
 +    #        job is already running, it is correct to queue
 +    #        a new one, it is incorrect to have more than one
 +    #        of these jobs pending at a given time, though.
 +    #
 +    def check_cache_size(self):
 +        job = CacheSizeJob(self, 'cache_size', 'cache_size/cache_size',
 +                           resources=[ResourceType.CACHE,
 +                                      ResourceType.PROCESS],
 +                           complete_cb=self._run_cleanup)
 +        self.schedule_jobs([job])
++
      #######################################################
      #                  Local Private Methods              #
      #######################################################
@@ -314,26 +333,32 @@ class Scheduler():
          self.schedule_jobs(ready)
          self._sched()
 +    # _run_cleanup()
 +    #
 +    # Schedules the cache cleanup job if the passed size
 +    # exceeds the cache quota.
 +    #
 +    # Args:
 +    #    cache_size (int): The calculated cache size (ignored)
 +    #
 +    # NOTE: This runs in response to completion of the cache size
 +    #       calculation job lauched by Scheduler.check_cache_size(),
 +    #       which will report the calculated cache size.
 +    #
      def _run_cleanup(self, cache_size):
          platform = Platform.get_platform()
 -        if cache_size and cache_size < platform.artifactcache.cache_quota:
 +        artifacts = platform.artifactcache
++
 +        if not artifacts.get_quota_exceeded():
              return
 -        job = CleanupJob(self, 'cleanup', 'cleanup',
 +        job = CleanupJob(self, 'cleanup', 'cleanup/cleanup',
                           resources=[ResourceType.CACHE,
                                      ResourceType.PROCESS],
                           exclusive_resources=[ResourceType.CACHE],
                           complete_cb=None)
          self.schedule_jobs([job])
 -    def _check_cache_size_real(self):
 -        job = CacheSizeJob(self, 'cache_size', 'cache_size/cache_size',
 -                           resources=[ResourceType.CACHE,
 -                                      ResourceType.PROCESS],
 -                           exclusive_resources=[ResourceType.CACHE],
 -                           complete_cb=self._run_cleanup)
 -        self.schedule_jobs([job])
+-
      # _suspend_jobs()
+     #
      # Suspend all ongoing jobs.

buildstream/element.py

@@ -231,7 +231,6 @@ class Element(Plugin):
          self.__staged_sources_directory = None  # Location where Element.stage_sources() was called
          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_size = None             # The size of data committed to the artifact cache
          # hash tables of loaded artifact metadata, hashed by key
          self.__metadata_keys = {}                     # Strong and weak keys for this key
@@ -1454,6 +1453,9 @@ class Element(Plugin):
      #   - Call the public abstract methods for the build phase
      #   - Cache the resulting artifact
+     #
 +    # Returns:
 +    #    (int): The size of the newly cached artifact
 +    #
      def _assemble(self):
          # Assert call ordering
@@ -1573,12 +1575,14 @@ class Element(Plugin):
                  }), os.path.join(metadir, 'workspaced-dependencies.yaml'))
                  with self.timed_activity("Caching artifact"):
 -                    self.__artifact_size = utils._get_dir_size(assembledir)
 +                    artifact_size = utils._get_dir_size(assembledir)
                      self.__artifacts.commit(self, assembledir, self.__get_cache_keys_for_commit())
              # Finally cleanup the build dir
              cleanup_rootdir()
 +        return artifact_size
++
      # _pull_pending()
+     #
      # Check whether the artifact will be pulled.
@@ -1817,25 +1821,6 @@ class Element(Plugin):
          workspaces = self._get_context().get_workspaces()
          return workspaces.get_workspace(self._get_full_name())
 -    # _get_artifact_size()
 -    #
 -    # Get the size of the artifact produced by this element in the
 -    # current pipeline - if this element has not been assembled or
 -    # pulled, this will be None.
 -    #
 -    # Note that this is the size of an artifact *before* committing it
 -    # to the cache, the size on disk may differ. It can act as an
 -    # approximate guide for when to do a proper size calculation.
 -    #
 -    # Returns:
 -    #    (int|None): The size of the artifact
 -    #
 -    def _get_artifact_size(self):
 -        return self.__artifact_size
+-
 -    def _get_artifact_cache(self):
 -        return self.__artifacts
+-
      # _write_script():
+     #
      # Writes a script to the given directory.

tests/artifactcache/expiry.py

@@ -175,6 +175,22 @@ def test_keep_dependencies(cli, datafiles, tmpdir):
  # Assert that we never delete a dependency required for a build tree
 +#
 +# NOTE: This test expects that a build will fail if it attempts to
 +#       put more artifacts in the cache than the quota can hold,
 +#       and expects that the last two elements which don't fit into
 +#       the quota wont even be built.
 +#
 +#       In real life, this will not be the case, since once we reach
 +#       the estimated quota we launch a cache size calculation job and
 +#       only launch a cleanup job when the size is calculated; and
 +#       other build tasks will be scheduled while the cache size job
 +#       is running.
 +#
 +#       This test only passes because we configure `builders` to 1,
 +#       ensuring that the cache size job runs exclusively since it
 +#       also requires a compute resource (a "builder").
 +#
  @pytest.mark.datafiles(DATA_DIR)
  def test_never_delete_dependencies(cli, datafiles, tmpdir):
      project = os.path.join(datafiles.dirname, datafiles.basename)
@@ -183,6 +199,9 @@ def test_never_delete_dependencies(cli, datafiles, tmpdir):
      cli.configure({
          'cache': {
              'quota': 10000000
 +        },
 +        'scheduler': {
 +            'builders': 1
+         }
      })

[Notes] [Git][BuildStream/buildstream][bst-1.2] 15 commits: tests/artifactcache/expiry.py: Fix test case expectations.

Tristan Van Berkom pushed to branch bst-1.2 at BuildStream / buildstream

Commits:

12 changed files:

Changes: