[Notes] [Git][BuildStream/buildstream][juerg/command-batching] 9 commits

Jürg Billeter pushed to branch juerg/command-batching at BuildStream / buildstream

Commits:

770f061f

by Jürg Billeter at 2018-11-19T07:08:58Z

sandbox/sandbox.py: Add command batching API

This adds the batch() context manager.

226434fe

by Jürg Billeter at 2018-11-19T07:08:58Z

element.py: Defer workspace.prepared setting in case of pending commands

5d572d28

by Jürg Billeter at 2018-11-19T07:08:58Z

element.py: Support command batching for integration commands

ddc76a38

by Jürg Billeter at 2018-11-19T07:08:58Z

utils.py: Add _SplitContextManager class

A context manager wrapper that allows using a context manager across two
`with` statements.

9ed7dfc2

by Jürg Billeter at 2018-11-19T07:08:58Z

buildelement.py: Support batching for integration and build commands

165986ba

by Jürg Billeter at 2018-11-19T07:08:58Z

scriptelement.py: Support batching for integration and build commands

6b21aae9

by Jürg Billeter at 2018-11-19T07:08:58Z

compose.py: Support batching for integration commands

03b191d8

by Jürg Billeter at 2018-11-19T07:08:58Z

sandbox/_sandboxremote.py: Implement command batching

Execute batched commands in a single shell script.

f94bc8a3
by Jürg Billeter at 2018-11-19T07:11:36Z
```
NEWS: Add entry for command batching
```

Changes:

NEWS

@@ -55,6 +55,9 @@ buildstream 1.3.1
      with cached artifacts, only 'complete' elements can be pushed. If the element
      is expected to have a populated build tree then it must be cached before pushing.
 +  o Add sandbox API for command batching and use it for build, script, and
 +    compose elements.
++
  =================
  buildstream 1.1.5

buildstream/buildelement.py

@@ -127,8 +127,9 @@ artifact collection purposes.
  """
  import os
 -from . import Element, Scope, ElementError
 +from . import Element, Scope
  from . import SandboxFlags
 +from . import utils
  # This list is preserved because of an unfortunate situation, we
@@ -207,6 +208,9 @@ class BuildElement(Element):
          # Setup environment
          sandbox.set_environment(self.get_environment())
 +        # Active sandbox batch context manager
 +        self.__sandbox_batch_cm = None  # pylint: disable=attribute-defined-outside-init
++
      def stage(self, sandbox):
          # Stage deps in the sandbox root
@@ -215,7 +219,7 @@ class BuildElement(Element):
          # Run any integration commands provided by the dependencies
          # once they are all staged and ready
 -        with self.timed_activity("Integrating sandbox"):
 +        with sandbox.batch(0, label="Integrating sandbox"):
              for dep in self.dependencies(Scope.BUILD):
                  dep.integrate(sandbox)
@@ -223,16 +227,23 @@ class BuildElement(Element):
          self.stage_sources(sandbox, self.get_variable('build-root'))
      def assemble(self, sandbox):
 +        # Use the batch context manager from prepare(), if available
 +        batch_cm = self.__sandbox_batch_cm
 +        self.__sandbox_batch_cm = None  # pylint: disable=attribute-defined-outside-init
 -        # Run commands
 -        for command_name in _command_steps:
 -            commands = self.__commands[command_name]
 -            if not commands or command_name == 'configure-commands':
 -                continue
 +        if not batch_cm:
 +            batch_cm = sandbox.batch(SandboxFlags.ROOT_READ_ONLY)
++
 +        with batch_cm:
 +            # Run commands
 +            for command_name in _command_steps:
 +                commands = self.__commands[command_name]
 +                if not commands or command_name == 'configure-commands':
 +                    continue
 -            with self.timed_activity("Running {}".format(command_name)):
 -                for cmd in commands:
 -                    self.__run_command(sandbox, cmd, command_name)
 +                with sandbox.batch(SandboxFlags.ROOT_READ_ONLY, label="Running {}".format(command_name)):
 +                    for cmd in commands:
 +                        self.__run_command(sandbox, cmd, command_name)
          # %{install-root}/%{build-root} should normally not be written
          # to - if an element later attempts to stage to a location
@@ -252,11 +263,17 @@ class BuildElement(Element):
          return self.get_variable('install-root')
      def prepare(self, sandbox):
 -        commands = self.__commands['configure-commands']
 -        if commands:
 -            with self.timed_activity("Running configure-commands"):
 -                for cmd in commands:
 -                    self.__run_command(sandbox, cmd, 'configure-commands')
 +        # Allow use of single batch context manager across prepare and assemble
 +        batch_cm = utils._SplitContextManager(sandbox.batch(SandboxFlags.ROOT_READ_ONLY))
++
 +        with batch_cm:
 +            commands = self.__commands['configure-commands']
 +            if commands:
 +                with sandbox.batch(SandboxFlags.ROOT_READ_ONLY, label="Running configure-commands"):
 +                    for cmd in commands:
 +                        self.__run_command(sandbox, cmd, 'configure-commands')
++
 +        self.__sandbox_batch_cm = batch_cm  # pylint: disable=attribute-defined-outside-init
      def generate_script(self):
          script = ""
@@ -282,13 +299,9 @@ class BuildElement(Element):
          return commands
      def __run_command(self, sandbox, cmd, cmd_name):
 -        self.status("Running {}".format(cmd_name), detail=cmd)
+-
          # Note the -e switch to 'sh' means to exit with an error
          # if any untested command fails.
+         #
 -        exitcode = sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 -                               SandboxFlags.ROOT_READ_ONLY)
 -        if exitcode != 0:
 -            raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode),
 -                               collect=self.get_variable('install-root'))
 +        sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 +                    SandboxFlags.ROOT_READ_ONLY,
 +                    label=cmd)

buildstream/element.py

@@ -769,13 +769,13 @@ class Element(Plugin):
          environment = self.get_environment()
          if bstdata is not None:
 -            commands = self.node_get_member(bstdata, list, 'integration-commands', [])
 -            for i in range(len(commands)):
 -                cmd = self.node_subst_list_element(bstdata, 'integration-commands', [i])
 -                self.status("Running integration command", detail=cmd)
 -                exitcode = sandbox.run(['sh', '-e', '-c', cmd], 0, env=environment, cwd='/')
 -                if exitcode != 0:
 -                    raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode))
 +            with sandbox.batch(0):
 +                commands = self.node_get_member(bstdata, list, 'integration-commands', [])
 +                for i in range(len(commands)):
 +                    cmd = self.node_subst_list_element(bstdata, 'integration-commands', [i])
++
 +                    sandbox.run(['sh', '-e', '-c', cmd], 0, env=environment, cwd='/',
 +                                label=cmd)
      def stage_sources(self, sandbox, directory):
          """Stage this element's sources to a directory in the sandbox
@@ -2073,7 +2073,12 @@ class Element(Plugin):
              self.prepare(sandbox)
              if workspace:
 -                workspace.prepared = True
 +                def mark_workspace_prepared():
 +                    workspace.prepared = True
++
 +                # Defer workspace.prepared setting until pending batch commands
 +                # have been executed.
 +                sandbox._callback(mark_workspace_prepared)
      def __is_cached(self, keystrength):
          if keystrength is None:

buildstream/plugins/elements/compose.py

@@ -122,8 +122,9 @@ class ComposeElement(Element):
                      snapshot = set(vbasedir.list_relative_paths())
                      vbasedir.mark_unmodified()
 -                for dep in self.dependencies(Scope.BUILD):
 -                    dep.integrate(sandbox)
 +                with sandbox.batch(0):
 +                    for dep in self.dependencies(Scope.BUILD):
 +                        dep.integrate(sandbox)
                  if require_split:
                      # Calculate added, modified and removed files

buildstream/sandbox/_sandboxremote.py

@@ -19,11 +19,13 @@
  #        Jim MacArthur <jim macarthur codethink co uk>
  import os
 +import shlex
  from urllib.parse import urlparse
  import grpc
  from . import Sandbox
 +from .sandbox import _SandboxBatch
  from ..storage._filebaseddirectory import FileBasedDirectory
  from ..storage._casbaseddirectory import CasBasedDirectory
  from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
@@ -238,3 +240,69 @@ class SandboxRemote(Sandbox):
          self.process_job_output(action_result.output_directories, action_result.output_files)
          return 0
++
 +    def _create_batch(self, main_group, flags):
 +        return _SandboxRemoteBatch(self, main_group, flags)
++
++
 +# _SandboxRemoteBatch()
 +#
 +# Command batching by shell script generation.
 +#
 +class _SandboxRemoteBatch(_SandboxBatch):
++
 +    def __init__(self, sandbox, main_group, flags):
 +        super().__init__(sandbox, main_group, flags)
++
 +        self.script = None
 +        self.first_command = None
 +        self.cwd = None
 +        self.env = None
++
 +    def execute(self):
 +        self.script = ""
++
 +        self.main_group.execute(self)
++
 +        first = self.first_command
 +        if first and self.sandbox.run(['sh', '-c', '-e', self.script], self.flags, cwd=first.cwd, env=first.env) != 0:
 +            raise SandboxError("Command execution failed", reason="command-failed")
++
 +    def execute_group(self, group):
 +        group.execute_children(self)
++
 +    def execute_command(self, command):
 +        if self.first_command is None:
 +            # First command in batch
 +            # Initial working directory and environment of script already matches
 +            # the command configuration.
 +            self.first_command = command
 +        else:
 +            # Change working directory for this command
 +            if command.cwd != self.cwd:
 +                self.script += "mkdir -p {}\n".format(command.cwd)
 +                self.script += "cd {}\n".format(command.cwd)
++
 +            # Update environment for this command
 +            for key in self.env.keys():
 +                if key not in command.env:
 +                    self.script += "unset {}\n".format(key)
 +            for key, value in command.env.items():
 +                if key not in self.env or self.env[key] != value:
 +                    self.script += "export {}={}\n".format(key, shlex.quote(value))
++
 +        # Keep track of current working directory and environment
 +        self.cwd = command.cwd
 +        self.env = command.env
++
 +        # Actual command execution
 +        cmdline = ' '.join(shlex.quote(cmd) for cmd in command.command)
 +        self.script += "(set -ex; {})".format(cmdline)
++
 +        # Error handling
 +        label = command.label or cmdline
 +        quoted_label = shlex.quote("'{}'".format(label))
 +        self.script += " || (echo Command {} failed with exitcode $? >&2 ; exit 1)\n".format(quoted_label)
++
 +    def execute_call(self, call):
 +        raise SandboxError("SandboxRemote does not support callbacks in command batches")

buildstream/sandbox/sandbox.py

+ #
  #  Copyright (C) 2017 Codethink Limited
 +#  Copyright (C) 2018 Bloomberg Finance LP
+ #
  #  This program is free software; you can redistribute it and/or
  #  modify it under the terms of the GNU Lesser General Public
@@ -29,7 +30,12 @@ See also: :ref:`sandboxing`.
  """
  import os
 -from .._exceptions import ImplError, BstError
 +import shlex
 +import contextlib
 +from contextlib import contextmanager
++
 +from .._exceptions import ImplError, BstError, SandboxError
 +from .._message import Message, MessageType
  from ..storage._filebaseddirectory import FileBasedDirectory
  from ..storage._casbaseddirectory import CasBasedDirectory
@@ -94,6 +100,13 @@ class Sandbox():
          self.__mount_sources = {}
          self.__allow_real_directory = kwargs['allow_real_directory']
 +        # Plugin ID for logging
 +        plugin = kwargs.get('plugin', None)
 +        if plugin:
 +            self.__plugin_id = plugin._get_unique_id()
 +        else:
 +            self.__plugin_id = None
++
          # Configuration from kwargs common to all subclasses
          self.__config = kwargs['config']
          self.__stdout = kwargs['stdout']
@@ -121,6 +134,9 @@ class Sandbox():
          # directory via get_directory.
          self._never_cache_vdirs = False
 +        # Pending command batch
 +        self.__batch = None
++
      def get_directory(self):
          """Fetches the sandbox root directory
@@ -209,9 +225,16 @@ class Sandbox():
              'artifact': artifact
          })
 -    def run(self, command, flags, *, cwd=None, env=None):
 +    def run(self, command, flags, *, cwd=None, env=None, label=None):
          """Run a command in the sandbox.
 +        If this is called outside a batch context, the command is immediately
 +        executed.
++
 +        If this is called in a batch context, the command is added to the batch
 +        for later execution. If the command fails, later commands will not be
 +        executed. Command flags must match batch flags.
++
          Args:
              command (list): The command to run in the sandboxed environment, as a list
                              of strings starting with the binary to run.
@@ -219,9 +242,10 @@ class Sandbox():
              cwd (str): The sandbox relative working directory in which to run the command.
              env (dict): A dictionary of string key, value pairs to set as environment
                          variables inside the sandbox environment.
 +            label (str): An optional label for the command, used for logging.
          Returns:
 -            (int): The program exit code.
 +            (int|None): The program exit code, or None if running in batch context.
          Raises:
              (:class:`.ProgramNotFoundError`): If a host tool which the given sandbox
@@ -245,7 +269,62 @@ class Sandbox():
          if isinstance(command, str):
              command = [command]
 -        return self._run(command, flags, cwd=cwd, env=env)
 +        if self.__batch:
 +            if flags != self.__batch.flags:
 +                raise SandboxError("Inconsistent sandbox flags in single command batch")
++
 +            batch_command = _SandboxBatchCommand(command, cwd=cwd, env=env, label=label)
++
 +            current_group = self.__batch.current_group
 +            current_group.append(batch_command)
 +            return None
 +        else:
 +            return self._run(command, flags, cwd=cwd, env=env)
++
 +    @contextmanager
 +    def batch(self, flags, *, label=None):
 +        """Context manager for command batching
++
 +        This provides a batch context that defers execution of commands until
 +        the end of the context. If a command fails, the batch will be aborted
 +        and subsequent commands will not be executed.
++
 +        Command batches may be nested. Execution will start only when the top
 +        level batch context ends.
++
 +        Args:
 +            flags (:class:`.SandboxFlags`): The flags for this command batch.
 +            label (str): An optional label for the batch group, used for logging.
++
 +        Raises:
 +            (:class:`.SandboxError`): If a command fails.
 +        """
++
 +        group = _SandboxBatchGroup(label=label)
++
 +        if self.__batch:
 +            # Nested batch
 +            if flags != self.__batch.flags:
 +                raise SandboxError("Inconsistent sandbox flags in single command batch")
++
 +            parent_group = self.__batch.current_group
 +            parent_group.append(group)
 +            self.__batch.current_group = group
 +            try:
 +                yield
 +            finally:
 +                self.__batch.current_group = parent_group
 +        else:
 +            # Top-level batch
 +            batch = self._create_batch(group, flags)
++
 +            self.__batch = batch
 +            try:
 +                yield
 +            finally:
 +                self.__batch = None
++
 +            batch.execute()
      #####################################################
      #    Abstract Methods for Sandbox implementations   #
@@ -270,6 +349,18 @@ class Sandbox():
          raise ImplError("Sandbox of type '{}' does not implement _run()"
                          .format(type(self).__name__))
 +    # _create_batch()
 +    #
 +    # Abstract method for creating a batch object. Subclasses can override
 +    # this method to instantiate a subclass of _SandboxBatch.
 +    #
 +    # Args:
 +    #    main_group (:class:`_SandboxBatchGroup`): The top level batch group.
 +    #    flags (:class:`.SandboxFlags`): The flags for commands in this batch.
 +    #
 +    def _create_batch(self, main_group, flags):
 +        return _SandboxBatch(self, main_group, flags)
++
      ################################################
      #               Private methods                #
      ################################################
@@ -418,3 +509,137 @@ class Sandbox():
                  return True
          return False
++
 +    # _get_plugin_id()
 +    #
 +    # Get the plugin's unique identifier
 +    #
 +    def _get_plugin_id(self):
 +        return self.__plugin_id
++
 +    # _callback()
 +    #
 +    # If this is called outside a batch context, the specified function is
 +    # invoked immediately.
 +    #
 +    # If this is called in a batch context, the function is added to the batch
 +    # for later invocation.
 +    #
 +    # Args:
 +    #    callback (callable): The function to invoke
 +    #
 +    def _callback(self, callback):
 +        if self.__batch:
 +            batch_call = _SandboxBatchCall(callback)
++
 +            current_group = self.__batch.current_group
 +            current_group.append(batch_call)
 +        else:
 +            callback()
++
++
 +# _SandboxBatch()
 +#
 +# A batch of sandbox commands.
 +#
 +class _SandboxBatch():
++
 +    def __init__(self, sandbox, main_group, flags):
 +        self.sandbox = sandbox
 +        self.main_group = main_group
 +        self.current_group = main_group
 +        self.flags = flags
++
 +    def execute(self):
 +        self.main_group.execute(self)
++
 +    def execute_group(self, group):
 +        if group.label:
 +            context = self.sandbox._get_context()
 +            cm = context.timed_activity(group.label, unique_id=self.sandbox._get_plugin_id())
 +        else:
 +            cm = contextlib.suppress()
++
 +        with cm:
 +            group.execute_children(self)
++
 +    def execute_command(self, command):
 +        if command.label:
 +            context = self.sandbox._get_context()
 +            message = Message(self.sandbox._get_plugin_id(), MessageType.STATUS,
 +                              'Running {}'.format(command.label))
 +            context.message(message)
++
 +        exitcode = self.sandbox._run(command.command, self.flags, cwd=command.cwd, env=command.env)
 +        if exitcode != 0:
 +            cmdline = ' '.join(shlex.quote(cmd) for cmd in command.command)
 +            label = command.label or cmdline
 +            raise SandboxError("Command '{}' failed with exitcode {}".format(label, exitcode),
 +                               reason="command-failed")
++
 +    def execute_call(self, call):
 +        call.callback()
++
++
 +# _SandboxBatchItem()
 +#
 +# An item in a command batch.
 +#
 +class _SandboxBatchItem():
++
 +    def __init__(self, *, label=None):
 +        self.label = label
++
++
 +# _SandboxBatchCommand()
 +#
 +# A command item in a command batch.
 +#
 +class _SandboxBatchCommand(_SandboxBatchItem):
++
 +    def __init__(self, command, *, cwd, env, label=None):
 +        super().__init__(label=label)
++
 +        self.command = command
 +        self.cwd = cwd
 +        self.env = env
++
 +    def execute(self, batch):
 +        batch.execute_command(self)
++
++
 +# _SandboxBatchGroup()
 +#
 +# A group in a command batch.
 +#
 +class _SandboxBatchGroup(_SandboxBatchItem):
++
 +    def __init__(self, *, label=None):
 +        super().__init__(label=label)
++
 +        self.children = []
++
 +    def append(self, item):
 +        self.children.append(item)
++
 +    def execute(self, batch):
 +        batch.execute_group(self)
++
 +    def execute_children(self, batch):
 +        for item in self.children:
 +            item.execute(batch)
++
++
 +# _SandboxBatchCall()
 +#
 +# A call item in a command batch.
 +#
 +class _SandboxBatchCall(_SandboxBatchItem):
++
 +    def __init__(self, callback):
 +        super().__init__()
++
 +        self.callback = callback
++
 +    def execute(self, batch):
 +        batch.execute_call(self)

buildstream/scriptelement.py

@@ -226,10 +226,11 @@ class ScriptElement(Element):
                                           .format(build_dep.name), silent_nested=True):
                      build_dep.stage_dependency_artifacts(sandbox, Scope.RUN, path="/")
 -            for build_dep in self.dependencies(Scope.BUILD, recurse=False):
 -                with self.timed_activity("Integrating {}".format(build_dep.name), silent_nested=True):
 -                    for dep in build_dep.dependencies(Scope.RUN):
 -                        dep.integrate(sandbox)
 +            with sandbox.batch(0):
 +                for build_dep in self.dependencies(Scope.BUILD, recurse=False):
 +                    with self.timed_activity("Integrating {}".format(build_dep.name), silent_nested=True):
 +                        for dep in build_dep.dependencies(Scope.RUN):
 +                            dep.integrate(sandbox)
          else:
              # If layout, follow its rules.
              for item in self.__layout:
@@ -251,37 +252,38 @@ class ScriptElement(Element):
                          virtual_dstdir.descend(item['destination'].lstrip(os.sep).split(os.sep), create=True)
                          element.stage_dependency_artifacts(sandbox, Scope.RUN, path=item['destination'])
 -            for item in self.__layout:
 +            with sandbox.batch(0):
 +                for item in self.__layout:
 -                # Skip layout members which dont stage an element
 -                if not item['element']:
 -                    continue
 +                    # Skip layout members which dont stage an element
 +                    if not item['element']:
 +                        continue
 -                element = self.search(Scope.BUILD, item['element'])
 +                    element = self.search(Scope.BUILD, item['element'])
 -                # Integration commands can only be run for elements staged to /
 -                if item['destination'] == '/':
 -                    with self.timed_activity("Integrating {}".format(element.name),
 -                                             silent_nested=True):
 -                        for dep in element.dependencies(Scope.RUN):
 -                            dep.integrate(sandbox)
 +                    # Integration commands can only be run for elements staged to /
 +                    if item['destination'] == '/':
 +                        with self.timed_activity("Integrating {}".format(element.name),
 +                                                 silent_nested=True):
 +                            for dep in element.dependencies(Scope.RUN):
 +                                dep.integrate(sandbox)
          install_root_path_components = self.__install_root.lstrip(os.sep).split(os.sep)
          sandbox.get_virtual_directory().descend(install_root_path_components, create=True)
      def assemble(self, sandbox):
 -        for groupname, commands in self.__commands.items():
 -            with self.timed_activity("Running '{}'".format(groupname)):
 -                for cmd in commands:
 -                    self.status("Running command", detail=cmd)
 -                    # Note the -e switch to 'sh' means to exit with an error
 -                    # if any untested command fails.
 -                    exitcode = sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 -                                           SandboxFlags.ROOT_READ_ONLY if self.__root_read_only else 0)
 -                    if exitcode != 0:
 -                        raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode),
 -                                           collect=self.__install_root)
 +        sandbox_flags = SandboxFlags.ROOT_READ_ONLY if self.__root_read_only else 0
++
 +        with sandbox.batch(sandbox_flags):
 +            for groupname, commands in self.__commands.items():
 +                with sandbox.batch(sandbox_flags, label="Running '{}'".format(groupname)):
 +                    for cmd in commands:
 +                        # Note the -e switch to 'sh' means to exit with an error
 +                        # if any untested command fails.
 +                        sandbox.run(['sh', '-c', '-e', cmd + '\n'],
 +                                    sandbox_flags,
 +                                    label=cmd)
          # Return where the result can be collected from
          return self.__install_root

buildstream/utils.py

@@ -1199,3 +1199,27 @@ def _deduplicate(iterable, key=None):
  def _get_link_mtime(path):
      path_stat = os.lstat(path)
      return path_stat.st_mtime
++
++
 +# _SplitContextManager():
 +#
 +# A context manager wrapper that allows using a context manager across two
 +# `with` statements.
 +#
 +class _SplitContextManager():
 +    def __init__(self, cm):
 +        self._cm = cm
 +        self._stage1_complete = False
++
 +    def __enter__(self):
 +        if not self._stage1_complete:
 +            return self._cm.__enter__()
 +        else:
 +            return None
++
 +    def __exit__(self, exception_type, exception_value, traceback):
 +        if not self._stage1_complete and exception_type is None:
 +            self._stage1_complete = True
 +            return False
 +        else:
 +            return self._cm.__exit__(exception_type, exception_value, traceback)

tests/integration/sandbox-bwrap.py

@@ -58,5 +58,5 @@ def test_sandbox_bwrap_return_subprocess(cli, tmpdir, datafiles):
      })
      result = cli.run(project=project, args=['build', element_name])
 -    result.assert_task_error(error_domain=ErrorDomain.ELEMENT, error_reason=None)
 +    result.assert_task_error(error_domain=ErrorDomain.SANDBOX, error_reason="command-failed")
      assert "sandbox-bwrap/command-exit-42.bst|Command 'exit 42' failed with exitcode 42" in result.stderr

[Notes] [Git][BuildStream/buildstream][juerg/command-batching] 9 commits: sandbox/sandbox.py: Add command batching API

Jürg Billeter pushed to branch juerg/command-batching at BuildStream / buildstream

Commits:

9 changed files:

Changes: