VehicleMapping HD AV Spatial Data

Building Async Sensor Fusion Pipelines with Celery

Configure Celery to orchestrate non-blocking sensor fusion tasks — ingesting LiDAR sweeps, synchronized camera tensors, and GNSS/IMU odometry — without coupling worker memory or latency to the vehicle's primary control loop. This step assembles the distributed task layer of an asynchronous data pipeline architecture so that gigabyte-scale point clouds never transit the broker, workers recycle deterministically after heavy registration jobs, and hardware timestamps propagate intact through a strictly ordered fusion chain.

Celery's out-of-the-box configuration is fundamentally misaligned with this workload: its default pickle serializer, persistent worker processes, and broker-resident payloads all break down once a single task carries a multi-million-point cloud. The result is broker frame-size violations, heap fragmentation from long-lived NumPy and Eigen buffers, and blocking AsyncResult.get() calls that stall the worker. The configuration below corrects each of these for a production sensor fusion deployment.

Out-of-band transport keeps gigabyte payloads off the broker; only URIs flow through the task chain:

Out-of-band payload transport for an async sensor fusion pipeline The ingestion driver splits each frame into two paths: lightweight task kwargs carrying only URIs and hardware timestamps flow through the msgpack broker (RabbitMQ or Redis), while gigabyte-scale binary point clouds and image tensors are written to a shared store (/dev/shm or MinIO/S3). Prefork workers, recycled after one task, pull a task from the broker and fetch its binary payload by URI from the shared store. Each worker runs a strictly ordered task chain — ingest_raw, temporal_sync, coordinate_transform, spatial_fusion — emitting a fused HD-map tile. Ingestion driver hardware timestamps (ns) Broker · msgpack RabbitMQ / Redis · ≤256 KiB Shared payload store /dev/shm · MinIO / S3 prefork workers max_tasks_per_child = 1 kwargs = URIs + ts deliver task binary payloads fetch by URI (mmap) Ordered task chain (one worker) ingest_raw temporal_sync coordinate_transform spatial_fusion → fused HD-map tile

Prerequisites #

  • Python 3.10+ with celery==5.3.6, msgpack==1.0.8, redis==5.0.4 (result backend), and open3d==0.18.0 for the registration payloads.
  • Broker: RabbitMQ 3.13 or Redis 7.2. Examples assume RabbitMQ for task routing and Redis for results.
  • Shared payload store: either a tmpfs mount at /dev/shm (single-node) or a MinIO/S3 bucket (multi-node).
  • Upstream stage: an ingestion driver that already emits hardware-level timestamps in nanoseconds and writes raw sensor_msgs/PointCloud2 and image frames to the payload store. Timestamp discipline itself is covered in deterministic LiDAR and camera timestamp alignment in ROS; this page assumes those stamps are trustworthy.
  • Downstream consumer: the registration stage from production-grade ICP registration in Python, invoked inside coordinate_transform and spatial_fusion.

Step-by-step #

1. Configure the worker pool for native numerical workloads #

Spatial libraries (Open3D, PDAL, PCL bindings) release the GIL and rely on C-level threading, which conflicts with the cooperative gevent/eventlet runtimes. Force prefork for OS-level process isolation, and recycle each worker after a single heavy task so fragmented heaps from Eigen/NumPy buffers are returned to the OS.

python
# celeryconfig.py
broker_url = "amqp://av:av@rabbitmq:5672//"
result_backend = "redis://redis:6379/0"

# Native numerical workloads — prefork only, no cooperative concurrency.
worker_pool = "prefork"
worker_concurrency = 4

# Recycle after every heavy registration job to defeat heap fragmentation.
worker_max_tasks_per_child = 1
worker_prefetch_multiplier = 1   # one in-flight task per worker; no greedy prefetch

worker_prefetch_multiplier = 1 prevents a worker from reserving a queue of multi-gigabyte tasks it cannot hold in RAM simultaneously. Expected effect: each worker handles exactly one fusion job, then exits and is respawned with a clean address space.

2. Replace pickle with msgpack and move payloads out-of-band #

pickle is slow and an arbitrary-code-execution surface; raw point clouds must never traverse the broker at all. Serialize only metadata and URIs with msgpack, and write binary payloads to the shared store.

python
# celeryconfig.py (continued)
task_serializer = "msgpack"
result_serializer = "msgpack"
accept_content = ["msgpack", "json"]

# Cap broker frame size so an accidental large payload fails fast, loudly.
broker_transport_options = {"max_message_bytes": 262_144}  # 256 KiB
python
# payload_io.py — write binary, pass an immutable URI through the chain
import uuid, numpy as np, open3d as o3d

PAYLOAD_ROOT = "/dev/shm/fusion"

def stash_cloud(cloud: o3d.geometry.PointCloud) -> str:
    """Persist a cloud to shared memory, return a URI for task kwargs."""
    uri = f"{PAYLOAD_ROOT}/{uuid.uuid4().hex}.npy"
    np.save(uri, np.asarray(cloud.points, dtype=np.float32))
    return uri

def load_cloud(uri: str) -> o3d.geometry.PointCloud:
    pts = np.load(uri, mmap_mode="r")          # memory-mapped, zero-copy read
    pc = o3d.geometry.PointCloud()
    pc.points = o3d.utility.Vector3dVector(np.asarray(pts, dtype=np.float64))
    return pc

Switching coordinate and calibration matrices to msgpack cuts serialization latency by roughly 40–60% versus pickle; the 256 KiB broker frame cap guarantees that any payload mistakenly passed inline raises a kombu frame error instead of silently bloating the broker.

3. Build the strictly ordered fusion chain with bound retries #

Temporal synchronization must precede spatial alignment. Define idempotent tasks bound with retries and exponential backoff, then wire them into a chain so order is enforced regardless of arrival jitter. Pass hardware timestamps and payload URIs as immutable kwargs — never the binary data.

python
# tasks.py
from celery import Celery
from celery.utils.log import get_task_logger
from payload_io import load_cloud, stash_cloud

app = Celery("fusion")
app.config_from_object("celeryconfig")
log = get_task_logger(__name__)

@app.task(bind=True, max_retries=3, retry_backoff=True, acks_late=True)
def temporal_sync(self, *, cloud_uri, image_uri, t_lidar_ns, t_cam_ns):
    drift_ns = abs(t_lidar_ns - t_cam_ns)
    if drift_ns > 8_000_000:                    # 8 ms hard drift bound
        raise self.retry(exc=ValueError(f"drift {drift_ns} ns exceeds 8 ms"))
    return {"cloud_uri": cloud_uri, "image_uri": image_uri,
            "t_ref_ns": min(t_lidar_ns, t_cam_ns)}

@app.task(bind=True, max_retries=3, retry_backoff=True, acks_late=True)
def coordinate_transform(self, synced, *, extrinsics_uri):
    cloud = load_cloud(synced["cloud_uri"])     # mmap fetch by URI
    # apply ISO 8855 vehicle-frame extrinsics here...
    synced["cloud_uri"] = stash_cloud(cloud)
    return synced

@app.task(bind=True, max_retries=3, retry_backoff=True, acks_late=True)
def spatial_fusion(self, synced):
    cloud = load_cloud(synced["cloud_uri"])     # ICP / fusion runs here
    tile_uri = stash_cloud(cloud)
    return {"tile_uri": tile_uri, "t_ref_ns": synced["t_ref_ns"]}
python
# dispatch.py — order is guaranteed by the chain, not by arrival time
from celery import chain
from tasks import temporal_sync, coordinate_transform, spatial_fusion

def submit_fusion(cloud_uri, image_uri, extrinsics_uri, t_lidar_ns, t_cam_ns):
    workflow = chain(
        temporal_sync.s(cloud_uri=cloud_uri, image_uri=image_uri,
                        t_lidar_ns=t_lidar_ns, t_cam_ns=t_cam_ns),
        coordinate_transform.s(extrinsics_uri=extrinsics_uri),
        spatial_fusion.s(),
    )
    return workflow.apply_async()               # returns AsyncResult, non-blocking

acks_late=True keeps a task on the queue until it actually completes, so a worker killed mid-registration redelivers rather than losing the frame. apply_async() returns immediately — the dispatcher never blocks the ingestion loop.

4. Emit spatial telemetry into structured logs #

Embed spatial bounds and frame identity into every log line so a stalled tile can be traced to its sensor and timestamp. Inject custom fields into Celery's task log format and route worker metrics to Prometheus.

python
# celeryconfig.py (continued)
task_log_format = (
    "%(asctime)s %(levelname)s %(task_id)s %(task_name)s "
    "queue=%(processName)s host=%(hostname)s :: %(message)s"
)
worker_send_task_events = True   # enable celery-exporter / Flower metrics
task_send_sent_event = True
python
# inside spatial_fusion, before returning:
log.info("fused tile bbox_wgs84=%s frame_ts_ns=%d sensor_id=%s",
         bbox_wgs84, synced["t_ref_ns"], sensor_id)

Scrape task_latency, queue_depth, and worker_memory_rss from celery-exporter to catch broker backpressure and memory pressure before they cascade.

Verification & acceptance criteria #

Run a soak test that submits 500 fusion chains and confirm each invariant:

python
# verify.py
import psutil, time
from dispatch import submit_fusion

results = [submit_fusion(*frame) for frame in load_test_frames(500)]
for r in results:
    out = r.get(timeout=30)                     # acceptable in the TEST harness only
    assert out["tile_uri"].endswith(".npy")
    assert r.state == "SUCCESS"

# Memory must return to baseline — no monotonic RSS growth across the run.
rss_mb = psutil.Process().memory_info().rss / 1e6
assert rss_mb < 1500, f"worker leak suspected: {rss_mb:.0f} MB"

Acceptance gates:

  • Broker isolation: rabbitmqctl list_queues name message_bytes reports per-message bytes ≤ 256 KiB — no point-cloud binaries on the broker.
  • Memory recycling: worker memory_rss returns to baseline between tasks; the 500-frame run shows no monotonic upward RSS trend.
  • Temporal coherence: every accepted chain has drift_ns ≤ 8_000_000 (8 ms); frames exceeding it appear in the retry/dead-letter path, not in output.
  • Throughput: dispatch (apply_async) latency stays ≤ 5 ms, confirming the ingestion loop never blocks on Celery.

Common errors & fixes #

kombu.exceptions.OperationalError: Message exceeds maximum size — a raw cloud was passed as a kwarg instead of a URI. Confirm the producer calls stash_cloud() and that task kwargs contain only strings and scalars. The 256 KiB max_message_bytes cap is doing its job; do not raise it to "fix" this.

Silent RSS growth across tasks (OOM-killed workers after hours) — worker_max_tasks_per_child is unset or gevent/eventlet is active, so fragmented Eigen/NumPy heaps are never reclaimed. Set worker_max_tasks_per_child = 1 and worker_pool = "prefork", then re-run the soak test.

Can't decode content-type 'application/x-python-serialize' — a producer is still emitting pickle while consumers accept only msgpack. Pin task_serializer, result_serializer, and accept_content identically across every process and redeploy them together.

Workers stall and queue_depth climbs without throughput — a blocking AsyncResult.get() is being called inside a task, deadlocking the prefork worker. Replace in-task .get() with a chain/chord so dependencies resolve through the broker; reserve .get() for the dispatcher or test harness only.

Up one level: Asynchronous Data Pipeline Architecture for HD mapping and spatial processing.