Pool API Reference

ProcessPoolExecutor(max_workers=None, mp_context=None, initializer=None, initargs=(),
                    *, max_tasks_per_child=None)          # max_tasks_per_child: 3.11
executor.submit(fn, /, *args, **kwargs) -> Future
executor.map(fn, *iterables, timeout=None, chunksize=1, buffersize=None)   # buffersize: 3.14
executor.shutdown(wait=True, *, cancel_futures=False)     # cancel_futures: 3.9
executor.terminate_workers() / executor.kill_workers()    # 3.14

import concurrent.futures as cf
import multiprocessing as mp


def score(prompt: str) -> int:
    return sum(map(ord, prompt)) % 97


if __name__ == "__main__":
    prompts = [f"prompt {i}" for i in range(100)]
    with cf.ProcessPoolExecutor(max_workers=4, mp_context=mp.get_context("spawn")) as ex:
        ordered = list(ex.map(score, prompts, chunksize=25))            # input order
        futs = {ex.submit(score, p): i for i, p in enumerate(prompts)}
        by_index = {futs[f]: f.result() for f in cf.as_completed(futs)}  # completion order
    assert ordered == [by_index[i] for i in range(100)]
    print(len(ordered))   # 100

multiprocessing.Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None, context=None)
pool.map(func, iterable, chunksize=None)               pool.map_async(func, iterable, chunksize=None,
pool.imap(func, iterable, chunksize=1)                                callback=None, error_callback=None)
pool.imap_unordered(func, iterable, chunksize=1)       pool.starmap(func, iterable, chunksize=None)
pool.apply(func, args=(), kwds={})                     pool.starmap_async(..., callback=None, error_callback=None)
pool.apply_async(func, args=(), kwds={}, callback=None, error_callback=None)
pool.close(); pool.join(); pool.terminate()

import multiprocessing as mp


def count_words(line: str) -> int:
    return len(line.split())


if __name__ == "__main__":
    lines = (f"word {i} " * (i % 5) for i in range(10_000))        # a generator: never materialized
    total = 0
    with mp.get_context("spawn").Pool(4) as pool:
        for n in pool.imap_unordered(count_words, lines, chunksize=500):
            total += n
    print(total)   # 40000

import multiprocessing as mp

results, errors = [], []


def parse(doc_id: int) -> tuple:
    if doc_id % 7 == 0:
        raise ValueError(f"doc {doc_id} is encrypted")
    return doc_id, doc_id * 10


if __name__ == "__main__":
    with mp.get_context("spawn").Pool(4) as pool:
        handles = [pool.apply_async(parse, (i,), callback=results.append, error_callback=errors.append)
                   for i in range(1, 30)]
        pool.close()
        pool.join()                     # wait INSIDE the block; callbacks have all run after join()
    print(len(results), sorted(str(e) for e in errors)[:2])   # 25 ['doc 14 is encrypted', 'doc 21 is encrypted']

from concurrent.futures import ProcessPoolExecutor


def batched(items: list, size: int):
    for i in range(0, len(items), size):
        yield items[i:i + size]


def embed_batch(texts: list) -> list:
    return [[float(len(t))] for t in texts]          # stand-in for model.encode(texts, batch_size=64)


if __name__ == "__main__":
    texts = [f"chunk {i}" for i in range(1000)]
    with ProcessPoolExecutor(max_workers=4) as ex:
        vectors = [v for batch in ex.map(embed_batch, batched(texts, 64)) for v in batch]
    print(len(vectors))   # 1000

import concurrent.futures as cf
import itertools


def work(x: int) -> int:
    return x * x


def bounded_map(ex, fn, items, max_in_flight: int):
    it = iter(items)
    pending = {ex.submit(fn, x) for x in itertools.islice(it, max_in_flight)}
    while pending:
        done, pending = cf.wait(pending, return_when=cf.FIRST_COMPLETED)
        for fut in done:
            yield fut.result()
        for x in itertools.islice(it, len(done)):
            pending.add(ex.submit(fn, x))


if __name__ == "__main__":
    with cf.ProcessPoolExecutor(max_workers=4) as ex:
        print(sum(bounded_map(ex, work, range(10_000), max_in_flight=64)))   # 333283335000

import concurrent.futures as cf


def check(x: int) -> int:
    if x == 13:
        raise RuntimeError("bad shard 13")
    return x


if __name__ == "__main__":
    with cf.ProcessPoolExecutor(max_workers=4) as ex:
        futs = [ex.submit(check, i) for i in range(50)]
        done, not_done = cf.wait(futs, return_when=cf.FIRST_EXCEPTION)
        for f in not_done:
            f.cancel()                                   # pending ones are dropped
        errors = [f.exception() for f in done if f.exception()]
    print([str(e) for e in errors])   # ['bad shard 13']

import sys

if sys.version_info >= (3, 14):
    from concurrent.futures import InterpreterPoolExecutor

    def fib(n: int) -> int:
        return n if n < 2 else fib(n - 1) + fib(n - 2)

    if __name__ == "__main__":
        with InterpreterPoolExecutor(max_workers=4) as ex:
            print(list(ex.map(fib, [20, 21, 22])))   # [6765, 10946, 17711]

ctx.Process(target=None, name=None, args=(), kwargs={}, *, daemon=None)
p.start(); p.join(timeout=None); p.is_alive(); p.exitcode; p.terminate() (SIGTERM); p.kill() (SIGKILL, 3.7)
p.interrupt()  (SIGINT, 3.14); p.close() (release resources, 3.7)
ctx.Queue(maxsize=0)   put/get(block, timeout), close(), join_thread(), cancel_join_thread()
ctx.SimpleQueue()      put/get, close() (3.9); no feeder thread
ctx.JoinableQueue()    task_done(), join()
ctx.Pipe(duplex=True)  -> (conn1, conn2); send/recv, poll(timeout)
ctx.Event / Lock / RLock / Semaphore / Condition / Barrier / Value / Array

import multiprocessing as mp

STOP = None


def parser(inbox, outbox) -> None:
    while (item := inbox.get(timeout=60)) is not STOP:
        outbox.put((item, len(item)))
    outbox.put(STOP)


if __name__ == "__main__":
    ctx = mp.get_context("spawn")
    inbox, outbox = ctx.Queue(maxsize=100), ctx.Queue()
    workers = [ctx.Process(target=parser, args=(inbox, outbox)) for _ in range(3)]
    for w in workers:
        w.start()
    for doc in (f"doc-{i}" for i in range(500)):
        inbox.put(doc, timeout=60)          # bounded: back-pressure on the producer
    for _ in workers:
        inbox.put(STOP)
    results, stopped = [], 0
    while stopped < len(workers):           # drain BEFORE join
        msg = outbox.get(timeout=60)
        if msg is STOP:
            stopped += 1
        else:
            results.append(msg)
    for w in workers:
        w.join(timeout=10)
    print(len(results), [w.exitcode for w in workers])   # 500 [0, 0, 0]