diff --git a/Test/E2E/suite/run_e2e.sh b/Test/E2E/suite/run_e2e.sh index cda42bb..26c8c2c 100755 --- a/Test/E2E/suite/run_e2e.sh +++ b/Test/E2E/suite/run_e2e.sh @@ -10,7 +10,9 @@ # against the analytic/fed oracle, renders plots, and aggregates results.json. # Artifacts: Build/x86-linux/E2E/chain/ (report) and /tmp/chain_e2e/ (scratch). # -# Usage: ./run_e2e.sh [--skip-build] [--only ] [--pdf-only] +# Usage: ./run_e2e.sh [--skip-build] [--only ] [--pdf-only] [--cpp-coverage] +# [--skip-coverage] [--skip-stress] [--skip-datasources] +# [--skip-recorder] [--skip-debug] [--skip-tcplogger] set -u SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" @@ -25,13 +27,24 @@ SKIP_BUILD=0 ONLY="" PDF_ONLY=0 CPP_COV=1 +SKIP_STRESS=0 +SKIP_DATASOURCES=0 +SKIP_RECORDER=0 +SKIP_DEBUG=0 +SKIP_TCPLOGGER=0 while [ $# -gt 0 ]; do case "$1" in --skip-build) SKIP_BUILD=1 ;; --only) shift; ONLY="$1" ;; --pdf-only) PDF_ONLY=1 ;; --cpp-coverage) CPP_COV=1 ;; - --help|-h) echo "Usage: $0 [--skip-build] [--only ] [--pdf-only] [--cpp-coverage]"; exit 0 ;; + --skip-coverage) CPP_COV=0 ;; + --skip-stress) SKIP_STRESS=1 ;; + --skip-datasources) SKIP_DATASOURCES=1 ;; + --skip-recorder) SKIP_RECORDER=1 ;; + --skip-debug) SKIP_DEBUG=1 ;; + --skip-tcplogger) SKIP_TCPLOGGER=1 ;; + --help|-h) echo "Usage: $0 [--skip-build] [--only ] [--pdf-only] [--cpp-coverage] [--skip-coverage] [--skip-stress] [--skip-datasources] [--skip-recorder] [--skip-debug] [--skip-tcplogger]"; exit 0 ;; *) echo "unknown arg $1" >&2; exit 2 ;; esac shift @@ -99,15 +112,23 @@ fi # direct: kind|id|cfg|-|-|-|-|- # recorder: kind|id|marte_cfg|hub_cfg|-|-|-|- # debug/tcplogger: kind|id|cfg|cmd_port|udp_port|log_port|-|- -LIST="$(${PY} - "${ONLY}" <<'PY' +SKIP_KINDS="" +[ "${SKIP_DATASOURCES}" -eq 1 ] && SKIP_KINDS="${SKIP_KINDS}direct," +[ "${SKIP_RECORDER}" -eq 1 ] && SKIP_KINDS="${SKIP_KINDS}recorder," +[ "${SKIP_DEBUG}" -eq 1 ] && SKIP_KINDS="${SKIP_KINDS}debug," +[ "${SKIP_TCPLOGGER}" -eq 1 ] && SKIP_KINDS="${SKIP_KINDS}tcplogger," +LIST="$(${PY} - "${ONLY}" "${SKIP_KINDS}" <<'PY' import sys, os sys.path.insert(0, os.path.dirname(os.path.abspath("Test/E2E/suite/scenarios.py"))) sys.path.insert(0, os.path.join(os.getcwd(), "Test/E2E/suite")) import scenarios as S only = sys.argv[1] if len(sys.argv) > 1 else "" +skip_kinds = set(sys.argv[2].split(",")) if len(sys.argv) > 2 and sys.argv[2] else set() for s in S.SCENARIOS: if only and s["id"] != only: continue + if s["kind"] in skip_kinds: + continue kind = s["kind"] if kind == "chain": trig = s.get("trig_signal") or "" @@ -129,6 +150,16 @@ PY if [ -z "${LIST}" ]; then echo "no scenarios selected"; exit 1; fi +# run_scenario_matrix — runs the full scenario matrix (already computed into +# ${LIST}) against whatever binaries are currently built. Invoked once for the +# authoritative pass (normal binaries) and again, inside a subshell with +# OUT_DIR overridden, for the coverage-only re-run against instrumented +# binaries (see Phase 4 below). Running the coverage pass in a subshell means +# its SCEN_IDS/OUT_DIR/HUB_PID/APP_PID/trap mutations never leak back into the +# parent shell, so the primary pass's SCEN_IDS (consumed by the results.json +# aggregation step further down) is unaffected regardless of call order. +run_scenario_matrix() { + local RESULTS_SUFFIX="$1" SCEN_IDS="" HUB_PID=""; APP_PID="" cleanup() { @@ -329,6 +360,18 @@ done <<< "${LIST}" trap - EXIT cleanup +} +run_scenario_matrix "primary" + +# ── Stress (Phase 3: normal binaries, always uninstrumented) ──────────────── +if [ "${SKIP_STRESS}" -eq 0 ]; then + echo "" + echo "── Stress matrix ──" + STRESS_OUT="${OUT_DIR}/stress" + mkdir -p "${STRESS_OUT}" + ${PY} "${SCRIPT_DIR}/stress_run.py" --marte "${MARTE_APP}" --hub "${STREAMHUB_EX}" \ + --client "${CLIENT}" --work "/tmp/chain_stress" --out "${STRESS_OUT}" || true +fi # ── Aggregate results.json ─────────────────────────────────────────────────── # Scenarios carrying a `known_issue` marker exercise a documented, not-yet-fixed @@ -388,16 +431,38 @@ echo "── Unit tests + coverage ──" # restore a clean build so later --skip-build runs aren't left instrumented. CPP_COV_FLAG="" if [ "${CPP_COV}" -eq 1 ]; then - echo " building instrumented (gcov) libraries + GTest ..." + echo " building instrumented (gcov) libraries + apps + GTest ..." COV_O="--coverage" COV_L="-Wl,--no-as-needed -fPIC --coverage" make -C "${REPO_ROOT}" -f Makefile.gcc clean >/dev/null 2>&1 || true - make -C "${REPO_ROOT}" -f Makefile.gcc core TARGET="${TARGET}" \ + # `apps` (StreamHub.ex) must be rebuilt here too, not just `core`: the + # coverage-pass E2E re-run below launches StreamHub, and `make clean` just + # removed the non-instrumented StreamHub.ex — without this it would 404 the + # WS port for every chain/recorder scenario in the coverage pass. + make -C "${REPO_ROOT}" -f Makefile.gcc core apps TARGET="${TARGET}" \ OPTIM="${COV_O}" LFLAGS="${COV_L}" 2>&1 | tail -1 for d in Test/Components/DataSources/UDPStreamer Test/Components/DataSources/UDPStreamerClient Test/Applications/StreamHub Test/GTest Test/Integration; do make -C "${REPO_ROOT}/${d}" -f Makefile.gcc TARGET="${TARGET}" \ OPTIM="${COV_O}" LFLAGS="${COV_L}" 2>&1 | tail -1 done + + # Coverage-only scenario re-run: the instrumented binaries above are only + # exercised so far by collect.py's unit-test binaries (MainGTest/Integration), + # which never touch the E2E-scenario-only code paths (e.g. StreamHub code + # only hit via the direct/recorder/debug E2E flows). Re-running the full + # scenario matrix here — against the SAME instrumented binaries, before + # collect.py's lcov capture — lets those .gcda files accumulate E2E-path + # coverage too. Run in a subshell with OUT_DIR overridden so the coverage + # pass's logs/results land in a discardable subdirectory and none of its + # SCEN_IDS/OUT_DIR/trap state leaks back into this shell (the authoritative + # results.json was already written above, from the primary pass). + echo " re-running scenario matrix under instrumented binaries (coverage pass) ..." + ( + OUT_DIR="${OUT_DIR}/coverage_pass" + mkdir -p "${OUT_DIR}" + run_scenario_matrix "coverage" + ) || true + CPP_COV_FLAG="--cpp-coverage" fi diff --git a/Test/E2E/suite/stress.py b/Test/E2E/suite/stress.py index 4915e0b..5f8cee7 100644 --- a/Test/E2E/suite/stress.py +++ b/Test/E2E/suite/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, @@ -125,11 +134,27 @@ def _gate(min_frames=5, marte_rss=512.0, hub_rss=1024.0, zoom_p95=1000.0): # Single source + single hub; the load lands on the UDPStreamer serialise/send # path and is read back through one WS client (proc_perf measures marte). -# size: one float32 array, growing element count → bigger single-datagram packet. +# size: one float32 array, growing element count → bigger packet. _DS_SIZE = [ mk_stress(f"ds_size_{e}", "ds_signal_elements", e, [_source("src", 1, e)], gate=_gate()) - for e in (1000, 4000, 8000, 15000) # 4 KB → 60 KB packets (sub-64 KB cap) + for e in (1000, 4000, 8000, 15000) # 4 KB → 60 KB packets (original single-datagram cases) +] + +# 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 @@ -137,7 +162,7 @@ _DS_SIZE = [ _DS_COUNT = [ mk_stress(f"ds_count_{n}", "ds_signal_count", n, [_source("src", n, 1000)], gate=_gate()) - for n in (1, 4, 8, 15) # 15*4 KB ≈ 60 KB packet (sub-64 KB cap) + for n in (1, 4, 8, 15) # 15*4 KB ≈ 60 KB packet (original single-datagram case) ] # clients (fan-out): one source, M subscriber hubs (ds_fanout shape). The @@ -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, @@ -198,16 +231,18 @@ def validate_case(c): f"(UDPStreamer max unicast clients)") if c["shape"] == "hub" and (st["clients"] < 1): errs.append(f"{c['id']}: hub clients must be >= 1") - # Single-datagram ceiling: keep each source's packet < 64 KB so it never - # needs reassembly (the deliverable-packet cap). + # Enforce 1 MiB deliverable cap (UDPS_CLIENT_MAX_PACKET_BYTES): each source's + # reassembled DATA payload must stay under this. Note: the size axis includes + # multi-fragment cases that intentionally exceed 64 KB (reassembled by UDPSClient). for src in c["sources"]: n_data = sum(1 for s in src["signals"] if not s["is_time"]) 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