From 92c820f1a38539c0f7d7da710c0c20fd9be6a0df Mon Sep 17 00:00:00 2001 From: William Emfinger Date: Sun, 1 Nov 2026 19:48:36 -0600 Subject: [PATCH 1/3] test(rtps): interop cases for lease liveliness and a busy domain, on FastDDS and CycloneDDS Two defect classes found against CycloneDDS peers had no coverage because every ROS 2 pair in the matrix proves a single sample within a few seconds: - Liveliness past the remote participant lease. A peer whose SPDP re-announcements are not delivered to the SPDP reader has its lease expire (10 s for CycloneDDS, which addresses the repeats to readerId ENTITYID_UNKNOWN) and loses every matched endpoint - discovery looked fine, data stopped after the lease. Both legs now run ~30 s and count only samples that arrive after 16 s: `rtps_interop_sub` gains a backward-compatible `count_after_s` argument, and a small rclpy `ros2_count_subscriber.py` does the same on the ROS 2 side. Run on the FastDDS pass as a control and on the CycloneDDS pass, where it fails until the SPDP-reader delivery fix (#871) is in. - A busy domain. Twelve unrelated ROS 2 participants are up before espp starts, then the publisher espp needs joins last and must still be matched: a participant beyond the remote-participant pool is ignored silently. The harness builds the host profile (32 slots) so this passes; the embedded profile's 10 slots need the pool raised for such a domain (#870's knob), which the note and README say. A live multi-homed-peer case (the #865 class) is described in the README as follow-up work: it needs a peer address espp cannot reach, which one shared network namespace cannot provide. Workflow timeout 30 -> 40 min for the ~3 min the four new runs add. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DFjjnq3XCTRAXENSxsCxJU --- .github/workflows/rtps_interop.yml | 2 +- components/rtps/interop/README.md | 15 +++- .../rtps/interop/ros2_count_subscriber.py | 60 ++++++++++++++++ components/rtps/interop/run_interop.sh | 71 +++++++++++++++++++ pc/tests/rtps_interop_sub.cpp | 32 +++++++-- 5 files changed, 171 insertions(+), 9 deletions(-) create mode 100644 components/rtps/interop/ros2_count_subscriber.py diff --git a/.github/workflows/rtps_interop.yml b/.github/workflows/rtps_interop.yml index 4252f2d22c..ac7e306093 100644 --- a/.github/workflows/rtps_interop.yml +++ b/.github/workflows/rtps_interop.yml @@ -28,7 +28,7 @@ concurrency: jobs: interop: runs-on: ubuntu-latest - timeout-minutes: 30 + timeout-minutes: 40 steps: - uses: actions/checkout@v7 with: diff --git a/components/rtps/interop/README.md b/components/rtps/interop/README.md index 57a000af5e..69e845a839 100644 --- a/components/rtps/interop/README.md +++ b/components/rtps/interop/README.md @@ -28,7 +28,20 @@ touched. | espp_pub->ros2_echo | espp reliable writer -> ROS 2 subscriber (`rt/chatter`, `std_msgs::msg::dds_::String_`) | | ros2_pub->espp_sub | ROS 2 publisher -> espp reliable reader | | espp_be_pub->ros2_be_echo | best-effort pairing | -| cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` | +| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 16 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control | +| many_participants_late_joiner | twelve unrelated ROS 2 participants are on the domain before espp starts, then the publisher espp needs joins last and must still be matched. Guards the remote-participant pool (`SPDP_MAX_NUMBER_FOUND_PARTICIPANTS`, 32 on the host profile built here, 10 on the embedded profile - raise it via menuconfig for a domain this busy): a participant beyond the pool is ignored silently | +| cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) and the two cases above again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` | + +## Not covered (yet): a multi-homed peer + +A FastDDS or CycloneDDS peer on several networks announces one unicast locator +per interface, and espp must pick the one on its own network (`locator_subnet` +checks the selection logic on canned announcements). Reproducing it live needs +a peer address espp cannot reach, which a single shared network namespace cannot +provide: every local address is reachable there. The design would be a second +Docker network, the ROS 2 side attached to both and espp to one, so the ROS 2 +participant announces a locator on the network espp is not on; that is a +two-container `run.sh` and is left for a follow-up. ## Notes diff --git a/components/rtps/interop/ros2_count_subscriber.py b/components/rtps/interop/ros2_count_subscriber.py new file mode 100644 index 0000000000..4663d79804 --- /dev/null +++ b/components/rtps/interop/ros2_count_subscriber.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Count std_msgs/String samples from a ROS 2 subscriber for a fixed time. + +Used by the interop harness as the ROS 2 side of the liveliness cases: +`ros2 topic echo --once` proves one sample arrived, this proves samples KEEP +arriving - samples received earlier than `count_after_s` seconds after start +are reported but not counted, so a peer that stops being served once its +participant lease has expired cannot pass. + +Usage: ros2_count_subscriber.py TOPIC SECONDS REQUIRED COUNT_AFTER_S [reliable|best_effort] +Exits 0 iff at least REQUIRED samples arrived after COUNT_AFTER_S seconds. +""" + +import sys +import time + +import rclpy +from rclpy.qos import QoSProfile, ReliabilityPolicy +from std_msgs.msg import String + + +def main() -> int: + topic = sys.argv[1] if len(sys.argv) > 1 else "/chatter" + seconds = float(sys.argv[2]) if len(sys.argv) > 2 else 25.0 + required = int(sys.argv[3]) if len(sys.argv) > 3 else 10 + count_after_s = float(sys.argv[4]) if len(sys.argv) > 4 else 0.0 + reliable = (sys.argv[5] if len(sys.argv) > 5 else "reliable") != "best_effort" + + rclpy.init() + node = rclpy.create_node("countsub") + qos = QoSProfile(depth=10) + qos.reliability = ReliabilityPolicy.RELIABLE if reliable else ReliabilityPolicy.BEST_EFFORT + start = time.time() + counts = {"early": 0, "counted": 0} + + def on_sample(_msg: String) -> None: + if time.time() - start < count_after_s: + counts["early"] += 1 + else: + counts["counted"] += 1 + if counts["counted"] <= 3 or counts["counted"] % 25 == 0: + print(f"received {counts['counted']} (t={time.time() - start:.1f}s)", flush=True) + + node.create_subscription(String, topic, on_sample, qos) + while time.time() - start < seconds: + rclpy.spin_once(node, timeout_sec=0.1) + + ok = counts["counted"] >= required + print( + f"{'PASS' if ok else 'FAIL'} received={counts['counted']} required={required} " + f"(not counted, before {count_after_s:.0f}s: {counts['early']})", + flush=True, + ) + node.destroy_node() + rclpy.shutdown() + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/components/rtps/interop/run_interop.sh b/components/rtps/interop/run_interop.sh index 5cf964aa67..bceb822dc6 100755 --- a/components/rtps/interop/run_interop.sh +++ b/components/rtps/interop/run_interop.sh @@ -22,6 +22,71 @@ result() { # name exit_code else echo "RESULT FAIL: $1"; FAIL=$((FAIL+1)); fi } +# --- Liveliness past the remote participant lease ---------------------------- +# A peer keeps re-announcing itself over SPDP; if those re-announcements are not +# delivered to the SPDP reader, espp lets the peer's lease expire (10 s for +# CycloneDDS, which addresses the repeats to readerId ENTITYID_UNKNOWN) and +# drops every matched endpoint with it - discovery looked fine, data stopped +# after the lease. `ros2 topic echo --once` cannot see that, so both legs here +# only count samples that arrive after the lease would have expired +# (count_after_s, see rtps_interop_sub / ros2_count_subscriber.py). The +# FastDDS pass is the control: it names the SPDP reader in every announcement. +lease_liveliness_cases() { # prefix log-tag + local prefix="$1" tag="$2" + note "${prefix}ROS 2 publisher -> espp subscriber, samples still arriving 16 s in (liveliness past the lease)" + "$BIN"/rtps_interop_sub rt/lease std_msgs::msg::dds_::String_ 1 10 30 "" 0 2 16 > "/tmp/${tag}leasesub.log" 2>&1 & + local sub=$! + sleep 3 + timeout 30 ros2 topic pub -r 5 /lease std_msgs/msg/String "data: 'still alive'" > "/tmp/${tag}leaserospub.log" 2>&1 & + local rospub=$! + wait $sub + local sub_rc=$? + kill $rospub 2>/dev/null; wait $rospub 2>/dev/null + tail -2 "/tmp/${tag}leasesub.log" + result "${prefix}ros2_pub->espp_sub_liveliness" $sub_rc + + note "${prefix}espp publisher -> ROS 2 subscriber, samples still arriving 16 s in (liveliness past the lease)" + "$BIN"/rtps_interop_pub rt/lease2 std_msgs::msg::dds_::String_ 1 160 200 > "/tmp/${tag}leasepub.log" 2>&1 & + local pub=$! + sleep 3 + timeout 32 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 27 10 16 reliable \ + > "/tmp/${tag}leasecount.log" 2>&1 + local count_rc=$? + kill $pub 2>/dev/null; wait $pub 2>/dev/null + tail -2 "/tmp/${tag}leasecount.log" + result "${prefix}espp_pub->ros2_sub_liveliness" $count_rc +} + +# --- A dozen participants on the domain, the publisher joins last ------------- +# Every ROS 2 process is a DDS participant, and espp tracks remote participants +# in a fixed pool (SPDP_MAX_NUMBER_FOUND_PARTICIPANTS: 32 on the host profile +# this harness builds, 10 on the embedded profile). A participant beyond the +# pool is ignored silently - never matched, its data dropped. With twelve +# unrelated participants already on the domain, the publisher espp actually +# needs joins last and must still be matched. On an ESP32 the same scenario +# needs the pool raised (menuconfig RTPS -> Custom capacity overrides). +many_participants_case() { # prefix log-tag + local prefix="$1" tag="$2" + note "${prefix}espp subscriber with twelve other participants on the domain; the publisher joins last" + local crowd=() + local i + for i in $(seq 1 12); do + timeout 45 ros2 topic pub -r 1 "/crowd$i" std_msgs/msg/String "data: 'crowd'" > /dev/null 2>&1 & + crowd+=("$!") + done + sleep 8 # let the crowd come up (each is its own participant) before espp starts + "$BIN"/rtps_interop_sub rt/crowd std_msgs::msg::dds_::String_ 1 3 30 > "/tmp/${tag}crowdsub.log" 2>&1 & + local sub=$! + sleep 3 + timeout 32 ros2 topic pub -r 5 /crowd std_msgs/msg/String "data: 'late joiner'" > "/tmp/${tag}crowdpub.log" 2>&1 & + local pub=$! + wait $sub + local sub_rc=$? + kill $pub "${crowd[@]}" 2>/dev/null; wait $pub "${crowd[@]}" 2>/dev/null + tail -2 "/tmp/${tag}crowdsub.log" + result "${prefix}many_participants_late_joiner" $sub_rc +} + note "Build espp lib + host binaries (linux)" cmake -S lib -B lib/build -DCMAKE_BUILD_TYPE=Release -DESPP_INSTALL=ON -DCMAKE_INSTALL_PREFIX=/tmp/espp/install > /tmp/cmake_lib.log 2>&1 \ && cmake --build lib/build -j"$(nproc)" --target install > /tmp/build_lib.log 2>&1 \ @@ -233,6 +298,9 @@ kill $ROSBAND_PID 2>/dev/null; wait $ROSBAND_PID 2>/dev/null cat /tmp/subband.log result "ros2_pub->espp_banded_sub" $subband_rc +lease_liveliness_cases "" "f" +many_participants_case "" "f" + note "espp best-effort publisher -> ROS 2 best-effort subscriber" "$BIN"/rtps_interop_pub rt/chatter std_msgs::msg::dds_::String_ 0 60 200 > /tmp/pub2.log 2>&1 & ESPP_PID=$! @@ -395,6 +463,9 @@ kill $ROS_PID 2>/dev/null; wait $ROS_PID 2>/dev/null tail -2 /tmp/csub2.log result "cyclone: ros2_be_pub->espp_be_sub" $csub2_rc +lease_liveliness_cases "cyclone: " "c" +many_participants_case "cyclone: " "c" + if grep -l "malformed packet received from vendor" /tmp/c*.log > /dev/null 2>&1; then echo "CycloneDDS reported malformed packets from espp:"; grep -h "malformed packet" /tmp/c*.log | head -3 fi diff --git a/pc/tests/rtps_interop_sub.cpp b/pc/tests/rtps_interop_sub.cpp index aee4af1eda..3b00237509 100644 --- a/pc/tests/rtps_interop_sub.cpp +++ b/pc/tests/rtps_interop_sub.cpp @@ -6,7 +6,7 @@ // std_msgs/String on /chatter. // // Usage: rtps_interop_sub [topic] [type] [reliable(0|1)] [required] [timeout_s] -// [interface_ip] [payload_bytes] [band] +// [interface_ip] [payload_bytes] [band] [count_after_s] // When payload_bytes > 0, each received String is verified byte-exact against the // deterministic payload_bytes-long pattern (proving fragmented >64 KB samples are // reassembled correctly); only byte-exact receptions count toward `required`. @@ -15,7 +15,13 @@ // announced via its SEDP per-endpoint unicast locator - the interop matrix uses // this to prove a FastDDS/ROS 2 peer honors that locator and delivers the // topic's traffic to the dedicated port. -// Exits 0 once `required` samples arrive within `timeout_s`. +// count_after_s (default 0): samples that arrive earlier than this many seconds +// after start are reported but NOT counted toward `required`. The interop +// matrix uses this as a liveliness gate: a peer whose participant we dropped +// (e.g. its lease expired because its re-announcements were not delivered) +// stops being matched, so nothing counts after the lease - only samples that +// keep arriving past that point prove the peer stayed alive. +// Exits 0 once `required` (counted) samples arrive within `timeout_s`. #include #include @@ -61,6 +67,7 @@ int main(int argc, char **argv) { const char *interface_ip = (argc > 6) ? argv[6] : ""; // "" -> auto-detect const std::size_t payload_bytes = (argc > 7) ? std::strtoul(argv[7], nullptr, 10) : 0; const int band_arg = (argc > 8) ? std::atoi(argv[8]) : static_cast(espp::QosBand::Normal); + const int count_after_s = (argc > 9) ? std::atoi(argv[9]) : 0; // Validate before casting: an out-of-range value would index band arrays. if (band_arg < static_cast(espp::QosBand::Critical) || band_arg > static_cast(espp::QosBand::Low)) { @@ -72,6 +79,9 @@ int main(int argc, char **argv) { const std::string expected = payload_bytes > 0 ? make_pattern(payload_bytes) : std::string{}; std::atomic received{0}; + std::atomic early{0}; // arrived before count_after_s; reported, not counted + const auto start = std::chrono::steady_clock::now(); + const auto count_from = start + std::chrono::seconds(count_after_s); espp::RtpsParticipant participant({ .interface_address = interface_ip, @@ -87,11 +97,18 @@ int main(int argc, char **argv) { .type_name = type, .reliability = reliable ? Reliability::RELIABLE : Reliability::BEST_EFFORT, .on_sample = - [&received, payload_bytes, &expected](std::span cdr_payload) { + [&received, &early, count_from, payload_bytes, + &expected](std::span cdr_payload) { auto msg = cdr::deserialize(std::as_bytes(cdr_payload)); if (!msg) { return; } + if (std::chrono::steady_clock::now() < count_from) { + const int n = early.fetch_add(1) + 1; + std::printf("received (not counted yet) %d\n", n); + std::fflush(stdout); + return; + } if (payload_bytes > 0) { // Large-payload mode: only a byte-exact reassembly counts. const bool exact = msg->data.size() == payload_bytes && msg->data == expected; @@ -115,17 +132,18 @@ int main(int argc, char **argv) { std::printf("FAIL: add_reader\n"); return 1; } - std::printf("interop_sub: topic=%s type=%s reliable=%d required=%d timeout=%ds band=%d\n", topic, - type, reliable ? 1 : 0, required, timeout_s, band_arg); + std::printf("interop_sub: topic=%s type=%s reliable=%d required=%d timeout=%ds band=%d " + "count_after=%ds\n", + topic, type, reliable ? 1 : 0, required, timeout_s, band_arg, count_after_s); - const auto start = std::chrono::steady_clock::now(); while (received.load() < required && std::chrono::steady_clock::now() - start < std::chrono::seconds(timeout_s)) { std::this_thread::sleep_for(100ms); } const int n = received.load(); - std::printf("%s received=%d required=%d\n", n >= required ? "PASS" : "FAIL", n, required); + std::printf("%s received=%d required=%d (not counted, before %ds: %d)\n", + n >= required ? "PASS" : "FAIL", n, required, count_after_s, early.load()); participant.stop(); return n >= required ? 0 : 1; } From 735c9b651d84088aa118a08623abd877c3d83f67 Mon Sep 17 00:00:00 2001 From: William Emfinger Date: Sun, 1 Nov 2026 20:34:13 -0600 Subject: [PATCH 2/3] test(rtps): interop liveliness window past the worst-case expiry; lift CycloneDDS's participant-index cap Count from 22 s in, not 16: discovery, the 10 s lease and the liveliness check cadence all fit before that, so a peer dropped at its lease cannot contribute to the count. CycloneDDS allows ten participants per host per domain by default (MaxAutoParticipantIndex 9); the busy-domain case creates thirteen, so the CycloneDDS pass raises it through CYCLONEDDS_URI. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DFjjnq3XCTRAXENSxsCxJU --- components/rtps/interop/README.md | 2 +- components/rtps/interop/run_interop.sh | 25 ++++++++++++++++--------- 2 files changed, 17 insertions(+), 10 deletions(-) diff --git a/components/rtps/interop/README.md b/components/rtps/interop/README.md index 69e845a839..3f4ce4da81 100644 --- a/components/rtps/interop/README.md +++ b/components/rtps/interop/README.md @@ -28,7 +28,7 @@ touched. | espp_pub->ros2_echo | espp reliable writer -> ROS 2 subscriber (`rt/chatter`, `std_msgs::msg::dds_::String_`) | | ros2_pub->espp_sub | ROS 2 publisher -> espp reliable reader | | espp_be_pub->ros2_be_echo | best-effort pairing | -| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 16 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control | +| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 22 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control | | many_participants_late_joiner | twelve unrelated ROS 2 participants are on the domain before espp starts, then the publisher espp needs joins last and must still be matched. Guards the remote-participant pool (`SPDP_MAX_NUMBER_FOUND_PARTICIPANTS`, 32 on the host profile built here, 10 on the embedded profile - raise it via menuconfig for a domain this busy): a participant beyond the pool is ignored silently | | cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) and the two cases above again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` | diff --git a/components/rtps/interop/run_interop.sh b/components/rtps/interop/run_interop.sh index bceb822dc6..fd7a1fb55d 100755 --- a/components/rtps/interop/run_interop.sh +++ b/components/rtps/interop/run_interop.sh @@ -28,16 +28,18 @@ result() { # name exit_code # CycloneDDS, which addresses the repeats to readerId ENTITYID_UNKNOWN) and # drops every matched endpoint with it - discovery looked fine, data stopped # after the lease. `ros2 topic echo --once` cannot see that, so both legs here -# only count samples that arrive after the lease would have expired -# (count_after_s, see rtps_interop_sub / ros2_count_subscriber.py). The -# FastDDS pass is the control: it names the SPDP reader in every announcement. +# only count samples that arrive from 22 s in: discovery (a few seconds), the +# 10 s lease and the liveliness check cadence all fit before that, so a +# dropped peer contributes nothing to the count (count_after_s, see +# rtps_interop_sub / ros2_count_subscriber.py). The FastDDS pass is the +# control: it names the SPDP reader in every announcement. lease_liveliness_cases() { # prefix log-tag local prefix="$1" tag="$2" - note "${prefix}ROS 2 publisher -> espp subscriber, samples still arriving 16 s in (liveliness past the lease)" - "$BIN"/rtps_interop_sub rt/lease std_msgs::msg::dds_::String_ 1 10 30 "" 0 2 16 > "/tmp/${tag}leasesub.log" 2>&1 & + note "${prefix}ROS 2 publisher -> espp subscriber, samples still arriving 22 s in (liveliness past the lease)" + "$BIN"/rtps_interop_sub rt/lease std_msgs::msg::dds_::String_ 1 10 34 "" 0 2 22 > "/tmp/${tag}leasesub.log" 2>&1 & local sub=$! sleep 3 - timeout 30 ros2 topic pub -r 5 /lease std_msgs/msg/String "data: 'still alive'" > "/tmp/${tag}leaserospub.log" 2>&1 & + timeout 36 ros2 topic pub -r 5 /lease std_msgs/msg/String "data: 'still alive'" > "/tmp/${tag}leaserospub.log" 2>&1 & local rospub=$! wait $sub local sub_rc=$? @@ -45,11 +47,11 @@ lease_liveliness_cases() { # prefix log-tag tail -2 "/tmp/${tag}leasesub.log" result "${prefix}ros2_pub->espp_sub_liveliness" $sub_rc - note "${prefix}espp publisher -> ROS 2 subscriber, samples still arriving 16 s in (liveliness past the lease)" - "$BIN"/rtps_interop_pub rt/lease2 std_msgs::msg::dds_::String_ 1 160 200 > "/tmp/${tag}leasepub.log" 2>&1 & + note "${prefix}espp publisher -> ROS 2 subscriber, samples still arriving 22 s in (liveliness past the lease)" + "$BIN"/rtps_interop_pub rt/lease2 std_msgs::msg::dds_::String_ 1 220 200 > "/tmp/${tag}leasepub.log" 2>&1 & local pub=$! sleep 3 - timeout 32 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 27 10 16 reliable \ + timeout 40 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 34 10 22 reliable \ > "/tmp/${tag}leasecount.log" 2>&1 local count_rc=$? kill $pub 2>/dev/null; wait $pub 2>/dev/null @@ -422,6 +424,10 @@ result "espp_action_client->ros2_server" $actclient_rc # that espp's services (and rmw_fastrtps_cpp) use. note "CycloneDDS (rmw_cyclonedds_cpp): ROS 2 pub/sub pairs" export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp +# CycloneDDS caps the participants one host may create on a domain through its +# participant-index range (MaxAutoParticipantIndex, default 9 = ten +# participants); the busy-domain case below creates thirteen, so lift it. +export CYCLONEDDS_URI='60' ros2 daemon stop > /dev/null 2>&1 "$BIN"/rtps_interop_pub rt/chatter std_msgs::msg::dds_::String_ 1 60 200 > /tmp/cpub1.log 2>&1 & @@ -470,6 +476,7 @@ if grep -l "malformed packet received from vendor" /tmp/c*.log > /dev/null 2>&1; echo "CycloneDDS reported malformed packets from espp:"; grep -h "malformed packet" /tmp/c*.log | head -3 fi export RMW_IMPLEMENTATION=rmw_fastrtps_cpp +unset CYCLONEDDS_URI # --- Python bindings: functional round-trip of every RMI/AMI mechanism -------- # Build the espp wheel in-container and run the demo (multicast discovery works From bf0b16ed322586aa19d7c69595426896fcd41f67 Mon Sep 17 00:00:00 2001 From: William Emfinger Date: Mon, 2 Nov 2026 07:00:34 -0600 Subject: [PATCH 3/3] test(rtps): interop review follow-ups; note why only the espp-publisher liveliness leg detects a dropped peer - cleanup `wait`s in the new cases tolerate an already-reaped PID - rtps_interop_sub: count_after_s < 0 is a usage error (exit 1); early samples are verified byte-exact and reported like counted ones, only the count differs - ros2_count_subscriber.py: argparse with typed, non-negative arguments - README: CI showed only `espp_pub->ros2_sub_liveliness` failing on CycloneDDS. A HEARTBEAT reaching a matched espp reader refreshes the sending participant's liveliness, so a CycloneDDS writer keeps its participant alive through its heartbeats even with its SPDP re-announcements lost; an espp writer only gets ACKNACKs, which do not, and the peer expires. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DFjjnq3XCTRAXENSxsCxJU --- components/rtps/interop/README.md | 2 +- .../rtps/interop/ros2_count_subscriber.py | 45 ++++++++++++++++--- components/rtps/interop/run_interop.sh | 8 ++-- pc/tests/rtps_interop_sub.cpp | 32 +++++++++---- 4 files changed, 67 insertions(+), 20 deletions(-) diff --git a/components/rtps/interop/README.md b/components/rtps/interop/README.md index 3f4ce4da81..e8d742c278 100644 --- a/components/rtps/interop/README.md +++ b/components/rtps/interop/README.md @@ -28,7 +28,7 @@ touched. | espp_pub->ros2_echo | espp reliable writer -> ROS 2 subscriber (`rt/chatter`, `std_msgs::msg::dds_::String_`) | | ros2_pub->espp_sub | ROS 2 publisher -> espp reliable reader | | espp_be_pub->ros2_be_echo | best-effort pairing | -| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 22 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control | +| ros2_pub->espp_sub_liveliness, espp_pub->ros2_sub_liveliness | samples keep arriving 22 s in, i.e. past a remote participant lease (CycloneDDS: 10 s). Only samples after that point count (`rtps_interop_sub ... count_after_s`, `ros2_count_subscriber.py`), so a peer whose SPDP re-announcements were not delivered - and whose lease therefore expired, dropping every matched endpoint - fails. CycloneDDS addresses those re-announcements to readerId ENTITYID_UNKNOWN; the FastDDS pass is the control. Only the espp-publisher leg detects a dropped peer: a HEARTBEAT that reaches a matched espp reader also refreshes the sending participant's liveliness (`MessageReceiver::processHeartbeat`), so a CycloneDDS *writer* keeps its participant alive through its heartbeats even when its SPDP re-announcements are lost, while an espp writer only gets ACKNACKs back, which do not refresh it, and the peer expires | | many_participants_late_joiner | twelve unrelated ROS 2 participants are on the domain before espp starts, then the publisher espp needs joins last and must still be matched. Guards the remote-participant pool (`SPDP_MAX_NUMBER_FOUND_PARTICIPANTS`, 32 on the host profile built here, 10 on the embedded profile - raise it via menuconfig for a domain this busy): a participant beyond the pool is ignored silently | | cyclone: ... | the ROS 2 pub/sub pairs (reliable and best-effort, both directions) and the two cases above again on `rmw_cyclonedds_cpp`. Services are not covered: rmw_cyclonedds_cpp puts a `{guid, seq}` request header in the payload instead of `related_sample_identity` | diff --git a/components/rtps/interop/ros2_count_subscriber.py b/components/rtps/interop/ros2_count_subscriber.py index 4663d79804..f77e2fce1e 100644 --- a/components/rtps/interop/ros2_count_subscriber.py +++ b/components/rtps/interop/ros2_count_subscriber.py @@ -7,10 +7,13 @@ are reported but not counted, so a peer that stops being served once its participant lease has expired cannot pass. -Usage: ros2_count_subscriber.py TOPIC SECONDS REQUIRED COUNT_AFTER_S [reliable|best_effort] -Exits 0 iff at least REQUIRED samples arrived after COUNT_AFTER_S seconds. +Usage: ros2_count_subscriber.py TOPIC [--seconds S] [--required N] [--count-after S] + [--reliability reliable|best_effort] +Exits 0 iff at least REQUIRED samples arrived after COUNT_AFTER seconds, 2 on +bad arguments. """ +import argparse import sys import time @@ -19,12 +22,40 @@ from std_msgs.msg import String +def non_negative(text: str) -> float: + value = float(text) + if value < 0: + raise argparse.ArgumentTypeError(f"must be >= 0, got {text}") + return value + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Count std_msgs/String samples on a topic for a fixed time; " + "samples before --count-after are reported but not counted." + ) + parser.add_argument("topic", nargs="?", default="/chatter") + parser.add_argument("--seconds", type=non_negative, default=25.0, help="how long to listen") + parser.add_argument("--required", type=int, default=10, help="counted samples needed to pass") + parser.add_argument( + "--count-after", type=non_negative, default=0.0, help="seconds before counting starts" + ) + parser.add_argument( + "--reliability", choices=("reliable", "best_effort"), default="reliable" + ) + args = parser.parse_args(argv) + if args.required < 0: + parser.error(f"--required must be >= 0, got {args.required}") + return args + + def main() -> int: - topic = sys.argv[1] if len(sys.argv) > 1 else "/chatter" - seconds = float(sys.argv[2]) if len(sys.argv) > 2 else 25.0 - required = int(sys.argv[3]) if len(sys.argv) > 3 else 10 - count_after_s = float(sys.argv[4]) if len(sys.argv) > 4 else 0.0 - reliable = (sys.argv[5] if len(sys.argv) > 5 else "reliable") != "best_effort" + args = parse_args(sys.argv[1:]) + topic = args.topic + seconds = args.seconds + required = args.required + count_after_s = args.count_after + reliable = args.reliability == "reliable" rclpy.init() node = rclpy.create_node("countsub") diff --git a/components/rtps/interop/run_interop.sh b/components/rtps/interop/run_interop.sh index fd7a1fb55d..8d6db6e99f 100755 --- a/components/rtps/interop/run_interop.sh +++ b/components/rtps/interop/run_interop.sh @@ -43,7 +43,7 @@ lease_liveliness_cases() { # prefix log-tag local rospub=$! wait $sub local sub_rc=$? - kill $rospub 2>/dev/null; wait $rospub 2>/dev/null + kill $rospub 2>/dev/null; wait $rospub 2>/dev/null || true tail -2 "/tmp/${tag}leasesub.log" result "${prefix}ros2_pub->espp_sub_liveliness" $sub_rc @@ -51,10 +51,10 @@ lease_liveliness_cases() { # prefix log-tag "$BIN"/rtps_interop_pub rt/lease2 std_msgs::msg::dds_::String_ 1 220 200 > "/tmp/${tag}leasepub.log" 2>&1 & local pub=$! sleep 3 - timeout 40 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 34 10 22 reliable \ + timeout 40 python3 /work/components/rtps/interop/ros2_count_subscriber.py /lease2 --seconds 34 --required 10 --count-after 22 --reliability reliable \ > "/tmp/${tag}leasecount.log" 2>&1 local count_rc=$? - kill $pub 2>/dev/null; wait $pub 2>/dev/null + kill $pub 2>/dev/null; wait $pub 2>/dev/null || true tail -2 "/tmp/${tag}leasecount.log" result "${prefix}espp_pub->ros2_sub_liveliness" $count_rc } @@ -84,7 +84,7 @@ many_participants_case() { # prefix log-tag local pub=$! wait $sub local sub_rc=$? - kill $pub "${crowd[@]}" 2>/dev/null; wait $pub "${crowd[@]}" 2>/dev/null + kill $pub "${crowd[@]}" 2>/dev/null; wait $pub "${crowd[@]}" 2>/dev/null || true tail -2 "/tmp/${tag}crowdsub.log" result "${prefix}many_participants_late_joiner" $sub_rc } diff --git a/pc/tests/rtps_interop_sub.cpp b/pc/tests/rtps_interop_sub.cpp index 3b00237509..7b9d5bbbe0 100644 --- a/pc/tests/rtps_interop_sub.cpp +++ b/pc/tests/rtps_interop_sub.cpp @@ -15,8 +15,10 @@ // announced via its SEDP per-endpoint unicast locator - the interop matrix uses // this to prove a FastDDS/ROS 2 peer honors that locator and delivers the // topic's traffic to the dedicated port. -// count_after_s (default 0): samples that arrive earlier than this many seconds -// after start are reported but NOT counted toward `required`. The interop +// count_after_s (default 0, must be >= 0 - a negative value is a usage error +// and exits 1): samples that arrive earlier than this many seconds after start +// are reported (and, in large-payload mode, still verified byte-exact) but +// NOT counted toward `required`. The interop // matrix uses this as a liveliness gate: a peer whose participant we dropped // (e.g. its lease expired because its re-announcements were not delivered) // stops being matched, so nothing counts after the lease - only samples that @@ -68,6 +70,10 @@ int main(int argc, char **argv) { const std::size_t payload_bytes = (argc > 7) ? std::strtoul(argv[7], nullptr, 10) : 0; const int band_arg = (argc > 8) ? std::atoi(argv[8]) : static_cast(espp::QosBand::Normal); const int count_after_s = (argc > 9) ? std::atoi(argv[9]) : 0; + if (count_after_s < 0) { + std::printf("FAIL: count_after_s must be >= 0, got %d\n", count_after_s); + return 1; + } // Validate before casting: an out-of-range value would index band arrays. if (band_arg < static_cast(espp::QosBand::Critical) || band_arg > static_cast(espp::QosBand::Low)) { @@ -103,12 +109,9 @@ int main(int argc, char **argv) { if (!msg) { return; } - if (std::chrono::steady_clock::now() < count_from) { - const int n = early.fetch_add(1) + 1; - std::printf("received (not counted yet) %d\n", n); - std::fflush(stdout); - return; - } + // Early samples (before count_after_s) are verified and + // reported exactly like counted ones; only the count differs. + const bool counts = std::chrono::steady_clock::now() >= count_from; if (payload_bytes > 0) { // Large-payload mode: only a byte-exact reassembly counts. const bool exact = msg->data.size() == payload_bytes && msg->data == expected; @@ -118,10 +121,23 @@ int main(int argc, char **argv) { std::fflush(stdout); return; } + if (!counts) { + const int n = early.fetch_add(1) + 1; + std::printf("received (not counted yet) %d: %zu bytes byte-exact\n", n, + msg->data.size()); + std::fflush(stdout); + return; + } const int n = received.fetch_add(1) + 1; std::printf("received %d: %zu bytes byte-exact\n", n, msg->data.size()); std::fflush(stdout); } else { + if (!counts) { + const int n = early.fetch_add(1) + 1; + std::printf("received (not counted yet) %d: '%s'\n", n, msg->data.c_str()); + std::fflush(stdout); + return; + } const int n = received.fetch_add(1) + 1; std::printf("received %d: '%s'\n", n, msg->data.c_str()); std::fflush(stdout);