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.

Why the payload leaves the broker. A point cloud pushed through the message body is copied at least four times before a worker touches it; a reference is copied once and the cloud is read straight from object storage:

In-Band against Out-of-Band Payload Transport Two horizontal flows from producer through broker to worker, the first carrying the full payload at every hop, the second carrying only a small reference while the payload travels once to object storage. payload in the message body producer broker worker full cloud full cloud serialize · store · forward · deserialize — four copies, and broker memory scales with backlog payload out of band, reference in the message producer broker worker ≤256 KiB ref ≤256 KiB ref object store write once read once

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.

Retries have to be bounded and the chain strictly ordered for the same reason: a fusion step that runs twice, or runs before its predecessor, produces a pose that is wrong in a way no downstream check can distinguish from sensor noise:

A Retried Task Overtaking Its Own First Attempt Two timelines of the same three-task chain: a correct sequential run, and a run where a retried middle task overlaps the original, delivering two results to the final task. correct: one attempt per link, strictly ordered undistort register fuse unbounded retry with early acknowledgement undistort register — attempt 1 slow, not dead register — retry different worker fuse — receives both bound max_retries, acknowledge late, and key results by frame id so a duplicate is detected rather than merged

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.

Why the ordering constraint is not negotiable #

One design point is worth restating because a task queue makes it easy to lose. The fusion chain is strictly ordered — undistort, then register, then fuse — and the ordering is a data dependency rather than a preference: a register step that runs on an uncompensated cloud produces a transform that is wrong by the ego motion across the sweep, and the fuse step has no way to detect that.

Celery's chain primitive expresses that ordering, and two settings decide whether it holds under failure. Acknowledging late means a worker that dies mid-task returns the message to the queue rather than losing it. Bounding max_retries means a slow-but-alive task cannot be retried indefinitely alongside itself. Together they turn a chain from a hopeful sequence into an ordering that survives a worker restart, which is the condition under which the pipeline actually runs.

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