Sizing Workers, Chunksize and Memory

How many workers, how big a chunk, how much memory, and what speedup to expect. Ends with the exact model used by scripts/pool_tool.py plan, so its numbers can be checked and recalibrated.

#Contents

• 1. Measure before you size
• 2. Counting usable cores
• 3. Hyperthreading and physical cores
• 4. Thread oversubscription inside workers
• 5. Memory per worker
• 6. Chunksize and overhead math
• 7. Where the speedup goes (Amdahl for pools)
• 8. The planner's model
• 9. Calibrating the constants
• 10. Worked examples

#1. Measure before you size

The planner needs five numbers. Measure them on real data, single process:

import pickle
import statistics
import time


def task(text: str) -> list:                     # stand-in for parse/chunk/tokenize
    return [w for w in text.split() if len(w) > 3]


samples = [" ".join(f"word{i}" for i in range(200)) for _ in range(100)]
for s in samples[:5]:
    task(s)                                       # warm-up
times = []
for s in samples:
    t = time.perf_counter()
    out = task(s)
    times.append((time.perf_counter() - t) * 1000)
print(f"--item-ms {statistics.median(times):.3f}",
      f"--payload-kb {len(pickle.dumps(samples[0], protocol=5)) / 1024:.1f}",
      f"--result-kb {len(pickle.dumps(out, protocol=5)) / 1024:.1f}")

If item times vary a lot (PDFs of 1 vs 400 pages), use the mean for throughput and keep chunks small (the tail is set by the slowest chunk). Sort items largest-first to shorten the tail.

#2. Counting usable cores

A Kubernetes pod with limits.cpu: 2 on a 64-core node sees os.cpu_count() == 64. A pool of 64 workers is then throttled by the CFS quota: every worker gets 1/32 of a core in bursts, latency spikes and memory is 32x higher than needed.

import math
import os


def usable_cores() -> int:
    if hasattr(os, "process_cpu_count"):
        n = os.process_cpu_count() or 1                  # 3.13+
    elif hasattr(os, "sched_getaffinity"):
        n = len(os.sched_getaffinity(0))
    else:
        n = os.cpu_count() or 1
    try:
        with open("/sys/fs/cgroup/cpu.max") as fh:       # cgroup v2: "max 100000" or "200000 100000"
            quota, period = fh.read().split()
        if quota != "max":
            n = min(n, max(1, math.ceil(int(quota) / int(period))))
    except (OSError, ValueError):
        pass
    return n


print(usable_cores() >= 1)   # True

Leave headroom when the parent does real work (writing results, a web server) or other services share the box: workers = usable - 1 is a common choice for services; batch jobs use all.

#3. Hyperthreading and physical cores

Two hardware threads share one core's execution units. Pure-Python interpreter loops gain little from the second hardware thread (often 0-30%); memory-bound or branchy code sometimes more. There is no stdlib way to count physical cores (psutil.cpu_count(logical=False) does). Benchmark workers = physical vs workers = logical on the real task; pick the smaller when within 10%, since each worker costs memory.

#4. Thread oversubscription inside workers

numpy (OpenBLAS/MKL), torch, onnxruntime, faiss, polars and HF tokenizers start their own thread pools sized to the core count. W workers x C threads each on C cores means W x C runnable threads: context switching, cache thrash, often slower than one worker.

Set limits before the library is imported in the worker (spawn imports fresh, so the initializer or environment works):

import os


def init_worker() -> None:
    for var in ("OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS", "NUMEXPR_NUM_THREADS"):
        os.environ[var] = "1"
    os.environ["TOKENIZERS_PARALLELISM"] = "false"
    try:
        import torch
        torch.set_num_threads(1)
        torch.set_num_interop_threads(1)        # only valid before any parallel work in this process
    except ImportError:
        pass


init_worker()
print(os.environ["OMP_NUM_THREADS"])   # 1

Rules of thumb: either many single-threaded workers, or one process with a multi-threaded library, or a few workers with cores // workers threads each. threadpoolctl.threadpool_limits(1) (third-party) limits BLAS pools at runtime.

#5. Memory per worker

workers_by_memory = floor(0.85 x memory_limit / peak_rss_per_worker)        (planner: MEM_HEADROOM = 0.85)
total_memory      = parent_rss + workers x peak_rss_per_worker

What counts in a worker's RSS:

• The interpreter and imports: 15-30 MB bare, 300-800 MB with torch + transformers imported.
• Models: weights in RAM (a 110M-parameter fp32 encoder is ~440 MB; int8 ONNX ~110 MB).
• The working set of the in-flight chunk (input + output), times chunksize.
• With fork: pages shared with the parent are counted in each RSS but exist once (use PSS/USS: smem, /proc/<pid>/smaps_rollup). They become private when written, including by refcount updates.
• Allocator fragmentation that grows over time: cap with maxtasksperchild / max_tasks_per_child.

Where the memory limit comes from: cgroup v2 memory.max, v1 memory.limit_in_bytes, otherwise MemAvailable in /proc/meminfo (the planner reads these when --mem-limit-mb is omitted). When the limit is hit, the kernel OOM-killer sends SIGKILL to one worker: ProcessPoolExecutor raises BrokenProcessPool, Pool.map hangs (see the debugging reference).

#6. Chunksize and overhead math

Per chunk, the stdlib pools pay a fixed cost: pickling the call frame, a write and a read on a pipe, lock acquisitions on both queues, waking a worker and the result-handler thread. Per item they pay pickling of argument and result proportional to size and object count.

per_chunk_fixed   ~ 35-90 us   (measured below; Pool > ProcessPoolExecutor)
per_KB_pickled    ~ 0.4 us (bytes) | 1 us (str) | 2 us (list of floats) | 13 us (list of small dicts)
overhead_share    = (n_chunks x per_chunk + N x KB x per_KB) / (N x item_time + overhead)

Choosing chunksize:

1. Floor (overhead): chunk work >= 100 x per-chunk cost keeps overhead <= 1%. With ~150 us budgeted per chunk that is >= 15 ms; the planner uses 20 ms (MIN_CHUNK_S).
2. Ceiling (balance): at least ~4 chunks per worker so a slow chunk does not leave the other workers idle: chunksize <= ceil(N / (4 x workers)) (the Pool.map default).
3. Ceiling (latency/memory): results of a chunk arrive together, so progress bars jump and memory holds a whole chunk of results.
4. When the floor exceeds the ceiling, items are too small for this many workers: batch them into bigger items or use fewer workers.

#7. Where the speedup goes (Amdahl for pools)

T_parallel = startup + init + max(T_workers, T_parent) + stragglers + shutdown
T_workers  = (N x (item + ipc_item) + n_chunks x worker_chunk_cost) / workers
T_parent   = N x (per_item_bookkeeping + ipc_item) + n_chunks x per_chunk_cost
speedup    = T_serial / T_parallel

T_parent is the hidden serial part: the parent pickles every argument and unpickles every result on its own (task-feeder and result-handler threads share the GIL). If T_parent > T_workers, adding workers changes nothing. Remedies: smaller payloads (paths instead of contents), smaller results (write to disk in the worker, return ids), bigger items, shared memory for arrays.

#8. The planner's model

pool_tool.py plan evaluates every worker count w from 1 to the cap and, for each, every candidate chunksize c in {1, 2, 4, ..., ceil(N / w)} U {ceil(N / (4w))}. All times in seconds.

Inputs: N items, t = item_ms / 1000, P payload KB, R result KB, S init seconds, method m.

Constants (in pool_tool.py):

Formulas:

ipc_item   = (P + R) x IPC_US_PER_KB / 1e6
n_chunks   = ceil(N / c)
parent     = N x (ITEM_OVERHEAD_US / 1e6 + ipc_item) + n_chunks x CHUNK_OVERHEAD_US / 1e6
workers    = (N x (t + ipc_item) + n_chunks x WORKER_CHUNK_US / 1e6) / w
rounds     = ceil(n_chunks / w)
balanced   = rounds x c x (t + ipc_item) + rounds x WORKER_CHUNK_US / 1e6      # quantization: last round
tail       = TAIL_FRACTION x c x (t + ipc_item)      if w > 1 and n_chunks > 1 else 0
compute    = max(workers, balanced, parent) + tail
startup    = w x LAUNCH_S[m] + BOOT_S[m] + S
wall(w, c) = startup + compute + SHUTDOWN_S
serial     = N x t + S
ipc_share  = overhead / (overhead + N x t),
             overhead = 2 x N x ipc_item + N x ITEM_OVERHEAD_US / 1e6 + n_chunks x (CHUNK_OVERHEAD_US + WORKER_CHUNK_US) / 1e6

Selection:

1. Worker cap = min(cores, N, floor(MEM_HEADROOM x mem_limit / mem_per_worker)) (memory term only when --mem-per-worker-mb is given; the limit defaults to cgroup / MemAvailable).
2. For each w: among chunk candidates within 1% of that w's fastest wall time, take the smallest whose chunk lasts >= MIN_CHUNK_S; if none lasts that long, the largest near-fastest one.
3. Recommended w: the smallest worker count whose wall time is within 5% of the overall best (saves memory and cores for the same speed).

Warnings emitted:

What the model ignores: per-item variance beyond the tail term, memory bandwidth saturation, CPU frequency scaling with all cores busy (expect 10-20% less than linear), NUMA, disk contention, and GIL contention between the parent's feeder/handler threads and a busy parent main thread. It is a sizing aid; confirm with a benchmark on 1-5% of the data.

#9. Calibrating the constants

Measured on a 12-core Linux machine, Python 3.14, forkserver, 4 workers:

Reproduce on your hardware:

import concurrent.futures as cf
import multiprocessing as mp
import time


def ident(x):
    return x


def per_chunk_us(n: int = 5000) -> float:
    with cf.ProcessPoolExecutor(2, mp_context=mp.get_context("spawn")) as ex:
        list(ex.map(ident, range(8)))                          # warm up workers
        t = time.perf_counter()
        list(ex.map(ident, range(n), chunksize=1))
        return (time.perf_counter() - t) / n * 1e6


if __name__ == "__main__":
    print(f"{per_chunk_us():.0f} us per chunk")               # e.g. 30-150 us

Note that the same object repeated in one chunk is pickled once (pickle memoizes within a message), so benchmark with distinct objects.

#10. Worked examples

Tokenizing 200k documents, 4 ms each, 8 KB in, 2 KB out, 8 cores: plan --items 200000 --item-ms 4 --payload-kb 8 --result-kb 2 --cores 8 -> 8 workers, chunksize 8, ~101 s vs 13.3 min serial (7.9x), IPC 1.7%. Fine as is.

Row-level feature extraction, 1M rows at 0.02 ms: plan --items 1000000 --item-ms 0.02 --cores 16 -> 3 workers, ~2.2x, IPC 40%, "parent is the bottleneck". Fix the shape, not the pool: map over batches of 10k rows (--items 100 --item-ms 200) or vectorize.

CPU reranker, 5000 queries at 120 ms, 6 s model load, 1.8 GB per worker, 8 GB pod, spawn: plan --items 5000 --item-ms 120 --init-s 6 --mem-per-worker-mb 1800 --mem-limit-mb 8192 --start-method spawn --cores 16 -> memory caps workers at 3, ~2.9x. Options: quantize to shrink RSS, or share one model across threads when the runtime releases the GIL (onnxruntime and torch inference do).

Embedding with large payloads, 50k items at 30 ms, 2 MB in, fork: the parent spends ~3.5 min pickling: "parent is the bottleneck" and "payload >= 1 MB". Send file paths and let workers read, or place arrays in shared memory.