# Copyright 2012 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Module for running cbuildbot stages in the background."""

import collections
import contextlib
import errno
import functools
import logging
import multiprocessing
from multiprocessing.managers import SyncManager
import os
import queue as Queue
import signal
import sys
import time
import traceback

from chromite.cbuildbot import cbuildbot_alerts
from chromite.lib import cros_build_lib
from chromite.lib import failures_lib
from chromite.lib import osutils
from chromite.lib import results_lib
from chromite.lib import signals
from chromite.lib import timeout_util
from chromite.utils import prctl


_BUFSIZE = 1024


class HackTimeoutSyncManager(SyncManager):
    """Increase the process join timeout in SyncManager.

    The timeout for the manager process to join in the core library is
    too low. The process is often killed before shutting down properly,
    resulting in temporary directories (pymp-xxx) not being cleaned
    up. This class increases the default timeout.
    """

    @staticmethod
    def _finalize_manager(process, *args, **kwargs):
        """Shutdown the manager process."""

        def _join(functor, *args, **kwargs):
            timeout = kwargs.get("timeout")
            if not timeout is None and timeout < 1:
                kwargs["timeout"] = 1

            functor(*args, **kwargs)

        process.join = functools.partial(_join, process.join)
        SyncManager._finalize_manager(process, *args, **kwargs)


def IgnoreSigintAndSigterm():
    """Ignores any future SIGINTs and SIGTERMs."""
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)


def WrapMultiprocessing(callback, *args, **kwargs):
    """Wraps a callback with a short value of TMPDIR.

    This is intended to wrap calls to multiprocessing to ensure any sockets
    created during initialization are created under the /tmp tree rather than in
    a custom temp directory. This is needed because TMPDIR might be really long,
    and named sockets are limited to 108 characters.

    Examples:
        errors = WrapMultiprocessing(multiprocessing.Value, 'i')

    Returns:
        The return value of the callback function with the specified args
    """
    # Use a short directory in /tmp due to the UNIX socket name limit
    # (see b/249124285 for an example of long path to $TMPDIR).
    # Do not use /tmp directly to keep these
    # temporary files together and because certain environments do not like too
    # many files in /tmp (see crbug.com/945523).
    # Make it mode 1777 to mirror /tmp, so that we don't have failures when
    # root calls parallel first, and some other user calls it later.
    tmp_dir = "/tmp/chromite.parallel.%d" % os.geteuid()
    osutils.SafeMakedirs(tmp_dir, mode=0o1777)
    old_tempdir_value, old_tempdir_env = osutils.SetGlobalTempDir(tmp_dir)
    try:
        return callback(*args, **kwargs)
    finally:
        osutils.SetGlobalTempDir(old_tempdir_value, old_tempdir_env)


def _get_sync_manager():
    """Create a SyncManager for use in Manager below

    This is put into a helper function so it can be used inside
    WrapMultiprocessing to adjust TMPDIR to avoid paths too long for UNIX
    sockets.
    """
    m = HackTimeoutSyncManager()
    # SyncManager doesn't handle KeyboardInterrupt exceptions well; pipes get
    # broken and E_NOENT or E_PIPE errors are thrown from various places. We
    # can just ignore SIGINT in the SyncManager and things will close properly
    # when the enclosing with-statement exits.
    # TODO(b/236161656): Fix.
    # pylint: disable-next=consider-using-with
    m.start(IgnoreSigintAndSigterm)
    return m


def Manager():
    """Create a background process for managing interprocess communication.

    This manager wraps multiprocessing.Manager() and ensures that any sockets
    created during initialization are created under the /tmp tree rather than in
    a custom temp directory. This is needed because TMPDIR might be really long,
    and named sockets are limited to 108 characters.

    Examples:
        with Manager() as manager:
        queue = manager.Queue()
        ...

    Returns:
        The return value of multiprocessing.Manager()
    """
    return WrapMultiprocessing(_get_sync_manager)


class BackgroundFailure(failures_lib.CompoundFailure):
    """Exception to show a step failed while running in a background process."""


class ProcessExitTimeout(Exception):
    """Raised if a process cannot exit within the timeout."""


class ProcessUnexpectedExit(Exception):
    """Raised if a process exits unexpectedly."""


class ProcessSilentTimeout(Exception):
    """Raised when there is no output for a prolonged period of time."""


class UnexpectedException(Exception):
    """Raised when exception occurs at an unexpected place."""


class _BackgroundTask(multiprocessing.Process):
    """Run a task in the background.

    This task may be the 'Run' function from a buildbot stage or just a plain
    function. It will be run in the background. Output from this task is saved
    to a temporary file and is printed when the 'Wait' function is called.
    """

    # The time we give Python to start up and exit.
    STARTUP_TIMEOUT = 60 * 5
    EXIT_TIMEOUT = 60 * 10

    # The time we allow processes to be silent. This is in place so that we
    # eventually catch hanging processes, and print the remainder of our output.
    # Do not increase this. Instead, adjust your program to print regular
    # progress updates, so that cbuildbot (and buildbot) can know that it has
    # not hung.
    SILENT_TIMEOUT = 60 * 145

    # The amount by which we reduce the SILENT_TIMEOUT every time we launch
    # a subprocess. This helps ensure that children get a chance to enforce the
    # SILENT_TIMEOUT prior to the parents enforcing it.
    SILENT_TIMEOUT_STEP = 30
    MINIMUM_SILENT_TIMEOUT = 60 * 135

    # The time before terminating or killing a task.
    SIGTERM_TIMEOUT = 30
    SIGKILL_TIMEOUT = 60

    # How long we allow debug commands to run (so we don't hang while trying to
    # recover from a hang).
    DEBUG_CMD_TIMEOUT = 60

    # Interval we check for updates from print statements.
    PRINT_INTERVAL = 1

    def __init__(
        self, task, queue, semaphore=None, task_args=None, task_kwargs=None
    ):
        """Create a new _BackgroundTask object.

        If semaphore is supplied, it will be acquired for the duration of the
        steps that are run in the background. This can be used to limit the
        number of simultaneous parallel tasks.

        Args:
            task: The task (a functor) to run in the background.
            queue: A queue to be used for managing communication between the
                parent and child process. This queue must be valid for the
                length of the life of the child process, until the parent has
                collected its status.
            semaphore: The lock to hold while |task| runs.
            task_args: A list of args to pass to the |task|.
            task_kwargs: A dict of optional args to pass to the |task|.
        """
        multiprocessing.Process.__init__(self)
        self._task = task
        self._queue = queue
        self._semaphore = semaphore
        self._started = multiprocessing.Event()
        self._killing = multiprocessing.Event()
        self._output = None
        self._parent_pid = None
        self._task_args = task_args if task_args else ()
        self._task_kwargs = task_kwargs if task_kwargs else {}

    def _WaitForStartup(self):
        # TODO(davidjames): Use python-2.7 syntax to simplify this.
        self._started.wait(self.STARTUP_TIMEOUT)
        msg = "Process failed to start in %d seconds" % self.STARTUP_TIMEOUT
        assert self._started.is_set(), msg

    def _DecodeToUtf8(self, raw_bytes):
        """Decode buffer to utf-8 in python2 or python3."""
        try:
            buf = raw_bytes.decode("utf-8", "backslashreplace")
        except TypeError:
            # For python < 3.5 we may encounter errors decoding. One approximate
            # workaround is to decode using latin1, then re-encode and decode
            # with backslashreplace so that something like 0x80 ends up as
            # u'\\x80' and not u'\x80'.
            buf = (
                raw_bytes.decode("latin1")
                .encode("ascii", "backslashreplace")
                .decode("ascii")
            )
        return buf

    @classmethod
    def _DebugRunCommand(cls, cmd, **kwargs):
        """Swallow any exception run raises.

        Since these commands are for purely informational purposes, we don't
        random issues causing the bot to die.

        Returns:
            Stdout on success
        """
        log_level = kwargs["debug_level"]
        try:
            with timeout_util.Timeout(cls.DEBUG_CMD_TIMEOUT):
                return cros_build_lib.run(cmd, **kwargs).stdout
        except (cros_build_lib.RunCommandError, timeout_util.TimeoutError) as e:
            logging.log(log_level, "Running %s failed: %s", cmd[0], str(e))
            return ""

    # Debug commands to run in gdb.  A class member so tests can stub it out.
    GDB_COMMANDS = (
        "info proc all",
        "info threads",
        "thread apply all py-list",
        "thread apply all py-bt",
        "thread apply all bt",
        "detach",
    )

    @classmethod
    def _DumpDebugPid(cls, log_level: int, pid: str) -> None:
        """Dump debug info about the hanging |pid|."""
        commands = (
            ("pstree", "-Apals", pid),
            ("lsof", "-p", pid),
        )
        for cmd in commands:
            cls._DebugRunCommand(
                cmd,
                debug_level=log_level,
                check=False,
                log_output=True,
                encoding="utf-8",
            )

        stdin = "\n".join(
            ["echo \\n>>> %s\\n\n%s" % (x, x) for x in cls.GDB_COMMANDS]
        )
        cmd = (
            "gdb",
            "--nx",
            "-q",
            "-p",
            pid,
            "-ex",
            "set prompt",
        )
        cls._DebugRunCommand(
            cmd,
            debug_level=log_level,
            check=False,
            log_output=True,
            input=stdin,
            encoding="utf-8",
        )

    def Kill(self, sig, log_level, first=False):
        """Kill process with signal, ignoring if the process is dead.

        Args:
            sig: Signal to send.
            log_level: The log level of log messages.
            first: Whether this is the first signal we've sent.
        """
        self._killing.set()
        self._WaitForStartup()
        if logging.getLogger().isEnabledFor(log_level):
            # Dump debug information about the hanging process.
            logging.log(
                log_level,
                "Killing %r (sig=%r %s)",
                self.pid,
                sig,
                signals.StrSignal(sig),
            )

            if first:
                ppid = str(self.pid)
                output = self._DebugRunCommand(
                    ("pgrep", "-P", ppid),
                    debug_level=log_level,
                    print_cmd=False,
                    check=False,
                    capture_output=True,
                    encoding="utf-8",
                )
                for pid in [ppid] + output.splitlines():
                    self._DumpDebugPid(log_level, pid)

        try:
            os.kill(self.pid, sig)
        except OSError as ex:
            if ex.errno != errno.ESRCH:
                raise

    def Cleanup(self, silent=False):
        """Wait for a process to exit."""
        if os.getpid() != self._parent_pid or self._output is None:
            return
        try:
            # Print output from subprocess.
            if not silent and logging.getLogger().isEnabledFor(logging.DEBUG):
                # Read as binary to avoid utf-8 decoding issues.
                with open(self._output.name, "rb") as f:
                    for line in f:
                        textline = self._DecodeToUtf8(line)
                        logging.debug(textline.rstrip("\n"))
        finally:
            # Clean up our temporary file.
            osutils.SafeUnlink(self._output.name)
            self._output.close()
            self._output = None

    def Wait(self):
        """Wait for the task to complete.

        Output from the task is printed as it runs.

        If an exception occurs, return a string containing the traceback.
        """
        try:
            # Flush stdout and stderr to be sure no output is interleaved.
            sys.stdout.flush()
            sys.stderr.flush()

            # File position pointers are shared across processes, so we must
            # open our own file descriptor to ensure output is not lost.
            self._WaitForStartup()
            silent_death_time = time.time() + self.SILENT_TIMEOUT
            results = []
            # Read as binary to avoid utf-8 decoding issues.
            with open(self._output.name, "rb") as output:
                pos = 0
                running, exited_cleanly, task_errors, run_errors = (
                    True,
                    False,
                    [],
                    [],
                )
                while running:
                    # Check whether the process is still alive.
                    running = self.is_alive()

                    try:
                        errors, results = self._queue.get(
                            True, self.PRINT_INTERVAL
                        )
                        if errors:
                            task_errors.extend(errors)

                        running = False
                        exited_cleanly = True
                    except Queue.Empty:
                        pass

                    if not running:
                        # Wait for the process to actually exit. If the child
                        # doesn't exit in a timely fashion, kill it.
                        self.join(self.EXIT_TIMEOUT)
                        if self.exitcode is None:
                            msg = "%r hung for %r seconds" % (
                                self,
                                self.EXIT_TIMEOUT,
                            )
                            run_errors.extend(
                                failures_lib.CreateExceptInfo(
                                    ProcessExitTimeout(msg), ""
                                )
                            )
                            self._KillChildren([self])
                        elif not exited_cleanly:
                            msg = "%r exited unexpectedly with code %s" % (
                                self,
                                self.exitcode,
                            )
                            run_errors.extend(
                                failures_lib.CreateExceptInfo(
                                    ProcessUnexpectedExit(msg), ""
                                )
                            )

                    # Read output from process.
                    output.seek(pos)
                    buf = self._DecodeToUtf8(output.read(_BUFSIZE))
                    if buf:
                        silent_death_time = time.time() + self.SILENT_TIMEOUT
                    elif running and time.time() > silent_death_time:
                        msg = "No output from %r for %r seconds" % (
                            self,
                            self.SILENT_TIMEOUT,
                        )
                        run_errors.extend(
                            failures_lib.CreateExceptInfo(
                                ProcessSilentTimeout(msg), ""
                            )
                        )
                        self._KillChildren([self])

                        # Read remaining output from the process.
                        output.seek(pos)
                        buf = self._DecodeToUtf8(output.read(_BUFSIZE))
                        running = False

                    # Print output so far.
                    while buf:
                        sys.stdout.write(buf)
                        pos += len(buf)
                        if len(buf) < _BUFSIZE:
                            break
                        buf = self._DecodeToUtf8(output.read(_BUFSIZE))

                    # Print error messages if anything exceptional occurred.
                    if run_errors:
                        cbuildbot_alerts.PrintBuildbotStepFailure()
                        traceback.print_stack()
                        logging.warning(
                            "\n".join(x.str for x in run_errors if x)
                        )
                        logging.info("\n".join(x.str for x in task_errors if x))

                    sys.stdout.flush()
                    sys.stderr.flush()

            # Propagate any results.
            for result in results:
                results_lib.Results.Record(*result)

        finally:
            self.Cleanup(silent=True)

        # If an error occurred, return it.
        return run_errors + task_errors

    def start(self):
        """Invoke multiprocessing.Process.start after flushing output/err."""
        if self.SILENT_TIMEOUT < self.MINIMUM_SILENT_TIMEOUT:
            raise AssertionError(
                "Maximum recursion depth exceeded in %r" % self
            )

        sys.stdout.flush()
        sys.stderr.flush()
        tmp_dir = "/tmp/chromite.parallel.%d" % os.geteuid()
        osutils.SafeMakedirs(tmp_dir, mode=0o1777)
        self._output = cros_build_lib.UnbufferedNamedTemporaryFile(
            delete=False, dir=tmp_dir, prefix="chromite-parallel-"
        )
        self._parent_pid = os.getpid()
        return multiprocessing.Process.start(self)

    def run(self):
        """Run the list of steps."""
        if self._semaphore is not None:
            self._semaphore.acquire()

        errors = failures_lib.CreateExceptInfo(
            UnexpectedException("Unexpected exception in %r" % self), ""
        )
        pid = os.getpid()
        try:
            errors = self._Run()
        finally:
            if not self._killing.is_set() and os.getpid() == pid:
                results = results_lib.Results.Get()
                self._queue.put((errors, results))
                if self._semaphore is not None:
                    self._semaphore.release()

    def _Run(self):
        """Internal method for running the list of steps."""

        # Register a handler for a signal that is rarely used.
        def trigger_bt(_sig_num, frame):
            logging.error(
                "pre-kill notification (SIGXCPU); traceback:\n%s",
                "".join(traceback.format_stack(frame)),
            )

        signal.signal(signal.SIGXCPU, trigger_bt)

        sys.stdout.flush()
        sys.stderr.flush()
        errors = []
        # Send all output to a named temporary file.
        with open(self._output.name, "wb", 0) as output:
            # Back up sys.std{err,out}. These aren't used, but we keep a copy so
            # that they aren't garbage collected. We intentionally don't restore
            # the old stdout and stderr at the end, because we want shutdown
            # errors to also be sent to the same log file.
            _orig_stdout, _orig_stderr = sys.stdout, sys.stderr

            # Replace std{out,err} with unbuffered file objects.
            os.dup2(output.fileno(), sys.__stdout__.fileno())
            os.dup2(output.fileno(), sys.__stderr__.fileno())
            sys.stdout = os.fdopen(sys.__stdout__.fileno(), "w", closefd=False)
            sys.stderr = os.fdopen(sys.__stderr__.fileno(), "w", closefd=False)

            try:
                self._started.set()
                results_lib.Results.Clear()

                # Reduce the silent timeout by the prescribed amount.
                cls = self.__class__
                cls.SILENT_TIMEOUT -= cls.SILENT_TIMEOUT_STEP

                # Actually launch the task.
                self._task(*self._task_args, **self._task_kwargs)
            except failures_lib.StepFailure as ex:
                errors.extend(
                    failures_lib.CreateExceptInfo(ex, traceback.format_exc())
                )
            except BaseException as ex:
                errors.extend(
                    failures_lib.CreateExceptInfo(ex, traceback.format_exc())
                )
                if self._killing.is_set():
                    traceback.print_exc()
            finally:
                sys.stdout.flush()
                sys.stderr.flush()

        return errors

    @classmethod
    def _KillChildren(cls, bg_tasks, log_level=logging.WARNING):
        """Kill a deque of background tasks.

        This is needed to prevent hangs in the case where child processes refuse
        to exit.

        Args:
            bg_tasks: A list filled with _BackgroundTask objects.
            log_level: The log level of log messages.
        """
        logging.log(log_level, "Killing tasks: %r", bg_tasks)
        siglist = (
            (signal.SIGXCPU, cls.SIGTERM_TIMEOUT),
            (signal.SIGTERM, cls.SIGKILL_TIMEOUT),
            (signal.SIGKILL, None),
        )
        first = True
        for sig, timeout in siglist:
            # Send signal to all tasks.
            for task in bg_tasks:
                task.Kill(sig, log_level, first)
            first = False

            # Wait for all tasks to exit, if requested.
            if timeout is None:
                for task in bg_tasks:
                    task.join()
                    task.Cleanup()
                break

            # Wait until timeout expires.
            end_time = time.time() + timeout
            while bg_tasks:
                time_left = end_time - time.time()
                if time_left <= 0:
                    break
                task = bg_tasks[-1]
                task.join(time_left)
                if task.exitcode is not None:
                    task.Cleanup()
                    bg_tasks.pop()

    @classmethod
    @contextlib.contextmanager
    def ParallelTasks(cls, steps, max_parallel=None, halt_on_error=False):
        """Run a list of functions in parallel.

        This function launches the provided functions in the background, yields,
        and then waits for the functions to exit.

        The output from the functions is saved to a temporary file and printed
        as if they were run in sequence.

        If exceptions occur in the steps, we join together the tracebacks and
        print them after all parallel tasks have finished running. Further, a
        BackgroundFailure is raised with full stack traces of all exceptions.

        Args:
            steps: A list of functions to run.
            max_parallel: The maximum number of simultaneous tasks to run in
                parallel. By default, run all tasks in parallel.
            halt_on_error: After the first exception occurs, halt any running
                steps, and squelch any further output, including any exceptions
                that might occur.
        """

        semaphore = None
        if max_parallel is not None:
            semaphore = multiprocessing.Semaphore(max_parallel)

        # First, start all the steps.
        with Manager() as manager:
            bg_tasks = collections.deque()
            for step in steps:
                task = cls(step, queue=manager.Queue(), semaphore=semaphore)
                task.start()
                bg_tasks.append(task)

            foreground_except = None
            try:
                yield
            except BaseException:
                foreground_except = sys.exc_info()
            finally:
                errors = []
                skip_bg_wait = halt_on_error and foreground_except is not None
                # Wait for each step to complete.
                while not skip_bg_wait and bg_tasks:
                    task = bg_tasks.popleft()
                    task_errors = task.Wait()
                    if task_errors:
                        errors.extend(task_errors)
                        if halt_on_error:
                            break

                # If there are still tasks left, kill them.
                if bg_tasks:
                    cls._KillChildren(bg_tasks, log_level=logging.DEBUG)

                # Propagate any exceptions; foreground exceptions take
                # precedence.
                if foreground_except is not None:
                    # contextlib ignores caught exceptions unless explicitly
                    # re-raised.
                    raise foreground_except[1].with_traceback(
                        foreground_except[2]
                    )
                if errors:
                    raise BackgroundFailure(exc_infos=errors)

    @staticmethod
    def TaskRunner(queue, task, onexit=None, task_args=None, task_kwargs=None):
        """Run task(*input) for each input in the queue.

        Returns when it encounters an _AllTasksComplete object on the queue.
        If exceptions occur, save them off and re-raise them as a
        BackgroundFailure once we've finished processing the items in the queue.

        Args:
            queue: A queue of tasks to run. Add tasks to this queue, and they
                will be run.
            task: Function to run on each queued input.
            onexit: Function to run after all inputs are processed.
            task_args: A list of args to pass to the |task|.
            task_kwargs: A dict of optional args to pass to the |task|.
        """
        if task_args is None:
            task_args = []
        elif not isinstance(task_args, list):
            task_args = list(task_args)
        if task_kwargs is None:
            task_kwargs = {}

        errors = []
        while True:
            # Wait for a new item to show up on the queue. This is a blocking
            # wait, so if there's nothing to do, we just sit here.
            x = queue.get()
            if isinstance(x, _AllTasksComplete):
                # All tasks are complete, so we should exit.
                break
            elif not isinstance(x, list):
                x = task_args + list(x)
            else:
                x = task_args + x

            # If no tasks failed yet, process the remaining tasks.
            if not errors:
                try:
                    task(*x, **task_kwargs)
                except BaseException as ex:
                    errors.extend(
                        failures_lib.CreateExceptInfo(
                            ex, traceback.format_exc()
                        )
                    )

        # Run exit handlers.
        if onexit:
            onexit()

        # Propagate any exceptions.
        if errors:
            raise BackgroundFailure(exc_infos=errors)


def RunParallelSteps(
    steps, max_parallel=None, halt_on_error=False, return_values=False
):
    """Run a list of functions in parallel.

    This function blocks until all steps are completed.

    The output from the functions is saved to a temporary file and printed as if
    they were run in sequence.

    If exceptions occur in the steps, we join together the tracebacks and print
    them after all parallel tasks have finished running. Further, a
    BackgroundFailure is raised with full stack traces of all exceptions.

    Examples:
        # This snippet will execute in parallel:
        #   somefunc()
        #   anotherfunc()
        #   funcfunc()
        steps = [somefunc, anotherfunc, funcfunc]
        RunParallelSteps(steps)
        # Blocks until all calls have completed.

    Args:
        steps: A list of functions to run.
        max_parallel: The maximum number of simultaneous tasks to run in
            parallel. By default, run all tasks in parallel.
        halt_on_error: After the first exception occurs, halt any running steps,
            and squelch any further output, including any exceptions that might
            occur.
        return_values: If set to True, RunParallelSteps returns a list
            containing the return values of the steps.  Defaults to False.

    Returns:
        If |return_values| is True, the function will return a list containing
        the return values of the steps.
    """

    def ReturnWrapper(queue, fn):
        """Put the return value of |fn| into |queue|."""
        queue.put(fn())

    full_steps = []
    queues = []
    with contextlib.ExitStack() as stack:
        if return_values:
            # We use a managed queue here, because the child process will wait
            # for the queue(pipe) to be flushed (i.e., when items are read from
            # the queue) before exiting, and with a regular queue this may
            # result in hangs for large return values.  But with a managed
            # queue, the manager process will read the items and hold on to them
            # until the managed queue goes out of scope and is cleaned up.
            manager = stack.enter_context(Manager())
            for step in steps:
                queue = manager.Queue()
                queues.append(queue)
                full_steps.append(functools.partial(ReturnWrapper, queue, step))
        else:
            full_steps = steps

        with _BackgroundTask.ParallelTasks(
            full_steps, max_parallel=max_parallel, halt_on_error=halt_on_error
        ):
            pass

        if return_values:
            return [queue.get_nowait() for queue in queues]


class _AllTasksComplete:
    """Sentinel object to indicate that all tasks are complete."""


@contextlib.contextmanager
def BackgroundTaskRunner(task, *args, **kwargs):
    """Run the specified task on each queued input in a pool of processes.

    This context manager starts a set of workers in the background, who each
    wait for input on the specified queue. For each input on the queue, these
    workers run task(*args + *input, **kwargs). Note that certain kwargs will
    not pass through to the task (see Args below for the list).

    The output from these tasks is saved to a temporary file. When control
    returns to the context manager, the background output is printed in order,
    as if the tasks were run in sequence.

    If exceptions occur in the steps, we join together the tracebacks and print
    them after all parallel tasks have finished running. Further, a
    BackgroundFailure is raised with full stack traces of all exceptions.

    Examples:
        # This will run somefunc(1, 'small', 'cow', foo='bar') in the background
        # as soon as data is added to the queue (i.e. queue.put() is called).

        def somefunc(arg1, arg2, arg3, foo=None):
            ...

        with BackgroundTaskRunner(somefunc, 1, foo='bar') as queue:
            ... do random stuff ...
            queue.put(['small', 'cow'])
            ... do more random stuff while somefunc() runs ...
        # Exiting the with statement will block until all calls have completed.

    Args:
        task: Function to run on each queued input.
        queue: A queue of tasks to run. Add tasks to this queue, and they will
            be run in the background.  If None, one will be created on the fly.
        processes: Number of processes to launch.
        onexit: Function to run in each background process after all inputs are
            processed.
        halt_on_error: After the first exception occurs, halt any running steps,
            and squelch any further output, including any exceptions that might
            occur. Halts on exceptions in any of the background processes, or in
            the foreground process using the BackgroundTaskRunner.
    """

    queue = kwargs.pop("queue", None)
    processes = kwargs.pop("processes", None)
    onexit = kwargs.pop("onexit", None)
    halt_on_error = kwargs.pop("halt_on_error", False)

    with contextlib.ExitStack() as stack:
        if queue is None:
            manager = stack.enter_context(Manager())
            queue = manager.Queue()

        if not processes:
            processes = multiprocessing.cpu_count()

        child = functools.partial(
            _BackgroundTask.TaskRunner,
            queue,
            task,
            onexit=onexit,
            task_args=args,
            task_kwargs=kwargs,
        )
        steps = [child] * processes
        with _BackgroundTask.ParallelTasks(steps, halt_on_error=halt_on_error):
            try:
                yield queue
            finally:
                for _ in range(processes):
                    queue.put(_AllTasksComplete())


def RunTasksInProcessPool(task, inputs, processes=None, onexit=None):
    """Run |task| with each supplied input in a pool of processes.

    This function runs task(*x) for x in inputs in a pool of processes. This
    function blocks until all tasks are completed.

    The output from these tasks is saved to a temporary file. When control
    returns to the context manager, the background output is printed in order,
    as if the tasks were run in sequence.

    If exceptions occur in the steps, we join together the tracebacks and print
    them after all parallel tasks have finished running. Further, a
    BackgroundFailure is raised with full stack traces of all exceptions.

    Examples:
        # This snippet will execute in parallel:
        #   somefunc('hi', 'fat', 'code')
        #   somefunc('foo', 'bar', 'cow')

        def somefunc(arg1, arg2, arg3):
            ...
        ...
        inputs = [
            ['hi', 'fat', 'code'],
            ['foo', 'bar', 'cow'],
        ]
        RunTasksInProcessPool(somefunc, inputs)
        # Blocks until all calls have completed.

    Args:
        task: Function to run on each input.
        inputs: List of inputs.
        processes: Number of processes, at most, to launch.
        onexit: Function to run in each background process after all inputs are
            processed.

    Returns:
        Returns a list containing the return values of the task for each input.
    """
    if not processes:
        # Use >=16 processes by default, in case it's a network-bound operation.
        # Try to use all the CPUs, in case it's a CPU-bound operation.
        processes = min(max(16, multiprocessing.cpu_count()), len(inputs))

    with Manager() as manager:
        # Set up output queue.
        out_queue = manager.Queue()
        fn = lambda idx, task_args: out_queue.put((idx, task(*task_args)))

        # Micro-optimization: Set up the queue so that BackgroundTaskRunner
        # doesn't have to set up another Manager process.
        queue = manager.Queue()

        with BackgroundTaskRunner(
            fn, queue=queue, processes=processes, onexit=onexit
        ) as queue:
            for idx, input_args in enumerate(inputs):
                queue.put((idx, input_args))

        return [
            x[1] for x in sorted(out_queue.get() for _ in range(len(inputs)))
        ]


def ExitWithParent(sig=signal.SIGHUP):
    """Sets this process to receive |sig| when the parent dies.

    Note: this uses libc, so it only works on linux.

    Args:
        sig: Signal to receive. Defaults to SIGHUP.

    Returns:
        Whether we were successful in setting the deathsignal flag
    """
    return prctl.prctl(prctl.Option.SET_PDEATHSIG, sig) is not None