Building Async Pipelines for Batch Floor Plan Processing

This guide covers the async ingestion layer in detail — the bounded-queue, backpressure-aware front end of the Async Batch Processing Pipelines architecture that streams DWG, DXF, and SVG floor plans into CPU-bound vectorization without starving the event loop or exhausting memory.

Concept Definition: The Async Ingestion Layer

The async ingestion layer is the stage that turns a directory (or object-store prefix) full of heterogeneous floor plan exports into a steady, memory-safe stream of work units, then hands the CPU-heavy geometry off the event loop. It is defined by three invariants:

  1. Bounded buffering. A producer (the directory scanner) and a set of consumer coroutines (the workers) communicate through a single asyncio.Queue with a fixed maxsize. The buffer holds metadata, never file bytes — a path, a source format, a floor identifier, and a trace id. Queue depth therefore bounds metadata, not megabytes.
  2. Backpressure for free. When workers fall behind, the queue fills and the scanner’s await queue.put() blocks. Ingestion automatically slows to match consumption — no manual rate limiter, no token bucket. This is the single most important property: an unbounded glob() + eager open() loop has no backpressure, so it exhausts file descriptors and is OOM-killed on a campus of multi-storey exports.
  3. Loop-safe offloading. Reads are non-blocking (aiofiles), and the synchronous geometry kernel (ezdxf, shapely, lxml) runs inside a ProcessPoolExecutor reached through loop.run_in_executor. The worker awaits the executor and yields the loop to its peers while a separate process chews through the polygonization.
Resident memory under bounded and unbounded ingestion Two curves against the number of floor plans discovered in a batch. The unbounded strategy, which globs the directory and opens every file eagerly, rises linearly at about 1.9 MB per open plan and crosses a 2 GB worker ceiling at roughly 800 plans, where it is OOM-killed. The bounded strategy holds a queue of at most twice the worker count, so memory rises only until the queue fills at 32 manifests and then stays flat at roughly 210 MB no matter how large the batch grows. Memory is bounded by queue depth, not by batch size 0 1000 2000 3000 0 500 1000 1500 floor plans discovered in the batch resident set size (MB) unbounded glob() + eager open() bounded queue (maxsize = 2 x workers) OOM-killed here on a 2 GB worker

Why the queue is bounded. The unbounded path holds one open handle per discovered plan, so resident memory tracks batch size and a large campus export kills the worker. A bounded queue caps residency at the queue depth, so a 40-plan batch and a 4,000-plan batch cost the same memory.

The contrast with a naive synchronous batch is sharp: blocking geometry work inside an async def freezes the whole loop, so health checks, progress reporting, and queue draining all stall behind one polyline triangulation. The ingestion layer exists precisely to keep that CPU work off the loop thread.

Loop-safe async ingestion: bounded queue with backpressure, then offloaded geometry The diagram is split into two horizontal bands. The upper band is the event loop thread, which must never block. In it, left to right: a Directory Scanner walks DWG, DXF, and SVG files and emits a lightweight TaskManifest (path, source format, floor level, trace id); it calls await queue.put() into a bounded asyncio.Queue whose maxsize is two times the worker count and which buffers manifests, not file bytes. When that queue is full, put() blocks, shown by a dashed backpressure arrow looping back to the scanner. N worker coroutines call await queue.get() under a Semaphore that caps concurrent open handles, then perform a non-blocking aiofiles read that yields the loop while waiting. Each worker writes its result as a GeoJSON FeatureCollection carrying floor_level, trace_id, and topology_hash. The lower band is the worker processes, off the loop: a vertical run_in_executor arrow crosses the band boundary, carrying raw bytes down to a ProcessPoolExecutor that runs the synchronous ezdxf, shapely, and lxml geometry kernel with one process per core, and a return arrow carries the extracted geometry back up. Because the heavy work runs in a separate process, a slow polygon triangulation never freezes the event loop. Loop-safe ingestion — bounded queue, backpressure, offloaded geometry backpressure — put() blocks while the queue is full EVENT LOOP THREAD — non-blocking, never stalls WORKER PROCESSES — off the loop A separate OS process runs the synchronous geometry kernel, so a slow polygon triangulation can never freeze the loop. put() get() write 1 2 3 Directory Scanner rglob / object-store prefix DWG · DXF · SVG emits TaskManifest path · format · level · trace_id asyncio.Queue maxsize = 2 × workers buffers manifests, never file bytes Worker Coroutines × N async with sem: → caps open handles await aiofiles.read() → raw bytes non-blocking I/O — the await yields the loop GeoJSON FeatureCollection floor_level · trace_id topology_hash · source_format run_in_executor raw bytes ↓ geometry ↑ 4 ProcessPoolExecutor ezdxf shapely lxml one process per core · CPU-bound polygonization

Minimal Working Example

The snippet below is the smallest complete ingestion layer that demonstrates the pattern end to end: a bounded queue, semaphore-throttled non-blocking reads, and a process pool for the blocking parse. It scans a directory, reads each file off the loop, and counts entities in a worker process.

import asyncio
import logging
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path

import aiofiles

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("floorplan.ingest")

def parse_entities(raw: bytes, suffix: str) -> int:
    """CPU-bound kernel: runs in a worker process, never touches the loop."""
    return raw.count(b"LINE") if suffix == ".dxf" else raw.count(b"<path")

async def worker(wid: int, queue: asyncio.Queue, sem: asyncio.Semaphore,
                 pool: ProcessPoolExecutor) -> None:
    loop = asyncio.get_running_loop()
    while (fp := await queue.get()) is not None:
        try:
            async with sem:                                   # cap concurrent reads
                async with aiofiles.open(fp, "rb") as fh:
                    raw = await fh.read()                     # non-blocking I/O
            count = await loop.run_in_executor(pool, parse_entities, raw, fp.suffix)
            logger.info("[w%d] %s -> %d entities", wid, fp.name, count)
        except (OSError, ValueError) as exc:
            logger.error("[w%d] ingest failed | %s | %s", wid, fp.name, exc)
        finally:
            queue.task_done()

async def run(scan_dir: Path, workers: int = 4) -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=workers * 2)  # bounded buffer
    sem = asyncio.Semaphore(workers)
    with ProcessPoolExecutor(max_workers=workers) as pool:
        consumers = [asyncio.create_task(worker(i, queue, sem, pool)) for i in range(workers)]
        for fp in (*scan_dir.rglob("*.dxf"), *scan_dir.rglob("*.svg")):
            await queue.put(fp)                               # blocks when full -> backpressure
        await queue.join()
        for _ in consumers:
            await queue.put(None)                             # one sentinel per worker
        await asyncio.gather(*consumers)

if __name__ == "__main__":
    asyncio.run(run(Path("./floor_plans_input")))

Each worker exits cleanly on a None sentinel — one per worker — so the pipeline drains without forced cancellation or half-written output. In production the integer count is replaced by a full geometry extraction that emits a GeoJSON FeatureCollection (see the integration note below).

Parameter & Spec Reference

Parameter Type Default Notes
Queue(maxsize=...) int workers * 2 Bounds in-flight manifests, not bytes. Too small starves workers; too large defeats backpressure and grows memory.
Semaphore(value) int = workers Caps concurrent open file handles. Critical against high-latency network/object storage that throttles.
ProcessPoolExecutor(max_workers=...) int physical cores The real parallelism cap for geometry work. In a memory-constrained container, set to min(cores, mem_budget // peak_file_RAM).
aiofiles.open(..., "rb") mode str "rb" Read CAD/SVG as bytes; decode inside the kernel so encoding failures stay off the loop.
sentinel (None) Optional one per worker Signals end-of-stream so each consumer exits on its own.
wait_for(queue.get(), timeout) float 30.0 Optional dead-man timer so a stalled producer cannot hang a worker forever.

Size the semaphore and the pool to the physical core count (or a tuned fraction) so total resident memory stays under the cgroup allocation. The queue maxsize is a buffer-tuning knob, not a parallelism knob — raising it does not add throughput, it only adds slack before backpressure engages.

Common Errors & Fixes

1. RuntimeError: cannot schedule new futures after shutdown — A worker called run_in_executor after the ProcessPoolExecutor was torn down, usually because the pool’s with block exited while a coroutine was still in flight. Keep the executor alive until every worker has joined; let the context manager close it last:

Choosing the execution context for each stage of the ingestion layer A single decision node asking where a unit of work should run, fanning out to three outcomes. Work that waits on I/O stays on the event loop and is expressed as awaits on aiofiles.open and queue.get. Work that is a blocking call into a C library which releases the GIL goes to a thread pool via asyncio.to_thread. Work that burns CPU in Python — the ezdxf, shapely and lxml geometry kernel — goes to a process pool through loop.run_in_executor, because a thread would hold the GIL and stall the loop. Loop safety is a routing decision, made per stage Where should this unit of work run? the one question that decides loop safety waits on I/O Event loop await aiofiles.open() await queue.get() blocking C call Thread pool to_thread() for libs that release the GIL burns CPU in Python Process pool run_in_executor() ezdxf, shapely, lxml

The rule that keeps the loop responsive. Only the middle branch is arguable; the outer two are not. Python-level CPU work on the loop freezes every other coroutine, and a thread does not help because it never releases the GIL.

async def run(scan_dir: Path, workers: int = 4) -> None:
    pool = ProcessPoolExecutor(max_workers=workers)
    try:
        consumers = [asyncio.create_task(worker(i, queue, sem, pool)) for i in range(workers)]
        await queue.join()
        await asyncio.gather(*consumers)
    finally:
        pool.shutdown(wait=True)   # only after all consumers have exited

2. Queue join() hangs forever. Every queue.get() must be matched by exactly one queue.task_done(), or join() blocks indefinitely. The bug is almost always a task_done() skipped when the parse raises. Guard it with try/finally (as in the worker above) so the counter is decremented even on a ValueError from a malformed DXF header.

3. BlockingIOError / file-descriptor exhaustion at scale. The scanner is opening files faster than workers close them, typically because reads bypassed the semaphore. Confirm every aiofiles.open sits inside async with sem:, and lower the semaphore value below your storage backend’s connection ceiling. If the source is an object store that throttles, treat OSError as retryable with bounded exponential backoff rather than failing the whole batch.

Integration Point

The ingestion layer is one stage in a longer chain. Upstream, per-format extraction is normalized before a manifest is enqueued — rasterized PDFs and vector DWG need different read strategies, which is why SVG/DWG Parsing Workflows run ahead of the queue. Downstream, the bytes a worker reads feed geometry extraction, then Wall & Door Detection Algorithms identify passable edges before a routing graph is built, and blueprint metadata is resolved by Attribute Mapping from Blueprints. Each worker emits the same envelope every other parser produces — a GeoJSON FeatureCollection whose metadata carries floor_level, trace_id, topology_hash, and source_format — so the validated geometry can later be projected into a consistent Indoor Coordinate Reference System, with multi-storey jobs relying on Level Mapping & Z-Axis Logic to keep floor levels distinct.

Frequently Asked Questions

What should `maxsize` on the queue actually be?

Twice the worker count is the working default, and the reasoning matters more than the number. The queue exists to keep every worker fed while the scanner is briefly slow, so it needs at least one queued manifest per worker plus a little slack — hence 2 * workers. Making it much larger buys nothing, because manifests are tiny and the real memory is held by in-flight reads rather than by the queue, while it does hide backpressure: with a queue of ten thousand the scanner races through the whole directory before put() ever blocks, and you are back to an unbounded scan with extra steps. Making it smaller than the worker count starves workers whenever the scanner stalls on a slow directory listing. If you are tuning this at all, measure whether queue.qsize() spends most of its time at zero (scanner-bound) or at maxsize (worker-bound), and fix that instead.

Why a `ProcessPoolExecutor` rather than threads?

Because the geometry kernel is Python-level CPU work. ezdxf parses entities into Python objects, shapely calls into GEOS but marshals through Python for every geometry, and lxml builds a Python tree — all of which hold the GIL. A thread pool would serialise those calls onto one core while adding context-switch overhead and making the failure modes harder to reason about. A process pool gives each worker its own interpreter and its own GIL, so a twelve-core machine actually runs twelve polygonisations at once. The cost is that arguments and results are pickled across the process boundary, which is why workers send raw bytes and receive a compact FeatureCollection rather than passing live shapely objects around. If a stage genuinely spends its time inside a C library that releases the GIL, asyncio.to_thread is the cheaper choice — but very little of a floor-plan pipeline qualifies.

How do I stop one corrupt drawing from killing the whole batch?

Catch per task, not per batch, and record the failure as data. Each worker wraps its run_in_executor call in a try/except that catches the two exceptions that actually fire — ezdxf.DXFStructureError for malformed drawings and OSError for unreadable files — logs the path and trace id, appends the failure to a report, and calls queue.task_done() before moving on. The batch then completes with, say, 1,197 successes and three named failures rather than dying at file 412 with a stack trace and no record of what had already been processed. Broad except Exception around the executor call is worth adding as an outer net as well, because a ProcessPoolExecutor can raise BrokenProcessPool if a worker segfaults inside a native geometry call — a condition no amount of input validation prevents.

This page is part of the Async Batch Processing Pipelines section, within the Automated Floor Plan Parsing & Vectorization area.