From 471b482af49b20a8d5eb45102fd4c913d5d5bc08 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Fri, 26 Jun 2026 09:15:52 +0200 Subject: [PATCH] test(e2e-stress): extend size axis into multi-fragment regime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add 50k/100k/250k-element size cases (~195 KB–954 KB packets) to the DS and hub size axes so the stress suite exercises UDPSClient multi-fragment reassembly under load, and lift the now-outdated 64 KB validation cap to the 1 MiB deliverable cap (UDPS_CLIENT_MAX_PACKET_BYTES). Slowed producers keep bandwidth realistic. Co-Authored-By: Claude Opus 4.6 --- Test/E2E/chain/stress.py | 58 +++++++++++++++++++++++++++++++--------- 1 file changed, 46 insertions(+), 12 deletions(-) diff --git a/Test/E2E/chain/stress.py b/Test/E2E/chain/stress.py index 4915e0b..275946a 100644 --- a/Test/E2E/chain/stress.py +++ b/Test/E2E/chain/stress.py @@ -41,10 +41,11 @@ gen_data.py / gen_cfg.py generators consume it unchanged): "max_hub_rss_mb": float, "max_zoom_p95_ms": float }, } } -The matrix keeps every datagram a single UDP fragment (payload < 64 KB): the -UDPSClient reassembly buffer caps a deliverable packet at ~64 KB, so sustained -high-rate streaming must stay below it (see scenarios.py s51). Cases therefore -sweep *count* and *rate*, not oversized single packets. +The matrix keeps every datagram a single UDP fragment for the count/rate axes, but +the *size* axis deliberately crosses the 64 KB single-datagram boundary into the +multi-fragment regime: the UDPSClient reassembles up to UDPS_CLIENT_MAX_PACKET_BYTES +(1 MiB) per packet, so the size sweep runs to ~954 KB packets to exercise that path +under load. """ import itertools @@ -59,6 +60,12 @@ _data = itertools.count(46000, 2) # float32 arrays sized so one cycle's array is a single sub-64 KB datagram. F32 = "float32" +# Deliverable-packet cap mirrored verbatim from +# Source/Components/Interfaces/UDPStream/UDPSClient.h +# (UDPS_CLIENT_MAX_PACKET_BYTES). A single source's reassembled DATA payload must +# stay under this; the UDPStreamer still fragments it into MaxPayloadSize chunks. +UDPS_CLIENT_MAX_PACKET_BYTES = 1048576 # 1 MiB + def _f32_arr(name, elements, sampling_rate=1.0e6): """A float32 array signal timestamped FirstSample off the shared ns anchor. @@ -70,12 +77,13 @@ def _f32_arr(name, elements, sampling_rate=1.0e6): time_signal="Tns", sampling_rate=sampling_rate, formula="ramp") -def _source(sid, n_signals, elements, multicast=False): +def _source(sid, n_signals, elements, multicast=False, row_dt=S.ROW_DT): """One UDPStreamer source: a uint64 ns anchor + n_signals float32 arrays. - sampling_rate = elements / row_dt(1 ms) = elements * 1000 keeps each array's - 1 ms window aligned to the 1 kHz cycle regardless of `elements`.""" - rate = elements * 1000.0 + sampling_rate = elements / row_dt keeps each array's per-cycle window equal to + one producer cycle (the First/LastSample monotonic-ring constraint) regardless + of `elements` or a slowed-down producer.""" + rate = elements / row_dt sigs = [S._sig("Tns", "uint64", 1, unit="ns", formula="time_ns", is_time=True)] sigs += [_f32_arr(f"S{i}", elements, rate) for i in range(n_signals)] @@ -94,7 +102,8 @@ def _packet_bytes(n_signals, elements): def mk_stress(sid, axis, level, sources, shape="hub", clients=1, hubs=1, reqrate=0.0, dur=6.0, network="unicast", publishing="Strict", - ratio=None, min_refresh_hz=None, gate=None): + ratio=None, min_refresh_hz=None, gate=None, + row_dt=None, num_rows=None, producer_hz=None): case = { "id": sid, "desc": f"stress {axis}={level}", "network": network, "publishing": publishing, @@ -103,7 +112,7 @@ def mk_stress(sid, axis, level, sources, shape="hub", clients=1, hubs=1, "ws_port": next(_ws), "sources": sources, "oracle": "analytic", "client_checks": ["live", "zoom"], "trig_signal": None, "known_issue": None, - "row_dt": None, "num_rows": None, "producer_hz": None, + "row_dt": row_dt, "num_rows": num_rows, "producer_hz": producer_hz, "shape": shape, "stress": { "axis": axis, "level": level, "clients": clients, "hubs": hubs, @@ -132,6 +141,22 @@ _DS_SIZE = [ for e in (1000, 4000, 8000, 15000) # 4 KB → 60 KB packets (sub-64 KB cap) ] +# Multi-fragment size cases: one float32 array large enough that the DATA packet +# spans several UDP datagrams (>64 KB), exercising the UDPSClient reassembly path +# (cap 1 MiB). A slowed-down producer keeps bandwidth realistic: +# 50k≈195 KB @100 Hz≈20 MB/s, 100k≈390 KB @100 Hz≈39 MB/s, 250k≈954 KB @50 Hz≈48 MB/s. +# row_dt sets the producer cycle so the FirstSample window equals one cycle; a small +# num_rows keeps the FileReader input file bounded (50*250k*4 ≈ 50 MB). +_BIG_SIZE = [(50000, 0.01, 100), (100000, 0.01, 100), (250000, 0.02, 50)] + +_DS_SIZE += [ + mk_stress(f"ds_size_{e}", "ds_signal_elements", e, + [_source("src", 1, e, row_dt=rdt)], + row_dt=rdt, num_rows=50, producer_hz=phz, + gate=_gate(marte_rss=1024.0, hub_rss=2048.0)) + for (e, rdt, phz) in _BIG_SIZE +] + # count: many 1000-element float32 arrays in one source → wider packet + more # per-cycle serialise work. _DS_COUNT = [ @@ -160,6 +185,14 @@ _HUB_SIZE = [ for e in (1000, 4000, 8000, 15000) ] +_HUB_SIZE += [ + mk_stress(f"hub_size_{e}", "hub_signal_elements", e, + [_source("src", 1, e, row_dt=rdt)], + row_dt=rdt, num_rows=50, producer_hz=phz, + gate=_gate(marte_rss=1024.0, hub_rss=2048.0)) + for (e, rdt, phz) in _BIG_SIZE +] + # sources: N independent UDPStreamer feeds into one hub (each its own udp_port). _HUB_SOURCES = [ mk_stress(f"hub_sources_{n}", "hub_source_count", n, @@ -205,9 +238,10 @@ def validate_case(c): elem = max((s["elements"] for s in src["signals"] if not s["is_time"]), default=0) pb = _packet_bytes(n_data, elem) - if pb >= 65536: + if pb >= UDPS_CLIENT_MAX_PACKET_BYTES: errs.append(f"{c['id']}: source {src['id']} packet {pb} B exceeds " - f"the 64 KB single-datagram cap") + f"the {UDPS_CLIENT_MAX_PACKET_BYTES} B deliverable cap " + f"(UDPS_CLIENT_MAX_PACKET_BYTES)") return errs