Class DefaultExecutionPool

deps/v8/tools/testrunner/local/pool.py:171–380

Distributes tasks to a number of worker processes. New tasks can be added dynamically even after the workers have been started. Requirement: Tasks can only be added from the parent process, e.g. while consuming the results generator.

class DefaultExecutionPool(ContextPool):
  """Distributes tasks to a number of worker processes.
  New tasks can be added dynamically even after the workers have been started.
  Requirement: Tasks can only be added from the parent process, e.g. while
  consuming the results generator."""

  # Factor to calculate the maximum number of items in the work/done queue.
  # Necessary to not overflow the queue's pipe if a keyboard interrupt happens.
  BUFFER_FACTOR = 4

  def __init__(self, os_context=None):
    super(DefaultExecutionPool, self).__init__()
    self.os_context = os_context
    self.processes = []
    self.terminated = False

    # Invariant: processing_count >= #work_queue + #done_queue. It is greater
    # when a worker takes an item from the work_queue and before the result is
    # submitted to the done_queue. It is equal when no worker is working,
    # e.g. when all workers have finished, and when no results are processed.
    # Count is only accessed by the parent process. Only the parent process is
    # allowed to remove items from the done_queue and to add items to the
    # work_queue.
    self.processing_count = 0

    # Disable sigint and sigterm to prevent subprocesses from capturing the
    # signals.
    with without_sig():
      self.work_queue = Queue()
      self.done_queue = Queue()

  def init(self, num_workers=1, heartbeat_timeout=1, notify_function=None):
    """
    Args:
      num_workers: Number of worker processes to run in parallel.
      heartbeat_timeout: Timeout in seconds for waiting for results. Each time
          the timeout is reached, a heartbeat is signalled and timeout is reset.
      notify_function: Callable called to signal some events like termination. The
          event name is passed as string.
    """
    self.num_workers = num_workers
    self.heartbeat_timeout = heartbeat_timeout
    self.notify = notify_function or (lambda x: x)

  def add_jobs(self, jobs):
    self.add(jobs)

  def results(self, requirement):
    return self.imap_unordered(
        fn=run_job,
        gen=[],
        process_context_fn=ProcessContext,
        process_context_args=[requirement],
    )

  def imap_unordered(self, fn, gen,
                     process_context_fn=None, process_context_args=None):
    """Maps function "fn" to items in generator "gen" on the worker processes