Bounding Concurrency for Async Spatial Scans

Dispatching every spatial query to a thread pool and awaiting them all is the shape most async DuckDB code takes, and it is the shape that turns a memory limit into an out-of-memory kill — because the limit is shared and the task count is not. This walkthrough, part of the async execution patterns reference, covers sizing a bound from the memory budget, bounding the result queue as well as the query count, and telling backpressure apart from a stall.

Root-Cause Analysis: why unbounded dispatch fails #

• The memory limit is shared, the task count is not. Every in-flight query draws on the same memory_limit. Ten concurrent overlays each expecting a gigabyte will exceed an eight-gigabyte ceiling regardless of how comfortable any one of them is alone.
• Each query already uses every core. DuckDB parallelises a single query across its own threads, so adding concurrent queries adds contention rather than parallelism once the cores are busy. The second concurrent query is usually the last one that helps.
• Results accumulate faster than they are consumed. A fast producer and a slow consumer with an unbounded queue between them holds every batch produced so far, and geometry batches are large. The memory is in the queue rather than in the engine, so memory_limit does not see it.
• Failures arrive after the memory is gone. A task that would have failed cheaply on its own instead contributes to a process-level kill, so the diagnostic is a dead process rather than a query error.
• The bound is usually set from cores. A semaphore sized from the CPU count is sized against the resource that is not scarce. Memory is what runs out, and cores are already fully used by one query.

The distinguishing question is what the heaviest concurrent query needs. Dividing the memory limit by that number gives the bound; the core count gives a number that is usually far too large.

Cores were saturated at one. Everything after that spends memory to buy nothing.

Deterministic Configuration #

import asyncio, duckdb

DB = duckdb.connect("gis.duckdb")
DB.execute("INSTALL spatial; LOAD spatial")
DB.execute("SET memory_limit = '8GB'")
DB.execute("SET threads = 8")
DB.execute("SET temp_directory = '/var/tmp/duckdb_async'")

# memory_limit ÷ peak per query, rounded down. Not the core count.
PEAK_PER_QUERY_GB = 2
SLOTS = asyncio.Semaphore(max(1, 8 // PEAK_PER_QUERY_GB - 1))   # leave headroom

# Bound the results as well as the queries: an unbounded queue holds every
# batch produced so far, and that memory is outside the engine's budget.
QUEUE_MAX_BATCHES = 4

• The peak working set of the heaviest concurrent query has been measured, not guessed.
• The semaphore is sized from memory rather than from the core count.
• Any queue between producer and consumer has a maxsize.
• Each task uses its own cursor over the shared database.

Optimized Execution Pattern #

The pattern is a semaphore around the dispatch and a bounded queue around the results, so both the engine memory and the Python-side memory have a ceiling.

# ANTI-PATTERN: every unit dispatched at once, results collected into a list.
# Two unbounded things at the same time: concurrent queries and held results.
results = await asyncio.gather(*[
    asyncio.to_thread(run_tile, tile) for tile in tiles      # all 400 at once
])

# PATTERN: bounded dispatch, bounded results, one cursor per task.
async def run_all(tiles):
    out = []
    async def one(tile):
        async with SLOTS:                                   # bounds engine memory
            cur = DB.cursor()                               # isolation, shared pool
            try:
                return await asyncio.to_thread(run_tile, cur, tile)
            finally:
                cur.close()

    for coro in asyncio.as_completed([one(t) for t in tiles]):
        out.append(await coro)                              # consumed as they finish
    return out

as_completed matters as much as the semaphore: gather holds every result until the last task finishes, so a bounded dispatch with an unbounded collection has simply moved the memory from the engine into the list.

Only the first row is inside the engine budget. The other three are Python’s problem.

Measuring the peak per query #

The bound is only as good as the number it is derived from, and the peak working set of a query is not something to estimate. Running the heaviest query alone and sampling resident memory gives it directly, and doing so takes a minute.

Where the workload has a range of query costs, size the bound against the heaviest rather than the median — the whole point is that the worst case does not exceed the ceiling. Where that makes the bound uselessly small for the cheap queries, the answer is two semaphores rather than one compromise, exactly as it is for a request handler serving mixed endpoints.

import resource, time

def peak_rss_gb() -> float:
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 / 1024

before = peak_rss_gb()
run_tile(DB.cursor(), heaviest_tile)          # alone, nothing else running
print(f"peak for the heaviest query: {peak_rss_gb() - before:.1f} GB")

Diagnostic Queries & Plan Validation #

Backpressure and a stall look similar from outside and are distinguished by whether anything is progressing.

# Sampled every few seconds. Queue full with the consumer advancing is
# backpressure working; queue full with nothing advancing is a stall.
print({
    "slots_free": SLOTS._value,
    "queue_depth": queue.qsize(),
    "completed": completed_count,
})

A full queue and a rising completed count is backpressure doing its job — the producer is waiting because the consumer is the bottleneck, which is correct. A full queue and a flat completed count is a stall, and the usual cause is a consumer awaiting something that will never happen.

The third row is the common case, and it argues for a smaller bound rather than a larger one.

Two bounds rather than one compromise #

A workload with a wide range of query costs is badly served by a single semaphore, because the bound has to be sized against the heaviest query and then applies to the cheapest. A batch of four hundred trivial tile summaries and a handful of full-layer dissolves should not share one slot count.

Splitting them is straightforward and makes the cost model explicit: a generous semaphore for the cheap class, a tight one for the expensive class, and a rule that a task acquires exactly one. The result is that the expensive queries cannot exhaust the ceiling and the cheap ones are not throttled to protect against a query they are nothing like.

CHEAP = asyncio.Semaphore(8)      # bbox summaries, a few hundred MB each
HEAVY = asyncio.Semaphore(2)      # dissolves and overlays, gigabytes each

async def dispatch(unit):
    gate = HEAVY if unit.is_overlay else CHEAP
    async with gate:
        cur = DB.cursor()
        try:
            return await asyncio.to_thread(run_unit, cur, unit)
        finally:
            cur.close()

Geometry Validation & Fallback Routing #

Where the bound is right and a single query still exceeds the ceiling, the fallback is to make that query smaller rather than to raise the limit.

# Split the heavy unit rather than granting it more memory. Each piece fits
# under the same per-query budget the bound was derived from.
async def run_heavy(unit):
    async with HEAVY:
        cur = DB.cursor()
        try:
            for sub in unit.split_by_grid(cell_m=5000):
                await asyncio.to_thread(run_unit, cur, sub)
        finally:
            cur.close()

Frequently Asked Questions #

How many concurrent queries should I allow? #

Divide the memory limit by the peak working set of the heaviest query and round down, leaving some headroom. For heavy spatial work that is frequently two or three even on a machine with many cores, because each query already uses all of them and the second concurrent one is usually the last that helps.

Why is the core count the wrong denominator? #

Because DuckDB parallelises a single query across every core, so the cores are saturated at a concurrency of one. Adding more queries adds contention for CPU and competition for one shared memory ceiling. Memory is the scarce resource, so memory is what the bound should be derived from.

My memory grows even though queries complete normally. Why? #

Results are accumulating outside the engine budget. memory_limit governs the engine, not the Python objects holding its output — an unbounded queue, or gather holding every result until the last task finishes, keeps everything produced so far. Bound the queue and consume incrementally.

What is the difference between backpressure and a stall? #

Whether anything is progressing. A full queue with a rising completion count is backpressure working correctly: the producer is waiting because the consumer is the bottleneck. A full queue with a flat completion count is a stall, and the usual cause is a consumer awaiting something that will never arrive.

Should every task have its own connection? #

Its own cursor, over one shared database. A separate connection per task multiplies nothing useful and re-pays the startup cost; a shared cursor serialises the tasks and interleaves their session state. A cursor per task is the arrangement that isolates and shares correctly.

What if one query alone exceeds the limit? #

Then the bound is not the problem and raising the limit only defers it. Split that query — by grid cell, by partition, by time window — so each piece fits under the per-query budget the bound was derived from. A single unbounded query is a query design problem rather than a concurrency one.

Related #

• Async execution patterns — what asyncio does and does not buy
• Cancelling async spatial scans safely — freeing a slot that a timeout should have released
• Pooling DuckDB connections in web services — the same bound, inside a request handler

Up: Async Execution Patterns for DuckDB Spatial Workloads