Helpers API¶

Overview

canfar.helpers.distributed partitions an ordered list across Session replicas. Each replica container receives REPLICA_ID and REPLICA_COUNT in its environment, and these functions read them by default.

Practical examples¶

Stripe: take every Nth item with an offset¶

from canfar.helpers import distributed

# Assume REPLICA_ID=2 and REPLICA_COUNT=4
# Replica 2 (1-based) will see indices 1, 5, 9, ...
items = list(range(12))
shard = list(distributed.stripe(items, replica=2, total=4))
print(shard)  # [1, 5, 9]

Chunk: contiguous chunks of roughly equal size¶

from canfar.helpers import distributed

# Assume 10 items, 4 replicas
items = list(range(10))
# Replica 1 gets [0,1], 2->[2,3], 3->[4,5], 4->[6,7,8,9] (last takes remainder)
print(list(distributed.chunk(items, replica=1, total=4)))
print(list(distributed.chunk(items, replica=4, total=4)))

Sparse distribution

When items < replicas, chunk assigns exactly one item to each of the first len(items) replicas, and later replicas get nothing. This avoids duplication.

Using container-provided environment variables¶

# Inside CANFAR container replicas, you can omit replica/total and read from env
from canfar.helpers import distributed
work = list(range(1000))
for item in distributed.chunk(work):
    process(item)

Validation and errors¶

chunk() requires total > 0 and 1 <= replica <= total, raising ValueError otherwise. stripe() intentionally keeps a looser compatibility contract: replica < 1 yields no items, even when total <= 0. For a positive replica, total <= 0 raises ValueError; otherwise replica > total yields no items. Both functions use REPLICA_ID and REPLICA_COUNT when their corresponding arguments are omitted, defaulting to one replica when the environment is absent.

API reference¶

canfar.helpers.distributed ¶

Helper functions for distributed computing.

stripe ¶

stripe(
    iterable: Iterable[T], replica: int | None = None, total: int | None = None
) -> Iterator[T]

Return every total-th item from iterable with a replica offset.

Uses 1-based replica indexing to match REPLICA_ID in CANFAR containers. Replica 1 receives indices 0, total, 2 * total, …; replica 2 receives indices 1, total + 1, …; and so on.

Unlike chunk, stripe returns no items when replica < 1, even when total <= 0. For a positive replica, total <= 0 raises ValueError; otherwise replica > total returns no items.

PARAMETER	DESCRIPTION
`iterable`	The iterable to stripe across replicas. TYPE: `Iterable[T]`
`replica`	The replica number (1-based). Defaults to `REPLICA_ID`. TYPE: `int \| None` DEFAULT: `None`
`total`	The total number of replicas. Defaults to `REPLICA_COUNT`. TYPE: `int \| None` DEFAULT: `None`

YIELDS	DESCRIPTION
`T`	Items assigned to this replica.

Examples:

>>> from canfar.helpers import distributed
>>> list(distributed.stripe(range(10), replica=1, total=3))
[0, 3, 6, 9]
>>> list(distributed.stripe(range(10), replica=2, total=3))
[1, 4, 7]
>>> list(distributed.stripe(range(100), replica=1, total=10))
[0, 10, 20, 30, 40, 50, 60, 70, 80, 90]

chunk ¶

chunk(
    iterable: Iterable[T], replica: int | None = None, total: int | None = None
) -> Iterator[T]

Return the replica-th contiguous chunk of iterable.

Splits iterable into total roughly equal contiguous chunks using 1-based replica indexing to match REPLICA_ID in CANFAR containers.

Distribution behavior:

Standard (len(items) >= total): Items are divided into equal-sized chunks; the last replica receives any remainder.
Sparse (len(items) < total): Each of the first len(items) replicas receives exactly one item; remaining replicas get nothing.

PARAMETER	DESCRIPTION
`iterable`	The iterable to distribute across replicas. TYPE: `Iterable[T]`
`replica`	The replica number (1-based). Must be >= 1 and <= `total`. Defaults to `REPLICA_ID`. TYPE: `int \| None` DEFAULT: `None`
`total`	The total number of replicas. Must be > 0. Defaults to `REPLICA_COUNT`. TYPE: `int \| None` DEFAULT: `None`

YIELDS	DESCRIPTION
`T`	Items assigned to this replica.

RAISES	DESCRIPTION
`ValueError`	If `replica` < 1, `replica` > `total`, or `total` <= 0.

Examples:

>>> from canfar.helpers import distributed
>>> list(distributed.chunk(range(12), replica=1, total=3))
[0, 1, 2, 3]
>>> list(distributed.chunk(range(10), replica=4, total=4))
[6, 7, 8, 9]
>>> list(distributed.chunk([1, 2, 3], replica=2, total=5))
[2]