Skip to content

Optimizing worker pool sizes for mixed I/O and CPU workloads

A pool sized for I/O thrashes on CPU work, and a pool sized for CPU starves under I/O wait. When a single workload mixes both — say, fetching a payload over the network (I/O), then parsing and hashing it (CPU) — a single max_workers number is wrong for at least half the task. The symptom is recognizable: CPU pins at 100% across all cores while the queue keeps backing up, or cores sit idle while throughput stalls behind network latency. Both look like "the pool is too small" on a dashboard, but the fixes are opposite — add threads in the second case, remove them in the first — which is why guessing at max_workers rarely converges.

The root cause is that the two failure modes are governed by different physics. Thread concurrency helps only while threads are parked in I/O syscalls (where the GIL is released); stack more threads on CPU-bound work and they simply queue for the one interpreter lock, adding context-switch overhead for zero parallelism. Process concurrency gives true parallelism but caps usefully at core count and pays serialization cost on every task. So the right number depends entirely on where each task spends its time — and "each task" is rarely uniform in a real service. This guide walks through measuring the I/O-to-CPU ratio of your tasks, turning that ratio into concrete pool sizes with established concurrency formulas, and replacing static sizing with a feedback-driven controller that resizes safely at runtime.

Prerequisites

  • Python 3.11+ for asyncio.timeout() and modern TaskGroup idioms in the surrounding pool code.
  • Standard library only here: concurrent.futures, multiprocessing, time, tracemalloc, sys. A profiler such as py-spy or cProfile is useful for validation but not required to run the snippets.
  • Working knowledge of the pool mechanics this builds on — the worker-coroutine-plus-queue structure and the executor-backed variant from worker pool implementations, and the loop-blocking constraint from the parent Concurrent Execution & Worker Patterns overview.
Sizing by I/O-to-CPU ratio Wall-clock time of a task is split into I/O wait and CPU compute; the ratio decides routing to a thread or process pool and the worker count. Size by the I/O-to-CPU ratio wall_clock I/O wait (W) CPU (C) W = wall_clock − cpu_time W/C > 1 → ThreadPool N = cores × (1 + W/C) threads yield during I/O C/W > 1 → ProcessPool N = min(cores, cores + 1) cap to avoid thrashing

Step 1 — Profile each task's I/O-to-CPU ratio

Before you can size anything, measure the split between time spent waiting on I/O and time spent burning CPU, per task category. The trick is comparing two clocks: time.perf_counter() measures wall-clock (which includes I/O wait), while time.process_time() measures only CPU time charged to the process. The difference is I/O wait. This is deliberately cheaper than a sampling profiler — you can leave it running in production behind a flag, accumulating a rolling mean per task type, because two clock reads per call cost essentially nothing.

The rolling-mean design matters: a single sample is noisy (one GC pause or one slow DNS lookup skews it), so the profiler below smooths over the last samples calls. Keep a separate profiler instance per task category (e.g. one for "fetch", one for "parse"), not one global instance, because the whole point is that categories have different ratios and averaging across them reproduces the original problem.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import time
import functools
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from typing import Callable, Any


class WorkloadProfiler:
    def __init__(self, samples: int = 100):
        self.samples = samples
        self.w_ratio = 0.0   # rolling mean I/O wait
        self.c_ratio = 0.0   # rolling mean CPU time

    def profile_task(self, func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            wall_start = time.perf_counter()
            cpu_start = time.process_time()
            try:
                return func(*args, **kwargs)
            finally:
                self._update(
                    time.perf_counter() - wall_start,
                    time.process_time() - cpu_start,
                )
        return wrapper

    def _update(self, wall_time: float, cpu_time: float) -> None:
        io_wait = max(0.0, wall_time - cpu_time)
        self.w_ratio = (self.w_ratio * (self.samples - 1) + io_wait) / self.samples
        self.c_ratio = (self.c_ratio * (self.samples - 1) + cpu_time) / self.samples

    @property
    def optimal_executor(self) -> type:
        if self.c_ratio == 0:
            return ThreadPoolExecutor
        return ThreadPoolExecutor if (self.w_ratio / self.c_ratio) > 1.0 else ProcessPoolExecutor

Verify: decorate a representative task, run it a few hundred times against real dependencies, then inspect profiler.w_ratio and profiler.c_ratio. A network-bound task should show w_ratio several times larger than c_ratio; a parsing/hashing task the reverse. If a "network" task shows high c_ratio, you have hidden CPU work (TLS handshakes, JSON deserialization, regex compilation) masquerading as I/O — confirm with cProfile.

Step 2 — Convert the ratio into pool sizes

With the ratio in hand, apply the standard concurrency formulas. Thread pools scale above core count because threads spend most of their time parked on I/O and yielding the GIL; process pools cap at core count because each process genuinely competes for a physical core and extra processes only add context-switch and memory overhead. The thread formula traces back to Brian Goetz's concurrency work: a thread that waits W and computes C keeps a core busy only C / (W + C) of the time, so to keep all cores busy you need cores × (W + C) / C = cores × (1 + W/C) threads. The formula is a starting point, not a final answer — it assumes the GIL is fully released during the wait, which holds for socket and disk I/O but not for CPU work mislabeled as I/O (see the pitfalls).

  • Thread pool: N_threads = cores × (1 + W/C) — the more a task waits on I/O relative to compute, the more threads you can profitably stack.
  • Process pool: N_processes = min(cores, cores + 1) — capping at core count prevents thrashing; the +1 only helps if one process is reliably I/O-bound.
  • Never mix: route I/O and CPU tasks to separate pools via a dispatcher. A single shared executor guarantees GIL contention drags down the I/O tasks.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import multiprocessing
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def calculate_pool_sizes(w_c_ratio: float, cpu_cores: int | None = None) -> dict[str, int]:
    cores = cpu_cores or os.cpu_count() or 4
    thread_workers = max(1, int(cores * (1 + w_c_ratio)))
    process_workers = min(cores, cores + 1)
    return {"threads": thread_workers, "processes": process_workers}


def initialize_executors(thread_workers: int, process_workers: int):
    thread_pool = ThreadPoolExecutor(
        max_workers=thread_workers, thread_name_prefix="io_worker"
    )
    # "spawn" avoids fork-related deadlocks with threads/locks held at fork time.
    process_pool = ProcessPoolExecutor(
        max_workers=process_workers,
        mp_context=multiprocessing.get_context("spawn"),
    )
    return thread_pool, process_pool

Verify: feed the measured w_ratio / c_ratio into calculate_pool_sizes. For an I/O-heavy workload (ratio ~4 on 8 cores) you should get ~40 threads and 8 processes; for a CPU-heavy one (ratio ~0.2) ~9 threads and 8 processes. Sanity-check that threads is not absurd — a ratio implying hundreds of threads usually means the work belongs on the loop as coroutines instead.

Step 3 — Replace static sizing with an adaptive controller

Static sizes are correct only at the load you measured. Real traffic shifts network latency and arrival rate, so wrap the pool in a controller that tracks rolling latency and queue depth and resizes within bounds. Because concurrent.futures cannot hot-swap max_workers — the attribute is fixed at construction and the thread set is managed internally — resizing means draining and recreating the executor. That is an expensive operation (shutdown(wait=True) blocks until in-flight tasks complete), so it must be rare: gate it behind a cooldown so at most one resize happens per window, and use hysteresis (a scale-up threshold well above the scale-down threshold) so the controller cannot oscillate across a single boundary. The controller resizes on two signals together — latency and queue depth — because either alone is ambiguous: high latency with an empty queue is just slow work, not under-provisioning, and only the combination of high latency and a backed-up queue justifies adding workers.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import asyncio
import threading
import time
from collections import deque
from concurrent.futures import Future, ThreadPoolExecutor


class AdaptivePoolController:
    def __init__(self, base_workers: int, max_workers: int, cooldown: float = 5.0):
        self.base_workers = base_workers
        self.max_workers = max_workers
        self.cooldown = cooldown
        self._executor = ThreadPoolExecutor(max_workers=base_workers)
        self._lock = threading.Lock()
        self._last_resize = 0.0
        self._latency = deque(maxlen=100)
        self._queue_depth = 0

    def _resize(self, workers: int) -> None:
        self._executor.shutdown(wait=True)       # drain in-flight cleanly
        self._executor = ThreadPoolExecutor(
            max_workers=workers, thread_name_prefix="adaptive_io"
        )
        self._last_resize = time.monotonic()

    def _adjust(self) -> None:
        avg = sum(self._latency) / max(1, len(self._latency))
        cur = self._executor._max_workers
        if avg > 0.5 and self._queue_depth > cur * 1.5 and cur < self.max_workers:
            self._resize(min(self.max_workers, cur + 2))     # scale up on backlog
        elif avg < 0.1 and self._queue_depth < cur * 0.3 and cur > self.base_workers:
            self._resize(max(self.base_workers, cur - 1))    # scale down when idle

    def submit(self, fn, *args, **kwargs) -> Future:
        with self._lock:
            if time.monotonic() - self._last_resize > self.cooldown:
                self._adjust()
            return self._executor.submit(fn, *args, **kwargs)

    def record_latency(self, duration: float) -> None:
        self._latency.append(duration)

    def record_queue_depth(self, depth: int) -> None:
        self._queue_depth = depth

Verify: drive synthetic load with rising latency and confirm worker count steps up by 2 only after the cooldown, then steps back toward base_workers when latency falls. Log every _resize call; in steady state you should see zero resizes — constant flapping means your up/down thresholds overlap. One subtlety to confirm: because submit() and _adjust() share self._lock, a resize cannot race a submission, but the shutdown(wait=True) inside _resize will hold that lock for the duration of the drain. If your tasks are long-running, that blocks new submissions for the whole drain — acceptable for occasional resizes, but a reason to keep the cooldown generous and the per-task duration bounded.

Verification

You cannot validate a pool size by reasoning alone — the GIL, the OS scheduler, and the downstream dependency all interact in ways the formulas approximate. Run a controlled load test and tie the three steps together, watching for these signals:

  • Throughput plateau finds the knee. Sweep worker count upward; throughput rises, then flattens, then drops. The flat point just before the drop is your size. Past it, you are paying context-switch overhead for nothing.
  • CPU vs queue depth disambiguates the bottleneck. Cores at ~100% with a growing queue means CPU-starved — shift work to processes or reduce threads. Cores idle with a growing queue means you are I/O-bound and can add threads.
  • IPC handoff stays cheap. Benchmark per-task serialization for the process pool; overhead above ~15ms per task means payloads are too large — move to shared memory or coarser task granularity rather than more processes.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import sys
import tracemalloc
import time


def measure_handoff_latency(executor, payload_size_kb: int) -> float:
    """Round-trip ms for submitting a payload to an executor."""
    payload = b"x" * (payload_size_kb * 1024)
    start = time.perf_counter_ns()
    executor.submit(len, payload).result()
    return (time.perf_counter_ns() - start) / 1e6


def dump_thread_stacks() -> dict[int, list[str]]:
    """Snapshot every thread's stack to find workers stuck on I/O or locks."""
    snapshot = {}
    for tid, frame in sys._current_frames().items():
        stack, f = [], frame
        while f:
            stack.append(f"{f.f_code.co_filename}:{f.f_lineno} ({f.f_code.co_name})")
            f = f.f_back
        snapshot[tid] = stack
    return snapshot

Diagnostic Hook: In production, export three series: average task latency (from record_latency), live queue_depth, and the controller's current _max_workers. The healthy picture is latency and depth fluctuating while worker count stays flat. A staircase of resizes correlated with steady latency means your thresholds are mis-tuned; a flat worker count with monotonically rising depth means you have hit max_workers and must scale out (more processes/hosts), not up. Periodically capture dump_thread_stacks() in staging — many threads parked on pthread_cond_wait/sem_wait confirm pool exhaustion or over-synchronization.

Pitfalls & edge cases

  • Setting thread pool size to os.cpu_count() for I/O-heavy work caps you far below capacity; those threads spend most of their life parked, so cores × (1 + W/C) is the right target, not cores.
  • Running CPU-bound tasks in a thread pool serializes them on the GIL — they execute one at a time on a single core no matter how many threads you add. Route them to a ProcessPoolExecutor.
  • Over-provisioning the process pool duplicates interpreter memory per process and adds OS context-switch overhead; past core count you lose throughput and risk OOM.
  • Hidden CPU work inside "I/O" tasks (TLS, JSON parsing, regex compilation, decompression) inflates c_ratio and silently makes a thread pool GIL-bound. Profile with cProfile and reclassify.
  • Resizing without a cooldown turns the adaptive controller into a thrash machine — each drain-and-recreate is expensive, and overlapping up/down thresholds make it oscillate forever. Always gap the thresholds and enforce the cooldown.

Frequently Asked Questions

Should I use threads or processes for mixed I/O and CPU workloads?

Route I/O-bound tasks to ThreadPoolExecutor and CPU-bound tasks to ProcessPoolExecutor. Mixing both in a single pool causes GIL contention and suboptimal utilization. Use a dispatcher that classifies tasks by their measured W/C ratio and routes accordingly.

How do I calculate the ideal thread pool size for high-latency network requests?

Use N = cores times (1 + W/C), where W/C is the ratio of average I/O wait time to CPU processing time. Start around 2 to 4 times cores for moderate latency and scale up as network round-trip time increases, validating against measured queue depth.

Can asyncio replace worker pools for mixed workloads?

asyncio excels at I/O but still runs CPU work on the loop thread, which blocks every other task. Offload CPU work to a separate process pool via loop.run_in_executor() and keep blocking CPU operations out of async handlers entirely.