You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by ro...@apache.org on 2020/12/29 13:39:02 UTC
[buildstream] branch tiagogomes/issue-573 created (now 23b54e6)
This is an automated email from the ASF dual-hosted git repository.
root pushed a change to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git.
at 23b54e6 artifactcache: add some logging to cleanup operation
This branch includes the following new commits:
new 3bd589b artifactcache: return overhead required to store an artifact
new fccb722 artifactcache: dynamically update size all time
new 2d9cb0c artifactcache: replace set_cache_size()
new 372c48f scheduler: remove CacheSize job
new e2af2d5 artifactcache: ensure method is called from main process
new 23b54e6 artifactcache: add some logging to cleanup operation
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[buildstream] 05/06: artifactcache: ensure method is called from
main process
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit e2af2d519c9c6f3f87019f3c1f76a6995945d2e9
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Tue Sep 11 16:32:22 2018 +0100
artifactcache: ensure method is called from main process
Ensure that add_artifact_size() is called from the main process.
---
buildstream/_artifactcache/artifactcache.py | 2 ++
1 file changed, 2 insertions(+)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index 412a815..f1cd117 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -292,6 +292,8 @@ class ArtifactCache():
# artifact_size (int): The artifact size to add.
#
def add_artifact_size(self, artifact_size):
+ assert utils._is_main_process()
+
self._cache_size = self.get_cache_size() + artifact_size
self._write_cache_size(self._cache_size)
[buildstream] 03/06: artifactcache: replace set_cache_size()
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 2d9cb0c0360d8d10bdae387533e61a13da3f5f1f
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Tue Sep 11 15:10:40 2018 +0100
artifactcache: replace set_cache_size()
Replace set_cache_size() with subtract_cache_size(), for symmetry with
add_artifact_size().
---
buildstream/_artifactcache/artifactcache.py | 47 +++++++++++------------------
buildstream/_scheduler/jobs/cleanupjob.py | 2 +-
2 files changed, 18 insertions(+), 31 deletions(-)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index 047587b..ea8a323 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -234,7 +234,7 @@ class ArtifactCache():
# Clean the artifact cache as much as possible.
#
# Returns:
- # (int): The size of the cache after having cleaned up
+ # (int): Amount of bytes cleaned from the cache.
#
def clean(self):
artifacts = self.list_artifacts()
@@ -252,7 +252,7 @@ class ArtifactCache():
])
# Do a real computation of the cache size once, just in case
- self.compute_cache_size()
+ old_cache_size = self.compute_cache_size()
while self.get_cache_size() >= self._cache_lower_threshold:
try:
@@ -280,12 +280,9 @@ class ArtifactCache():
# Remove the actual artifact, if it's not required.
size = self.remove(to_remove)
+ 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.get_cache_size()
+ return old_cache_size - self._cache_size
# compute_cache_size()
#
@@ -302,17 +299,24 @@ class ArtifactCache():
# add_artifact_size()
#
- # Adds the reported size of a newly cached artifact to the
- # current cache size.
+ # Adds given artifact size to the cache size
#
# Args:
- # artifact_size (int): The size to add.
+ # artifact_size (int): The artifact size to add.
#
def add_artifact_size(self, artifact_size):
- cache_size = self.get_cache_size()
- cache_size += artifact_size
+ self._cache_size = self.get_cache_size() + artifact_size
+ self._write_cache_size(self._cache_size)
- self.set_cache_size(cache_size)
+ # subtract_artifact_size()
+ #
+ # Subtracts given artifact size from the cache size
+ #
+ # Args:
+ # artifact_size (int): The artifact size to subtract.
+ #
+ def subtract_artifact_size(self, artifact_size):
+ self.add_artifact_size(artifact_size * -1)
# get_cache_size()
#
@@ -330,23 +334,6 @@ class ArtifactCache():
return self._cache_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)
-
# has_quota_exceeded()
#
# Checks if the current artifact cache size exceeds the quota.
diff --git a/buildstream/_scheduler/jobs/cleanupjob.py b/buildstream/_scheduler/jobs/cleanupjob.py
index 399435a..1f1cb14 100644
--- a/buildstream/_scheduler/jobs/cleanupjob.py
+++ b/buildstream/_scheduler/jobs/cleanupjob.py
@@ -32,4 +32,4 @@ class CleanupJob(Job):
def parent_complete(self, success, result):
if success:
- self._artifacts.set_cache_size(result)
+ self._artifacts.subtract_artifact_size(result)
[buildstream] 04/06: scheduler: remove CacheSize job
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 372c48f22d328f8f8bf0a8c186c2909869a2f7bb
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Tue Sep 11 12:34:00 2018 +0100
scheduler: remove CacheSize job
Is no longer used now that we update the cache size dynamically all the
time.
Also the ArtifactCache.compute_cache_size() is no longer used externally
and can be removed.
---
buildstream/_artifactcache/artifactcache.py | 15 +----------
buildstream/_scheduler/jobs/__init__.py | 1 -
buildstream/_scheduler/jobs/cachesizejob.py | 42 -----------------------------
3 files changed, 1 insertion(+), 57 deletions(-)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index ea8a323..412a815 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -252,7 +252,7 @@ class ArtifactCache():
])
# Do a real computation of the cache size once, just in case
- old_cache_size = self.compute_cache_size()
+ old_cache_size = self._cache_size = self.calculate_cache_size()
while self.get_cache_size() >= self._cache_lower_threshold:
try:
@@ -284,19 +284,6 @@ class ArtifactCache():
return old_cache_size - self._cache_size
- # compute_cache_size()
- #
- # Computes the real artifact cache size by calling
- # the abstract calculate_cache_size() method.
- #
- # 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()
#
# Adds given artifact size to the cache size
diff --git a/buildstream/_scheduler/jobs/__init__.py b/buildstream/_scheduler/jobs/__init__.py
index 4b0b11d..e7e9575 100644
--- a/buildstream/_scheduler/jobs/__init__.py
+++ b/buildstream/_scheduler/jobs/__init__.py
@@ -18,5 +18,4 @@
# Tristan Maat <tr...@codethink.co.uk>
from .elementjob import ElementJob
-from .cachesizejob import CacheSizeJob
from .cleanupjob import CleanupJob
diff --git a/buildstream/_scheduler/jobs/cachesizejob.py b/buildstream/_scheduler/jobs/cachesizejob.py
deleted file mode 100644
index 68cd913..0000000
--- a/buildstream/_scheduler/jobs/cachesizejob.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# Copyright (C) 2018 Codethink Limited
-#
-# This program is free software; you can redistribute it and/or
-# modify it under the terms of the GNU Lesser General Public
-# License as published by the Free Software Foundation; either
-# version 2 of the License, or (at your option) any later version.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public
-# License along with this library. If not, see <http://www.gnu.org/licenses/>.
-#
-# Author:
-# Tristan Daniël Maat <tr...@codethink.co.uk>
-#
-from .job import Job
-from ..._platform import Platform
-
-
-class CacheSizeJob(Job):
- def __init__(self, *args, complete_cb, **kwargs):
- super().__init__(*args, **kwargs)
- self._complete_cb = complete_cb
-
- platform = Platform.get_platform()
- self._artifacts = platform.artifactcache
-
- def child_process(self):
- return self._artifacts.compute_cache_size()
-
- def parent_complete(self, success, result):
- if success:
- self._artifacts.set_cache_size(result)
-
- if self._complete_cb:
- self._complete_cb(result)
-
- def child_process_data(self):
- return {}
[buildstream] 02/06: artifactcache: dynamically update size all time
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit fccb722af621c4f6725d2162cb3659c70d75e2aa
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Tue Sep 11 15:17:47 2018 +0100
artifactcache: dynamically update size all time
Change pull jobs to dynamically update the cache size, as build jobs
were already doing it.
The pathways between pull and build jobs are similiar and the code can
be updated appropriately.
As the cache size is always dynamically updated, we no longer need to
run the CacheSize job.
Fixes #573.
---
buildstream/_artifactcache/artifactcache.py | 25 +++++---------
buildstream/_scheduler/queues/buildqueue.py | 36 ++++++--------------
buildstream/_scheduler/queues/pullqueue.py | 18 ++++++----
buildstream/_scheduler/scheduler.py | 53 +++++++----------------------
buildstream/element.py | 2 +-
5 files changed, 44 insertions(+), 90 deletions(-)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index ce93079..047587b 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -1,5 +1,5 @@
#
-# Copyright (C) 2017-2018 Codethink Limited
+# Copyright (C) 2018 Codethink Limited
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -16,6 +16,7 @@
#
# Authors:
# Tristan Maat <tr...@codethink.co.uk>
+# Tiago Gomes <ti...@codethink.co.uk>
import os
import string
@@ -88,7 +89,7 @@ class ArtifactCache():
self.project_remote_specs = {}
self._required_elements = set() # The elements required for this session
- self._cache_size = None # The current cache size, sometimes it's an estimate
+ self._cache_size = None # The current cache size
self._cache_quota = None # The cache quota
self._cache_lower_threshold = None # The target cache size for a cleanup
@@ -302,7 +303,7 @@ class ArtifactCache():
# add_artifact_size()
#
# Adds the reported size of a newly cached artifact to the
- # overall estimated size.
+ # current cache size.
#
# Args:
# artifact_size (int): The size to add.
@@ -315,25 +316,17 @@ class ArtifactCache():
# 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 the size of the artifact cache.
#
# Returns:
- # (int) An approximation of the artifact cache size.
+ # (int): The size of the artifact cache.
#
def get_cache_size(self):
+ if self._cache_size is None:
+ self._cache_size = self._read_cache_size()
- # 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._cache_size = stored_size
- else:
- self.compute_cache_size()
+ self._cache_size = self.calculate_cache_size()
return self._cache_size
diff --git a/buildstream/_scheduler/queues/buildqueue.py b/buildstream/_scheduler/queues/buildqueue.py
index 0c74b36..8478881 100644
--- a/buildstream/_scheduler/queues/buildqueue.py
+++ b/buildstream/_scheduler/queues/buildqueue.py
@@ -1,5 +1,5 @@
#
-# Copyright (C) 2016 Codethink Limited
+# Copyright (C) 2018 Codethink Limited
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -87,31 +87,17 @@ class BuildQueue(Queue):
return QueueStatus.READY
- def _check_cache_size(self, job, element, artifact_size):
-
- # 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
-
- artifacts.add_artifact_size(artifact_size)
-
- # If the estimated size outgrows the quota, ask the scheduler
- # to queue a job to actually check the real cache size.
- #
- if artifacts.has_quota_exceeded():
- self._scheduler.check_cache_size()
-
def done(self, job, element, result, success):
+ if not success:
+ return False
+
+ element._assemble_done()
- if success:
- # Inform element in main process that assembly is done
- element._assemble_done()
+ artifacts = Platform.get_platform().artifactcache
+ artifacts.add_artifact_size(result)
- # 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)
+ # This has to be done after _assemble_done, such that the
+ # element may register its cache key as required
+ self._scheduler.check_cache_size()
- return True
+ return success
diff --git a/buildstream/_scheduler/queues/pullqueue.py b/buildstream/_scheduler/queues/pullqueue.py
index e18967c..36a5401 100644
--- a/buildstream/_scheduler/queues/pullqueue.py
+++ b/buildstream/_scheduler/queues/pullqueue.py
@@ -1,5 +1,5 @@
#
-# Copyright (C) 2016 Codethink Limited
+# Copyright (C) 2018 Codethink Limited
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -21,6 +21,7 @@
# Local imports
from . import Queue, QueueStatus
from ..resources import ResourceType
+from ..._platform import Platform
# A queue which pulls element artifacts
@@ -52,18 +53,21 @@ class PullQueue(Queue):
else:
return QueueStatus.SKIP
- def done(self, _, element, result, success):
-
+ def done(self, job, element, result, success):
if not success:
return False
element._pull_done()
- # 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.
+ pulled, artifact_size = result
+
+ artifacts = Platform.get_platform().artifactcache
+ artifacts.add_artifact_size(artifact_size)
+
+ # This has to be done after _pull_done, such that the
+ # element may register its cache key as required
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.
- return result
+ return pulled
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index 8faf58e..cb820c9 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -1,5 +1,5 @@
#
-# Copyright (C) 2016 Codethink Limited
+# Copyright (C) 2018 Codethink Limited
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -28,7 +28,7 @@ from contextlib import contextmanager
# Local imports
from .resources import Resources, ResourceType
-from .jobs import CacheSizeJob, CleanupJob
+from .jobs import CleanupJob
from .._platform import Platform
@@ -243,22 +243,18 @@ class Scheduler():
# 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.
+ # Queues a cleanup job if the size of the artifact cache exceeded
+ # the quota
#
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])
+ artifacts = Platform.get_platform().artifactcache
+
+ if artifacts.has_quota_exceeded():
+ job = CleanupJob(self, 'cleanup', 'cleanup/cleanup',
+ resources=[ResourceType.CACHE,
+ ResourceType.PROCESS],
+ exclusive_resources=[ResourceType.CACHE])
+ self.schedule_jobs([job])
#######################################################
# Local Private Methods #
@@ -335,31 +331,6 @@ 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()
- artifacts = platform.artifactcache
-
- if not artifacts.has_quota_exceeded():
- return
-
- job = CleanupJob(self, 'cleanup', 'cleanup/cleanup',
- resources=[ResourceType.CACHE,
- ResourceType.PROCESS],
- exclusive_resources=[ResourceType.CACHE])
- self.schedule_jobs([job])
-
# _suspend_jobs()
#
# Suspend all ongoing jobs.
diff --git a/buildstream/element.py b/buildstream/element.py
index da28cf5..9cd289b 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -1761,7 +1761,7 @@ class Element(Plugin):
display_key = self._get_brief_display_key()
self.info("Downloaded artifact {}".format(display_key))
- return pulled
+ return pulled, artifact_size
# _skip_push():
#
[buildstream] 01/06: artifactcache: return overhead required to
store an artifact
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 3bd589b91d18db208699cb29120ffce516509720
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Sun Aug 12 02:01:49 2018 +0100
artifactcache: return overhead required to store an artifact
Track amount of bytes required to commit or pull an artifact and return
this data on pull() and commit().
The commit() return value is more accurate than computing the size of
the artifact staging directory, as it takes deduplication into account.
This will be used in a following commit to dynamically update the cache
size.
---
buildstream/_artifactcache/artifactcache.py | 9 +++-
buildstream/_artifactcache/cascache.py | 72 +++++++++++++++++++++--------
buildstream/_artifactcache/casserver.py | 2 +-
buildstream/element.py | 51 ++++++++++----------
buildstream/storage/_casbaseddirectory.py | 7 +--
5 files changed, 91 insertions(+), 50 deletions(-)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index eb7759d..ce93079 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -460,6 +460,10 @@ class ArtifactCache():
# content (str): The element's content directory
# keys (list): The cache keys to use
#
+ # Returns:
+ # (int): Disk size overhead in bytes required to cache the
+ # artifact
+ #
def commit(self, element, content, keys):
raise ImplError("Cache '{kind}' does not implement commit()"
.format(kind=type(self).__name__))
@@ -531,8 +535,9 @@ class ArtifactCache():
# progress (callable): The progress callback, if any
#
# Returns:
- # (bool): True if pull was successful, False if artifact was not available
- #
+ # (bool): True if pull was successful, False if artifact was not available
+ # (int): Disk size overhead in bytes required to cache the
+ # artifact
def pull(self, element, key, *, progress=None):
raise ImplError("Cache '{kind}' does not implement pull()"
.format(kind=type(self).__name__))
diff --git a/buildstream/_artifactcache/cascache.py b/buildstream/_artifactcache/cascache.py
index 9cf83a2..24c47ca 100644
--- a/buildstream/_artifactcache/cascache.py
+++ b/buildstream/_artifactcache/cascache.py
@@ -16,6 +16,7 @@
#
# Authors:
# Jürg Billeter <ju...@codethink.co.uk>
+# Tiago Gomes <ti...@codethink.co.uk>
import hashlib
import itertools
@@ -117,11 +118,13 @@ class CASCache(ArtifactCache):
def commit(self, element, content, keys):
refs = [self.get_artifact_fullname(element, key) for key in keys]
- tree = self._commit_directory(content)
+ tree, size = self._commit_directory(content)
for ref in refs:
self.set_ref(ref, tree)
+ return size
+
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)
@@ -239,12 +242,12 @@ class CASCache(ArtifactCache):
tree.hash = response.digest.hash
tree.size_bytes = response.digest.size_bytes
- self._fetch_directory(remote, tree)
+ size = self._fetch_directory(remote, tree)
self.set_ref(ref, tree)
# no need to pull from additional remotes
- return True
+ return True, size
except grpc.RpcError as e:
if e.code() != grpc.StatusCode.NOT_FOUND:
@@ -258,7 +261,7 @@ class CASCache(ArtifactCache):
remote.spec.url, element._get_brief_display_key())
))
- return False
+ return False, 0
def pull_tree(self, project, digest):
""" Pull a single Tree rather than an artifact.
@@ -437,6 +440,7 @@ class CASCache(ArtifactCache):
#
# Returns:
# (Digest): The digest of the added object
+ # (int): The amount of bytes required to store the object
#
# Either `path` or `buffer` must be passed, but not both.
#
@@ -465,22 +469,38 @@ class CASCache(ArtifactCache):
out.flush()
+ file_size = os.fstat(out.fileno()).st_size
+
digest.hash = h.hexdigest()
- digest.size_bytes = os.fstat(out.fileno()).st_size
+ digest.size_bytes = file_size
# Place file at final location
objpath = self.objpath(digest)
- os.makedirs(os.path.dirname(objpath), exist_ok=True)
+ dirpath = os.path.dirname(objpath)
+
+ # Track the increased size on the parent directory caused by
+ # adding a new entry, as these directories can contain a large
+ # number of files.
+ new_dir_size = 0
+ old_dir_size = 0
+ try:
+ os.makedirs(dirpath)
+ except FileExistsError:
+ old_dir_size = os.stat(dirpath).st_size
+ else:
+ new_dir_size = os.stat(dirpath).st_size
+
os.link(out.name, objpath)
+ new_dir_size = os.stat(dirpath).st_size - old_dir_size
except FileExistsError as e:
# We can ignore the failed link() if the object is already in the repo.
- pass
+ file_size = 0
except OSError as e:
raise ArtifactError("Failed to hash object: {}".format(e)) from e
- return digest
+ return digest, file_size + new_dir_size
# set_ref():
#
@@ -489,6 +509,8 @@ class CASCache(ArtifactCache):
# Args:
# ref (str): The name of the ref
#
+ # Note: as setting a ref has very low disk size overhead, don't
+ # bother to track this.
def set_ref(self, ref, tree):
refpath = self._refpath(ref)
os.makedirs(os.path.dirname(refpath), exist_ok=True)
@@ -678,8 +700,10 @@ class CASCache(ArtifactCache):
#
# Returns:
# (Digest): Digest object for the directory added.
+ # (int): Bytes required to cache local directory
#
def _commit_directory(self, path, *, dir_digest=None):
+ size = 0
directory = remote_execution_pb2.Directory()
for name in sorted(os.listdir(path)):
@@ -688,11 +712,11 @@ class CASCache(ArtifactCache):
if stat.S_ISDIR(mode):
dirnode = directory.directories.add()
dirnode.name = name
- self._commit_directory(full_path, dir_digest=dirnode.digest)
+ size += self._commit_directory(full_path, dir_digest=dirnode.digest)[1]
elif stat.S_ISREG(mode):
filenode = directory.files.add()
filenode.name = name
- self.add_object(path=full_path, digest=filenode.digest)
+ size += self.add_object(path=full_path, digest=filenode.digest)[1]
filenode.is_executable = (mode & stat.S_IXUSR) == stat.S_IXUSR
elif stat.S_ISLNK(mode):
symlinknode = directory.symlinks.add()
@@ -704,8 +728,10 @@ class CASCache(ArtifactCache):
else:
raise ArtifactError("Unsupported file type for {}".format(full_path))
- return self.add_object(digest=dir_digest,
- buffer=directory.SerializeToString())
+ dir_digest, dir_object_size = self.add_object(
+ digest=dir_digest, buffer=directory.SerializeToString())
+
+ return dir_digest, size + dir_object_size
def _get_subdir(self, tree, subdir):
head, name = os.path.split(subdir)
@@ -860,11 +886,15 @@ class CASCache(ArtifactCache):
# remote (Remote): The remote to use.
# dir_digest (Digest): Digest object for the directory to fetch.
#
+ # Returns:
+ # (int): Bytes required to cache fetched directory
+ #
def _fetch_directory(self, remote, dir_digest):
+ size = 0
objpath = self.objpath(dir_digest)
if os.path.exists(objpath):
# already in local cache
- return
+ return 0
with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
self._fetch_blob(remote, dir_digest, out)
@@ -883,18 +913,24 @@ class CASCache(ArtifactCache):
with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
self._fetch_blob(remote, filenode.digest, f)
- digest = self.add_object(path=f.name)
+ digest, obj_size = self.add_object(path=f.name)
+ size += obj_size
assert digest.hash == filenode.digest.hash
for dirnode in directory.directories:
- self._fetch_directory(remote, dirnode.digest)
+ size += self._fetch_directory(remote, dirnode.digest)
# Place directory blob only in final location when we've
# downloaded all referenced blobs to avoid dangling
# references in the repository.
- digest = self.add_object(path=out.name)
+ digest, obj_size = self.add_object(path=out.name)
+
assert digest.hash == dir_digest.hash
+ size += obj_size
+
+ return size
+
def _fetch_tree(self, remote, digest):
# download but do not store the Tree object
with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
@@ -916,13 +952,13 @@ class CASCache(ArtifactCache):
with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
self._fetch_blob(remote, filenode.digest, f)
- added_digest = self.add_object(path=f.name)
+ added_digest = self.add_object(path=f.name)[0]
assert added_digest.hash == filenode.digest.hash
# place directory blob only in final location when we've downloaded
# all referenced blobs to avoid dangling references in the repository
dirbuffer = directory.SerializeToString()
- dirdigest = self.add_object(buffer=dirbuffer)
+ dirdigest = self.add_object(buffer=dirbuffer)[0]
assert dirdigest.size_bytes == len(dirbuffer)
return dirdigest
diff --git a/buildstream/_artifactcache/casserver.py b/buildstream/_artifactcache/casserver.py
index 8c3ece2..5bdf7a5 100644
--- a/buildstream/_artifactcache/casserver.py
+++ b/buildstream/_artifactcache/casserver.py
@@ -210,7 +210,7 @@ class _ByteStreamServicer(bytestream_pb2_grpc.ByteStreamServicer):
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
return response
out.flush()
- digest = self.cas.add_object(path=out.name)
+ digest = self.cas.add_object(path=out.name)[0]
if digest.hash != client_digest.hash:
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
return response
diff --git a/buildstream/element.py b/buildstream/element.py
index 13d76db..da28cf5 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -1659,8 +1659,8 @@ class Element(Plugin):
}), os.path.join(metadir, 'workspaced-dependencies.yaml'))
with self.timed_activity("Caching artifact"):
- artifact_size = utils._get_dir_size(assembledir)
- self.__artifacts.commit(self, assembledir, self.__get_cache_keys_for_commit())
+ artifact_size = self.__artifacts.commit(
+ self, assembledir, self.__get_cache_keys_for_commit())
if collect is not None and collectvdir is None:
raise ElementError(
@@ -1712,31 +1712,31 @@ class Element(Plugin):
self._update_state()
def _pull_strong(self, *, progress=None):
- weak_key = self._get_cache_key(strength=_KeyStrength.WEAK)
-
key = self.__strict_cache_key
- if not self.__artifacts.pull(self, key, progress=progress):
- return False
+ pulled, artifact_size = self.__artifacts.pull(self, key,
+ progress=progress)
- # update weak ref by pointing it to this newly fetched artifact
- self.__artifacts.link_key(self, key, weak_key)
+ if pulled:
+ # update weak ref by pointing it to this newly fetched artifact
+ weak_key = self._get_cache_key(strength=_KeyStrength.WEAK)
+ self.__artifacts.link_key(self, key, weak_key)
- return True
+ return pulled, artifact_size
def _pull_weak(self, *, progress=None):
weak_key = self._get_cache_key(strength=_KeyStrength.WEAK)
+ pulled, artifact_size = self.__artifacts.pull(self, weak_key,
+ progress=progress)
- if not self.__artifacts.pull(self, weak_key, progress=progress):
- return False
-
- # extract strong cache key from this newly fetched artifact
- self._pull_done()
+ if pulled:
+ # extract strong cache key from this newly fetched artifact
+ self._pull_done()
- # create tag for strong cache key
- key = self._get_cache_key(strength=_KeyStrength.STRONG)
- self.__artifacts.link_key(self, weak_key, key)
+ # create tag for strong cache key
+ key = self._get_cache_key(strength=_KeyStrength.STRONG)
+ self.__artifacts.link_key(self, weak_key, key)
- return True
+ return pulled, artifact_size
# _pull():
#
@@ -1751,18 +1751,17 @@ class Element(Plugin):
self.status(message)
# Attempt to pull artifact without knowing whether it's available
- pulled = self._pull_strong(progress=progress)
+ pulled, artifact_size = self._pull_strong(progress=progress)
if not pulled and not self._cached() and not context.get_strict():
- pulled = self._pull_weak(progress=progress)
+ pulled, artifact_size = self._pull_weak(progress=progress)
- if not pulled:
- return False
+ if pulled:
+ # Notify successfull download
+ display_key = self._get_brief_display_key()
+ self.info("Downloaded artifact {}".format(display_key))
- # Notify successfull download
- display_key = self._get_brief_display_key()
- self.info("Downloaded artifact {}".format(display_key))
- return True
+ return pulled
# _skip_push():
#
diff --git a/buildstream/storage/_casbaseddirectory.py b/buildstream/storage/_casbaseddirectory.py
index d580635..a72116f 100644
--- a/buildstream/storage/_casbaseddirectory.py
+++ b/buildstream/storage/_casbaseddirectory.py
@@ -111,7 +111,7 @@ class CasBasedDirectory(Directory):
the parent).
"""
- self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())
+ self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())[0]
if caller:
old_dir = self._find_pb2_entry(caller.filename)
self.cas_cache.add_object(digest=old_dir.digest, buffer=caller.pb2_directory.SerializeToString())
@@ -130,9 +130,10 @@ class CasBasedDirectory(Directory):
self.index[entry.name].buildstream_object._recalculate_recursing_down(entry)
if parent:
- self.ref = self.cas_cache.add_object(digest=parent.digest, buffer=self.pb2_directory.SerializeToString())
+ self.ref = self.cas_cache.add_object(digest=parent.digest,
+ buffer=self.pb2_directory.SerializeToString())[0]
else:
- self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())
+ self.ref = self.cas_cache.add_object(buffer=self.pb2_directory.SerializeToString())[0]
# We don't need to do anything more than that; files were already added ealier, and symlinks are
# part of the directory structure.
[buildstream] 06/06: artifactcache: add some logging to cleanup
operation
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
root pushed a commit to branch tiagogomes/issue-573
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 23b54e60fd3189ef9f6125aa3cd8b56ef750e926
Author: Tiago Gomes <ti...@codethink.co.uk>
AuthorDate: Mon Aug 13 17:00:04 2018 +0100
artifactcache: add some logging to cleanup operation
---
buildstream/_artifactcache/artifactcache.py | 8 ++++++++
buildstream/_scheduler/scheduler.py | 3 ++-
2 files changed, 10 insertions(+), 1 deletion(-)
diff --git a/buildstream/_artifactcache/artifactcache.py b/buildstream/_artifactcache/artifactcache.py
index f1cd117..09b7149 100644
--- a/buildstream/_artifactcache/artifactcache.py
+++ b/buildstream/_artifactcache/artifactcache.py
@@ -281,6 +281,14 @@ class ArtifactCache():
# Remove the actual artifact, if it's not required.
size = self.remove(to_remove)
self._cache_size -= size
+ self._message(MessageType.DEBUG,
+ "Removed artifact {} ({})".format(
+ to_remove[:-(len(key) - self.context.log_key_length)],
+ utils._pretty_size(size)))
+
+ self._message(MessageType.INFO,
+ "New artifact cache size: {}".format(
+ utils._pretty_size(self._cache_size)))
return old_cache_size - self._cache_size
diff --git a/buildstream/_scheduler/scheduler.py b/buildstream/_scheduler/scheduler.py
index cb820c9..f66777d 100644
--- a/buildstream/_scheduler/scheduler.py
+++ b/buildstream/_scheduler/scheduler.py
@@ -250,7 +250,8 @@ class Scheduler():
artifacts = Platform.get_platform().artifactcache
if artifacts.has_quota_exceeded():
- job = CleanupJob(self, 'cleanup', 'cleanup/cleanup',
+ job = CleanupJob(self, 'Clean artifact cache',
+ 'cleanup/cleanup',
resources=[ResourceType.CACHE,
ResourceType.PROCESS],
exclusive_resources=[ResourceType.CACHE])