Article

Limiting Async Concurrency with Semaphores

Wrap each stage in its own asyncio.Semaphore, sized to the resource that stage exhausts, and wrap the whole read-transform-write unit in one more so overlapping stages cannot push open descriptors past the process limit. A single global semaphore cannot express this, because the stages saturate different things. It builds on the Async I/O for Raster Processing guide, part of the broader Spatial Batch Processing & Async Workflows reference.

Prerequisites

  • Python 3.10 or later
  • pip install aiohttp rasterio numpy
  • A pipeline whose stages you can name — fetch, decode, transform, write

Why One Number Is Not Enough

Each stage runs out of something different Fetching exhausts sockets and the remote rate limit and tolerates high concurrency. Decoding exhausts CPU and should match the core count. Reprojection exhausts memory and must be derived from the working-set size. Writing exhausts disk throughput and is usually best kept low. fetch sockets, and the remote service's rate limit tens decode CPU, in a process pool one per core reproject resident memory, per live buffer budget ÷ working set write disk throughput and fsync latency low single digits

Setting one global limit means picking a number that is wrong for three of the four. Too high and the reprojection stage exhausts memory; too low and the fetch stage sits idle waiting for the network it could have saturated.

The two ways of bounding work answer different questions, and a pipeline over variable-sized inputs usually needs both.

Bounding by count, or by size A count semaphore protects a resource with a fixed number of slots: sockets, cores, file descriptors. A byte budget protects memory, where the resource consumed per task varies by orders of magnitude across geospatial inputs. a count semaphore slots are interchangeable sockets, cores, descriptors simple and sufficient for fixed-cost stages for fixed resources a byte budget tasks consume unequally memory, where one tile may be 100x another admits on measured size for memory

Applying a count semaphore to memory is the mistake that survives testing and fails in production, because the test corpus is uniform and the real one is not.

Complete Working Implementation

# gistools/async_limits.py
from __future__ import annotations

import asyncio
import os
from contextlib import asynccontextmanager
from dataclasses import dataclass

import aiohttp
import numpy as np
import rasterio
from rasterio.io import MemoryFile


@dataclass
class Limits:
    fetch: asyncio.Semaphore
    cpu: asyncio.Semaphore
    memory: asyncio.Semaphore
    write: asyncio.Semaphore
    unit: asyncio.Semaphore        # bounds descriptors across overlapping stages

    @classmethod
    def build(cls, *, memory_budget_mb: int, per_tile_mb: int,
              descriptor_budget: int = 400) -> "Limits":
        cores = os.cpu_count() or 4
        by_memory = max(1, memory_budget_mb // max(per_tile_mb, 1))
        return cls(
            fetch=asyncio.Semaphore(32),
            cpu=asyncio.Semaphore(cores),
            memory=asyncio.Semaphore(by_memory),
            write=asyncio.Semaphore(4),
            # Each in-flight unit may hold a source and a destination handle.
            unit=asyncio.Semaphore(descriptor_budget // 2),
        )


@asynccontextmanager
async def held(*semaphores: asyncio.Semaphore):
    """Acquire several semaphores in a fixed order, release in reverse."""
    acquired = []
    try:
        for sem in semaphores:
            await sem.acquire()
            acquired.append(sem)
        yield
    finally:
        for sem in reversed(acquired):
            sem.release()


async def process_tile(session: aiohttp.ClientSession, url: str,
                       dst_path: str, dst_crs: str, limits: Limits) -> str:
    async with held(limits.unit):
        async with held(limits.fetch):
            async with session.get(url) as response:
                response.raise_for_status()
                payload = await response.read()

        async with held(limits.cpu, limits.memory):
            array, profile = await asyncio.to_thread(_decode_and_warp,
                                                     payload, dst_crs)

        async with held(limits.write):
            await asyncio.to_thread(_write, dst_path, array, profile)

    return dst_path


def _decode_and_warp(payload: bytes, dst_crs: str):
    with MemoryFile(payload) as memfile:
        with memfile.open() as src:
            profile = src.profile | {"crs": dst_crs}
            return src.read(), profile


def _write(path: str, array: np.ndarray, profile: dict) -> None:
    with rasterio.open(path, "w", **profile) as dst:
        dst.write(array)

Step Annotations

  1. held acquires in a fixed order. Acquiring two semaphores in different orders in different code paths is a deadlock waiting for load; a single helper with a fixed order removes it.
  2. The unit semaphore wraps everything. Without it, a unit holding a source handle while waiting for the write semaphore still counts against the descriptor limit — the accumulation described in Processing 100k GeoJSON Files with Python asyncio.
  3. asyncio.to_thread for the blocking work. Decoding and writing block; running them on the event loop stalls every other task, which is the most common way an async pipeline ends up slower than a synchronous one.
  4. CPU and memory semaphores are held together. The decode allocates the array the transform then uses, so releasing the memory bound between them would let more arrays exist than the budget allows.
  5. raise_for_status before reading. An error body is small and would otherwise be decoded as a raster, producing a confusing failure two stages later.

One Named Gotcha: A Semaphore Does Not Bound Memory Already Allocated

A semaphore limits how many tasks are inside a block, not how much they allocate. A limit derived from a two-hundred-megabyte tile does nothing when a tile turns out to be two gigabytes, and the process is killed with the semaphore at exactly its intended value.

Bounding by size rather than by count fixes it:

class ByteSemaphore:
    """Admit tasks until a byte budget is reached, rather than a count."""

    def __init__(self, budget_bytes: int):
        self._budget = budget_bytes
        self._used = 0
        self._cond = asyncio.Condition()

    @asynccontextmanager
    async def hold(self, size: int):
        async with self._cond:
            await self._cond.wait_for(lambda: self._used + size <= self._budget)
            self._used += size
        try:
            yield
        finally:
            async with self._cond:
                self._used -= size
                self._cond.notify_all()

The size is available before the work starts — from the object’s content length, or from the window dimensions and dtype — so admitting on it costs nothing and turns an occasional kill into a task that simply waits its turn.

Verification

# Descriptors stay bounded through the run.
while pgrep -f 'gistools fetch' > /dev/null; do
  ls /proc/$(pgrep -f 'gistools fetch' | head -1)/fd | wc -l
  sleep 2
done | sort -n | tail -1

# Concurrency is actually being used — a flat count of 1 means a stage is serialising.
jq -r 'select(.event=="tile_started") | .ts' run.jsonl | cut -c1-19 | uniq -c | head

A descriptor count that climbs steadily means the unit semaphore is missing or its bound is too high. A start-timestamp histogram showing one tile per second means something is holding a semaphore across an await it should not.

Two symptoms distinguish a pipeline that is bounded correctly from one that is merely slow.

Reading the symptom back to the bound A stage sitting at its limit while others idle means that bound is the bottleneck. All stages below their limits means something outside the semaphores is serialising. A climbing descriptor count means the unit-level bound is missing. one stage pinned at its limit that bound is the bottleneck — raise it or shard expected, if intended every stage below its limit something outside the semaphores serialises look for a blocking call descriptors climbing the unit-level bound is missing or too high the leak case

The middle row is almost always a blocking call left on the event loop, which no amount of semaphore tuning will fix.

Choosing the Numbers

Every semaphore above has a number in it, and picking them by intuition produces a pipeline that works on the machine it was tuned on. Three of the four can be derived.

The CPU bound is the core count, or the core count minus one if the event loop has real work to do between tasks. There is no reason to guess.

The memory bound is the budget divided by the per-task working set, both of which are measurable — the budget from the container limit, the working set from one instrumented run. That gives a number that follows the machine rather than a constant that does not.

The write bound is usually small and flat. Parallel writes to one device rarely help and often hurt, because the device serialises them anyway and the queue depth adds latency. Two to four is a reasonable starting point for spinning or network storage and can go higher on local NVMe.

Only the fetch bound genuinely needs measuring, because it depends on the remote service. The method is to raise it until throughput stops improving or the service starts returning throttling responses, and then back off. Doing that once against a representative endpoint gives a number good for the life of the pipeline.

When a Semaphore Is the Wrong Tool

Two situations look like concurrency limits and are better solved elsewhere.

The first is a rate limit expressed per second rather than in flight — a service allowing a hundred requests per second regardless of how long each takes. A semaphore bounds concurrency, not rate, and a pipeline with fast responses will exceed the limit at any semaphore value. A token bucket is the right shape there, and it composes with a semaphore rather than replacing it.

The second is back-pressure through a pipeline of stages. When a fast producer feeds a slow consumer, bounding the consumer does not stop the producer building an unbounded queue in between. A bounded asyncio.Queue between the stages does, because a full queue blocks the producer — which is the behaviour you want and is not what a semaphore provides.

Both are worth recognising because the symptom is the same: a pipeline that respects its limits and still runs out of something. If the limits are held and memory grows anyway, the problem is between the stages rather than inside them.

Finally, remember that a semaphore acquired inside a task is released when the task is cancelled only if the release happens in a finally block. The held helper above does exactly that, which is why it is worth using rather than acquiring inline: a cancellation partway through a batch otherwise leaks a permit per cancelled task, and the pipeline quietly stops admitting work.

One last practical note: name the semaphores after the resource rather than the stage. A field called fetch invites someone to reuse it for a second kind of fetch that saturates something else, whereas one called connections makes the mistake obvious at the call site. The same applies to the memory bound, which is about bytes rather than about reprojection and will be needed by any stage that allocates. Naming by resource also makes the sizing arithmetic self-documenting, because the number beside connections is plainly a count of connections and the number beside memory_mb is plainly a budget.