Helpers API¶

Overview

The Helpers API provides utility functions to partition work across replicas of a CANFAR session. Containers receive REPLICA_ID and REPLICA_COUNT environment variables, and these helpers make using them simple and correct.

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¶

replica must be >= 1 and <= total
total must be > 0

API Reference¶

canfar.helpers.distributed ¶

Helper functions for distributed computing.

stripe ¶

stripe(
    iterable: Iterable[T],
    replica: int = int(os.environ.get("REPLICA_ID", "1")),
    total: int = int(os.environ.get("REPLICA_COUNT", "1")),
) -> 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 does not validate replica or total. When replica is less than 1, or greater than total, the result is empty. When total is zero, ValueError is raised.

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

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 = int(os.environ.get("REPLICA_ID", "1")),
    total: int = int(os.environ.get("REPLICA_COUNT", "1")),
) -> 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` DEFAULT: `int(get('REPLICA_ID', '1'))`
`total`	The total number of replicas. Must be > 0. Defaults to `REPLICA_COUNT`. TYPE: `int` DEFAULT: `int(get('REPLICA_COUNT', '1'))`

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]