[Notes] [Git][BuildStream/buildstream][jmac/remote_execution

Jim MacArthur pushed to branch jmac/remote_execution_rebase at BuildStream / buildstream

Commits:

a5a8472b

by Jim MacArthur at 2018-08-03T11:02:48Z

_casbaseddirectory: Alter _save to save the serialised digest object, not the text digest

eba5f422

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache: Restructure of push() - was e7e00999

ba56ae6e

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache.py: modify _required_blobs so non-Directory objects have no dependencies rather than throwing an exception

feab2af4

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache.py: yBetter way of managing dependencies (was a77938b6)

dae4fa9f

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache: Correction to earlier work - imports and info statement

00388453

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache: Reorder push/push_refs/verify, also fix return value of push

9061d051

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache: Add two methods to verify if a ref exists on the remote.

These are necessary to tell if something has previous been pushed.
push() returns false if the ref already exists on the remote,
so we need another method to verify existence on the remote.

bfa58c98

by Jim MacArthur at 2018-08-03T11:02:48Z

cascache.py: Add pull_key, a method to pull single objects without references.

8c840025

by Jim MacArthur at 2018-08-03T11:02:48Z

sandbox.py: Allow setting the virtual directory (for use after remote execution has finished)

27487d45

by Jim MacArthur at 2018-08-03T11:02:48Z

sandbox.py: Add set_output_directory to sandboxes and set it

5a3bab8b

by Jim MacArthur at 2018-08-03T11:02:48Z

sandbox.py: Fix error in comment for set_environment

b5661046
by Jim MacArthur at 2018-08-03T11:13:41Z
```
_sandboxremote.py: New file.
```

e5fdde84

by Jim MacArthur at 2018-08-03T11:13:41Z

_platform/linux.py: Make Remote the default sandbox for Linux

63fc7a08

by Jim MacArthur at 2018-08-03T12:04:37Z

Actually don't make Remote the default sandbox yet.

This reverts commit e5fdde84c3f3bb2843a1f10234cb817a2aa5e6d3.

840e867a

by Jim MacArthur at 2018-08-03T12:04:37Z

cascache.py: Make 'element' argument to push_refs optional

5 changed files:

buildstream/_artifactcache/cascache.py
buildstream/buildelement.py
+ buildstream/sandbox/_sandboxremote.py
buildstream/sandbox/sandbox.py
buildstream/storage/_casbaseddirectory.py

Changes:

buildstream/_artifactcache/cascache.py

@@ -213,6 +213,30 @@ class CASCache(ArtifactCache):
              remotes_for_project = self._remotes[element._get_project()]
              return any(remote.spec.push for remote in remotes_for_project)
++
 +    def pull_key(self, key, size_bytes, project):
 +        """ Pull a single key rather than an artifact.
 +        Does not update local refs. """
++
 +        for remote in self._remotes[project]:
 +            try:
 +                remote.init()
++
 +                tree = remote_execution_pb2.Digest()
 +                tree.hash = key
 +                tree.size_bytes = size_bytes
++
 +                self._fetch_directory(remote, tree)
++
 +                # no need to pull from additional remotes
 +                return True
++
 +            except grpc.RpcError as e:
 +                if e.code() != grpc.StatusCode.NOT_FOUND:
 +                    raise
++
 +        return False
++
      def pull(self, element, key, *, progress=None):
          ref = self.get_artifact_fullname(element, key)
@@ -254,10 +278,93 @@ class CASCache(ArtifactCache):
          self.set_ref(newref, tree)
 +    def _push_refs_to_remote(self, refs, remote, may_have_dependencies):
 +        skipped_remote = True
++
 +        try:
 +            for ref in refs:
 +                tree = self.resolve_ref(ref)
++
 +                # Check whether ref is already on the server in which case
 +                # there is no need to push the artifact
 +                try:
 +                    request = buildstream_pb2.GetReferenceRequest()
 +                    request.key = ref
 +                    response = remote.ref_storage.GetReference(request)
++
 +                    if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
 +                        # ref is already on the server with the same tree
 +                        continue
++
 +                except grpc.RpcError as e:
 +                    if e.code() != grpc.StatusCode.NOT_FOUND:
 +                        # Intentionally re-raise RpcError for outer except block.
 +                        raise
++
 +                missing_blobs = {}
 +                required_blobs = self._required_blobs(tree)
++
 +                # Limit size of FindMissingBlobs request
 +                for required_blobs_group in _grouper(required_blobs, 512):
 +                    request = remote_execution_pb2.FindMissingBlobsRequest()
++
 +                    for required_digest in required_blobs_group:
 +                        d = request.blob_digests.add()
 +                        d.hash = required_digest.hash
 +                        d.size_bytes = required_digest.size_bytes
++
 +                    response = remote.cas.FindMissingBlobs(request)
 +                    for digest in response.missing_blob_digests:
 +                        d = remote_execution_pb2.Digest()
 +                        d.hash = digest.hash
 +                        d.size_bytes = digest.size_bytes
 +                        missing_blobs[d.hash] = d
++
 +                # Upload any blobs missing on the server
 +                skipped_remote = False
 +                for digest in missing_blobs.values():
 +                    def request_stream():
 +                        resource_name = os.path.join(digest.hash, str(digest.size_bytes))
 +                        with open(self.objpath(digest), 'rb') as f:
 +                            assert os.fstat(f.fileno()).st_size == digest.size_bytes
 +                            offset = 0
 +                            finished = False
 +                            remaining = digest.size_bytes
 +                            while not finished:
 +                                chunk_size = min(remaining, 64 * 1024)
 +                                remaining -= chunk_size
++
 +                                request = bytestream_pb2.WriteRequest()
 +                                request.write_offset = offset
 +                                # max. 64 kB chunks
 +                                request.data = f.read(chunk_size)
 +                                request.resource_name = resource_name
 +                                request.finish_write = remaining <= 0
 +                                yield request
 +                                offset += chunk_size
 +                                finished = request.finish_write
 +                    response = remote.bytestream.Write(request_stream())
++
 +                request = buildstream_pb2.UpdateReferenceRequest()
 +                request.keys.append(ref)
 +                request.digest.hash = tree.hash
 +                request.digest.size_bytes = tree.size_bytes
 +                remote.ref_storage.UpdateReference(request)
++
 +        except grpc.RpcError as e:
 +            if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
 +                raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
++
 +        return not skipped_remote
++
      def push(self, element, keys):
 +        keys = list(keys)
          refs = [self.get_artifact_fullname(element, key) for key in keys]
          project = element._get_project()
 +        return self.push_refs(refs, project, element=element)
++
 +    def push_refs(self, refs, project, may_have_dependencies=True, element=None):
          push_remotes = [r for r in self._remotes[project] if r.spec.push]
@@ -265,94 +372,53 @@ class CASCache(ArtifactCache):
          for remote in push_remotes:
              remote.init()
 -            skipped_remote = True
 -            element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
+-
 -            try:
 -                for ref in refs:
 -                    tree = self.resolve_ref(ref)
+-
 -                    # Check whether ref is already on the server in which case
 -                    # there is no need to push the artifact
 -                    try:
 -                        request = buildstream_pb2.GetReferenceRequest()
 -                        request.key = ref
 -                        response = remote.ref_storage.GetReference(request)
+-
 -                        if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
 -                            # ref is already on the server with the same tree
 -                            continue
+-
 -                    except grpc.RpcError as e:
 -                        if e.code() != grpc.StatusCode.NOT_FOUND:
 -                            # Intentionally re-raise RpcError for outer except block.
 -                            raise
+-
 -                    missing_blobs = {}
 -                    required_blobs = self._required_blobs(tree)
+-
 -                    # Limit size of FindMissingBlobs request
 -                    for required_blobs_group in _grouper(required_blobs, 512):
 -                        request = remote_execution_pb2.FindMissingBlobsRequest()
+-
 -                        for required_digest in required_blobs_group:
 -                            d = request.blob_digests.add()
 -                            d.hash = required_digest.hash
 -                            d.size_bytes = required_digest.size_bytes
+-
 -                        response = remote.cas.FindMissingBlobs(request)
 -                        for digest in response.missing_blob_digests:
 -                            d = remote_execution_pb2.Digest()
 -                            d.hash = digest.hash
 -                            d.size_bytes = digest.size_bytes
 -                            missing_blobs[d.hash] = d
+-
 -                    # Upload any blobs missing on the server
 -                    skipped_remote = False
 -                    for digest in missing_blobs.values():
 -                        def request_stream():
 -                            resource_name = os.path.join(digest.hash, str(digest.size_bytes))
 -                            with open(self.objpath(digest), 'rb') as f:
 -                                assert os.fstat(f.fileno()).st_size == digest.size_bytes
 -                                offset = 0
 -                                finished = False
 -                                remaining = digest.size_bytes
 -                                while not finished:
 -                                    chunk_size = min(remaining, 64 * 1024)
 -                                    remaining -= chunk_size
+-
 -                                    request = bytestream_pb2.WriteRequest()
 -                                    request.write_offset = offset
 -                                    # max. 64 kB chunks
 -                                    request.data = f.read(chunk_size)
 -                                    request.resource_name = resource_name
 -                                    request.finish_write = remaining <= 0
 -                                    yield request
 -                                    offset += chunk_size
 -                                    finished = request.finish_write
 -                        response = remote.bytestream.Write(request_stream())
+-
 -                    request = buildstream_pb2.UpdateReferenceRequest()
 -                    request.keys.append(ref)
 -                    request.digest.hash = tree.hash
 -                    request.digest.size_bytes = tree.size_bytes
 -                    remote.ref_storage.UpdateReference(request)
+-
 -                    pushed = True
+-
 -            except grpc.RpcError as e:
 -                if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
 -                    raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
+-
 -            if skipped_remote:
 +            if self._push_refs_to_remote(refs, remote, may_have_dependencies):
 +                pushed = True
 +            elif element:
                  self.context.message(Message(
                      None,
                      MessageType.SKIPPED,
                      "Remote ({}) already has {} cached".format(
                          remote.spec.url, element._get_brief_display_key())
                  ))
++
 +        return pushed
++
 +    def verify_key_pushed(self, key, project):
 +        ref = key
 +        push_remotes = [r for r in self._remotes[project] if r.spec.push]
++
 +        pushed = False
++
 +        for remote in push_remotes:
 +            remote.init()
++
 +            if self._verify_ref_on_remote(ref, remote):
 +                pushed = True
++
          return pushed
 +    def _verify_ref_on_remote(self, ref, remote):
 +        pushed = False
 +        tree = self.resolve_ref(ref)
++
 +        # Check whether ref is already on the server in which case
 +        # there is no need to push the artifact
 +        try:
 +            request = buildstream_pb2.GetArtifactRequest()
 +            request.key = ref
 +            response = remote.artifact_cache.GetArtifact(request)
++
 +            if response.artifact.hash == tree.hash and response.artifact.size_bytes == tree.size_bytes:
 +                # ref is already on the server with the same tree
 +                return True
++
 +        except grpc.RpcError as e:
 +            if e.code() != grpc.StatusCode.NOT_FOUND:
 +                raise
++
 +        return False
++
      ################################################
      #                API Private Methods           #
      ################################################
@@ -726,26 +792,27 @@ class CASCache(ArtifactCache):
+             #
              q.put(str(e))
 -    def _required_blobs(self, tree):
 +    def _required_blobs(self, tree, may_have_dependencies=True):
          # parse directory, and recursively add blobs
          d = remote_execution_pb2.Digest()
          d.hash = tree.hash
          d.size_bytes = tree.size_bytes
          yield d
 -        directory = remote_execution_pb2.Directory()
 +        if may_have_dependencies:
 +            directory = remote_execution_pb2.Directory()
 -        with open(self.objpath(tree), 'rb') as f:
 -            directory.ParseFromString(f.read())
 +            with open(self.objpath(tree), 'rb') as f:
 +                directory.ParseFromString(f.read())
 -        for filenode in directory.files:
 -            d = remote_execution_pb2.Digest()
 -            d.hash = filenode.digest.hash
 -            d.size_bytes = filenode.digest.size_bytes
 -            yield d
 +            for filenode in directory.files:
 +                d = remote_execution_pb2.Digest()
 +                d.hash = filenode.digest.hash
 +                d.size_bytes = filenode.digest.size_bytes
 +                yield d
 -        for dirnode in directory.directories:
 -            yield from self._required_blobs(dirnode.digest)
 +            for dirnode in directory.directories:
 +                yield from self._required_blobs(dirnode.digest)
      def _fetch_blob(self, remote, digest, out):
          resource_name = os.path.join(digest.hash, str(digest.size_bytes))

buildstream/buildelement.py

@@ -155,6 +155,9 @@ class BuildElement(Element):
              command_dir = build_root
          sandbox.set_work_directory(command_dir)
 +        # Tell sandbox which directory is preserved in the finished artifact
 +        sandbox.set_output_directory(install_root)
++
          # Setup environment
          sandbox.set_environment(self.get_environment())

buildstream/sandbox/_sandboxremote.py

 +#!/usr/bin/env python3
 +#
 +#  Copyright (C) 2016 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/>.
 +#
 +#  Authors:
 +#        Andrew Leeming <andrew leeming codethink co uk>
 +#        Tristan Van Berkom <tristan vanberkom codethink co uk>
 +import os
 +import sys
 +import time
 +import errno
 +import signal
 +import subprocess
 +import shutil
 +from contextlib import ExitStack
++
 +import grpc
 +import psutil
++
 +from .. import utils, _signals
 +from ._mount import MountMap
 +from . import Sandbox, SandboxFlags
 +from ..storage._filebaseddirectory import FileBasedDirectory
 +from ..storage._casbaseddirectory import CasBasedDirectory
 +from .._protos.google.bytestream import bytestream_pb2, bytestream_pb2_grpc
 +from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
 +from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
 +from .._protos.google.longrunning import operations_pb2, operations_pb2_grpc
++
 +from .._artifactcache.cascache import CASCache
++
 +class SandboxError(Exception):
 +    pass
++
 +# SandboxRemote()
 +#
 +# This isn't really a sandbox, it's a stub which sends all the source to a remote server and retrieves the results from it.
 +#
 +class SandboxRemote(Sandbox):
++
 +    def __init__(self, *args, **kwargs):
 +        super().__init__(*args, **kwargs)
 +        self.user_ns_available = kwargs['user_ns_available']
 +        self.die_with_parent_available = kwargs['die_with_parent_available']
 +        self.cascache = None
++
 +    def _get_cascache(self):
 +        if self.cascache is None:
 +            self.cascache = CASCache(self._get_context())
 +            self.cascache.setup_remotes(use_config=True)
 +        return self.cascache
++
 +    def __run_remote_command(self, cascache, command, input_root_digest, environment):
 +        environment_variables = []
 +        for(k,v) in environment.items():
 +            environment_variables.append(remote_execution_pb2.Command.EnvironmentVariable(name=k, value=v))
 +        remote_command = remote_execution_pb2.Command(arguments=command, environment_variables = environment_variables)
++
 +        # Serialise this into the cascache...
 +        command_digest = cascache.add_object(buffer=remote_command.SerializeToString())
++
 +        command_ref = 'worker-command/{}'.format(command_digest.hash)
 +        cascache.set_ref(command_ref, command_digest)
++
 +        command_push_successful = cascache.push_refs([command_ref], self._get_project(), may_have_dependencies=False)
 +        if command_push_successful or cascache.verify_key_pushed(command_ref, self._get_project()):
 +            # Next, try to create a communication channel
 +            port = 50051
 +            channel = grpc.insecure_channel('dekatron.office.codethink.co.uk:{}'.format(port))
 +            stub = remote_execution_pb2_grpc.ExecutionStub(channel)
 +            ops_stub = operations_pb2_grpc.OperationsStub(channel)
++
 +            # Having done that, create and send the action.
++
 +            action = remote_execution_pb2.Action(command_digest = command_digest,
 +                                                 input_root_digest = input_root_digest,
 +                                                 output_files = [],
 +                                                 output_directories = [self._output_directory],
 +                                                 platform = None,
 +                                                 timeout = None,
 +                                                 do_not_cache = True)
++
 +            request = remote_execution_pb2.ExecuteRequest(instance_name = 'default',
 +                                                          action = action,
 +                                                          skip_cache_lookup = True)
++
 +            operation = stub.Execute(request) # Returns Operation
 +            job_name = operation.name
 +        else:
 +            # Source push failed
 +            return None
 +        while True:
 +            # TODO: Timeout
 +            # Refresh the operation data periodically using the name
 +            request = operations_pb2.GetOperationRequest(name=job_name)
 +            operation = ops_stub.GetOperation(request)
 +            sys.stderr.write("Operation {} is in stage <{}>\n".format(operation.name, operation.metadata))
 +            sys.stderr.write("......... {} has response <{}>\n".format(operation.name, operation.response))
 +            time.sleep(1)
 +            if operation.done:
 +                break
 +        return operation
++
 +    """ output_directories is an array of OutputDirectory objects
 +    output_files is an array of OutputFile objects """
 +    def process_job_output(self, output_directories, output_files):
 +        # We only specify one output_directory, so it's an error
 +        # for there to be any output files or more than one directory at the moment.
++
 +        if len(output_files)>0:
 +            raise SandboxError("Output files were returned when we didn't request any.")
 +        if len(output_directories)>1:
 +            raise SandboxError("More than one output directory was returned from the build server: {}".format(output_directories))
++
 +        digest = output_directories[0].tree_digest
 +        if digest is None or digest.hash is None or digest.hash=="":
 +            raise SandboxError("Output directory structure had no digest attached.")
++
 +        # Now do a pull to ensure we have the necessary parts.
 +        cascache = self._get_cascache()
 +        cascache.pull_key(digest.hash, digest.size_bytes, self._get_project())
 +        path_components = os.path.split(self._output_directory)
++
 +        # Now what we have is a digest for the output. Once we return, the calling process will
 +        # attempt to descend into our directory and find that directory, so we need to overwrite
 +        # that.
++
 +        if len(path_components)==0:
 +            # The artifact wants the whole directory; we could just return the returned hash in its
 +            # place, but we don't have a means to do that yet.
 +            raise SandboxError("Unimplemented: Output directory is empty or equal to the sandbox root.")
++
 +        # At the moment, we will get the whole directory back in the first directory argument and we need
 +        # to replace the sandbox's virtual directory with that. Creating a new virtual directory object
 +        # from another hash will be interesting, though...
++
 +        new_dir = CasBasedDirectory(self._get_context(), ref=digest)
 +        self.set_virtual_directory(new_dir)
++
 +    def run(self, command, flags, *, cwd=None, env=None):
 +        stdout, stderr = self._get_output()
 +        sys.stderr.write("Attempting run with remote sandbox...\n")
 +        # Upload sources
 +        upload_vdir = self.get_virtual_directory()
 +        if isinstance(upload_vdir, FileBasedDirectory):
 +            # Make a new temporary directory to put source in
 +            upload_vdir = CasBasedDirectory(self._get_context(), ref=None)
 +            upload_vdir.import_files(self.get_virtual_directory().get_underlying_directory())
++
 +        # Now, push that key (without necessarily needing a ref) to the remote.
 +        cascache = self._get_cascache()
++
 +        ref = 'worker-source/{}'.format(upload_vdir.ref.hash)
 +        upload_vdir._save(ref)
 +        source_push_successful = cascache.push_refs([ref], self._get_project())
 +        # Fallback to the sandbox default settings for
 +        # the cwd and environment.
++
 +        if env is None:
 +            env = self._get_environment()
++
 +        # We want command args as a list of strings
 +        if isinstance(command, str):
 +            command = [command]
++
 +        # Now transmit the command to execute
 +        if source_push_successful or cascache.verify_key_pushed(ref, self._get_project()):
 +            response = self.__run_remote_command(cascache, command, upload_vdir.ref, env)
++
 +            if response is None or response.HasField("error"):
 +                # Build failed, so return a failure code
 +                return 1
 +            else:
++
 +                # At the moment, response can either be an ExecutionResponse containing an ActionResult, or an ActionResult directly.
 +                executeResponse = remote_execution_pb2.ExecuteResponse()
 +                if response.response.Is(executeResponse.DESCRIPTOR):
 +                    # Unpack ExecuteResponse and set response to its response
 +                    response.response.Unpack(executeResponse)
 +                    response = executeResponse
++
 +                actionResult = remote_execution_pb2.ActionResult()
 +                if response.response.Is(actionResult.DESCRIPTOR):
 +                    response.response.Unpack(actionResult)
 +                    self.process_job_output(actionResult.output_directories, actionResult.output_files)
 +                else:
 +                    sys.stderr.write("Received unknown message from server.\n")
 +                    return 1
 +        else:
 +            sys.stderr.write("Failed to verify source on remote artifact cache.\n")
 +            return 1
 +        # TODO: Pull the results
 +        sys.stderr.write("Completed remote run with sandbox.\n")
 +        return 0

buildstream/sandbox/sandbox.py

@@ -99,9 +99,11 @@ class Sandbox():
          self.__stdout = kwargs['stdout']
          self.__stderr = kwargs['stderr']
 -        # Setup the directories. Root should be available to subclasses, hence
 -        # being single-underscore. The others are private to this class.
 +        # Setup the directories. Root and output_directory should be
 +        # available to subclasses, hence being single-underscore. The
 +        # others are private to this class.
          self._root = os.path.join(directory, 'root')
 +        self._output_directory = None
          self.__directory = directory
          self.__scratch = os.path.join(self.__directory, 'scratch')
          for directory_ in [self._root, self.__scratch]:
@@ -142,11 +144,29 @@ class Sandbox():
                  self._vdir = FileBasedDirectory(self._root)
          return self._vdir
 +    def set_virtual_directory(self, vdir):
 +        """ Sets virtual directory. Useful after remote execution
 +        has rewritten the working directory. """
 +        self.__vdir = vdir
++
 +    def get_virtual_toplevel_directory(self):
 +        """Fetches the sandbox's toplevel directory
++
 +        The toplevel directory contains 'root', 'scratch' and later
 +        'artifact' where output is copied to.
++
 +        Returns:
 +           (str): The sandbox toplevel directory
++
 +        """
 +        # For now, just create a new Directory every time we're asked
 +        return FileBasedDirectory(self.__directory)
++
      def set_environment(self, environment):
          """Sets the environment variables for the sandbox
          Args:
 -           directory (dict): The environment variables to use in the sandbox
 +           environment (dict): The environment variables to use in the sandbox
          """
          self.__env = environment
@@ -158,6 +178,15 @@ class Sandbox():
          """
          self.__cwd = directory
 +    def set_output_directory(self, directory):
 +        """Sets the output directory - the directory which is preserved
 +        as an artifact after assembly.
++
 +        Args:
 +           directory (str): An absolute path within the sandbox
 +        """
 +        self._output_directory = directory
++
      def mark_directory(self, directory, *, artifact=False):
          """Marks a sandbox directory and ensures it will exist

buildstream/storage/_casbaseddirectory.py

@@ -343,6 +343,20 @@ class CasBasedDirectory(Directory):
                      result.files_written.append(relative_pathname)
          return result
 +    def _save(self, name):
 +        """ Saves this directory into the content cache as a named ref. This function is not
 +        currently in use, but may be useful later. """
 +        self._recalculate_recursing_up()
 +        self._recalculate_recursing_down()
 +        (rel_refpath, refname) = os.path.split(name)
 +        refdir = os.path.join(self.cas_directory, 'refs', 'heads', rel_refpath)
 +        refname = os.path.join(refdir, refname)
++
 +        if not os.path.exists(refdir):
 +            os.makedirs(refdir)
 +        with open(refname, "wb") as f:
 +            f.write(self.ref.SerializeToString())
++
      def import_files(self, external_pathspec, *, files=None,
                       report_written=True, update_utimes=False,
                       can_link=False):

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_rebase] 15 commits: _casbaseddirectory: Alter _save to save the serialised digest object, not the text digest

Jim MacArthur pushed to branch jmac/remote_execution_rebase at BuildStream / buildstream

Commits:

5 changed files:

Changes: