[Notes] [Git][BuildStream/buildstream][raoul/802-refactor-artifactcache]

Raoul Hidalgo Charman pushed to branch raoul/802-refactor-artifactcache at BuildStream / buildstream

Commits:

8221e090

by Raoul Hidalgo Charman at 2018-12-14T13:35:19Z

Add tmpdir which is passed to CASRemote

Part of #802

8d4b0bf9

by Raoul Hidalgo Charman at 2018-12-14T13:35:24Z

_cas: Move pull logic into CASRemote

Seperates the pull logic into a remote/local API , so that artifact cache
iterates over blob digests checks whether it has them, and then downloads them
if not. Currently doesn't use batching.

Part of #802

Changes:

buildstream/_artifactcache.py

@@ -28,7 +28,7 @@ from ._message import Message, MessageType
  from . import utils
  from . import _yaml
 -from ._cas.casremote import CASRemote, CASRemoteSpec
 +from ._cas.casremote import BlobNotFound, CASRemote, CASRemoteSpec
  CACHE_SIZE_FILE = "cache_size"
@@ -374,7 +374,7 @@ class ArtifactCache():
          q = multiprocessing.Queue()
          for remote_spec in remote_specs:
 -            error = CASRemote.check_remote(remote_spec, q)
 +            error = CASRemote.check_remote(remote_spec, self.context.tmpdir, q)
              if error and on_failure:
                  on_failure(remote_spec.url, error)
@@ -385,7 +385,7 @@ class ArtifactCache():
                  if remote_spec.push:
                      self._has_push_remotes = True
 -                remotes[remote_spec.url] = CASRemote(remote_spec)
 +                remotes[remote_spec.url] = CASRemote(remote_spec, self.context.tmpdir)
          for project in self.context.get_projects():
              remote_specs = self.global_remote_specs
@@ -634,7 +634,7 @@ class ArtifactCache():
      # Returns:
      #   (bool): True if pull was successful, False if artifact was not available
+     #
 -    def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=None):
 +    def pull(self, element, key, *, progress=None, subdir=None, excluded_subdirs=[]):
          ref = self.get_artifact_fullname(element, key)
          project = element._get_project()
@@ -644,13 +644,31 @@ class ArtifactCache():
                  display_key = element._get_brief_display_key()
                  element.status("Pulling artifact {} <- {}".format(display_key, remote.spec.url))
 -                if self.cas.pull(ref, remote, progress=progress, subdir=subdir, excluded_subdirs=excluded_subdirs):
 -                    element.info("Pulled artifact {} <- {}".format(display_key, remote.spec.url))
 +                root_digest = remote.get_reference(ref)
++
 +                if root_digest:
 +                    try:
 +                        for blob_digest in remote.yield_blob_digests(
 +                                root_digest, progress=progress, subdir=subdir,
 +                                excluded_subdirs=excluded_subdirs):
 +                            if self.cas.check_blob(blob_digest):
 +                                continue
 +                            remote.request_blob(blob_digest)
 +                            for blob_file in remote.get_blobs():
 +                                self.cas.add_object(path=blob_file.name, link_directly=True)
 +                        self.cas.set_ref(ref, root_digest)
 +                    except BlobNotFound:
 +                        element.info("Remote ({}) is missing blobs for {}".format(
 +                            remote.spec.url, element._get_brief_display_key()))
 +                        continue
++
                      if subdir:
                          # Attempt to extract subdir into artifact extract dir if it already exists
                          # without containing the subdir. If the respective artifact extract dir does not
                          # exist a complete extraction will complete.
                          self.extract(element, key, subdir)
++
 +                    element.info("Pulled artifact {} <- {}".format(display_key, remote.spec.url))
                      # no need to pull from additional remotes
                      return True
                  else:

buildstream/_cas/cascache.py

@@ -33,7 +33,7 @@ from .._protos.buildstream.v2 import buildstream_pb2
  from .. import utils
  from .._exceptions import CASError
 -from .casremote import BlobNotFound, _CASBatchRead, _CASBatchUpdate
 +from .casremote import _CASBatchUpdate
  # A CASCache manages a CAS repository as specified in the Remote Execution API.
@@ -183,50 +183,6 @@ class CASCache():
          return modified, removed, added
 -    # pull():
 -    #
 -    # Pull a ref from a remote repository.
 -    #
 -    # Args:
 -    #     ref (str): The ref to pull
 -    #     remote (CASRemote): The remote repository to pull from
 -    #     progress (callable): The progress callback, if any
 -    #     subdir (str): The optional specific subdir to pull
 -    #     excluded_subdirs (list): The optional list of subdirs to not pull
 -    #
 -    # Returns:
 -    #   (bool): True if pull was successful, False if ref was not available
 -    #
 -    def pull(self, ref, remote, *, progress=None, subdir=None, excluded_subdirs=None):
 -        try:
 -            remote.init()
+-
 -            request = buildstream_pb2.GetReferenceRequest()
 -            request.key = ref
 -            response = remote.ref_storage.GetReference(request)
+-
 -            tree = remote_execution_pb2.Digest()
 -            tree.hash = response.digest.hash
 -            tree.size_bytes = response.digest.size_bytes
+-
 -            # Check if the element artifact is present, if so just fetch the subdir.
 -            if subdir and os.path.exists(self.objpath(tree)):
 -                self._fetch_subdir(remote, tree, subdir)
 -            else:
 -                # Fetch artifact, excluded_subdirs determined in pullqueue
 -                self._fetch_directory(remote, tree, excluded_subdirs=excluded_subdirs)
+-
 -            self.set_ref(ref, tree)
+-
 -            return True
 -        except grpc.RpcError as e:
 -            if e.code() != grpc.StatusCode.NOT_FOUND:
 -                raise CASError("Failed to pull ref {}: {}".format(ref, e)) from e
 -            else:
 -                return False
 -        except BlobNotFound as e:
 -            return False
+-
      # pull_tree():
+     #
      # Pull a single Tree rather than a ref.
@@ -591,6 +547,16 @@ class CASCache():
          reachable = set()
          self._reachable_refs_dir(reachable, tree, update_mtime=True)
 +    # Check to see if a blob is in the local CAS
 +    # return None if not
 +    def check_blob(self, digest):
 +        objpath = self.objpath(digest)
 +        if os.path.exists(objpath):
 +            # already in local repository
 +            return objpath
 +        else:
 +            return None
++
      ################################################
      #             Local Private Methods            #
      ################################################
@@ -805,103 +771,6 @@ class CASCache():
          return objpath
 -    def _batch_download_complete(self, batch):
 -        for digest, data in batch.send():
 -            with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
 -                f.write(data)
 -                f.flush()
+-
 -                added_digest = self.add_object(path=f.name, link_directly=True)
 -                assert added_digest.hash == digest.hash
+-
 -    # Helper function for _fetch_directory().
 -    def _fetch_directory_batch(self, remote, batch, fetch_queue, fetch_next_queue):
 -        self._batch_download_complete(batch)
+-
 -        # All previously scheduled directories are now locally available,
 -        # move them to the processing queue.
 -        fetch_queue.extend(fetch_next_queue)
 -        fetch_next_queue.clear()
 -        return _CASBatchRead(remote)
+-
 -    # Helper function for _fetch_directory().
 -    def _fetch_directory_node(self, remote, digest, batch, fetch_queue, fetch_next_queue, *, recursive=False):
 -        in_local_cache = os.path.exists(self.objpath(digest))
+-
 -        if in_local_cache:
 -            # Skip download, already in local cache.
 -            pass
 -        elif (digest.size_bytes >= remote.max_batch_total_size_bytes or
 -              not remote.batch_read_supported):
 -            # Too large for batch request, download in independent request.
 -            self._ensure_blob(remote, digest)
 -            in_local_cache = True
 -        else:
 -            if not batch.add(digest):
 -                # Not enough space left in batch request.
 -                # Complete pending batch first.
 -                batch = self._fetch_directory_batch(remote, batch, fetch_queue, fetch_next_queue)
 -                batch.add(digest)
+-
 -        if recursive:
 -            if in_local_cache:
 -                # Add directory to processing queue.
 -                fetch_queue.append(digest)
 -            else:
 -                # Directory will be available after completing pending batch.
 -                # Add directory to deferred processing queue.
 -                fetch_next_queue.append(digest)
+-
 -        return batch
+-
 -    # _fetch_directory():
 -    #
 -    # Fetches remote directory and adds it to content addressable store.
 -    #
 -    # Fetches files, symbolic links and recursively other directories in
 -    # the remote directory and adds them to the content addressable
 -    # store.
 -    #
 -    # Args:
 -    #     remote (Remote): The remote to use.
 -    #     dir_digest (Digest): Digest object for the directory to fetch.
 -    #     excluded_subdirs (list): The optional list of subdirs to not fetch
 -    #
 -    def _fetch_directory(self, remote, dir_digest, *, excluded_subdirs=None):
 -        fetch_queue = [dir_digest]
 -        fetch_next_queue = []
 -        batch = _CASBatchRead(remote)
 -        if not excluded_subdirs:
 -            excluded_subdirs = []
+-
 -        while len(fetch_queue) + len(fetch_next_queue) > 0:
 -            if not fetch_queue:
 -                batch = self._fetch_directory_batch(remote, batch, fetch_queue, fetch_next_queue)
+-
 -            dir_digest = fetch_queue.pop(0)
+-
 -            objpath = self._ensure_blob(remote, dir_digest)
+-
 -            directory = remote_execution_pb2.Directory()
 -            with open(objpath, 'rb') as f:
 -                directory.ParseFromString(f.read())
+-
 -            for dirnode in directory.directories:
 -                if dirnode.name not in excluded_subdirs:
 -                    batch = self._fetch_directory_node(remote, dirnode.digest, batch,
 -                                                       fetch_queue, fetch_next_queue, recursive=True)
+-
 -            for filenode in directory.files:
 -                batch = self._fetch_directory_node(remote, filenode.digest, batch,
 -                                                   fetch_queue, fetch_next_queue)
+-
 -        # Fetch final batch
 -        self._fetch_directory_batch(remote, batch, fetch_queue, fetch_next_queue)
+-
 -    def _fetch_subdir(self, remote, tree, subdir):
 -        subdirdigest = self._get_subdir(tree, subdir)
 -        self._fetch_directory(remote, subdirdigest)
+-
      def _fetch_tree(self, remote, digest):
          # download but do not store the Tree object
          with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:

buildstream/_cas/casremote.py

@@ -3,6 +3,7 @@ import io
  import os
  import multiprocessing
  import signal
 +import tempfile
  from urllib.parse import urlparse
  import uuid
@@ -77,7 +78,7 @@ class BlobNotFound(CASError):
  # Represents a single remote CAS cache.
+ #
  class CASRemote():
 -    def __init__(self, spec):
 +    def __init__(self, spec, tmpdir):
          self.spec = spec
          self._initialized = False
          self.channel = None
@@ -89,6 +90,11 @@ class CASRemote():
          self.capabilities = None
          self.max_batch_total_size_bytes = None
 +        self.tmpdir = tmpdir
 +        os.makedirs(tmpdir, exist_ok=True)
++
 +        self.__tmp_downloads = []  # files in the tmpdir waiting to be added to local caches
++
      def init(self):
          if not self._initialized:
              url = urlparse(self.spec.url)
@@ -170,11 +176,11 @@ class CASRemote():
      # in the main BuildStream process
      # See https://github.com/grpc/grpc/blob/master/doc/fork_support.md for details
      @classmethod
 -    def check_remote(cls, remote_spec, q):
 +    def check_remote(cls, remote_spec, tmpdir, q):
          def __check_remote():
              try:
 -                remote = cls(remote_spec)
 +                remote = cls(remote_spec, tmpdir)
                  remote.init()
                  request = buildstream_pb2.StatusRequest()
@@ -252,6 +258,70 @@ class CASRemote():
          return message_digest
 +    # get_reference():
 +    #
 +    # Args:
 +    #    ref (str): The ref to request
 +    #
 +    def get_reference(self, ref):
 +        try:
 +            self.init()
++
 +            request = buildstream_pb2.GetReferenceRequest()
 +            request.key = ref
 +            return self.ref_storage.GetReference(request).digest
 +        except grpc.RpcError as e:
 +            if e.code() != grpc.StatusCode.NOT_FOUND:
 +                raise CASError("Failed to find ref {}: {}".format(ref, e)) from e
 +            else:
 +                return None
++
 +    # yield_blob_digests():
 +    #
 +    # Iterate over blobs digests from a reference
 +    #
 +    # Args:
 +    #     root_digest (str): The root_digest to get a tree of
 +    #     progress (callable): The progress callback, if any
 +    #     subdir (str): The optional specific subdir to pull
 +    #     excluded_subdirs (list): The optional list of subdirs to not pull
 +    #
 +    # Returns:
 +    #   (iter): True if pull was successful, False if ref was not available
 +    #
 +    def yield_blob_digests(self, root_digest, *, progress=None, subdir=None, excluded_subdirs=[]):
 +        self.init()
++
 +        # TODO add subdir stuff
 +        # Fetch artifact, excluded_subdirs determined in pullqueue
 +        yield from self._yield_directory_digests(root_digest, excluded_subdirs=excluded_subdirs)
++
 +    # request_blob():
 +    #
 +    # Request blob and returns path to tmpdir location
 +    #
 +    # Args:
 +    #    digest (Digest): digest of the requested blob
 +    #    path (str): tmpdir locations of downloaded blobs
 +    #
 +    def request_blob(self, digest):
 +        # TODO expand for adding to batches some other logic
 +        f = tempfile.NamedTemporaryFile(dir=self.tmpdir)
 +        self._fetch_blob(digest, f)
 +        self.__tmp_downloads.append(f)
 +        return f.name
++
 +    # get_blobs():
 +    #
 +    # Yield over downloaded blobs in the tmp file locations, causing the files
 +    # to be deleted once they go out of scope.
 +    #
 +    # Returns:
 +    #    iterator over NamedTemporaryFile
 +    def get_blobs(self):
 +        while self.__tmp_downloads:
 +            yield self.__tmp_downloads.pop()
++
      ################################################
      #             Local Private Methods            #
      ################################################
@@ -266,6 +336,36 @@ class CASRemote():
          assert digest.size_bytes == os.fstat(stream.fileno()).st_size
 +    # _yield_directory_digests():
 +    #
 +    # Fetches remote directory and adds it to content addressable store.
 +    #
 +    # Fetches files, symbolic links and recursively other directories in
 +    # the remote directory and adds them to the content addressable
 +    # store.
 +    #
 +    # Args:
 +    #     dir_digest (Digest): Digest object for the directory to fetch.
 +    #     excluded_subdirs (list): The optional list of subdirs to not fetch
 +    #
 +    def _yield_directory_digests(self, dir_digest, *, excluded_subdirs=[]):
++
 +        objpath = self.request_blob(dir_digest)
++
 +        directory = remote_execution_pb2.Directory()
++
 +        with open(objpath, 'rb') as f:
 +            directory.ParseFromString(f.read())
++
 +        yield dir_digest
 +        for filenode in directory.files:
 +            yield filenode.digest
++
 +        for dirnode in directory.directories:
 +            if dirnode.name not in excluded_subdirs:
 +                yield dirnode.digest
 +                yield from self._yield_directory_digests(dirnode.digest)
++
      def _send_blob(self, digest, stream, u_uid=uuid.uuid4()):
          resource_name = '/'.join(['uploads', str(u_uid), 'blobs',
                                    digest.hash, str(digest.size_bytes)])

buildstream/_context.py

@@ -182,10 +182,11 @@ class Context():
          _yaml.node_validate(defaults, [
              'sourcedir', 'builddir', 'artifactdir', 'logdir',
              'scheduler', 'artifacts', 'logging', 'projects',
 -            'cache', 'prompt', 'workspacedir',
 +            'cache', 'prompt', 'workspacedir', 'tmpdir'
          ])
 -        for directory in ['sourcedir', 'builddir', 'artifactdir', 'logdir', 'workspacedir']:
 +        for directory in ['sourcedir', 'builddir', 'artifactdir', 'logdir',
 +                          'tmpdir', 'workspacedir']:
              # Allow the ~ tilde expansion and any environment variables in
              # path specification in the config files.
+             #

buildstream/data/userconfig.yaml

@@ -19,6 +19,9 @@ builddir: ${XDG_CACHE_HOME}/buildstream/build
  # Location to store local binary artifacts
  artifactdir: ${XDG_CACHE_HOME}/buildstream/artifacts
 +# tmp directory, used by casremote
 +tmpdir: ${XDG_CACHE_HOME}/buildstream/tmp
++
  # Location to store build logs
  logdir: ${XDG_CACHE_HOME}/buildstream/logs

tests/testutils/runcli.py

@@ -495,7 +495,8 @@ def cli_integration(tmpdir, integration_cache):
      # to avoid downloading the huge base-sdk repeatedly
      fixture.configure({
          'sourcedir': os.path.join(integration_cache, 'sources'),
 -        'artifactdir': os.path.join(integration_cache, 'artifacts')
 +        'artifactdir': os.path.join(integration_cache, 'artifacts'),
 +        'tmpdir': os.path.join(integration_cache, 'tmp')
      })
      return fixture
@@ -539,6 +540,8 @@ def configured(directory, config=None):
          config['builddir'] = os.path.join(directory, 'build')
      if not config.get('artifactdir', False):
          config['artifactdir'] = os.path.join(directory, 'artifacts')
 +    if not config.get('tmpdir', False):
 +        config['tmpdir'] = os.path.join(directory, 'tmp')
      if not config.get('logdir', False):
          config['logdir'] = os.path.join(directory, 'logs')

[Notes] [Git][BuildStream/buildstream][raoul/802-refactor-artifactcache] 2 commits: Add tmpdir which is passed to CASRemote

Raoul Hidalgo Charman pushed to branch raoul/802-refactor-artifactcache at BuildStream / buildstream

Commits:

6 changed files:

Changes: