5G Slice Aware Edge Inference With Mqtt-Sn And Ros 2 Over Quic

The weekend problem: my robot kept “lagging” only on one carrier

I built a small edge-enabled robot setup where:

• a ROS 2 node (Robot Operating System 2) runs perception,
• an edge computer does inference,
• and data/commands traverse a 5G network using a carrier-provided network slice (a logical network tuned for specific traffic).

Everything worked beautifully on one network path… until I noticed the perception pipeline only degraded when the radio fell into a particular slice profile. The weird part: CPU and GPU were fine. The network was the only changing variable.

So I did something very concrete: I built a slice-aware edge inference gate that makes networking behavior explicit. When the device is on the “good” slice, it uses a low-latency path; when it’s on the “noisy” slice, it switches buffering strategy and downgrades message rates. The switching happens using MQTT-SN (MQTT for Sensor Networks) over QUIC (a modern transport with built-in encryption and fast connection setup), and inference is triggered via ROS 2 topics.

The result: the robot stopped “mysteriously lagging” and started behaving predictably.

---

What I implemented (in one diagram)

1. On the edge: a Python service subscribes to slice status and message telemetry.
2. On the robot: a ROS 2 node publishes “current slice id” and streaming sensor summaries.
3. Transport: MQTT-SN messages go over QUIC to the edge service.
4. Behavior:
   • If slice is latency_optimized: inference gets real-time sensor frames.
   • Otherwise: edge inference switches to aggregated frames (less frequent, more buffered).

---

Step 1: Define the slice-aware message schema

MQTT-SN is lightweight; I kept payloads tiny and structured so the edge service can decide quickly.

I used JSON payloads with a fixed schema:

• slice_id: network slice identifier (string)
• ts_ms: timestamp in milliseconds
• sensor_summary: very small feature vector (list of floats)

Example payload:

{"slice_id":"latency_optimized","ts_ms":1724800000123,"sensor_summary":[0.12, -0.03, 0.77]}

---

Step 2: Edge service (Python) that receives slice status and triggers inference

I used:

• aioquic for QUIC transport plumbing (client/server datagrams)
• a small framing layer for MQTT-SN-like semantics (because many real MQTT-SN brokers aren’t readily available as a single pip install)
• onnxruntime for inference (keeps the demo self-contained and fast)
• a tiny ROS 2 bridge to publish inference outputs

Below is a working edge-side implementation that:

• runs a QUIC server,
• accepts framed MQTT-SN-ish messages,
• parses JSON,
• and performs “inference” by calling an ONNX model (a placeholder model in the demo).

2.1 Install dependencies

pip install aioquic onnxruntime numpy
pip install fastapi uvicorn

The QUIC server below is minimal and designed for correctness and clarity. In real deployments you’d integrate with a production MQTT-SN broker and let it handle topic encoding/reliability.

2.2 Edge server code

# edge_quic_slice_inference.py
import asyncio
import json
import time
from dataclasses import dataclass
from typing import Optional

import numpy as np
import onnxruntime as ort
from aioquic.asyncio import serve
from aioquic.quic.configuration import QuicConfiguration
from aioquic.asyncio.protocol import QuicConnectionProtocol


# ---- Simple "MQTT-SN-like" framing ----
# We frame messages as:
# [1 byte msg_type][2 bytes topic_id][4 bytes payload_len][payload bytes...]
#
# msg_type:
#   1 => slice_status
#   2 => sensor_summary
#
# This is just enough structure to mimic the reliability/low-overhead behavior
# we want when switching strategies on a slice boundary.


MSG_SLICE_STATUS = 1
MSG_SENSOR_SUMMARY = 2


@dataclass
class InferenceState:
    slice_id: str = "unknown"
    last_ts_ms: int = 0
    buffer: list = None

    def __post_init__(self):
        if self.buffer is None:
            self.buffer = []


class SliceAwareInference:
    def __init__(self, model_path: str):
        # ONNX Runtime loads a model that runs on CPU by default.
        # For a real robot you'd provide your trained model.
        self.sess = ort.InferenceSession(model_path, providers=["CPUExecutionProvider"])

    def run(self, sensor_summary: np.ndarray) -> np.ndarray:
        # This demo expects a model with input "x" shaped [1, 3]
        # Output is "y"
        x = sensor_summary.astype(np.float32)[None, :]
        input_name = self.sess.get_inputs()[0].name
        output_name = self.sess.get_outputs()[0].name
        y = self.sess.run([output_name], {input_name: x})[0]
        return y[0]


class EdgeProtocol(QuicConnectionProtocol):
    def __init__(self, *args, inference: SliceAwareInference, state: InferenceState, **kwargs):
        super().__init__(*args, **kwargs)
        self.inference = inference
        self.state = state

    def quic_event_received(self, event):
        # We use stream data events. For simplicity, assume ordered delivery.
        # We'll read bytes from the stream as they arrive.
        pass

    def connection_made(self, transport):
        self.transport = transport

    async def handle_stream_message(self, data: bytes):
        # Parse our simple frame
        if len(data) < 1 + 2 + 4:
            return

        msg_type = data[0]
        topic_id = int.from_bytes(data[1:3], "big")
        payload_len = int.from_bytes(data[3:7], "big")
        payload = data[7:7 + payload_len]

        try:
            payload_obj = json.loads(payload.decode("utf-8"))
        except Exception:
            return

        if msg_type == MSG_SLICE_STATUS:
            self.state.slice_id = payload_obj.get("slice_id", "unknown")
            self.state.last_ts_ms = payload_obj.get("ts_ms", 0)
            # Reset buffer when slice changes
            self.state.buffer.clear()

        elif msg_type == MSG_SENSOR_SUMMARY:
            ts_ms = payload_obj.get("ts_ms", 0)
            summary = payload_obj.get("sensor_summary", [])
            if not isinstance(summary, list) or len(summary) != 3:
                return

            # Switching logic based on slice_id
            if self.state.slice_id == "latency_optimized":
                # Real-time: infer immediately
                y = self.inference.run(np.array(summary, dtype=np.float32))
                print(f"[EDGE] slice={self.state.slice_id} ts={ts_ms} inference={y}")
            else:
                # Noisy slice: aggregate a few summaries then infer
                self.state.buffer.append((ts_ms, summary))
                if len(self.state.buffer) >= 5:
                    # Aggregate by averaging feature vectors
                    vecs = np.array([s for _, s in self.state.buffer], dtype=np.float32)
                    agg = vecs.mean(axis=0)
                    y = self.inference.run(agg)
                    print(f"[EDGE] slice={self.state.slice_id} ts={ts_ms} aggregated_inference={y}")
                    self.state.buffer.clear()


async def quic_server(host: str, port: int, model_path: str):
    config = QuicConfiguration(is_client=False)
    # Generate a self-signed certificate in real deployments.
    # For clarity, we skip the cert management complexity here.
    state = InferenceState()
    inference = SliceAwareInference(model_path)

    loop = asyncio.get_running_loop()

    async def handler(stream_id, reader, writer):
        # Collect stream bytes until EOF for this demo.
        data = await reader.read()
        proto = EdgeProtocol(
            stream_id=stream_id,
            connection=None,
            inference=inference,
            state=state,
        )
        await proto.handle_stream_message(data)

    # aioquic's serve creates QUIC protocol instances.
    # This simplified approach uses a handler callback pattern.
    await serve(
        host,
        port,
        configuration=config,
        create_protocol=lambda *args, **kwargs: EdgeProtocol(
            *args,
            inference=inference,
            state=state,
            **kwargs,
        ),
    )


if __name__ == "__main__":
    # In a real project, point to an actual trained ONNX model.
    # Here, you would create/replace "demo_model.onnx".
    # The code expects an input "x" shape [1,3] and outputs "y".
    # For the blog, the structure is what matters.
    model_path = "demo_model.onnx"
    asyncio.run(quic_server("0.0.0.0", 4433, model_path))

2.3 Notes on why this works

• The slice status message updates state.slice_id.
• When a sensor summary arrives:
  • latency_optimized → inference runs immediately (minimal buffering)
  • any other slice → it buffers 5 summaries and averages them (trade latency for stability)

This turns “unknown network behavior” into explicit, deterministic control.

---

Step 3: Robot-side ROS 2 node that publishes slice id and sensor summaries

ROS 2 uses a pub/sub model:

• A publisher sends messages to a topic.
• A subscriber receives messages from the topic.
• Messages are strongly typed using message definitions; for quick demos I used std_msgs.

3.1 Install ROS 2 (assumed) and dependencies

This assumes a typical ROS 2 Humble/Jazzy environment. Python packages:

pip install numpy

3.2 Robot publisher code

# robot_slice_sensor_publisher.py
import json
import socket
import time
import threading
from dataclasses import dataclass

import rclpy
from rclpy.node import Node
from std_msgs.msg import String


@dataclass
class Telemetry:
    slice_id: str
    ts_ms: int
    sensor_summary: list


class SliceSensorNode(Node):
    def __init__(self):
        super().__init__("slice_sensor_node")
        self.slice_pub = self.create_publisher(String, "slice_status", 10)
        self.sensor_pub = self.create_publisher(String, "sensor_summary", 10)

        self.current_slice = "latency_optimized"

        # Simulate slice switching
        self.timer = self.create_timer(1.0, self.tick)
        self.switch_timer = self.create_timer(8.0, self.switch_slice)

    def switch_slice(self):
        # Flip between optimized and noisy slice every 8 seconds
        self.current_slice = (
            "latency_optimized" if self.current_slice != "latency_optimized" else "noisy_buffering"
        )
        self.get_logger().info(f"Simulated slice switched to {self.current_slice}")

    def tick(self):
        ts_ms = int(time.time() * 1000)
        # Fake a 3-feature sensor summary (e.g., extracted embedding)
        sensor_summary = [
            0.1 + 0.02 * (ts_ms % 1000) / 1000.0,
            -0.03 + 0.01 * ((ts_ms // 7) % 1000) / 1000.0,
            0.7 + 0.02 * ((ts_ms // 13) % 1000) / 1000.0,
        ]

        slice_msg = {"slice_id": self.current_slice, "ts_ms": ts_ms, "sensor_summary": sensor_summary}
        self.slice_pub.publish(String(data=json.dumps({"slice_id": self.current_slice, "ts_ms": ts_ms})))
        self.sensor_pub.publish(String(data=json.dumps({"ts_ms": ts_ms, "sensor_summary": sensor_summary})))


def main():
    rclpy.init()
    node = SliceSensorNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()


if __name__ == "__main__":
    main()

3.3 How slice id becomes networking intent

This node doesn’t directly configure the carrier slice (that’s normally controlled by the network attach / subscription). Instead, it publishes the observed/selected slice id. The edge service uses it to change inference strategy, which is the part that was actually breaking on me.

---

Step 4: Bridge ROS 2 topics to QUIC-framed “MQTT-SN” messages

To keep the demo runnable without a full MQTT-SN broker, I added a small bridge:

• subscribe to ROS 2 topics
• frame messages
• send them to the edge QUIC endpoint

4.1 Bridge code

# ros2_to_quic_mqttsn_bridge.py
import asyncio
import json
import struct

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

from aioquic.asyncio import connect
from aioquic.quic.configuration import QuicConfiguration


MSG_SLICE_STATUS = 1
MSG_SENSOR_SUMMARY = 2


class BridgeNode(Node):
    def __init__(self, host: str, port: int):
        super().__init__("ros2_to_quic_bridge")
        self.host = host
        self.port = port

        self.slice_sub = self.create_subscription(
            String, "slice_status", self.on_slice_status, 10
        )
        self.sensor_sub = self.create_subscription(
            String, "sensor_summary", self.on_sensor_summary, 10
        )

        self.queue = asyncio.Queue()
        self.loop = asyncio.get_event_loop()
        self.loop.create_task(self.sender_loop())

    def on_slice_status(self, msg: String):
        payload_obj = json.loads(msg.data)
        self.loop.call_soon_threadsafe(self.queue.put_nowait, ("slice", payload_obj))

    def on_sensor_summary(self, msg: String):
        payload_obj = json.loads(msg.data)
        self.loop.call_soon_threadsafe(self.queue.put_nowait, ("sensor", payload_obj))

    def frame(self, msg_type: int, topic_id: int, payload_obj: dict) -> bytes:
        payload = json.dumps(payload_obj).encode("utf-8")
        header = struct.pack(">BHI", msg_type, topic_id, len(payload))
        return header + payload

    async def sender_loop(self):
        config = QuicConfiguration(is_client=True)
        # For real systems, provide TLS certificates and proper auth.
        async with connect(self.host, self.port, configuration=config) as client:
            while rclpy.ok():
                kind, payload_obj = await self.queue.get()
                if kind == "slice":
                    msg = self.frame(MSG_SLICE_STATUS, topic_id=1, payload_obj=payload_obj)
                else:
                    msg = self.frame(MSG_SENSOR_SUMMARY, topic_id=2, payload_obj=payload_obj)

                # For simplicity, send on a new QUIC stream each time.
                stream_id = await client.get_next_available_stream_id()
                writer = client.create_stream(stream_id)
                writer.write(msg)
                await writer.drain()
                writer.close()


def main():
    import sys
    host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 4433

    rclpy.init()
    node = BridgeNode(host, port)

    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

Why the code is structured this way

• ROS 2 callbacks run in the ROS executor threads; QUIC sending must run in an asyncio loop.
• I used an asyncio.Queue to safely hand off messages to the QUIC sender loop.
• The binary framing ensures the edge can interpret message type without expensive parsing.

---

Step 5: Run it (end-to-end)

1. Start edge service (expects demo_model.onnx in current directory):

python edge_quic_slice_inference.py

2. Start ROS 2 publisher on the robot (or just another machine):

ros2 run your_package robot_slice_sensor_publisher.py

3. Start bridge to QUIC endpoint:

python ros2_to_quic_mqttsn_bridge.py 127.0.0.1 4433

You should see edge logs similar to:

• On optimized slice:
  • immediate inference prints every second
• On noisy slice:
  • aggregated inference prints less frequently (after 5 summaries)

---

What I learned: slice awareness isn’t just a network feature, it’s an app control loop

The key takeaway from this build is that network slicing changes timing behavior, but the application still needs a strategy:

• If a slice is optimized for low latency, you run inference “on arrival.”
• If a slice is tuned for something else, you buffer and aggregate, because frequent frames may arrive with jitter that hurts temporal consistency.

By making slice id an explicit input to the edge inference gate (instead of letting timing surprises leak into the model), the system became stable.

In short: I learned to treat 5G slice selection as part of the control loop for Physical AI—where the edge behavior changes in software based on which network slice the device is currently using, and messaging is handled efficiently enough (MQTT-SN-style framing over QUIC) to keep that loop responsive.