pof: fix conditional delay buffer stalling under sustained reordered traffic

.gitignore

@@ -11,3 +11,10 @@ compile_commands.json
 
 #auto generated
 doc/protocols.md
+dnt_testbed/result_*.txt
+dnt_testbed/sweep_results.txt
+dnt_testbed/emulation_points.csv
+dnt_testbed/dnt_nxp*.log
+dnt_testbed/raw/
+# results_archive/ is intentionally NOT ignored: it preserves finished
+# measurement campaigns (commit it to keep them)

dnt_testbed/aggregate_results.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Aggregate sweep_results.txt into per-(D,H) means with 95% CIs.
+
+Usage:  python3 aggregate_results.py [sweep_results.txt] [-o emulation_points.csv]
+
+Output CSV columns (AoI in model units, 1u = 1 ms):
+  D,H,reps,comp_mean,comp_ci,avgAoI_mean,avgAoI_ci,peakAoI_mean,peakAoI_ci,ooo_mean
+
+Overlaying on the Letter's figure (matplotlib):
+
+  import pandas as pd
+  em = pd.read_csv("emulation_points.csv")
+  ax.errorbar(em.D, em.comp_mean, yerr=em.comp_ci, fmt="o", mfc="none",
+              ms=5, capsize=2, color="k", label="DNT emulation", zorder=5)
+  # same pattern with em.avgAoI_mean / em.peakAoI_mean on the AoI axis;
+  # if the x-axis is rho instead of D, map each D through eq. (Drho) inverse.
+"""
+import re, sys, csv, math
+from collections import defaultdict
+
+# two-sided 95% Student-t quantiles, df = reps-1
+T95 = {1: 12.71, 2: 4.303, 3: 3.182, 4: 2.776, 5: 2.571,
+       6: 2.447, 7: 2.365, 8: 2.306, 9: 2.262, 10: 2.228}
+
+def mean_ci(xs):
+    n = len(xs)
+    m = sum(xs) / n
+    if n < 2:
+        return m, float("nan")
+    s = math.sqrt(sum((x - m) ** 2 for x in xs) / (n - 1))
+    t = T95.get(n - 1, 1.96)
+    return m, t * s / math.sqrt(n)
+
+src = sys.argv[1] if len(sys.argv) > 1 and not sys.argv[1].startswith("-") else "sweep_results.txt"
+out = sys.argv[sys.argv.index("-o") + 1] if "-o" in sys.argv else "emulation_points.csv"
+
+pat = re.compile(
+    r"D=(\d+)\s+H=(\d+)\s+rep=\d+\s+delivered=\d+\s+completeness=([\d.]+)\s+"
+    r"avgAoI=([\d.]+)u\s+peakAoI=([\d.]+)u\s+outOfOrder=(\d+)")
+
+pts = defaultdict(lambda: {"comp": [], "avg": [], "peak": [], "ooo": []})
+for line in open(src):
+    m = pat.search(line)
+    if not m:
+        continue
+    k = (int(m.group(1)), int(m.group(2)))
+    pts[k]["comp"].append(float(m.group(3)))
+    pts[k]["avg"].append(float(m.group(4)))
+    pts[k]["peak"].append(float(m.group(5)))
+    pts[k]["ooo"].append(int(m.group(6)))
+
+with open(out, "w", newline="") as f:
+    w = csv.writer(f)
+    w.writerow(["D", "H", "reps", "comp_mean", "comp_ci",
+                "avgAoI_mean", "avgAoI_ci", "peakAoI_mean", "peakAoI_ci", "ooo_mean"])
+    for (D, H) in sorted(pts):
+        p = pts[(D, H)]
+        cm, cc = mean_ci(p["comp"]); am, ac = mean_ci(p["avg"]); pm, pc = mean_ci(p["peak"])
+        w.writerow([D, H, len(p["comp"]), f"{cm:.5f}", f"{cc:.5f}",
+                    f"{am:.4f}", f"{ac:.4f}", f"{pm:.4f}", f"{pc:.4f}",
+                    f"{sum(p['ooo'])/len(p['ooo']):.1f}"])
+print(f"wrote {out}: {len(pts)} operating points")

dnt_testbed/env_frer_aoi.sh

@@ -0,0 +1,60 @@
+#!/bin/bash
+# FRER_AoI Letter scenario on DNT (github.com/EricssonResearch/dnt)
+# Topology mirrors Fig. 1 of the manuscript:
+#
+#   talker ──head── nxp1 (SGF) ══branch A══ nxp2 (SRF+POF) ──tail── listener
+#                              ══branch B══
+#
+# Unit mapping: 1 model time unit = 1 ms  =>  T = 1 ms (1000 pkt/s)
+#   head/tail:  Dh = Dt = 1 ms  -> netem delay 1ms, no loss
+#   branch A :  Dprop=1ms, W~Exp(2/ms),   pA=0.97 -> delay (1+0.5)ms sigma 1ms expo, loss 3%
+#   branch B :  Dprop=3ms, W~Exp(0.4/ms), pB=0.99 -> delay (3+2.5)ms sigma 5ms expo, loss 1%
+# (mean of Exp(beta) = 1/beta = 0.5 ms and 2.5 ms; sigma = 2*mean, see gen_expo_dist.py)
+#
+# Run as root:  source env_frer_aoi.sh ; setup
+set -e
+export PATH=$PATH:/sbin:/usr/sbin
+
+setup() {
+  # exponential netem table -- must land in the dir tc actually searches
+  # (Debian/Ubuntu: /usr/lib/x86_64-linux-gnu/tc; others: /usr/lib/tc).
+  # Detect via a stock table; TC_LIB_DIR env var overrides for both us and tc.
+  tclib="${TC_LIB_DIR:-$(dirname "$(find /usr/lib -name normal.dist 2>/dev/null | head -1)" 2>/dev/null)}"
+  [ -d "$tclib" ] || tclib=/usr/lib/tc
+  python3 "$(dirname "${BASH_SOURCE[0]}")/gen_expo_dist.py" "$tclib/expo.dist"
+
+  for ns in talker listener nxp1 nxp2; do ip netns add $ns 2>/dev/null || true; done
+
+  ip link add eth0 netns talker   type veth peer uni  netns nxp1     # head
+  ip link add brA  netns nxp1     type veth peer brA  netns nxp2     # branch A
+  ip link add brB  netns nxp1     type veth peer brB  netns nxp2     # branch B
+  ip link add uni  netns nxp2     type veth peer eth0 netns listener # tail
+
+  for ns in talker listener nxp1 nxp2; do
+    ip netns exec $ns ip link set lo up
+    for d in $(ip netns exec $ns ls /sys/class/net | grep -v lo); do
+      ip netns exec $ns ip link set $d up
+      ip netns exec $ns ethtool -K $d rx off tx off rxvlan off txvlan off 2>/dev/null || true
+    done
+  done
+
+  # ---- impairments (egress qdiscs on the nxp1 side, head on talker side) ----
+  # head: Dh = 1 ms, lossless
+  netem_warn=0
+  ip netns exec talker tc qdisc replace dev eth0 root netem delay 1ms limit 10000 || netem_warn=1
+  # branch A: L_A = 1ms + Exp(mean 0.5ms), loss 3%
+  ip netns exec nxp1 tc qdisc replace dev brA root netem \
+      delay 1.5ms 1ms distribution expo loss random 3% limit 10000 || netem_warn=1
+  # branch B: L_B = 3ms + Exp(mean 2.5ms), loss 1%
+  ip netns exec nxp1 tc qdisc replace dev brB root netem \
+      delay 5.5ms 5ms distribution expo loss random 1% limit 10000 || netem_warn=1
+  # tail: Dt = 1 ms, lossless
+  ip netns exec nxp2 tc qdisc replace dev uni root netem delay 1ms limit 10000 || netem_warn=1
+
+  [ "$netem_warn" = 1 ] && echo "WARNING: netem unavailable (container?); impairments NOT applied"
+  echo "namespaces up. start DNT:"
+  echo "  ip netns exec nxp1 dnt nxp1.ini &"
+  echo "  ip netns exec nxp2 dnt nxp2.ini &   # edit MaxDelay/HistoryLength per sweep point"
+}
+
+teardown() { for ns in talker listener nxp1 nxp2; do ip netns del $ns 2>/dev/null || true; done; }

dnt_testbed/gen_expo_dist.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+"""Generate a netem distribution table for exponential residual delay.
+
+netem draws delay = mu + sigma * t/8192 with t an int16 from the table.
+We store t = 8192 * (Exp(1) - 1) / 2 and use sigma = 2 * mean, so
+delay = mu - mean + mean*Exp(1). With mu = Dprop + mean the netem line
+
+    delay (Dprop+mean) (2*mean) distribution expo
+
+realizes L = Dprop + W, W ~ Exp(1/mean), exactly eq. (3) of the Letter.
+int16 clipping truncates the Exp tail at ~9*mean (P ~ 1.2e-4).
+
+Table size: kernel netem rejects tables > 16384 entries (MAX_DIST in
+sch_netem.c); stock iproute2 tables use 4096, so we do too. Entries are
+exact Exp(1) quantiles (inverse CDF at midpoints), not random draws.
+"""
+import numpy as np, os, sys
+
+n = 4096
+p = (np.arange(n) + 0.5) / n                       # CDF midpoints
+x = (-np.log1p(-p) - 1.0) / 2.0                    # (Exp(1) quantile - 1)/2
+t = np.clip(np.round(x * 8192), -32768, 32767).astype(int)
+out = sys.argv[1] if len(sys.argv) > 1 else "/usr/lib/tc/expo.dist"
+os.makedirs(os.path.dirname(out), exist_ok=True)
+with open(out, "w") as f:
+    for i in range(0, n, 8):
+        f.write(" ".join(f"{v:6d}" for v in t[i:i+8]) + "\n")
+print(f"wrote {out}: mean={t.mean()/8192:.4f} (target 0), "
+      f"sigma-units, max={t.max()/8192:.2f}")

dnt_testbed/listener.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Receiver: logs (seq, gen_ts, rx_ts) per delivered frame; computes in-order
+completeness and time-average / mean-peak in-order AoI.
+Run: ip netns exec listener python3 listener.py N [raw.npz] > result.txt
+Optional 2nd arg: dump the raw per-frame (seq, gen, rx) arrays to a
+compressed .npz so distributional metrics (PAoI CCDF) can be computed
+offline; timestamps stay in CLOCK_REALTIME seconds."""
+import socket, struct, time, sys
+import numpy as np
+
+N = int(sys.argv[1]) if len(sys.argv) > 1 else 60_000
+RAW = sys.argv[2] if len(sys.argv) > 2 else None
+ETH_P_ALL = 3
+s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
+s.bind(("eth0", 0))
+s.settimeout(10.0)
+
+rec = []
+last = -1
+ooo = 0    # frames arriving out of order / duplicated (POF+SRF should make this 0)
+try:
+    while True:
+        try:
+            f = s.recv(2048)
+        except socket.timeout:
+            break
+        rx = time.clock_gettime(time.CLOCK_REALTIME)
+        # kernel may strip the 802.1Q tag into metadata on RX:
+        # tagged frame: eth | 8100 tci | 88b5 | payload  -> payload at 18
+        # stripped    : eth | 88b5 | payload             -> payload at 14
+        if len(f) < 30:
+            continue
+        if f[12:14] == b"\x81\x00" and f[16:18] == b"\x88\xb5":
+            off = 18
+        elif f[12:14] == b"\x88\xb5":
+            off = 14
+        else:
+            continue
+        if len(f) < off + 16:
+            continue
+        seq, gen = struct.unpack("!Qd", f[off:off+16])
+        if seq <= last:            # POF guarantees order; guard anyway
+            ooo += 1
+            continue
+        last = seq
+        rec.append((seq, gen, rx))
+        if seq >= N - 1:
+            break
+finally:
+    s.close()
+
+if RAW and rec:
+    a = np.array(rec, dtype=np.float64)
+    np.savez_compressed(RAW, seq=a[:, 0].astype(np.uint32),
+                        gen=a[:, 1], rx=a[:, 2])
+    print(f"raw samples -> {RAW} ({len(rec)} frames)", file=sys.stderr)
+
+if len(rec) < 2:
+    print(f"delivered={len(rec)}  completeness={len(rec)/N:.4f}  "
+          f"avgAoI=nan  peakAoI=nan  "
+          f"(too few frames received -- is dnt forwarding? check namespaces/impairments)")
+    sys.exit(0)
+
+seq = np.array([r[0] for r in rec]); g = np.array([r[1] for r in rec])
+u = np.array([r[2] for r in rec])
+comp = len(rec) / N
+A = u - g; dt = np.diff(u)
+avg = float(np.sum(A[:-1] * dt + 0.5 * dt**2) / (u[-1] - u[0]))
+peak = float(np.mean(u[1:] - g[:-1]))
+# report in model time units (1 unit = 1 ms)
+print(f"delivered={len(rec)}  completeness={comp:.4f}  "
+      f"avgAoI={avg/0.001:.3f}u  peakAoI={peak/0.001:.3f}u  outOfOrder={ooo}")

dnt_testbed/nxp1.ini

@@ -0,0 +1,18 @@
+; SGF node: sequence generation + replication onto two member streams
+[interfaces]
+uni = eth iface=uni
+uni:streams = compound_in
+nniA = eth iface=brA
+nniB = eth iface=brB
+
+[objects]
+Repl = Replicate
+Gen  = SeqGen InitSeqStart=0
+
+[streams]
+compound_in:packet  = eth, cvlan
+compound_in:match   = cvlan vid=10
+compound_in:actions = Gen, after cvlan add rtag, Repl memberA memberB
+
+memberA = edit cvlan.vid=100, send nniA
+memberB = edit cvlan.vid=200, send nniB

dnt_testbed/nxp2.ini

@@ -0,0 +1,24 @@
+; Recovery node: SRF (SeqRcvy, history H) then POF (hold-time D)
+; T = 1 ms  =>  MaxDelay = D in ms = D in units;  H* = ceil(D) capped at Hmax=16
+; run_point.sh rewrites MaxDelay and frerSeqRcvyHistoryLength per sweep point
+[interfaces]
+nniA = eth iface=brA
+nniA:streams = memberA_in
+nniB = eth iface=brB
+nniB:streams = memberB_in
+uni  = eth iface=uni
+
+[objects]
+Elim = SeqRcvy frerSeqRcvyAlgorithm=Vector frerSeqRcvyHistoryLength=16 frerSeqRcvyResetMSec=2000
+Ord  = Pof MaxDelay=16 BufferSize=64 TakeAnyTime=2000
+
+[streams]
+memberA_in:packet  = eth, cvlan, rtag
+memberA_in:match   = cvlan vid=100
+memberA_in:actions = readseq rtag, Elim compound_out
+
+memberB_in:packet  = eth, cvlan, rtag
+memberB_in:match   = cvlan vid=200
+memberB_in:actions = readseq rtag, Elim compound_out
+
+compound_out = Ord, del rtag, edit cvlan.vid=10, send uni

dnt_testbed/run_point.sh

@@ -0,0 +1,26 @@
+#!/bin/bash
+# Run one operating point: ./run_point.sh <D_ms> <H> <N>
+# Rewrites nxp2.ini, (re)starts DNT instances, drives talker/listener.
+D=${1:-4}; H=${2:-4}; N=${3:-60000}
+# Locate the dnt binary. sudo resets PATH to a secure_path that excludes the
+# project dir, so pass an explicit path. Override with: DNT=/path/to/dnt ./run_point.sh
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+DNT="${DNT:-$(command -v dnt || echo "$HERE/../dnt")}"
+if [ ! -x "$DNT" ]; then echo "ERROR: dnt binary not found/executable at '$DNT'" >&2; exit 1; fi
+sed -i "s/MaxDelay=[0-9]*/MaxDelay=$D/; s/frerSeqRcvyHistoryLength=[0-9]*/frerSeqRcvyHistoryLength=$H/" nxp2.ini
+pkill -f "dnt.*nxp[12]\.ini" 2>/dev/null; sleep 0.5
+# dnt logs go to files: keeps stdout clean AND releases the pipe so callers
+# using command substitution (sweep.sh) see EOF when this script exits
+ip netns exec nxp1 "$DNT" nxp1.ini > dnt_nxp1.log 2>&1 & sleep 0.5
+ip netns exec nxp2 "$DNT" nxp2.ini > dnt_nxp2.log 2>&1 & sleep 0.5
+# per-run raw capture (timestamped => reps never overwrite each other);
+# figs.py pools raw/D<D>_H16_*.npz into the PAoI CCDF overlay
+RAWDIR=${RAWDIR:-raw}; mkdir -p "$RAWDIR"
+RAWF="$RAWDIR/D${D}_H${H}_$(date +%Y%m%d-%H%M%S).npz"
+ip netns exec listener python3 listener.py $N "$RAWF" > "result_D${D}.txt" &
+LPID=$!
+# SCHED_FIFO keeps the 1 kHz source jitter low; fall back if chrt unavailable
+if ip netns exec talker chrt -f 1 true 2>/dev/null; then TK="chrt -f 80"; else TK=""; fi
+ip netns exec talker $TK python3 talker.py $N
+wait $LPID
+cat "result_D${D}.txt"

dnt_testbed/sweep.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+# Sweep the POF hold time D (ms): one run_point.sh call per (D, rep).
+# Run as root:  sudo ./sweep.sh
+# Override defaults via env:  D_LIST="2 4 6" R=10 N=60000 sudo -E ./sweep.sh
+#
+# Elimination history is FIXED at Hmax=16 across the sweep: the model's
+# Pareto front uses ideal dedup, and the POF hold-time D does all the
+# dropping (late frames are forwarded stale and discarded at the listener,
+# = the model's late-discard). Matching H to D would add non-model losses:
+# DNT's Vector recovery rogue-drops out-of-window packets, and H<=2 even
+# deadlocks into 2 s resets after a single branch-A loss.
+# The coupling H* = ceil(D_rho/T) is validated separately at the star point:
+#   sudo ./run_point.sh 6 7 60000   (rho=0.995 design point, expect comp>=rho)
+#   sudo ./run_point.sh 6 4 60000   (Hmax=4 infeasibility demo, comp<rho)
+cd "$(dirname "${BASH_SOURCE[0]}")"
+D_LIST=${D_LIST:-"0 1 2 3 4 5 6 8 10 12 16"}
+R=${R:-5}
+N=${N:-60000}          # keep < 65536 (16-bit R-TAG seq in DNT's POF)
+H=${H_FIXED:-16}
+OUT=${OUT:-sweep_results.txt}
+
+echo "# sweep $(date -Is)  N=$N R=$R H=$H D_LIST=[$D_LIST]" >> "$OUT"
+for D in $D_LIST; do
+  for r in $(seq 1 "$R"); do
+    line=$(./run_point.sh "$D" "$H" "$N" | tail -1)
+    echo "D=$D H=$H rep=$r  $line" | tee -a "$OUT"
+  done
+done
+echo "done -> $OUT  (aggregate with: python3 aggregate_results.py $OUT)"

dnt_testbed/talker.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+"""Periodic source: N frames at T=1 ms, 802.1Q vid=10, payload = seq + gen timestamp.
+Run: ip netns exec talker python3 talker.py [N]
+(pin timing jitter down with: chrt -f 80 python3 talker.py N)"""
+import socket, struct, time, sys
+
+N = int(sys.argv[1]) if len(sys.argv) > 1 else 60_000
+T = 0.001
+ETH_P_ALL = 3
+s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)
+s.bind(("eth0", 0))
+
+dst = bytes.fromhex("ffffffffffff")
+src = bytes.fromhex("020000000001")
+vlan = struct.pack("!HH", 0x8100, 0x000A) + struct.pack("!H", 0x88B5)  # vid=10, exp ethertype
+
+t0 = time.clock_gettime(time.CLOCK_REALTIME)
+for k in range(N):
+    target = t0 + k * T
+    while True:
+        now = time.clock_gettime(time.CLOCK_REALTIME)
+        if now >= target: break
+        time.sleep(min(0.0003, target - now))
+    payload = struct.pack("!Qd", k, now) + b"\x00" * 30
+    s.send(dst + src + vlan + payload)
+print(f"sent {N} frames at {1/T:.0f} Hz")