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

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

Commits:

e2817d89

by Raoul Hidalgo Charman at 2018-12-11T11:41:44Z

casremote: move subprocess initialization to a class method.

3 changed files:

buildstream/_artifactcache.py
buildstream/_cas/cascache.py
buildstream/_cas/casremote.py

Changes:

buildstream/_artifactcache.py

@@ -19,14 +19,12 @@
  import multiprocessing
  import os
 -import signal
  import string
  from collections.abc import Mapping
  from .types import _KeyStrength
  from ._exceptions import ArtifactError, CASError, LoadError, LoadErrorReason
  from ._message import Message, MessageType
 -from . import _signals
  from . import utils
  from . import _yaml
@@ -375,20 +373,8 @@ class ArtifactCache():
          remotes = {}
          q = multiprocessing.Queue()
          for remote_spec in remote_specs:
 -            # Use subprocess to avoid creation of gRPC threads in main BuildStream process
 -            # See https://github.com/grpc/grpc/blob/master/doc/fork_support.md for details
 -            p = multiprocessing.Process(target=self.cas.initialize_remote, args=(remote_spec, q))
 -            try:
 -                # Keep SIGINT blocked in the child process
 -                with _signals.blocked([signal.SIGINT], ignore=False):
 -                    p.start()
+-
 -                error = q.get()
 -                p.join()
 -            except KeyboardInterrupt:
 -                utils._kill_process_tree(p.pid)
 -                raise
 +            error = CASRemote.check_remote(remote_spec, q)
              if error and on_failure:
                  on_failure(remote_spec.url, error)

buildstream/_cas/cascache.py

@@ -35,7 +35,7 @@ from .._protos.buildstream.v2 import buildstream_pb2
  from .. import utils
  from .._exceptions import CASError
 -from .casremote import CASRemote, BlobNotFound, _CASBatchRead, _CASBatchUpdate, _MAX_PAYLOAD_BYTES
 +from .casremote import BlobNotFound, _CASBatchRead, _CASBatchUpdate, _MAX_PAYLOAD_BYTES
  # A CASCache manages a CAS repository as specified in the Remote Execution API.
@@ -185,29 +185,6 @@ class CASCache():
          return modified, removed, added
 -    def initialize_remote(self, remote_spec, q):
 -        try:
 -            remote = CASRemote(remote_spec)
 -            remote.init()
+-
 -            request = buildstream_pb2.StatusRequest()
 -            response = remote.ref_storage.Status(request)
+-
 -            if remote_spec.push and not response.allow_updates:
 -                q.put('CAS server does not allow push')
 -            else:
 -                # No error
 -                q.put(None)
+-
 -        except grpc.RpcError as e:
 -            # str(e) is too verbose for errors reported to the user
 -            q.put(e.details())
+-
 -        except Exception as e:               # pylint: disable=broad-except
 -            # Whatever happens, we need to return it to the calling process
 -            #
 -            q.put(str(e))
+-
      # pull():
+     #
      # Pull a ref from a remote repository.

buildstream/_cas/casremote.py

  from collections import namedtuple
  import os
 +import multiprocessing
 +import signal
  from urllib.parse import urlparse
  import grpc
@@ -8,9 +10,11 @@ from .. import _yaml
  from .._protos.google.rpc import code_pb2
  from .._protos.google.bytestream import 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_grpc
 +from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
  from .._exceptions import CASError, LoadError, LoadErrorReason
 +from .. import _signals
 +from .. import utils
  # The default limit for gRPC messages is 4 MiB.
  # Limit payload to 1 MiB to leave sufficient headroom for metadata.
@@ -157,6 +161,53 @@ class CASRemote():
              self._initialized = True
 +    # check_remote
 +    #
 +    # Used when checking whether remote_specs work in the buildstream main
 +    # thread, runs this in a seperate process to avoid creation of gRPC threads
 +    # 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():
 +            try:
 +                remote = cls(remote_spec)
 +                remote.init()
++
 +                request = buildstream_pb2.StatusRequest()
 +                response = remote.ref_storage.Status(request)
++
 +                if remote_spec.push and not response.allow_updates:
 +                    q.put('CAS server does not allow push')
 +                else:
 +                    # No error
 +                    q.put(None)
++
 +            except grpc.RpcError as e:
 +                # str(e) is too verbose for errors reported to the user
 +                q.put(e.details())
++
 +            except Exception as e:               # pylint: disable=broad-except
 +                # Whatever happens, we need to return it to the calling process
 +                #
 +                q.put(str(e))
++
 +        p = multiprocessing.Process(target=__check_remote)
++
 +        try:
 +            # Keep SIGINT blocked in the child process
 +            with _signals.blocked([signal.SIGINT], ignore=False):
 +                p.start()
++
 +            error = q.get()
 +            p.join()
 +        except KeyboardInterrupt:
 +            utils._kill_process_tree(p.pid)
 +            raise
++
 +        return error
++
  # Represents a batch of blobs queued for fetching.
+ #

[Notes] [Git][BuildStream/buildstream][raoul/802-refactor-artifactcache] casremote: move subprocess initialization to a class method.

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

Commits:

3 changed files:

Changes: