Your first distributed task

A task is one call to a function decorated with @eugo.hpc.distribute.

import eugo.hpc

@eugo.hpc.distribute
def transform(record):
    return {"id": record["id"], "score": record["value"] ** 2}

future = transform({"id": 1, "value": 7})
print(eugo.hpc.get(future))     # {'id': 1, 'score': 49}

Calling a decorated function returns a future at step 2, before any work has happened. The teal path is what get() waits on. Dispatch every task first, and all these round trips overlap; call get() inside the loop and they queue one behind another.

Tasks are stateless

Each call starts fresh. A task knows nothing about previous calls, and nothing it stores in a global persists for the next one.

counter = 0

@eugo.hpc.distribute
def broken_increment():
    global counter
    counter += 1        # mutates a COPY on some worker
    return counter

results = eugo.hpc.get([broken_increment() for _ in range(5)])
print(results)          # not [1, 2, 3, 4, 5]
print(counter)          # still 0 in your session

Each worker got its own copy of the module state. Nothing propagated back.

That is not a limitation to work around. Statelessness is what makes tasks safely parallelizable. When you genuinely need state that persists across calls, that is an actor.

What a task can rely on

• Its arguments. Serialized from your session, copied to the worker.
• Its imports. Anything importable on the compute node.
• Object storage. Reachable by every worker.

What it cannot

• Your session's variables, unless passed as arguments.
• Session-local files like /tmp paths, or workspace-only paths.
• Mutations reaching your session. Argument copies are one-way.

Arguments are copied, so mind the size

big = load_10gb_dataframe()

# Copies 10 GB to a worker, a thousand times over
futures = [process(big, i) for i in range(1000)]

If many tasks need the same large object, put it in object storage and have each task read the slice it needs, or hold it in an actor. Passing it as an argument moves far more data than the computation saves.

Return something small

Results travel back over the network too. Returning a 5 GB dataframe per task means gigabytes crossing the network before you see a value.

Prefer returning a summary, or writing the output to object storage and returning its path:

@eugo.hpc.distribute
def process_partition(path):
    df = read_partition(path)
    out = f"s3://bucket/processed/{basename(path)}"
    df.to_parquet(out)
    return {"path": out, "rows": len(df)}     # small

Exceptions surface at get()

An exception inside the body is raised when you resolve the future, with a traceback from inside the task. Read past the first frame. The useful line number is in the task body, not at the get() call.

For finding which input failed among many, see watching a run.

Task granularity

Dispatching a task costs something. When the body takes milliseconds, that overhead dominates:

# Bad: 1M tasks, each trivial
futures = [add_one(x) for x in range(1_000_000)]

# Better: 1000 tasks, each doing real work
futures = [add_one_batch(chunk) for chunk in chunks(range(1_000_000), 1000)]

Aim for task bodies measured in seconds rather than milliseconds.

---

Video: The distribute decorator, explained. The same material, with a transcript.