{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Local Parallel Execution: A Measured Comparison\n", "\n", "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/ContextLab/clustrix/blob/master/docs/source/notebooks/local_parallel_comparison.ipynb)\n", "\n", "Clustrix ships a `cluster_type=\"local\"` backend and a `LocalExecutor` class that\n", "wraps Python's `ThreadPoolExecutor` and `ProcessPoolExecutor`. This notebook\n", "measures what each of them actually buys you, on the machine that runs the\n", "notebook, against the obvious baseline: calling the function yourself.\n", "\n", "Every number below is printed by a cell in this notebook. Nothing is quoted from\n", "a previous run, and nothing is written by hand into the prose.\n", "\n", "Two questions get answered:\n", "\n", "1. Does decorating a function with `@cluster(cores=N)` under\n", " `cluster_type=\"local\"` make it faster? (Short answer: no, and the timings\n", " below show why.)\n", "2. If you want local parallelism, what is the thing that provides it, does it\n", " use threads or processes, and who decides?\n" ] }, { "cell_type": "code", "execution_count": 1, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:10.158970Z", "iopub.status.busy": "2026-08-22T21:56:10.158894Z", "iopub.status.idle": "2026-08-22T21:56:10.365225Z", "shell.execute_reply": "2026-08-22T21:56:10.364744Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "clustrix 0.2.0\n", "Python 3.12.10 (Darwin)\n", "os.cpu_count() 12\n", "mp start method spawn\n", "workers used below 8\n" ] } ], "source": [ "import multiprocessing\n", "import os\n", "import platform\n", "import statistics\n", "import sys\n", "import tempfile\n", "import time\n", "\n", "import clustrix\n", "\n", "CORES = min(8, os.cpu_count() or 2)\n", "\n", "print(f\"clustrix {clustrix.__version__}\")\n", "print(f\"Python {platform.python_version()} ({platform.system()})\")\n", "print(f\"os.cpu_count() {os.cpu_count()}\")\n", "print(f\"mp start method {multiprocessing.get_start_method()}\")\n", "print(f\"workers used below {CORES}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## What `cluster_type=\"local\"` does\n", "\n", "`local` is a real backend, but it is not a parallel one. `LocalJobManager.submit_job`\n", "(`clustrix/local_executor.py`) runs the deserialized function through\n", "`LocalExecutor.execute_single`, and `execute_single` is a two-line method that\n", "calls `func(*args, **kwargs)` in the calling thread. No pool is involved.\n", "\n", "The `@cluster` decorator does not even get that far in the common case. Its\n", "`_choose_execution_mode` (`clustrix/decorator.py:395`) returns `\"local\"` whenever\n", "`config.cluster_host` is unset, and the local branch calls the function directly in\n", "the current interpreter. So `cores=8` is a resource request that a local run has\n", "nobody to send to — there is no scheduler on the other end of it.\n", "\n", "That is worth measuring rather than asserting." ] }, { "cell_type": "code", "execution_count": 2, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:10.384847Z", "iopub.status.busy": "2026-08-22T21:56:10.384690Z", "iopub.status.idle": "2026-08-22T21:56:10.388050Z", "shell.execute_reply": "2026-08-22T21:56:10.387616Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "configured: local\n" ] } ], "source": [ "from clustrix import cluster, configure\n", "\n", "configure(cluster_type=\"local\", auto_parallel=False)\n", "\n", "\n", "def timed(fn, *args, **kwargs):\n", " \"\"\"Return (elapsed_seconds, result).\"\"\"\n", " start = time.perf_counter()\n", " result = fn(*args, **kwargs)\n", " return time.perf_counter() - start, result\n", "\n", "\n", "def best_of(n, fn, *args, **kwargs):\n", " \"\"\"Median wall time over n repetitions, plus the last result.\"\"\"\n", " times = []\n", " result = None\n", " for _ in range(n):\n", " elapsed, result = timed(fn, *args, **kwargs)\n", " times.append(elapsed)\n", " return statistics.median(times), result\n", "\n", "\n", "print(\"configured:\", clustrix.get_config().cluster_type)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## A workload that lives in a file\n", "\n", "Process pools have to get your function into the worker process. On macOS and\n", "Windows the default start method is `spawn`, so the child re-imports the module\n", "the function came from. A function typed into a notebook cell belongs to\n", "`__main__`, which in a Jupyter kernel is the kernel launcher — the child cannot\n", "find it there, and the task fails on unpickling.\n", "\n", "Writing the workloads to a real module sidesteps that, and it is what you would\n", "do in a project anyway. The module goes to a temporary directory so this notebook\n", "leaves nothing behind." ] }, { "cell_type": "code", "execution_count": 3, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:10.389396Z", "iopub.status.busy": "2026-08-22T21:56:10.389306Z", "iopub.status.idle": "2026-08-22T21:56:10.393033Z", "shell.execute_reply": "2026-08-22T21:56:10.392602Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "module written to /var/folders/tp/qtzc39jx5w556wl5w3dj21wr0000gn/T/clustrix_local_demo_z_72sp4q/workloads.py\n", "cpu_task iterations per call: 3000000\n" ] } ], "source": [ "WORKDIR = tempfile.mkdtemp(prefix=\"clustrix_local_demo_\")\n", "MODULE_PATH = os.path.join(WORKDIR, \"workloads.py\")\n", "\n", "MODULE_SOURCE = '''\n", "\"\"\"Workloads for the local-execution comparison notebook.\"\"\"\n", "\n", "import time\n", "\n", "CPU_ITERATIONS = 3_000_000\n", "\n", "\n", "def cpu_task(seed):\n", " \"\"\"Pure arithmetic. Holds the GIL for its whole run.\"\"\"\n", " total = 0\n", " for i in range(seed, seed + CPU_ITERATIONS):\n", " total += (i * i) % 9973\n", " return total\n", "\n", "\n", "def io_task(seed):\n", " \"\"\"Stands in for a network call or a slow disk read.\"\"\"\n", " time.sleep(0.25)\n", " return seed\n", "\n", "\n", "def tiny_task(x):\n", " \"\"\"Too small to be worth parallelizing. That is the point.\"\"\"\n", " return x * x\n", "\n", "\n", "def spin(_parallel_i=None):\n", " \"\"\"Shaped to satisfy clustrix's loop analyzer -- see the last section.\"\"\"\n", " for i in range(200_000):\n", " i ** 2\n", " return \"one call\"\n", "'''\n", "\n", "with open(MODULE_PATH, \"w\") as handle:\n", " handle.write(MODULE_SOURCE)\n", "\n", "sys.path.insert(0, WORKDIR)\n", "import workloads\n", "\n", "print(\"module written to\", MODULE_PATH)\n", "print(\"cpu_task iterations per call:\", workloads.CPU_ITERATIONS)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Baseline 1: the decorator against a bare call\n", "\n", "`@cluster(cores=N)` wrapping `cpu_task`, versus `cpu_task` itself. Five\n", "repetitions of each, median reported, so a single scheduling hiccup does not\n", "decide the answer." ] }, { "cell_type": "code", "execution_count": 4, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:10.394259Z", "iopub.status.busy": "2026-08-22T21:56:10.394184Z", "iopub.status.idle": "2026-08-22T21:56:11.961911Z", "shell.execute_reply": "2026-08-22T21:56:11.961389Z" } }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "@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_ keyword -- and even there it is an upper bound, not a promise that many workers will be busy.\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "plain call 0.157 s\n", "@cluster(cores=8) 0.156 s\n", "ratio 1.003x\n", "same answer: True\n", "\n", "difference (decorated - plain): -0.43 ms per call\n" ] } ], "source": [ "remote_style = cluster(cores=CORES)(workloads.cpu_task)\n", "\n", "plain_median, plain_result = best_of(5, workloads.cpu_task, 0)\n", "decorated_median, decorated_result = best_of(5, remote_style, 0)\n", "\n", "print(f\"plain call {plain_median:.3f} s\")\n", "print(f\"@cluster(cores={CORES}) {decorated_median:.3f} s\")\n", "print(f\"ratio {plain_median / decorated_median:.3f}x\")\n", "print(f\"same answer: {plain_result == decorated_result}\")\n", "print()\n", "delta_ms = (decorated_median - plain_median) * 1000\n", "print(f\"difference (decorated - plain): {delta_ms:+.2f} ms per call\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The two timings agree to within measurement noise, and they should: the\n", "decorated call is the undecorated call plus a config lookup and a couple of\n", "branches. Requesting eight cores changed nothing, because nothing in that path\n", "starts a second worker.\n", "\n", "The overhead becomes visible when the function is small. Here is the same\n", "comparison against `tiny_task`, which returns a single multiplication." ] }, { "cell_type": "code", "execution_count": 5, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:11.963283Z", "iopub.status.busy": "2026-08-22T21:56:11.963181Z", "iopub.status.idle": "2026-08-22T21:56:11.969142Z", "shell.execute_reply": "2026-08-22T21:56:11.968687Z" } }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "@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_ keyword -- and even there it is an upper bound, not a promise that many workers will be busy.\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "2000 bare calls 0.08 ms ( 0.04 us each)\n", "2000 decorated calls 2.91 ms ( 1.46 us each)\n", "cost of the decorator 1.41 us per call\n" ] } ], "source": [ "tiny_decorated = cluster(cores=CORES)(workloads.tiny_task)\n", "\n", "REPS = 2000\n", "plain_tiny, _ = timed(lambda: [workloads.tiny_task(i) for i in range(REPS)])\n", "dec_tiny, _ = timed(lambda: [tiny_decorated(i) for i in range(REPS)])\n", "\n", "print(f\"{REPS} bare calls {plain_tiny * 1e3:8.2f} ms\"\n", " f\" ({plain_tiny / REPS * 1e6:6.2f} us each)\")\n", "print(f\"{REPS} decorated calls {dec_tiny * 1e3:8.2f} ms\"\n", " f\" ({dec_tiny / REPS * 1e6:6.2f} us each)\")\n", "print(f\"cost of the decorator {(dec_tiny - plain_tiny) / REPS * 1e6:6.2f} us per call\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Threads or processes, and who chooses\n", "\n", "Local parallelism in clustrix comes from `LocalExecutor`\n", "(`clustrix/local_executor.py:19`), used directly. Its `use_threads` flag picks the\n", "pool: `True` builds a `ThreadPoolExecutor`, `False` a `ProcessPoolExecutor`\n", "(`_create_executor`, line 43). Nothing else in the class varies between the two.\n", "\n", "`create_local_executor` (line 434) fills that flag in when you leave it as `None`.\n", "It calls `choose_executor_type(func, args, kwargs)` (line 339), which decides in\n", "this order:\n", "\n", "1. **Picklability.** `pickle.dumps` is tried on the function, then on every\n", " positional and keyword argument. Any failure returns `True` — threads. A\n", " lambda, a closure, an open file handle, a database connection, a live socket:\n", " all of these force threads regardless of what the work looks like.\n", "2. **A substring scan of the source.** `inspect.getsource(func)` is lowercased and\n", " searched for `open(`, `requests.`, `urllib.`, `http.`, `ftp.`, `sql`,\n", " `database`, `time.sleep`, `threading.`. A hit returns `True` — threads. This is\n", " a text match on the function's own body only; a CPU-bound function that happens\n", " to call `open()` once takes the thread branch, and an I/O-bound function that\n", " reaches the network through a helper it calls does not.\n", "3. **Otherwise, processes.**\n", "\n", "You can override the whole thing by passing `use_threads=True` or\n", "`use_threads=False` explicitly — either to `create_local_executor` or to\n", "`LocalExecutor` directly. The auto-detection only runs when `use_threads is None`\n", "*and* a function was supplied.\n", "\n", "Run it on the three workloads and see." ] }, { "cell_type": "code", "execution_count": 6, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:11.970544Z", "iopub.status.busy": "2026-08-22T21:56:11.970455Z", "iopub.status.idle": "2026-08-22T21:56:11.974114Z", "shell.execute_reply": "2026-08-22T21:56:11.973720Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "cpu_task -> processes\n", "io_task -> threads (source contains 'time.sleep')\n", "tiny_task -> processes\n", "a lambda -> threads (unpicklable)\n", "cpu_task with an unpicklable argument -> threads\n", "\n", "auto-detected for cpu_task: use_threads=False\n", "explicit override: use_threads=True\n" ] } ], "source": [ "from clustrix.local_executor import LocalExecutor, choose_executor_type, create_local_executor\n", "\n", "\n", "def verdict(flag):\n", " return \"threads\" if flag else \"processes\"\n", "\n", "\n", "print(f\"cpu_task -> {verdict(choose_executor_type(workloads.cpu_task, (0,), {}))}\")\n", "print(f\"io_task -> {verdict(choose_executor_type(workloads.io_task, (0,), {}))}\"\n", " \" (source contains 'time.sleep')\")\n", "print(f\"tiny_task -> {verdict(choose_executor_type(workloads.tiny_task, (1,), {}))}\")\n", "print(f\"a lambda -> {verdict(choose_executor_type(lambda x: x * 2, (1,), {}))}\"\n", " \" (unpicklable)\")\n", "print(f\"cpu_task with an unpicklable argument -> \"\n", " f\"{verdict(choose_executor_type(workloads.cpu_task, (lambda: 1,), {}))}\")\n", "print()\n", "auto = create_local_executor(max_workers=CORES, func=workloads.cpu_task, args=(0,))\n", "forced = create_local_executor(max_workers=CORES, use_threads=True, func=workloads.cpu_task, args=(0,))\n", "print(f\"auto-detected for cpu_task: use_threads={auto.use_threads}\")\n", "print(f\"explicit override: use_threads={forced.use_threads}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## CPU-bound work: processes win, threads do not\n", "\n", "Twelve independent `cpu_task` calls. Serial first, then the same twelve through a\n", "thread pool and a process pool of `CORES` workers.\n", "\n", "The pools are warmed with a throwaway batch before the timed run. Process workers\n", "are spawned lazily on first submit, and folding that one-time cost into the\n", "measurement would understate steady-state throughput — so it is measured\n", "separately and reported on its own line." ] }, { "cell_type": "code", "execution_count": 7, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:11.975317Z", "iopub.status.busy": "2026-08-22T21:56:11.975243Z", "iopub.status.idle": "2026-08-22T21:56:16.059424Z", "shell.execute_reply": "2026-08-22T21:56:16.058874Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "serial 1.84 s\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "threads (8 workers) 1.85 s speedup 0.99x pool startup 0.00 s correct: True\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "processes (8 workers) 0.32 s speedup 5.65x pool startup 0.06 s correct: True\n", "\n", "os.cpu_count() is 12; the pool has 8 workers\n", "12 tasks over 8 workers means 2 rounds, so the best\n", "achievable speedup here is 6.00x, not 8.00x\n", "processes reached 94% of that ceiling\n" ] } ], "source": [ "N_TASKS = 12\n", "cpu_chunks = [{\"args\": (i * 1_000,), \"kwargs\": {}} for i in range(N_TASKS)]\n", "warmup_chunks = [{\"args\": (j,), \"kwargs\": {}} for j in range(CORES)]\n", "\n", "cpu_serial, serial_results = timed(\n", " lambda: [workloads.cpu_task(i * 1_000) for i in range(N_TASKS)]\n", ")\n", "print(f\"serial {cpu_serial:6.2f} s\")\n", "\n", "cpu_results = {}\n", "for use_threads in (True, False):\n", " with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:\n", " warmup, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))\n", " elapsed, results = timed(lambda: ex.execute_parallel(workloads.cpu_task, cpu_chunks))\n", " label = verdict(use_threads)\n", " cpu_results[label] = elapsed\n", " print(f\"{label:<10} ({CORES} workers) {elapsed:6.2f} s\"\n", " f\" speedup {cpu_serial / elapsed:5.2f}x\"\n", " f\" pool startup {warmup:5.2f} s\"\n", " f\" correct: {results == serial_results}\")\n", "\n", "rounds = -(-N_TASKS // CORES) # ceiling division\n", "ceiling = N_TASKS / rounds\n", "\n", "print()\n", "print(f\"os.cpu_count() is {os.cpu_count()}; the pool has {CORES} workers\")\n", "print(f\"{N_TASKS} tasks over {CORES} workers means {rounds} rounds, so the best\")\n", "print(f\"achievable speedup here is {ceiling:.2f}x, not {CORES:.2f}x\")\n", "print(f\"processes reached {cpu_serial / cpu_results['processes'] / ceiling * 100:.0f}% \"\n", " \"of that ceiling\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Threads land within noise of the serial time. That is the GIL doing exactly what\n", "it is documented to do: `cpu_task` is bytecode arithmetic with no C-level release\n", "point, so the threads take turns on one core and the batch finishes in about the\n", "time one core needs.\n", "\n", "Processes get real parallelism because each has its own interpreter and its own\n", "GIL. The measured speedup is below the worker count, and most of that gap is\n", "arithmetic rather than overhead — the cell prints the load-balancing ceiling that\n", "follows from the task count and the worker count, and the process run sits close\n", "to it. What is left after that is the actual cost of the pool: pickling arguments\n", "and return values across a pipe, plus whatever else the machine is doing." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## I/O-bound work: threads win, and processes pay for nothing\n", "\n", "`io_task` sleeps for 250 ms. Sleeping releases the GIL, so threads overlap\n", "perfectly and cost almost nothing to start." ] }, { "cell_type": "code", "execution_count": 8, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:16.060881Z", "iopub.status.busy": "2026-08-22T21:56:16.060772Z", "iopub.status.idle": "2026-08-22T21:56:20.275319Z", "shell.execute_reply": "2026-08-22T21:56:20.274706Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "serial 3.10 s\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "threads (8 workers) 0.52 s speedup 5.97x pool startup 0.00 s startup + run 0.52 s correct: True\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "processes (8 workers) 0.52 s speedup 5.95x pool startup 0.06 s startup + run 0.58 s correct: True\n" ] } ], "source": [ "io_chunks = [{\"args\": (i,), \"kwargs\": {}} for i in range(N_TASKS)]\n", "\n", "io_serial, io_serial_results = timed(\n", " lambda: [workloads.io_task(i) for i in range(N_TASKS)]\n", ")\n", "print(f\"serial {io_serial:6.2f} s\")\n", "\n", "io_results = {}\n", "io_startup = {}\n", "for use_threads in (True, False):\n", " with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:\n", " warmup, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))\n", " elapsed, results = timed(lambda: ex.execute_parallel(workloads.io_task, io_chunks))\n", " label = verdict(use_threads)\n", " io_results[label] = elapsed\n", " io_startup[label] = warmup\n", " print(f\"{label:<10} ({CORES} workers) {elapsed:6.2f} s\"\n", " f\" speedup {io_serial / elapsed:5.2f}x\"\n", " f\" pool startup {warmup:5.2f} s\"\n", " f\" startup + run {warmup + elapsed:5.2f} s\"\n", " f\" correct: {results == io_serial_results}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Both pools beat the serial loop by a similar margin, and once the pool is\n", "warm the two are close enough that the gap between them is not the thing to\n", "optimize. Sleeping releases the GIL, so threads overlap as well as separate\n", "interpreters do, and neither pool is doing arithmetic that the GIL would\n", "serialize.\n", "\n", "What separates them is everything around the run: the process pool costs\n", "something to start (printed above), it has to pickle every argument and every\n", "result, and it refuses connections, file handles and closures outright. For work\n", "that is only waiting, threads give the same overlap without asking for any of\n", "that.\n", "\n", "Note that `choose_executor_type` gets this one right by accident of spelling:\n", "`io_task` contains the literal text `time.sleep`. Rewrite it to call\n", "`sleep(0.25)` from a `from time import sleep` import and the substring scan\n", "misses, and the automatic choice flips to processes. If the answer matters, pass\n", "`use_threads` yourself." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Fan-out that is too small to be worth it\n", "\n", "Parallelism is not free. Each task submitted to a process pool has to be pickled,\n", "written to a pipe, unpickled, run, and have its result sent back. When the task\n", "itself takes microseconds, that overhead is the entire runtime." ] }, { "cell_type": "code", "execution_count": 9, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:20.276846Z", "iopub.status.busy": "2026-08-22T21:56:20.276728Z", "iopub.status.idle": "2026-08-22T21:56:20.378634Z", "shell.execute_reply": "2026-08-22T21:56:20.378151Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "serial, 400 tiny tasks 0.02 ms\n", "threads pool 3.03 ms speedup 0.006x\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "processes pool 27.80 ms speedup 0.001x\n", "\n", "A speedup below 1.00x means the pool made it slower.\n" ] } ], "source": [ "SMALL_N = 400\n", "small_chunks = [{\"args\": (i,), \"kwargs\": {}} for i in range(SMALL_N)]\n", "\n", "small_serial, _ = timed(lambda: [workloads.tiny_task(i) for i in range(SMALL_N)])\n", "print(f\"serial, {SMALL_N} tiny tasks {small_serial * 1e3:8.2f} ms\")\n", "\n", "for use_threads in (True, False):\n", " with LocalExecutor(max_workers=CORES, use_threads=use_threads) as ex:\n", " timed(lambda: ex.execute_parallel(workloads.tiny_task, warmup_chunks))\n", " elapsed, _ = timed(lambda: ex.execute_parallel(workloads.tiny_task, small_chunks))\n", " print(f\"{verdict(use_threads):<10} pool {elapsed * 1e3:8.2f} ms\"\n", " f\" speedup {small_serial / elapsed:6.3f}x\")\n", "\n", "print()\n", "print(\"A speedup below 1.00x means the pool made it slower.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The automatic loop parallelization, and why to leave it off\n", "\n", "`ClusterConfig.auto_parallel` defaults to `True`. Under local execution that turns\n", "on `_execute_local_parallel` (`clustrix/decorator.py:429`), which parses your\n", "function, looks for a `for` loop over a literal `range()` whose body reads no\n", "name other than the loop variable, splits that range into chunks, and submits one\n", "call per chunk.\n", "\n", "The splitting is done by passing a `_parallel_` keyword. Clustrix does\n", "not rewrite the loop — your function has to read that keyword and do less work\n", "because of it. `workloads.spin` accepts the keyword and ignores it, which is what\n", "most functions written without this contract in mind effectively do.\n", "\n", "Here is what that costs, measured." ] }, { "cell_type": "code", "execution_count": 10, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:20.380148Z", "iopub.status.busy": "2026-08-22T21:56:20.380037Z", "iopub.status.idle": "2026-08-22T21:56:20.480147Z", "shell.execute_reply": "2026-08-22T21:56:20.479725Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "workloads.spin() 0.01 s -> 'one call'\n", "@cluster(parallel=True) spin() 0.09 s -> list of length 16\n", "\n", "ratio: 15.8x the serial time\n", "The function ran once per chunk, in full, and the return value is now a\n", "list of per-chunk returns rather than what the function returns.\n" ] } ], "source": [ "parallel_spin = cluster(cores=CORES, parallel=True)(workloads.spin)\n", "\n", "spin_serial, spin_serial_result = timed(workloads.spin)\n", "spin_auto, spin_auto_result = timed(parallel_spin)\n", "\n", "print(f\"workloads.spin() {spin_serial:6.2f} s \"\n", " f\"-> {spin_serial_result!r}\")\n", "print(f\"@cluster(parallel=True) spin() {spin_auto:6.2f} s \"\n", " f\"-> {type(spin_auto_result).__name__} of length \"\n", " f\"{len(spin_auto_result) if isinstance(spin_auto_result, list) else 1}\")\n", "print()\n", "print(f\"ratio: {spin_auto / spin_serial:.1f}x the serial time\")\n", "print(\"The function ran once per chunk, in full, and the return value is now a\")\n", "print(\"list of per-chunk returns rather than what the function returns.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Set `auto_parallel=False` for local runs — as the second cell of this\n", "notebook does — unless you have written your function to the\n", "`_parallel_` contract and checked the result." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary of this run" ] }, { "cell_type": "code", "execution_count": 11, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:20.481803Z", "iopub.status.busy": "2026-08-22T21:56:20.481698Z", "iopub.status.idle": "2026-08-22T21:56:20.484524Z", "shell.execute_reply": "2026-08-22T21:56:20.484100Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "measurement speedup vs serial\n", "--------------------------------------------------\n", "CPU-bound, threads 0.99x\n", "CPU-bound, processes 5.65x\n", "I/O-bound, threads 5.97x\n", "I/O-bound, processes 5.95x\n", "@cluster(cores=8), local 1.00x\n", "--------------------------------------------------\n", "8 workers, os.cpu_count() = 12\n" ] } ], "source": [ "rows = [\n", " (\"CPU-bound, threads\", cpu_serial / cpu_results[\"threads\"]),\n", " (\"CPU-bound, processes\", cpu_serial / cpu_results[\"processes\"]),\n", " (\"I/O-bound, threads\", io_serial / io_results[\"threads\"]),\n", " (\"I/O-bound, processes\", io_serial / io_results[\"processes\"]),\n", " (f\"@cluster(cores={CORES}), local\", plain_median / decorated_median),\n", "]\n", "\n", "print(f\"{'measurement':<32}{'speedup vs serial':>18}\")\n", "print(\"-\" * 50)\n", "for name, value in rows:\n", " print(f\"{name:<32}{value:>17.2f}x\")\n", "print(\"-\" * 50)\n", "print(f\"{CORES} workers, os.cpu_count() = {os.cpu_count()}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## When to reach for this, and when not to\n", "\n", "Use `LocalExecutor` with `use_threads=False` when the work is CPU-bound, each task\n", "runs for at least tens of milliseconds, and the arguments and return values are\n", "picklable and small. That is the case the process numbers above are drawn from.\n", "\n", "Use `use_threads=True` when the tasks spend their time waiting — HTTP requests,\n", "database round trips, `subprocess` calls, large reads. Threads also let you pass\n", "things a process cannot receive: an open connection, a lambda, an object holding a\n", "file handle.\n", "\n", "Do not use either when:\n", "\n", "- The tasks are tiny. The fan-out section above shows the pool losing to a plain\n", " loop outright.\n", "- The work is already parallel underneath. NumPy, PyTorch and BLAS-backed code\n", " use their own thread pools; wrapping them in a process pool oversubscribes the\n", " machine and usually slows it down.\n", "- You need a shared mutable object across tasks. Process workers get copies, and\n", " writes do not come back.\n", "- The tasks are not independent. `LocalExecutor` gives no ordering or locking\n", " beyond returning results in submission order.\n", "\n", "Do not reach for an ordinary `@cluster(cores=N)` call with `cluster_type=\"local\"` expecting\n", "parallelism. It runs your function, once, in this interpreter. It is useful for\n", "keeping one code path while you switch `cluster_type` between `local` and a real\n", "cluster — the decorated function behaves identically either way, which is what\n", "makes a local run a valid dry run. It is not a parallel execution mode." ] }, { "cell_type": "code", "execution_count": 12, "metadata": { "execution": { "iopub.execute_input": "2026-08-22T21:56:20.485667Z", "iopub.status.busy": "2026-08-22T21:56:20.485573Z", "iopub.status.idle": "2026-08-22T21:56:20.488229Z", "shell.execute_reply": "2026-08-22T21:56:20.487821Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "cleaned up /var/folders/tp/qtzc39jx5w556wl5w3dj21wr0000gn/T/clustrix_local_demo_z_72sp4q\n" ] } ], "source": [ "import shutil\n", "\n", "sys.path.remove(WORKDIR)\n", "del sys.modules[\"workloads\"]\n", "shutil.rmtree(WORKDIR, ignore_errors=True)\n", "print(\"cleaned up\", WORKDIR)" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.10" } }, "nbformat": 4, "nbformat_minor": 4 }