feat(e2e): instrument-first coverage flow with double-run + new skip flags; port stress multi-fragment sizing
This commit is contained in:
Martino Ferrari
@@ -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 <id>] [--pdf-only]
|
||||
# Usage: ./run_e2e.sh [--skip-build] [--only <id>] [--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 <id>] [--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 <id>] [--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
|
||||
|
||||
|
||||
+52
-17
@@ -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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user