Skip to content

parallel

parallel #

The one thread pool every pmap runs on, and the parallel broadcast rule.

  1. Pool. One process-wide pool, sized to the CPU count. A pmap made from inside a worker runs sequentially, the outer pool being the parallelism, so nested fan-out never oversubscribes and never waits on the pool it occupies.
  2. Cap. parallelism(workers) bounds how many elements each pmap in the block has in flight at once; the default is the pool's size.
  3. Broadcast. pbroadcast is array.broadcast across the pool: the same results, the same failure rule, a progress step counting the elements as they finish, each worker in a copy of the caller's context so the active scratch, workspace and step carry over.

POOL_SIZE = os.cpu_count() or 1 module-attribute #

How many threads the pool runs, and the default cap on elements in flight.

parallelism(workers) #

Cap how many elements each pmap in the block has in flight at once.

with cg.parallelism(4):
    thumbnails = sessions.photo.pmap(session_thumbnail)

Parameters:

Name Type Description Default
workers int

The cap; 1 runs every pmap in the block sequentially.

 required
Source code in capturegraph-lib/capturegraph/types/containers/parallel.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@contextmanager
def parallelism(workers: int) -> Iterator[None]:
    """Cap how many elements each ``pmap`` in the block has in flight at once.

    ```python
    with cg.parallelism(4):
        thumbnails = sessions.photo.pmap(session_thumbnail)
    ```

    Args:
        workers: The cap; ``1`` runs every ``pmap`` in the block sequentially.
    """
    token = _in_flight.set(max(workers, 1))
    try:
        yield
    finally:
        _in_flight.reset(token)

pbroadcast(items, operation, function, label) #

broadcast across the pool: the parallel rule Array and Map share.

Parameters:

Name Type Description Default
items Iterable[I]

The elements to apply function to.

 required
operation str

Names this broadcast in the failure message.

 required
function Callable[[I], R]

The per-element operation.

 required
label str | None

What the step counting the elements shows; function's name when None.

 required

Returns:

Type Description
list[R]

What broadcast returns, in order.
Source code in capturegraph-lib/capturegraph/types/containers/parallel.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def pbroadcast[I, R](
    items: Iterable[I],
    operation: str,
    function: Callable[[I], R],
    label: str | None,
) -> list[R]:
    """``broadcast`` across the pool: the parallel rule ``Array`` and ``Map`` share.

    Args:
        items: The elements to apply ``function`` to.
        operation: Names this broadcast in the failure message.
        function: The per-element operation.
        label: What the step counting the elements shows; ``function``'s name
            when ``None``.

    Returns:
        What ``broadcast`` returns, in order.
    """
    from capturegraph.recipes.progress.step import label_of, step
    from capturegraph.types.containers.array import broadcast, isolated

    items = list(items)
    with step(label or label_of(function), total=len(items)) as progress:
        if _inside_pool.get() or len(items) < 2:
            return broadcast(progress.track(items), operation, function)
        pool = _shared_pool()
        cap = _in_flight.get()
        pending: deque[Future[R]] = deque()
        results: list[R] = []
        for item in items:
            if len(pending) == cap:
                results.append(pending.popleft().result())
            context = copy_context()
            context.run(_inside_pool.set, True)
            future = pool.submit(context.run, isolated, function, item)
            future.add_done_callback(lambda _: progress.advance())
            pending.append(future)
        results.extend(future.result() for future in pending)
        return guard_broadcast(results, operation)