Function execute_parallel

pex/jobs.py:531–655

Execute jobs for the given inputs in parallel subprocesses. :param int max_jobs: The maximum number of parallel jobs to spawn. :param inputs: An iterable of the data to parallelize over `spawn_func`. :param spawn_func: A function taking a single input and returning a :class:`SpawnedJob`

(
    inputs,  # type: Iterable[_I]
    spawn_func,  # type: Callable[[_I], SpawnedJob[_O]]
    error_handler=None,  # type: Optional[ErrorHandler[_I, _SE, _JE]]
    max_jobs=None,  # type: Optional[int]
)

def execute_parallel(
    inputs,  # type: Iterable[_I]
    spawn_func,  # type: Callable[[_I], SpawnedJob[_O]]
    error_handler=None,  # type: Optional[ErrorHandler[_I, _SE, _JE]]
    max_jobs=None,  # type: Optional[int]
):
    # type: (...) -> Iterator[Union[_O, _SE, _JE]]
    """Execute jobs for the given inputs in parallel subprocesses.

    :param int max_jobs: The maximum number of parallel jobs to spawn.
    :param inputs: An iterable of the data to parallelize over `spawn_func`.
    :param spawn_func: A function taking a single input and returning a :class:`SpawnedJob`.
    :param error_handler: An optional :class:`ErrorHandler`, defaults to :class:`Log`.
    :returns: An iterator over the spawned job results as they come in.
    :raises: A `raise_type` exception if any individual job errors and `raise_type` is not `None`.
    """
    handler = (
        error_handler or Log["_I", "_O"]()
    )  # type: Union[ErrorHandler[_I, _SE, _JE], Log[_I, _O]]
    size = _sanitize_max_jobs(max_jobs)
    TRACER.log(
        "Spawning a maximum of {} parallel jobs to process:\n  {}".format(
            size, "\n  ".join(map(str, inputs))
        ),
        V=9,
    )

    @attr.s(frozen=True)
    class Spawn(object):
        item = attr.ib()  # type: Any
        spawned_job = attr.ib()  # type: SpawnedJob

    @attr.s(frozen=True)
    class SpawnError(object):
        item = attr.ib()  # type: Any
        error = attr.ib()  # type: Exception

    stop = Event()  # Used as a signal to stop spawning further jobs once any one job fails.
    job_slots = BoundedSemaphore(value=size)

    class DoneSentinel(object):
        pass

    done_sentinel = DoneSentinel()
    spawn_queue = Queue()  # type: Queue[Union[Spawn, SpawnError, DoneSentinel]]

    def spawn_jobs():
        for item in inputs:
            if stop.is_set():
                break
            job_slots.acquire()
            try:
                result = Spawn(item, spawn_func(item))
            except Exception as e:
                result = SpawnError(item, e)
            finally:
                spawn_queue.put(result)
        spawn_queue.put(done_sentinel)