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..e8d742c278 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 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` | + +## 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..f77e2fce1e --- /dev/null +++ b/components/rtps/interop/ros2_count_subscriber.py @@ -0,0 +1,91 @@ +#!/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 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 + +import rclpy +from rclpy.qos import QoSProfile, ReliabilityPolicy +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: + 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") + 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..8d6db6e99f 100755 --- a/components/rtps/interop/run_interop.sh +++ b/components/rtps/interop/run_interop.sh @@ -22,6 +22,73 @@ 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 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 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 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=$? + 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 + + 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 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 || true + 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 || true + 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 +300,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=$! @@ -354,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 & @@ -395,10 +469,14 @@ 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 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 diff --git a/pc/tests/rtps_interop_sub.cpp b/pc/tests/rtps_interop_sub.cpp index aee4af1eda..7b9d5bbbe0 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,15 @@ // 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, 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 +// keep arriving past that point prove the peer stayed alive. +// Exits 0 once `required` (counted) samples arrive within `timeout_s`. #include #include @@ -61,6 +69,11 @@ 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; + 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)) { @@ -72,6 +85,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 +103,15 @@ 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; } + // 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; @@ -101,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); @@ -115,17 +148,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; }