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