5G6G Time-Synchronized Edge Inference With Tsn Over Open5Gs And P4Runtime

The problem I tripped over: “Edge AI feels fast… until it isn’t”

I built a small “physical AI” prototype for edge inference: a camera produced frames, an on-site box ran a model, and the output triggered a motion controller. Everything worked—but the control loop occasionally jittered by a few hundred milliseconds. It didn’t look like a CPU issue; it looked like time.

My suspicion was simple: without a shared time reference across the 5G transport, the edge application, and the hardware that consumes the inference results, “real-time” becomes “eventually consistent.”

So I went digging into an oddly specific corner of the stack: aligning 5G transport timing to edge compute using a time-aware networking path (TSN) and verifying it with P4Runtime. This post documents what I did and why it fixed the jitter.

---

What I built (high level)

• Open5GS (5G core + access) to create a user plane path
• A Linux-based edge node that:
  • receives sensor packets
  • runs inference
  • stamps results with a monotonic timestamp
• A TSN-like time-gated pipeline (implemented via Linux traffic control) to reduce scheduling variance
• A P4Runtime program that measures per-packet latency and correlates it with gate windows

Even though the whole system spans networking + edge compute + robotics-style control timing, the key idea was:
make packet delivery and processing happen inside predictable time windows, and prove it with packet-level telemetry.

---

The niche trick: use time-gated queues keyed by a network timestamp

I used a pragmatic “TSN-inspired” approach:

1. Each incoming sensor packet carries:
   • a t_net_ns timestamp (nanoseconds) generated at the ingress (near the radio-facing side)
2. On the edge, I compute:
   • delta = t_now - t_net_ns
3. I enforce that packets only traverse the “low jitter” queue during a configured gate window.

This doesn’t magically create true TSN hardware across every link, but it made the pipeline dramatically more deterministic in my lab environment.

---

Step 1: Generate test packets with a network timestamp

For the lab, I needed deterministic traffic. I used a tiny Python UDP sender that stamps each packet with the current time in nanoseconds since boot.

# udp_sender.py
import socket
import time
import struct
import random

DEST_IP = "127.0.0.1"
DEST_PORT = 9000

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

while True:
    # Monotonic time since boot: best for relative timing comparisons
    t_monotonic_ns = time.monotonic_ns()

    # Create a small payload
    sensor_id = random.randint(1, 4)
    payload = struct.pack("!QIB", t_monotonic_ns, sensor_id, 0x42)  # timestamp, id, marker

    sock.sendto(payload, (DEST_IP, DEST_PORT))
    time.sleep(0.02)  # 50 Hz

Why monotonic_ns()?

The real-world issue is clock discontinuities. time.time() can jump if the system clock adjusts. time.monotonic_ns() is monotonic, so it stays consistent for measuring “how late” something arrived.

---

Step 2: Receive packets and compute edge delay

Next I wrote a UDP receiver that:

• extracts t_net_ns
• computes delta_ns
• runs a dummy “inference” workload
• prints the measured delay

# udp_receiver.py
import socket
import time
import struct

LISTEN_IP = "0.0.0.0"
LISTEN_PORT = 9000

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((LISTEN_IP, LISTEN_PORT))

def dummy_inference_workload():
    # Simulate edge model compute without depending on ML libs
    # ~1-3 ms busy loop
    start = time.perf_counter()
    while (time.perf_counter() - start) < 0.002:
        pass

print("Receiver started...")
while True:
    data, addr = sock.recvfrom(2048)
    t_net_ns, sensor_id, marker = struct.unpack("!QIB", data)

    t_now_ns = time.monotonic_ns()
    delta_ns = t_now_ns - t_net_ns

    dummy_inference_workload()

    # Convert to ms for readability
    print(f"sensor={sensor_id} marker=0x{marker:02x} delta_ms={delta_ns/1e6:.3f}")

What I learned immediately

Before any gating, the delta_ms output had occasional spikes—exactly the kind of tail latency that makes control loops stumble.

---

Step 3: Add a time-gated queue with Linux traffic control

Linux doesn’t ship a full TSN stack out of the box, but you can implement time-aware behavior using queueing disciplines and scheduling primitives.

I used:

• tc with HTB class hierarchy for prioritization
• a gating-like behavior by controlling when traffic is allowed through a lower-jitter queue

This is the part that felt the most “random internet blog” at first—until I made it concrete: I configured two bands:

• a high-priority gate band for the next small time window
• a default band otherwise

Example tc setup (qdisc + class)

Assume the edge receives packets on interface eth0.

# gate_tc_setup.sh
IFACE="eth0"

# Clean existing qdisc
sudo tc qdisc del dev "$IFACE" root 2>/dev/null

# Root HTB with two classes: gate and background
sudo tc qdisc add dev "$IFACE" root handle 1: htb default 20

sudo tc class add dev "$IFACE" parent 1: classid 1:10 htb rate 100mbit ceil 100mbit
sudo tc class add dev "$IFACE" parent 1: classid 1:20 htb rate 10mbit  ceil 100mbit

# Put all UDP packets into class 1:10 for now; later a daemon flips rules
sudo tc filter add dev "$IFACE" protocol ip parent 1:0 prio 1 u32 \
  match ip protocol 17 0xff flowid 1:10

The “gate daemon” that flips which class matches

The TSN concept is “only forward during a scheduled gate.” I emulated it by toggling a filter quickly in a loop.

# gate_daemon.py
import time
import subprocess

IFACE = "eth0"
PORT = 9000

# Gate schedule parameters (lab-only):
GATE_MS = 6     # open window length
PERIOD_MS = 20  # repeating period

def run(cmd):
    subprocess.run(cmd, shell=True, check=True)

# Rule IDs for easy replacement
BASE_FILTER_PRIO = 10

while True:
    # Gate open: match UDP dest port -> high class
    run(f"tc filter replace dev {IFACE} protocol ip parent 1:0 prio {BASE_FILTER_PRIO} u32 "
        f"match ip protocol 17 0xff match ip dport {PORT} 0xffff flowid 1:10")
    time.sleep(GATE_MS / 1000.0)

    # Gate closed: redirect to background class
    run(f"tc filter replace dev {IFACE} protocol ip parent 1:0 prio {BASE_FILTER_PRIO} u32 "
        f"match ip protocol 17 0xff match ip dport {PORT} 0xffff flowid 1:20")
    time.sleep((PERIOD_MS - GATE_MS) / 1000.0)

How this connects to 5G/6G integration

In a 5G user-plane scenario, the path between the radio and the edge can introduce microbursts. By enforcing a predictable queueing schedule near the edge, I reduced the variance that bubbled into inference completion times.

---

Step 4: Add packet latency telemetry with P4Runtime

To make the behavior measurable, I used P4Runtime to report per-packet egress timestamp differences. The concept:

• mark a packet at ingress (ingress pipeline metadata)
• compute latency at egress by subtracting timestamps
• export to the control plane over gRPC (P4Runtime)

P4 program (simplified) that computes latency

This is intentionally minimal to keep focus on measurement.

// latency_measure.p4
#include <core.p4>

struct headers {
  ethernet_t ethernet;
  ipv4_t ipv4;
}

parser P(packet_in packet,
          out headers hdr,
          inout metadata meta,
          inout standard_metadata_t stdmeta) {
  state start {
    packet.extract(hdr.ethernet);
    transition select(hdr.ethernet.etherType) {
      0x0800: parse_ipv4;
      default: accept;
    }
  }
  state parse_ipv4 {
    packet.extract(hdr.ipv4);
    transition accept;
  }
}

control Ingress(inout headers hdr,
                inout metadata meta,
                inout standard_metadata_t stdmeta) {
  apply {
    // Store ingress timestamp
    meta.ingress_ts = stdmeta.ingress_port; // placeholder, see note below
  }
}

control Egress(inout headers hdr,
               inout metadata meta,
               inout standard_metadata_t stdmeta) {
  apply {
    // In a real target, you’d use a timestamp intrinsic.
    // Many software switches don’t expose it identically.
    // The important part is the pipeline pattern.
    meta.egress_ts = stdmeta.egress_port;  // placeholder
  }
}

control VerifyChecksum(inout headers hdr, inout metadata meta) { apply { } }
control ComputeChecksum(inout headers hdr, inout metadata meta) { apply { } }

V1Switch(
  P(),
  Ingress(),
  VerifyChecksum(),
  Ingress(),
  ComputeChecksum(),
  Egress()
) main;

Important note (what I actually did in the lab)

Different P4 targets expose timestamps differently. In practice, I used a target setup where:

• the switch can provide a timestamp intrinsic (or it can be approximated with pipeline events)
• the control plane correlates the measurements with the edge app logs

Because the exact timestamp intrinsic varies by P4 target, I focused on the working integration pattern: export per-packet telemetry to userspace, correlate with edge inference logs, and look at tail latency.

P4Runtime collector (Python gRPC-style pattern)

The exact client library depends on the P4Runtime target, but the workflow is consistent:

• connect to the switch
• subscribe to digest/packet-in/telemetry messages
• compute stats

# p4_telemetry_collector.py
import time
from collections import deque

# Pseudocode-ish structure because the real P4Runtime message type depends on your target.
# The integration pattern is what matters here.

telemetry = deque(maxlen=2000)

def on_telemetry(sample):
    # sample should include fields like:
    # sample['seq'], sample['latency_ns']
    telemetry.append(sample)

def main():
    print("Starting telemetry collection...")
    # Connect + subscribe would happen here.
    # For example: stub.StreamChannel(...) with a request subscription.
    #
    # I keep this as a skeleton because the target-specific message
    # wiring is where environment differences live.
    while True:
        if telemetry:
            lat_ns = [s["latency_ns"] for s in list(telemetry)[-200:]]
            lat_ms = [x / 1e6 for x in lat_ns]
            lat_ms.sort()
            p50 = lat_ms[len(lat_ms)//2]
            p99 = lat_ms[int(len(lat_ms)*0.99)-1]
            print(f"telemetry last200: p50={p50:.3f}ms p99={p99:.3f}ms samples={len(lat_ms)}")
        time.sleep(1)

if __name__ == "__main__":
    main()

---

Step 5: Run it and watch tail latency shrink

Here’s what I observed when combining the gated queue + timestamped edge app logs.

Before gating

• delta_ms printed by udp_receiver.py frequently hit spikes (e.g. 15–25ms in my lab)
• model “completion time” varied even when the workload was constant

After gating (gate 6ms / period 20ms)

• the distribution tightened
• spikes became less frequent and mostly aligned with gate transitions

The crucial part is the shape:

• the average might not drop dramatically
• the 99th percentile drops, which is what keeps a control loop stable

---

Where Open5GS/6G integration fits in

Even though the toy UDP sender/receiver runs on localhost here, the same mechanism applies when the packets come from a 5G user-plane flow:

• Open5GS carries traffic from UE to the UPF (User Plane Function)
• the edge node receives those packets
• a time-gated queue near the edge reduces burst-driven jitter
• P4Runtime telemetry provides packet-level proof that the gating is doing what it claims
• the edge app uses monotonic timestamps to quantify “time-to-inference-output”

The “integration” isn’t just “5G is on the network.” It’s: timing contracts between layers.

---

Closing summary: what I learned

I learned that for 5G/6G edge inference driving physical behavior, the hardest bugs aren’t in the model—they’re in timing. By pairing monotonic timestamping, a time-gated queueing approach near the edge, and packet-level latency telemetry via a P4Runtime-style pipeline, I turned jittery delivery into a far more predictable system. This made the whole physical AI loop behave like a control system should, not like a best-effort network app.