def _RunParallelTasks(target_arg_tuples, max_concurrency, get_task_manager,
                      parallel_exception_class):
  """Executes function calls concurrently in separate threads or processes.

  Args:
    target_arg_tuples: list of (target, args, kwargs) tuples. Each tuple
        contains the function to call and the arguments to pass it.
    max_concurrency: int or None. The maximum number of concurrent new
        threads or processes.
    get_task_manager: Callable that accepts an int max_concurrency arg and
        returns a _TaskManager.
    parallel_exception_class: Type of exception to raise upon an exception in
        one of the called functions.

  Returns:
    list of function return values in the order corresponding to the order of
    target_arg_tuples.

  Raises:
    parallel_exception_class: When an exception occurred in any of the called
        functions.
  """
  thread_context = _BackgroundTaskThreadContext()
  max_concurrency = min(max_concurrency, len(target_arg_tuples))
  error_strings = []
  started_task_count = 0
  active_task_count = 0
  with get_task_manager(max_concurrency) as task_manager:
    try:
      while started_task_count < len(target_arg_tuples) or active_task_count:
        if (started_task_count < len(target_arg_tuples) and
            active_task_count < max_concurrency):
          # Start a new task.
          target, args, kwargs = target_arg_tuples[started_task_count]
          task_manager.StartTask(target, args, kwargs, thread_context)
          started_task_count += 1
          active_task_count += 1
          continue

        # Wait for a task to complete.
        task_id = task_manager.AwaitAnyTask()
        active_task_count -= 1
        # If the task failed, it may still be a long time until all remaining
        # tasks complete. Log the failure immediately before continuing to wait
        # for other tasks.
        stacktrace = task_manager.tasks[task_id].traceback
        if stacktrace:
          msg = ('Exception occurred while calling {0}:{1}{2}'.format(
              _GetCallString(target_arg_tuples[task_id]), os.linesep,
              stacktrace))
          logging.error(msg)
          error_strings.append(msg)

    except KeyboardInterrupt:
      logging.error(
          'Received KeyboardInterrupt while executing parallel tasks. Waiting '
          'for %s tasks to clean up.', active_task_count)
      task_manager.HandleKeyboardInterrupt()
      raise

  if error_strings:
    # TODO(skschneider): Combine errors.VmUtil.ThreadException and
    # errors.VmUtil.CalledProcessException so this can be a single exception
    # type.
    raise parallel_exception_class(
        'The following exceptions occurred during parallel execution:'
        '{0}{1}'.format(os.linesep, os.linesep.join(error_strings)))
  results = [task.return_value for task in task_manager.tasks]
  assert len(target_arg_tuples) == len(results), (target_arg_tuples, results)
  return results
dataset/ETHPy150Open GoogleCloudPlatform/PerfKitBenchmarker/perfkitbenchmarker/background_tasks.py/_RunParallelTasks/original