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.

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: → Semaphore 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:

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.

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