Worker-Pool Sizing for Caption Batches

A caption-batch pipeline that sets its worker count to os.cpu_count() is a memory time bomb: on an 8-core host with 7 GB of RAM, eight workers each holding a 1.2 GB feature-length transcript in memory need 9.6 GB and the OS SIGKILLs the batch mid-run, losing every asset that had not yet flushed its result. The correct pool size is the smaller of two ceilings — the CPU count and how many copies of the peak per-asset resident set the memory budget can hold — expressed as min(usable_cpu, floor(mem_budget / peak_rss)). Get the memory ceiling wrong in either direction and you either OOM (too many workers) or leave cores idle and drag out a nightly batch past its window (too few). The peak per-asset RSS is not guessed; it is measured with resource.getrusage, and the budget must honour a container’s cgroup limit rather than the host’s physical RAM, because os.cpu_count() and /proc/meminfo both over-report inside a container. This is the concurrency-sizing detail underneath async batch caption processing.

import os
import math
import resource
from concurrent.futures import ProcessPoolExecutor, as_completed

# resource.getrusage reports ru_maxrss in KILOBYTES on Linux, BYTES on macOS.
_RSS_UNIT = 1 if os.uname().sysname == "Darwin" else 1024


def peak_rss_bytes() -> int:
    """High-water peak RSS of this process since start. Measure it around a
    representative asset to get the per-worker memory cost the formula needs."""
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * _RSS_UNIT


def usable_cpu() -> int:
    """cgroup-aware CPU count. os.cpu_count() reports host cores even when the
    container is pinned to fewer, so prefer the scheduler affinity mask."""
    if hasattr(os, "sched_getaffinity"):
        return len(os.sched_getaffinity(0))
    return os.cpu_count() or 1


def available_ram_bytes() -> int:
    """Honour a cgroup memory limit when present; a container's real ceiling is
    the cgroup max, not the host's physical RAM from SC_PHYS_PAGES."""
    for path in ("/sys/fs/cgroup/memory.max",                     # cgroup v2
                 "/sys/fs/cgroup/memory/memory.limit_in_bytes"):  # cgroup v1
        try:
            with open(path) as fh:
                raw = fh.read().strip()
        except FileNotFoundError:
            continue
        if raw and raw != "max":
            limit = int(raw)
            if limit < (1 << 62):        # cgroup v1 sentinel for "unlimited"
                return limit
    return os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES")


def pool_size(peak_rss: int,
              headroom: float = 0.20,
              reserve_bytes: int = 512 * 1024 * 1024) -> int:
    """min(usable_cpu, floor(mem_budget / peak_rss)); never below 1.

    reserve_bytes holds back the interpreter + OS working set; headroom keeps
    a safety margin against RSS growth spikes on the largest asset.
    """
    budget = (available_ram_bytes() - reserve_bytes) * (1.0 - headroom)
    if peak_rss <= 0 or budget <= 0:
        return 1
    mem_bound = math.floor(budget / peak_rss)
    return max(1, min(usable_cpu(), mem_bound))


def run_batch(assets, work_fn, measured_peak_rss: int) -> tuple[int, dict]:
    """Process a caption batch with a memory-bounded process pool."""
    workers = pool_size(measured_peak_rss)
    results: dict = {}
    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(work_fn, asset): asset for asset in assets}
        for fut in as_completed(futures):
            asset = futures[fut]
            try:
                results[asset] = fut.result()
            except Exception as exc:                 # one poisoned asset must not kill the batch
                results[asset] = {"error": repr(exc)}
    return workers, results

Code walkthrough

peak_rss_bytes is the input the whole formula turns on, and its one portability trap is the unit: ru_maxrss is kilobytes on Linux and bytes on macOS, so _RSS_UNIT normalizes both to bytes. Call it after processing one representative asset — ideally the largest expected, a feature-length program — to capture the true per-worker high-water mark rather than a mid-run snapshot, because getrusage reports the peak since process start, not the current resident set.

usable_cpu and available_ram_bytes exist because a container lies. os.cpu_count() returns the host’s physical core count even when the orchestrator has pinned the container to two cores, so os.sched_getaffinity(0) — the actual schedulable CPU mask — is the honest number. Likewise SC_PHYS_PAGES reports host RAM, but the process will be OOM-killed at the cgroup limit, which is read from memory.max (cgroup v2) or memory.limit_in_bytes (v1). The sentinel guard (< 1 << 62) filters the absurd “unlimited” value cgroup v1 reports so it does not masquerade as a real ceiling.

pool_size is the formula itself: min(usable_cpu, floor(budget / peak_rss)). The budget is not the raw available RAM — it first subtracts reserve_bytes for the interpreter and OS working set, then applies a headroom fraction so a single asset that spikes 20% above its measured peak does not tip the host into swap or a kill. The floor is deliberate: a fractional worker is a worker that will not fit, so it rounds down, and max(1, ...) guarantees the batch still makes progress on a host so tight that only one asset fits at a time.

run_batch binds the size to a ProcessPoolExecutor — processes, not threads, because caption parsing is CPU-bound and the GIL would serialize threads. as_completed drains results as they finish, and the per-future try/except ensures one malformed asset raising inside a worker is recorded as an error rather than propagating out of fut.result() and tearing down the pool. This is the same fault-isolation discipline the parent batch stage relies on when it fans thousands of files across runners.

Threshold reference table

Edge cases & known gotchas

• Feature-length assets set the ceiling, not the average. Peak RSS must be measured on the largest asset the batch can contain, not a typical 22-minute episode. A pool sized on the average will OOM the moment a two-hour film lands in the queue, because one oversized worker blows the per-worker budget the formula assumed.
• Matrix jobs share one host’s RAM. When a CI matrix runs several batch shards on the same runner, each shard’s available_ram_bytes sees the whole cgroup, so summing their pools oversubscribes memory. Divide the budget by the number of concurrent shards, or give each shard its own cgroup-limited container.
• cgroup limits beat os.cpu_count() and /proc/meminfo. Inside a container both the CPU count and physical RAM read the host, so sizing on them can request eight workers on a two-core, 2 GB slice. Always prefer sched_getaffinity and the cgroup memory file, which is what the process is actually killed against.
• arm64 vs x86-64 RSS differs. The same asset can carry a larger resident set on arm64 (16 KB pages on some kernels vs 4 KB on x86-64) and different wheel builds, so a peak-RSS figure measured on an x86 developer laptop under-counts on an arm CI runner. Re-measure per architecture rather than porting the constant.
• IO-bound stages want a different bound. This formula assumes CPU-bound parsing where processes beat threads. A stage dominated by network fetches from object storage is IO-bound and can oversubscribe cores with an async or thread pool, where the memory ceiling — not the core count — becomes the only real limit.

Integration hook

This sizing routine is the concurrency governor for async batch caption processing: that stage defines the streaming, fault-isolated worker model, and pool_size is the function that decides how wide it runs on a given host without an OOM kill. It pairs directly with the flow-control in cron-triggered batch runner with backpressure, which throttles how fast assets enter the pool this formula sizes, and it caps the parallelism of format conversions like converting SCC to SRT without timing loss so a conversion batch never collectively exhausts the runner.

Related

• Async batch caption processing — the streaming, fault-isolated worker model this formula sizes.
• Cron-triggered batch runner with backpressure — the intake throttle that feeds the sized pool without overrunning it.
• Converting SCC to SRT without timing loss — a per-asset conversion whose batch parallelism this ceiling bounds.

Part of: SRT, SCC & WebVTT Parsing Workflows — the broadcast caption parsing reference.