[Notes] [Git][BuildStream/buildstream][juerg/command-batching] 12 commit

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

Commits:

516e990e

by ctolentino8 at 2018-10-31T11:36:46Z

_frontend/app.py: Set correct element-path in interactive bst-init

b8a37a63

by Tristan Van Berkom at 2018-11-01T10:16:25Z

Merge branch 'chiaratolentino/bst-init-interactive-elementpath' into 'master'

_frontend/app.py: Set correct element-path in interactive bst-init

See merge request BuildStream/buildstream!912

10d8d994

by Jürg Billeter at 2018-11-01T11:00:08Z

sandbox: Add queue() and run_queue() methods for command batching

15d56e06
by Jürg Billeter at 2018-11-01T11:00:08Z
```
sandbox: Add _has_commands_queued() method
```
59b436df
by Jürg Billeter at 2018-11-01T11:00:08Z
```
sandbox: Add queue_context_manager() method
```

ecd9d5e4

by Jürg Billeter at 2018-11-01T11:00:08Z

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

1f16d9fd

by Jürg Billeter at 2018-11-01T11:00:08Z

element.py: Support command batching for integration commands

This adds a queue parameter to the integrate() method to keep command
batching optional for plugins that do not support command batching.

e1e4234d
by Jürg Billeter at 2018-11-01T11:00:08Z
```
NEWS: Add entry for command batching
```

ca6f10ce

by Jürg Billeter at 2018-11-01T11:00:08Z

buildelement.py: Use Sandbox queue API for command batching

2fbf0c6d

by Jürg Billeter at 2018-11-01T11:00:08Z

buildelement.py: Support batching for integration commands

7e0a641c

by Jürg Billeter at 2018-11-01T11:00:08Z

sandbox/_sandboxremote.py: Add initial support for stdout and stderr

9997c642

by Jürg Billeter at 2018-11-01T11:00:08Z

WIP: sandbox/_sandboxremote.py: Implement command batching

Changes:

NEWS

@@ -39,6 +39,8 @@ buildstream 1.3.1
      instead of just a specially-formatted build-root with a `root` and `scratch`
      subdirectory.
 +  o Add sandbox API for command batching.
++
  =================
  buildstream 1.1.5

buildstream/_frontend/app.py

@@ -305,7 +305,6 @@ class App():
          directory = self._main_options['directory']
          directory = os.path.abspath(directory)
          project_path = os.path.join(directory, 'project.conf')
 -        elements_path = os.path.join(directory, element_path)
          try:
              # Abort if the project.conf already exists, unless `--force` was specified in `bst init`
@@ -335,6 +334,7 @@ class App():
                  raise AppError("Error creating project directory {}: {}".format(directory, e)) from e
              # Create the elements sub-directory if it doesnt exist
 +            elements_path = os.path.join(directory, element_path)
              try:
                  os.makedirs(elements_path, exist_ok=True)
              except IOError as e:

buildstream/buildelement.py

@@ -217,7 +217,8 @@ class BuildElement(Element):
          # once they are all staged and ready
          with self.timed_activity("Integrating sandbox"):
              for dep in self.dependencies(Scope.BUILD):
 -                dep.integrate(sandbox)
 +                dep.integrate(sandbox, queue=True)
 +            sandbox.run_queue(0)
          # Stage sources in the build root
          self.stage_sources(sandbox, self.get_variable('build-root'))
@@ -230,9 +231,11 @@ class BuildElement(Element):
              if not commands or command_name == 'configure-commands':
                  continue
 -            with self.timed_activity("Running {}".format(command_name)):
 +            with sandbox.queue_context_manager(self.timed_activity("Running {}".format(command_name))):
                  for cmd in commands:
 -                    self.__run_command(sandbox, cmd, command_name)
 +                    self.__queue_command(sandbox, cmd, command_name)
++
 +        sandbox.run_queue(SandboxFlags.ROOT_READ_ONLY)
          # %{install-root}/%{build-root} should normally not be written
          # to - if an element later attempts to stage to a location
@@ -254,9 +257,9 @@ class BuildElement(Element):
      def prepare(self, sandbox):
          commands = self.__commands['configure-commands']
          if commands:
 -            with self.timed_activity("Running configure-commands"):
 +            with sandbox.queue_context_manager(self.timed_activity("Running configure-commands")):
                  for cmd in commands:
 -                    self.__run_command(sandbox, cmd, 'configure-commands')
 +                    self.__queue_command(sandbox, cmd, 'configure-commands')
      def generate_script(self):
          script = ""
@@ -281,14 +284,18 @@ class BuildElement(Element):
          return commands
 -    def __run_command(self, sandbox, cmd, cmd_name):
 -        self.status("Running {}".format(cmd_name), detail=cmd)
 +    def __queue_command(self, sandbox, cmd, cmd_name):
 +        def start_cb():
 +            self.status("Running {}".format(cmd_name), detail=cmd)
++
 +        def complete_cb(exitcode):
 +            if exitcode != 0:
 +                raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode),
 +                                   collect=self.get_variable('install-root'))
          # 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.queue(['sh', '-c', '-e', cmd + '\n'],
 +                      start_callback=start_cb,
 +                      complete_callback=complete_cb)

buildstream/element.py

@@ -79,6 +79,7 @@ import copy
  from collections import OrderedDict
  from collections.abc import Mapping
  from contextlib import contextmanager
 +from functools import partial
  import tempfile
  import shutil
@@ -753,11 +754,12 @@ class Element(Plugin):
                  detail += "  " + "  ".join(["/" + f + "\n" for f in value])
              self.warn("Ignored files", detail=detail)
 -    def integrate(self, sandbox):
 +    def integrate(self, sandbox, *, queue=False):
          """Integrate currently staged filesystem against this artifact.
          Args:
             sandbox (:class:`.Sandbox`): The build sandbox
 +           queue (bool): Whether to queue commands or immediately run them
          This modifies the sysroot staged inside the sandbox so that
          the sysroot is *integrated*. Only an *integrated* sandbox
@@ -768,14 +770,33 @@ class Element(Plugin):
          bstdata = self.get_public_data('bst')
          environment = self.get_environment()
 +        if not queue:
 +            # Element.integrate() must be called with queue=True if there are
 +            # commands pending. Otherwise these commands are run implicitly
 +            # by the call to run_queue() at the end of this method, which is
 +            # likely not expected by the plugin.
 +            assert not sandbox._has_commands_queued()
++
          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))
++
 +                def start_cb(cmd):
 +                    self.status("Running integration command", detail=cmd)
++
 +                def complete_cb(cmd, exitcode):
 +                    if exitcode != 0:
 +                        raise ElementError("Command '{}' failed with exitcode {}".format(cmd, exitcode))
++
 +                sandbox.queue(['sh', '-e', '-c', cmd], env=environment, cwd='/',
 +                              start_callback=partial(start_cb, cmd),
 +                              complete_callback=partial(complete_cb, cmd))
++
 +        if not queue:
 +            # The caller does not allow keeping pending commands in the queue.
 +            # Immediately run the pending commands.
 +            sandbox.run_queue(0)
      def stage_sources(self, sandbox, directory):
          """Stage this element's sources to a directory in the sandbox
@@ -2093,7 +2114,15 @@ class Element(Plugin):
              self.prepare(sandbox)
              if workspace:
 -                workspace.prepared = True
 +                def mark_workspace_prepared():
 +                    workspace.prepared = True
++
 +                if sandbox._has_commands_queued():
 +                    # Defer workspace.prepared setting until the queued
 +                    # prepare commands have been executed.
 +                    sandbox.queue(None, start_callback=mark_workspace_prepared)
 +                else:
 +                    mark_workspace_prepared()
      def __is_cached(self, keystrength):
          if keystrength is None:

buildstream/sandbox/_sandboxremote.py

@@ -19,11 +19,14 @@
  #        Jim MacArthur <jim macarthur codethink co uk>
  import os
 +import shlex
 +from collections import deque
  from urllib.parse import urlparse
  import grpc
  from . import Sandbox
 +from .sandbox import _QueueEntryType
  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
@@ -189,6 +192,8 @@ class SandboxRemote(Sandbox):
          self._set_virtual_directory(new_dir)
      def run(self, command, flags, *, cwd=None, env=None):
 +        stdout, stderr = self._get_output()
++
          # Upload sources
          upload_vdir = self.get_virtual_directory()
@@ -241,13 +246,90 @@ class SandboxRemote(Sandbox):
          action_result = execution_response.result
 +        if stdout:
 +            if action_result.stdout_raw:
 +                stdout.write(str(action_result.stdout_raw, 'utf-8', errors='ignore'))
 +        if stderr:
 +            if action_result.stderr_raw:
 +                stderr.write(str(action_result.stderr_raw, 'utf-8', errors='ignore'))
++
          if action_result.exit_code != 0:
              # A normal error during the build: the remote execution system
              # has worked correctly but the command failed.
 -            # action_result.stdout and action_result.stderr also contains
 -            # build command outputs which we ignore at the moment.
              return action_result.exit_code
          self.process_job_output(action_result.output_directories, action_result.output_files)
          return 0
++
 +    def run_queue(self, flags):
 +        # If a command fails, all remaining commands in the queue must be discarded.
 +        queue = self._queue
 +        self._queue = []
++
 +        script = ""
 +        i = 0
 +        first = None
 +        for entry in queue:
 +            if entry.type == _QueueEntryType.COMMAND and entry.command:
 +                if first is None:
 +                    # First command in queue
 +                    first = entry
 +                else:
 +                    if entry.cwd != cwd:
 +                        script += "mkdir -p {}\n".format(entry.cwd)
 +                        script += "cd {}\n".format(entry.cwd)
 +                    for key in env.keys():
 +                        if key not in entry.env:
 +                            script += "unset {}\n".format(key)
 +                    for key, value in entry.env.items():
 +                        if key not in env or env[key] != value:
 +                            script += "export {}={}\n".format(key, value)
 +                cwd = entry.cwd
 +                env = entry.env
++
 +                cmdline = ' '.join(shlex.quote(cmd) for cmd in entry.command)
 +                script += "({})\n".format(cmdline)
 +                script += "RETVAL=$?\n"
 +                script += "if [ $RETVAL -ne 0 ] ; then\n"
 +                # Report failing command and exit code to stderr (and then back to client)
 +                script += "  echo -e '\nbst-command-failure:' {} $RETVAL >&2\n".format(i)
 +                script += "  exit 1\n"
 +                script += "fi\n"
 +            i += 1
++
 +        if first:
 +            exit_code = self.run(['sh', '-c', script], flags, cwd=first.cwd, env=first.env)
 +        else:
 +            exit_code = 0
++
 +        if exit_code != 0:
 +            # TODO get failed command and exit code from stderr
 +            failed_command = 0
 +            command_exit_code = 1
++
 +        i = 0
 +        # Stack of pending context manager exit methods
 +        exit_stack = deque()
 +        try:
 +            for entry in queue:
 +                if entry.type == _QueueEntryType.COMMAND:
 +                    entry.start_callback()
 +                    if exit_code == 0 or i < failed_command:
 +                        # Command succeeded
 +                        entry.complete_callback(0)
 +                    else:
 +                        # Command failed
 +                        entry.complete_callback(command_exit_code)
 +                        break
 +                elif entry.type == _QueueEntryType.CONTEXT_MANAGER_ENTER:
 +                    entry.cm.__enter__()
 +                    exit_stack.append(entry.cm.__exit__)
 +                elif entry.type == _QueueEntryType.CONTEXT_MANAGER_EXIT:
 +                    exit_cb = exit_stack.pop()
 +                    exit_cb(None, None, None)
 +                i += 1
 +        finally:
 +            while exit_stack:
 +                exit_cb = exit_stack.pop()
 +                exit_cb(None, None, None)

buildstream/sandbox/sandbox.py

@@ -29,6 +29,9 @@ See also: :ref:`sandboxing`.
  """
  import os
 +from collections import deque, namedtuple
 +from contextlib import contextmanager
++
  from .._exceptions import ImplError, BstError
  from ..storage._filebaseddirectory import FileBasedDirectory
  from ..storage._casbaseddirectory import CasBasedDirectory
@@ -71,6 +74,22 @@ class SandboxFlags():
      """
 +# _QueueEntryType()
 +#
 +# A type for sandbox queue entries.
 +#
 +class _QueueEntryType():
++
 +    # A queued command.
 +    COMMAND = 1
++
 +    # A queued context manager enter.
 +    CONTEXT_MANAGER_ENTER = 2
++
 +    # A queued context manager exit.
 +    CONTEXT_MANAGER_EXIT = 3
++
++
  class Sandbox():
      """Sandbox()
@@ -121,6 +140,9 @@ class Sandbox():
          # directory via get_directory.
          self._never_cache_vdirs = False
 +        # Queued commands
 +        self._queue = []
++
      def get_directory(self):
          """Fetches the sandbox root directory
@@ -237,6 +259,97 @@ class Sandbox():
          raise ImplError("Sandbox of type '{}' does not implement run()"
                          .format(type(self).__name__))
 +    def queue(self, command, *, cwd=None, env=None, start_callback=None, complete_callback=None):
 +        """Queue a command to be run in the sandbox.
++
 +        If the command fails, commands queued later will not be executed.
 +        The callbacks are not guaranteed to be invoked in real time.
++
 +        Args:
 +            command (list): The command to run in the sandboxed environment, as a list
 +                            of strings starting with the binary to run.
 +            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.
 +            start_callback (callable): Called when the command starts.
 +            complete_callback (callble): Called when the command completes
 +                                         with the exit code as argument.
++
 +        .. note::
++
 +           The optional *cwd* argument will default to the value set with
 +           :func:`~buildstream.sandbox.Sandbox.set_work_directory`
++
 +        *Since: 1.4*
 +        """
 +        entry = namedtuple('QueueEntry', ['type', 'command', 'cwd', 'env', 'start_callback', 'complete_callback'])
 +        entry.type = _QueueEntryType.COMMAND
 +        entry.command = command
 +        entry.cwd = self._get_work_directory(cwd=cwd)
 +        entry.env = self._get_environment(cwd=cwd, env=env)
 +        entry.start_callback = start_callback
 +        entry.complete_callback = complete_callback
 +        self._queue.append(entry)
++
 +    @contextmanager
 +    def queue_context_manager(self, cm):
 +        """Queue entering and exiting a context manager
++
 +        Args:
 +           cm: A context manager
++
 +        *Since: 1.4*
 +        """
 +        entry = namedtuple('QueueEntry', ['type', 'cm'])
 +        entry.type = _QueueEntryType.CONTEXT_MANAGER_ENTER
 +        entry.cm = cm
 +        self._queue.append(entry)
 +        yield
 +        entry = namedtuple('QueueEntry', ['type'])
 +        entry.type = _QueueEntryType.CONTEXT_MANAGER_EXIT
 +        self._queue.append(entry)
++
 +    def run_queue(self, flags):
 +        """Run a command in the sandbox.
++
 +        Args:
 +            flags (:class:`.SandboxFlags`): The flags for running this command.
++
 +        Raises:
 +            (:class:`.ProgramNotFoundError`): If a host tool which the given sandbox
 +                                              implementation requires is not found.
++
 +        *Since: 1.4*
 +        """
 +        # If a command fails, all remaining commands in the queue must be discarded.
 +        queue = self._queue
 +        self._queue = []
++
 +        # Stack of pending context manager exit methods
 +        exit_stack = deque()
 +        try:
 +            for entry in queue:
 +                if entry.type == _QueueEntryType.COMMAND:
 +                    if entry.start_callback:
 +                        entry.start_callback()
 +                    if entry.command:
 +                        # pylint: disable=assignment-from-no-return
 +                        exit_code = self.run(entry.command, flags, cwd=entry.cwd, env=entry.env)
 +                    if entry.complete_callback:
 +                        entry.complete_callback(exit_code)
 +                    if exit_code != 0:
 +                        break
 +                elif entry.type == _QueueEntryType.CONTEXT_MANAGER_ENTER:
 +                    entry.cm.__enter__()
 +                    exit_stack.append(entry.cm.__exit__)
 +                elif entry.type == _QueueEntryType.CONTEXT_MANAGER_EXIT:
 +                    exit_cb = exit_stack.pop()
 +                    exit_cb(None, None, None)
 +        finally:
 +            while exit_stack:
 +                exit_cb = exit_stack.pop()
 +                exit_cb(None, None, None)
++
      ################################################
      #               Private methods                #
      ################################################
@@ -385,3 +498,12 @@ class Sandbox():
                  return True
          return False
++
 +    # _has_commands_queued()
 +    #
 +    # Returns whether the sandbox has a non-empty queue of pending commands.
 +    #
 +    # Returns:
 +    #    (bool): Whether the command queue is non-empty.
 +    def _has_commands_queued(self):
 +        return len(self._queue) > 0

tests/frontend/init.py

@@ -3,6 +3,7 @@ import pytest
  from tests.testutils import cli
  from buildstream import _yaml
 +from buildstream._frontend.app import App
  from buildstream._exceptions import ErrorDomain, LoadErrorReason
  from buildstream._versions import BST_FORMAT_VERSION
@@ -98,3 +99,34 @@ def test_bad_element_path(cli, tmpdir, element_path):
          'init', '--project-name', 'foo', '--element-path', element_path
      ])
      result.assert_main_error(ErrorDomain.APP, 'invalid-element-path')
++
++
 +@pytest.mark.parametrize("element_path", [('foo'), ('foo/bar')])
 +def test_element_path_interactive(cli, tmp_path, monkeypatch, element_path):
 +    project = tmp_path
 +    project_conf_path = project.joinpath('project.conf')
++
 +    class DummyInteractiveApp(App):
 +        def __init__(self, *args, **kwargs):
 +            super().__init__(*args, **kwargs)
 +            self.interactive = True
++
 +        @classmethod
 +        def create(cls, *args, **kwargs):
 +            return DummyInteractiveApp(*args, **kwargs)
++
 +        def _init_project_interactive(self, *args, **kwargs):
 +            return ('project_name', '0', element_path)
++
 +    monkeypatch.setattr(App, 'create', DummyInteractiveApp.create)
++
 +    result = cli.run(project=str(project), args=['init'])
 +    result.assert_success()
++
 +    full_element_path = project.joinpath(element_path)
 +    assert full_element_path.exists()
++
 +    project_conf = _yaml.load(str(project_conf_path))
 +    assert project_conf['name'] == 'project_name'
 +    assert project_conf['format-version'] == '0'
 +    assert project_conf['element-path'] == element_path

[Notes] [Git][BuildStream/buildstream][juerg/command-batching] 12 commits: _frontend/app.py: Set correct element-path in interactive bst-init

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

Commits:

7 changed files:

Changes: