When You Fan Out Streaming Sessions, Your Laptop's CPU Gives Out First — An Adaptive Throttle That Caps Concurrency by Measured Load

Measure and Smooth the Host CPU

The first step, before touching concurrency, is to observe host CPU continuously. Instantaneous readings jump around, so smooth them with an exponential moving average (EWMA).

import psutil
 
class CpuSampler:
    """Return host CPU usage smoothed with an EWMA."""
 
    def __init__(self, alpha: float = 0.3):
        self.alpha = alpha
        self.value = 0.0
        psutil.cpu_percent(interval=None)  # first call is a throwaway
 
    def sample(self) -> float:
        raw = psutil.cpu_percent(interval=None)
        self.value = self.alpha * raw + (1 - self.alpha) * self.value
        return self.value

A larger alpha reacts faster but follows noise; a smaller one is steadier but lags. Around 0.3 was my sweet spot: it ignores momentary spikes yet catches a real trend shift within a few seconds.

The Adaptive Gate — Move One Slot at a Time

Once you have measured CPU, adjust the number of concurrent sessions toward a target (say 70%). Drop one slot when overloaded, add one when there is headroom. Move gently — large jumps oscillate.

Because asyncio.Semaphore is awkward to resize on the fly, this small gate tracks the current allowance and in-flight count with a Condition.

import asyncio
from contextlib import asynccontextmanager
 
class AdaptiveCpuGate:
    """A gate that raises and lowers concurrency from measured host CPU."""
 
    def __init__(self, target_cpu=70.0, min_slots=1, max_slots=16, alpha=0.3):
        self.target_cpu = target_cpu
        self.min_slots = min_slots
        self.max_slots = max_slots
        self._slots = min_slots          # currently allowed concurrency
        self._in_flight = 0              # sessions running now
        self._sampler = CpuSampler(alpha)
        self._cond = asyncio.Condition()
 
    async def governor(self, stop: asyncio.Event, period: float = 2.0):
        """A resident loop that revises the limit periodically."""
        while not stop.is_set():
            cpu = self._sampler.sample()
            async with self._cond:
                if cpu > self.target_cpu:
                    self._slots = max(self.min_slots, self._slots - 1)
                elif cpu < self.target_cpu * 0.8:
                    self._slots = min(self.max_slots, self._slots + 1)
                self._cond.notify_all()   # wake waiters: a slot may have opened
            await asyncio.sleep(period)
 
    @asynccontextmanager
    async def slot(self):
        async with self._cond:
            # wait until in-flight drops below the allowance (backpressure)
            await self._cond.wait_for(lambda: self._in_flight < self._slots)
            self._in_flight += 1
        try:
            yield
        finally:
            async with self._cond:
                self._in_flight -= 1
                self._cond.notify_all()

Three things matter. First, the adjustment is asymmetric: drop one slot the moment you exceed the target, but only add one once you fall below 80% of it. Fast to shrink, cautious to grow — that keeps CPU off the ceiling. Second, wait_for naturally holds new sessions until a slot frees up, which is backpressure for free. Third, min_slots is at least one, so no matter how strained CPU gets, work never stops entirely.

Throttle and Keep Going, Don't Halt

The danger in a naive fixed-semaphore implementation is that overload detection tends to slide into halting everything. For unattended automation, running thin but steady gives a higher batch completion rate than stopping. Because AdaptiveCpuGate guarantees at least one slot, the worst case degrades to serial execution rather than dropping to zero.

The caller starts the gate alongside its resident governor:

async def run_one(site, gate, client):
    async with gate.slot():
        async with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": build_prompt(site)}],
        ) as stream:
            async for _ in stream.text_stream:
                pass
        return await stream.get_final_message()
 
async def main(sites, client):
    gate = AdaptiveCpuGate(target_cpu=70.0, min_slots=2, max_slots=12)
    stop = asyncio.Event()
    gov = asyncio.create_task(gate.governor(stop))
    try:
        await asyncio.gather(*(run_one(s, gate, client) for s in sites))
    finally:
        stop.set()
        await gov

Here client is an async Anthropic client. Pass the API key through an environment variable (ANTHROPIC_API_KEY); never hardcode it.

Before / After on My Own Box

Here is one comparison from my setup — a batch that runs article generation for several sites at once — using an older mini PC as the runner. These numbers are from my specific environment and will not be identical on every machine; read them as a directional guide.

The striking part: even after the gate cut concurrency nearly in half, total batch time grew only 12%. With CPU saturated, adding more sessions only made each one slower — throughput had barely improved. Pacing to a target CPU beat cramming sessions in, holding down heat and latency for a negligible difference in finish time.

Sharing One Script Across Machines

As an indie developer running this across several machines, the reason I rely on it is that the same script lets each machine pick its own concurrency to fit its own capacity. With a fixed constant, I had to revisit the number per machine; sharing a single policy — a target CPU percentage — lets each host drift toward its own optimum from live measurement. It is the kind of small operational lever I keep reaching for across the Dolice Labs sites.

A few field notes for putting this into production:

• Aim low. Target 70% and real peaks still swing to about 80%. On a machine shared with other scheduled jobs, starting around 60% errs on the safe side.
• Reconcile max_slots with your API rate limits. Spare CPU is useless if the server returns 429, so keep the ceiling consistent with your rate design.
• Governor period of 2–5 seconds. Too short oscillates; too long delays recovery from overload.
• Log the observations. Recording effective concurrency and CPU per line lets you later explain "why was that batch slow?" with a CPU plot.

Precisely because streaming itself is lighter now, the ceiling on "how many you can stack on one box" is set by your local host before the server. Next time you write a parallel batch, replace the fixed concurrency number with a single target CPU. You quietly stop re-tuning a constant every time you switch machines.

Thanks for reading. If you also stack many jobs solo, I hope this gives you a foothold for your own operational design.