Local Parallel Execution: A Measured Comparison

Clustrix ships a cluster_type="local" backend and a LocalExecutor class that wraps Python’s ThreadPoolExecutor and ProcessPoolExecutor. This notebook measures what each of them actually buys you, on the machine that runs the notebook, against the obvious baseline: calling the function yourself.

Every number below is printed by a cell in this notebook. Nothing is quoted from a previous run, and nothing is written by hand into the prose.

Two questions get answered:

1. Does decorating a function with @cluster(cores=N) under cluster_type="local" make it faster? (Short answer: no, and the timings below show why.)

2. If you want local parallelism, what is the thing that provides it, does it use threads or processes, and who decides?

import multiprocessing
import os
import platform
import statistics
import sys
import tempfile
import time

import clustrix

CORES = min(8, os.cpu_count() or 2)

print(f"clustrix           {clustrix.__version__}")
print(f"Python             {platform.python_version()} ({platform.system()})")
print(f"os.cpu_count()     {os.cpu_count()}")
print(f"mp start method    {multiprocessing.get_start_method()}")
print(f"workers used below {CORES}")

clustrix           0.2.0
Python             3.12.10 (Darwin)
os.cpu_count()     12
mp start method    spawn
workers used below 8

What cluster_type="local" does

local is a real backend, but it is not a parallel one. LocalJobManager.submit_job (clustrix/local_executor.py) runs the deserialized function through LocalExecutor.execute_single, and execute_single is a two-line method that calls func(*args, **kwargs) in the calling thread. No pool is involved.

The @cluster decorator does not even get that far in the common case. Its _choose_execution_mode (clustrix/decorator.py:395) returns "local" whenever config.cluster_host is unset, and the local branch calls the function directly in the current interpreter. So cores=8 is a resource request that a local run has nobody to send to — there is no scheduler on the other end of it.

That is worth measuring rather than asserting.

from clustrix import cluster, configure

configure(cluster_type="local", auto_parallel=False)


def timed(fn, *args, **kwargs):
    """Return (elapsed_seconds, result)."""
    start = time.perf_counter()
    result = fn(*args, **kwargs)
    return time.perf_counter() - start, result


def best_of(n, fn, *args, **kwargs):
    """Median wall time over n repetitions, plus the last result."""
    times = []
    result = None
    for _ in range(n):
        elapsed, result = timed(fn, *args, **kwargs)
        times.append(elapsed)
    return statistics.median(times), result


print("configured:", clustrix.get_config().cluster_type)

configured: local

A workload that lives in a file

Process pools have to get your function into the worker process. On macOS and Windows the default start method is spawn, so the child re-imports the module the function came from. A function typed into a notebook cell belongs to __main__, which in a Jupyter kernel is the kernel launcher — the child cannot find it there, and the task fails on unpickling.

Writing the workloads to a real module sidesteps that, and it is what you would do in a project anyway. The module goes to a temporary directory so this notebook leaves nothing behind.

WORKDIR = tempfile.mkdtemp(prefix="clustrix_local_demo_")
MODULE_PATH = os.path.join(WORKDIR, "workloads.py")

MODULE_SOURCE = '''
"""Workloads for the local-execution comparison notebook."""

import time

CPU_ITERATIONS = 3_000_000


def cpu_task(seed):
    """Pure arithmetic. Holds the GIL for its whole run."""
    total = 0
    for i in range(seed, seed + CPU_ITERATIONS):
        total += (i * i) % 9973
    return total


def io_task(seed):
    """Stands in for a network call or a slow disk read."""
    time.sleep(0.25)
    return seed


def tiny_task(x):
    """Too small to be worth parallelizing. That is the point."""
    return x * x


def spin(_parallel_i=None):
    """Shaped to satisfy clustrix's loop analyzer -- see the last section."""
    for i in range(200_000):
        i ** 2
    return "one call"
'''

with open(MODULE_PATH, "w") as handle:
    handle.write(MODULE_SOURCE)

sys.path.insert(0, WORKDIR)
import workloads

print("module written to", MODULE_PATH)
print("cpu_task iterations per call:", workloads.CPU_ITERATIONS)

module written to /var/folders/tp/qtzc39jx5w556wl5w3dj21wr0000gn/T/clustrix_local_demo_z_72sp4q/workloads.py
cpu_task iterations per call: 3000000

Baseline 1: the decorator against a bare call

@cluster(cores=N) wrapping cpu_task, versus cpu_task itself. Five repetitions of each, median reported, so a single scheduling hiccup does not decide the answer.

remote_style = cluster(cores=CORES)(workloads.cpu_task)

plain_median, plain_result = best_of(5, workloads.cpu_task, 0)
decorated_median, decorated_result = best_of(5, remote_style, 0)

print(f"plain call        {plain_median:.3f} s")
print(f"@cluster(cores={CORES})  {decorated_median:.3f} s")
print(f"ratio             {plain_median / decorated_median:.3f}x")
print(f"same answer:      {plain_result == decorated_result}")
print()
delta_ms = (decorated_median - plain_median) * 1000
print(f"difference (decorated - plain): {delta_ms:+.2f} ms per call")

@cluster(cores=8) has no effect here: the local backend runs the decorated function once, in this process. Locally, cores bounds the worker pool only when parallel=True finds a parallelizable loop and the function accepts the matching _parallel_<var> keyword -- and even there it is an upper bound, not a promise that many workers will be busy.

plain call        0.157 s
@cluster(cores=8)  0.156 s
ratio             1.003x
same answer:      True

difference (decorated - plain): -0.43 ms per call

The two timings agree to within measurement noise, and they should: the decorated call is the undecorated call plus a config lookup and a couple of branches. Requesting eight cores changed nothing, because nothing in that path starts a second worker.

The overhead becomes visible when the function is small. Here is the same comparison against tiny_task, which returns a single multiplication.

tiny_decorated = cluster(cores=CORES)(workloads.tiny_task)

REPS = 2000
plain_tiny, _ = timed(lambda: [workloads.tiny_task(i) for i in range(REPS)])
dec_tiny, _ = timed(lambda: [tiny_decorated(i) for i in range(REPS)])

print(f"{REPS} bare calls        {plain_tiny * 1e3:8.2f} ms"
      f"  ({plain_tiny / REPS * 1e6:6.2f} us each)")
print(f"{REPS} decorated calls   {dec_tiny * 1e3:8.2f} ms"
      f"  ({dec_tiny / REPS * 1e6:6.2f} us each)")
print(f"cost of the decorator   {(dec_tiny - plain_tiny) / REPS * 1e6:6.2f} us per call")

@cluster(cores=8) has no effect here: the local backend runs the decorated function once, in this process. Locally, cores bounds the worker pool only when parallel=True finds a parallelizable loop and the function accepts the matching _parallel_<var> keyword -- and even there it is an upper bound, not a promise that many workers will be busy.

2000 bare calls            0.08 ms  (  0.04 us each)
2000 decorated calls       2.91 ms  (  1.46 us each)
cost of the decorator     1.41 us per call

Threads or processes, and who chooses

Local parallelism in clustrix comes from LocalExecutor (clustrix/local_executor.py:19), used directly. Its use_threads flag picks the pool: True builds a ThreadPoolExecutor, False a ProcessPoolExecutor (_create_executor, line 43). Nothing else in the class varies between the two.

create_local_executor (line 434) fills that flag in when you leave it as None. It calls choose_executor_type(func, args, kwargs) (line 339), which decides in this order:

1. Picklability. pickle.dumps is tried on the function, then on every positional and keyword argument. Any failure returns True — threads. A lambda, a closure, an open file handle, a database connection, a live socket: all of these force threads regardless of what the work looks like.

2. A substring scan of the source. inspect.getsource(func) is lowercased and searched for open(, requests., urllib., http., ftp., sql, database, time.sleep, threading.. A hit returns True — threads. This is a text match on the function’s own body only; a CPU-bound function that happens to call open() once takes the thread branch, and an I/O-bound function that reaches the network through a helper it calls does not.

3. Otherwise, processes.

You can override the whole thing by passing use_threads=True or use_threads=False explicitly — either to create_local_executor or to LocalExecutor directly. The auto-detection only runs when use_threads is None and a function was supplied.

Run it on the three workloads and see.

from clustrix.local_executor import LocalExecutor, choose_executor_type, create_local_executor


def verdict(flag):
    return "threads" if flag else "processes"


print(f"cpu_task   -> {verdict(choose_executor_type(workloads.cpu_task, (0,), {}))}")
print(f"io_task    -> {verdict(choose_executor_type(workloads.io_task, (0,), {}))}"
      "   (source contains 'time.sleep')")
print(f"tiny_task  -> {verdict(choose_executor_type(workloads.tiny_task, (1,), {}))}")
print(f"a lambda   -> {verdict(choose_executor_type(lambda x: x * 2, (1,), {}))}"
      "   (unpicklable)")
print(f"cpu_task with an unpicklable argument -> "
      f"{verdict(choose_executor_type(workloads.cpu_task, (lambda: 1,), {}))}")
print()
auto = create_local_executor(max_workers=CORES, func=workloads.cpu_task, args=(0,))
forced = create_local_executor(max_workers=CORES, use_threads=True, func=workloads.cpu_task, args=(0,))
print(f"auto-detected for cpu_task: use_threads={auto.use_threads}")
print(f"explicit override:          use_threads={forced.use_threads}")

cpu_task   -> processes
io_task    -> threads   (source contains 'time.sleep')
tiny_task  -> processes
a lambda   -> threads   (unpicklable)
cpu_task with an unpicklable argument -> threads

auto-detected for cpu_task: use_threads=False
explicit override:          use_threads=True

CPU-bound work: processes win, threads do not

Twelve independent cpu_task calls. Serial first, then the same twelve through a thread pool and a process pool of CORES workers.

The pools are warmed with a throwaway batch before the timed run. Process workers are spawned lazily on first submit, and folding that one-time cost into the measurement would understate steady-state throughput — so it is measured separately and reported on its own line.

N_TASKS = 12
cpu_chunks = [{"args": (i * 1_000,), "kwargs": {}} for i in range(N_TASKS)]
warmup_chunks = [{"args": (j,), "kwargs": {}} for j in range(CORES)]

cpu_serial, serial_results = timed(
    lambda: [workloads.cpu_task(i * 1_000) for i in range(N_TASKS)]
)
print(f"serial                     {cpu_serial:6.2f} s")

cpu_results = {}
for use_threads in (True, False):
    with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:
        warmup, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))
        elapsed, results = timed(lambda: ex.execute_parallel(workloads.cpu_task, cpu_chunks))
    label = verdict(use_threads)
    cpu_results[label] = elapsed
    print(f"{label:<10} ({CORES} workers)   {elapsed:6.2f} s"
          f"   speedup {cpu_serial / elapsed:5.2f}x"
          f"   pool startup {warmup:5.2f} s"
          f"   correct: {results == serial_results}")

rounds = -(-N_TASKS // CORES)  # ceiling division
ceiling = N_TASKS / rounds

print()
print(f"os.cpu_count() is {os.cpu_count()}; the pool has {CORES} workers")
print(f"{N_TASKS} tasks over {CORES} workers means {rounds} rounds, so the best")
print(f"achievable speedup here is {ceiling:.2f}x, not {CORES:.2f}x")
print(f"processes reached {cpu_serial / cpu_results['processes'] / ceiling * 100:.0f}% "
      "of that ceiling")

serial                       1.84 s
threads    (8 workers)     1.85 s   speedup  0.99x   pool startup  0.00 s   correct: True
processes  (8 workers)     0.32 s   speedup  5.65x   pool startup  0.06 s   correct: True

os.cpu_count() is 12; the pool has 8 workers
12 tasks over 8 workers means 2 rounds, so the best
achievable speedup here is 6.00x, not 8.00x
processes reached 94% of that ceiling

Threads land within noise of the serial time. That is the GIL doing exactly what it is documented to do: cpu_task is bytecode arithmetic with no C-level release point, so the threads take turns on one core and the batch finishes in about the time one core needs.

Processes get real parallelism because each has its own interpreter and its own GIL. The measured speedup is below the worker count, and most of that gap is arithmetic rather than overhead — the cell prints the load-balancing ceiling that follows from the task count and the worker count, and the process run sits close to it. What is left after that is the actual cost of the pool: pickling arguments and return values across a pipe, plus whatever else the machine is doing.

I/O-bound work: threads win, and processes pay for nothing

io_task sleeps for 250 ms. Sleeping releases the GIL, so threads overlap perfectly and cost almost nothing to start.

io_chunks = [{"args": (i,), "kwargs": {}} for i in range(N_TASKS)]

io_serial, io_serial_results = timed(
    lambda: [workloads.io_task(i) for i in range(N_TASKS)]
)
print(f"serial                     {io_serial:6.2f} s")

io_results = {}
io_startup = {}
for use_threads in (True, False):
    with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:
        warmup, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))
        elapsed, results = timed(lambda: ex.execute_parallel(workloads.io_task, io_chunks))
    label = verdict(use_threads)
    io_results[label] = elapsed
    io_startup[label] = warmup
    print(f"{label:<10} ({CORES} workers)   {elapsed:6.2f} s"
          f"   speedup {io_serial / elapsed:5.2f}x"
          f"   pool startup {warmup:5.2f} s"
          f"   startup + run {warmup + elapsed:5.2f} s"
          f"   correct: {results == io_serial_results}")

serial                       3.10 s
threads    (8 workers)     0.52 s   speedup  5.97x   pool startup  0.00 s   startup + run  0.52 s   correct: True
processes  (8 workers)     0.52 s   speedup  5.95x   pool startup  0.06 s   startup + run  0.58 s   correct: True

Both pools beat the serial loop by a similar margin, and once the pool is warm the two are close enough that the gap between them is not the thing to optimize. Sleeping releases the GIL, so threads overlap as well as separate interpreters do, and neither pool is doing arithmetic that the GIL would serialize.

What separates them is everything around the run: the process pool costs something to start (printed above), it has to pickle every argument and every result, and it refuses connections, file handles and closures outright. For work that is only waiting, threads give the same overlap without asking for any of that.

Note that choose_executor_type gets this one right by accident of spelling: io_task contains the literal text time.sleep. Rewrite it to call sleep(0.25) from a from time import sleep import and the substring scan misses, and the automatic choice flips to processes. If the answer matters, pass use_threads yourself.

Fan-out that is too small to be worth it

Parallelism is not free. Each task submitted to a process pool has to be pickled, written to a pipe, unpickled, run, and have its result sent back. When the task itself takes microseconds, that overhead is the entire runtime.

SMALL_N = 400
small_chunks = [{"args": (i,), "kwargs": {}} for i in range(SMALL_N)]

small_serial, _ = timed(lambda: [workloads.tiny_task(i) for i in range(SMALL_N)])
print(f"serial, {SMALL_N} tiny tasks      {small_serial * 1e3:8.2f} ms")

for use_threads in (True, False):
    with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:
        timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))
        elapsed, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, small_chunks))
    print(f"{verdict(use_threads):<10} pool             {elapsed * 1e3:8.2f} ms"
          f"   speedup {small_serial / elapsed:6.3f}x")

print()
print("A speedup below 1.00x means the pool made it slower.")

serial, 400 tiny tasks          0.02 ms
threads    pool                 3.03 ms   speedup  0.006x
processes  pool                27.80 ms   speedup  0.001x

A speedup below 1.00x means the pool made it slower.

The automatic loop parallelization, and why to leave it off

ClusterConfig.auto_parallel defaults to True. Under local execution that turns on _execute_local_parallel (clustrix/decorator.py:429), which parses your function, looks for a for loop over a literal range() whose body reads no name other than the loop variable, splits that range into chunks, and submits one call per chunk.

The splitting is done by passing a _parallel_<loopvar> keyword. Clustrix does not rewrite the loop — your function has to read that keyword and do less work because of it. workloads.spin accepts the keyword and ignores it, which is what most functions written without this contract in mind effectively do.

Here is what that costs, measured.

parallel_spin = cluster(cores=CORES, parallel=True)(workloads.spin)

spin_serial, spin_serial_result = timed(workloads.spin)
spin_auto, spin_auto_result = timed(parallel_spin)

print(f"workloads.spin()                     {spin_serial:6.2f} s "
      f"-> {spin_serial_result!r}")
print(f"@cluster(parallel=True) spin()       {spin_auto:6.2f} s "
      f"-> {type(spin_auto_result).__name__} of length "
      f"{len(spin_auto_result) if isinstance(spin_auto_result, list) else 1}")
print()
print(f"ratio: {spin_auto / spin_serial:.1f}x the serial time")
print("The function ran once per chunk, in full, and the return value is now a")
print("list of per-chunk returns rather than what the function returns.")

workloads.spin()                       0.01 s -> 'one call'
@cluster(parallel=True) spin()         0.09 s -> list of length 16

ratio: 15.8x the serial time
The function ran once per chunk, in full, and the return value is now a
list of per-chunk returns rather than what the function returns.

Set auto_parallel=False for local runs — as the second cell of this notebook does — unless you have written your function to the _parallel_<loopvar> contract and checked the result.

Summary of this run

rows = [
    ("CPU-bound, threads", cpu_serial / cpu_results["threads"]),
    ("CPU-bound, processes", cpu_serial / cpu_results["processes"]),
    ("I/O-bound, threads", io_serial / io_results["threads"]),
    ("I/O-bound, processes", io_serial / io_results["processes"]),
    (f"@cluster(cores={CORES}), local", plain_median / decorated_median),
]

print(f"{'measurement':<32}{'speedup vs serial':>18}")
print("-" * 50)
for name, value in rows:
    print(f"{name:<32}{value:>17.2f}x")
print("-" * 50)
print(f"{CORES} workers, os.cpu_count() = {os.cpu_count()}")

measurement                      speedup vs serial
--------------------------------------------------
CPU-bound, threads                           0.99x
CPU-bound, processes                         5.65x
I/O-bound, threads                           5.97x
I/O-bound, processes                         5.95x
@cluster(cores=8), local                     1.00x
--------------------------------------------------
8 workers, os.cpu_count() = 12

When to reach for this, and when not to

Use LocalExecutor with use_threads=False when the work is CPU-bound, each task runs for at least tens of milliseconds, and the arguments and return values are picklable and small. That is the case the process numbers above are drawn from.

Use use_threads=True when the tasks spend their time waiting — HTTP requests, database round trips, subprocess calls, large reads. Threads also let you pass things a process cannot receive: an open connection, a lambda, an object holding a file handle.

Do not use either when:

• The tasks are tiny. The fan-out section above shows the pool losing to a plain loop outright.

• The work is already parallel underneath. NumPy, PyTorch and BLAS-backed code use their own thread pools; wrapping them in a process pool oversubscribes the machine and usually slows it down.

• You need a shared mutable object across tasks. Process workers get copies, and writes do not come back.

• The tasks are not independent. LocalExecutor gives no ordering or locking beyond returning results in submission order.

Do not reach for an ordinary @cluster(cores=N) call with cluster_type="local" expecting parallelism. It runs your function, once, in this interpreter. It is useful for keeping one code path while you switch cluster_type between local and a real cluster — the decorated function behaves identically either way, which is what makes a local run a valid dry run. It is not a parallel execution mode.

import shutil

sys.path.remove(WORKDIR)
del sys.modules["workloads"]
shutil.rmtree(WORKDIR, ignore_errors=True)
print("cleaned up", WORKDIR)

cleaned up /var/folders/tp/qtzc39jx5w556wl5w3dj21wr0000gn/T/clustrix_local_demo_z_72sp4q