feat(rtps): endpoint priority — banded channels, dedicated SEDP-advertised ports, deferred dispatch (phase 2)

[finger563]

Description

Phase 2 of priority-aware scheduling (#735 was phase 1): wires espp::QosBand into the RTPS stack so a high-priority subscriber/service actually preempts lower-priority traffic, end to end — from the socket to the user callback. Wire format is untouched (goldens byte-identical); the one intentional default-behavior change is called out below.

Per-channel transport bands

rtps::ChannelOptions{band, dscp} on the transport's receive ports; DomainConfig{metatraffic_band, user_traffic_band, enable_dedicated_endpoint_ports, max_prioritized_endpoint_ports} exposed on RtpsParticipant::Config. Metatraffic (SPDP/SEDP) now dispatches at QosBand::High by default so discovery stays responsive under user-traffic floods — the only default-behavior change; everything else is opt-in.

Per-endpoint priority via dedicated ports (the real enabler)

All user traffic for a participant normally shares ONE unicast port (demux happens in-engine), so socket-level priority alone can't distinguish endpoints. An endpoint configured with a non-default band (or a dscp) now gets:

• its own UDP port at 7400 + 250·domain + 100 + n (linear-probed, reuse-disabled bind; standard ports stay below the +100 offset for participant ids 0–44, and the range stays inside the domain's port block), registered on the reactor at the endpoint's band and optionally DSCP-marked (Socket::set_dscp);
• its SEDP announcement carrying that port via the standard PID_UNICAST_LOCATOR (already parsed/serialized by the engine — no new wire construct), which FastDDS/ROS 2 honor, so peers send that endpoint's traffic straight to the banded socket; outbound traffic also leaves from it (m_srcPort follows the locator);
• fd-budget rationing: max_prioritized_endpoint_ports (default 4, lwIP has ~10 sockets) — on exhaustion, warn + fall back to deferred dispatch below. Ports are released on endpoint deletion.

Deferred banded dispatch (shared-port fallback)

Banded endpoints without a dedicated port get their user callbacks re-submitted to the pool at their band instead of running inline on the receive worker: bounded queue (32/endpoint, newest-dropped with warn), single in-flight drain job so per-endpoint ordering is preserved (mirrors the reactor's one-shot pattern). Applies to reader on_sample, service-server handlers, and service-client reply delivery; default (Normal) endpoints keep today's inline path byte-for-byte.

Facade + python

band + optional dscp appended to Writer/Reader/Service/Action configs and all typed facades (a service applies them to request+reply; an action inherits to all sub-endpoints — a ROS action is ~8 endpoints, more than the default ration, so most defer). Python rtps_bindings.cpp: participant config fields + band/dscp kwargs on all creation methods.

Testing

• Wire safety: rtps_golden byte-identical; all 18 pre-existing rtps host suites + socket_reactor + thread_pool pass — 66/66 runs (22 binaries × 3), independently re-verified.
• New suites (×3 each): rtps_sedp_dedicated_locator (byte-exact locator param, port range, ration, release/reuse, disabled mode), rtps_banded_pubsub (delivery proves the dedicated socket received — the announcement carries only that locator), rtps_banded_deferred (30/30 strictly in order), rtps_banded_ration (cap=1, both readers deliver).
• Docker interop matrix: 32/32 PASS, including a new ros2_pub→espp_banded_sub entry — a QosBand::High espp reader on a dedicated port receiving from a live ROS 2 Jazzy / rmw_fastrtps publisher.
• esp32 rtps example builds; python config surface smoke-tested; cppcheck (CI flags) clean.

Reviewer notes

• Dedicated-port offsets only walk forward (released ports free the ration slot, not the offset) — ~150 banded-endpoint creations per Domain lifetime exhaust the range; documented.
• Deferred-queue drops are post-ack (not recovered by RTPS reliability) and only occur if a user callback stalls past 32 queued samples; warned on drop.
• Deferred closures capture shared_ptr<vector> payloads to work around a GCC 15 xtensa -Werror=free-nonheap-object false positive on moved-vector captures.

Phase 3 (flood-latency p99 acceptance test) remains as follow-up.

🤖 Generated with Claude Code

[github-actions]

✅Static analysis result - no issues found! ✅

[Copilot]

Pull request overview

This PR implements Phase 2 of priority-aware RTPS scheduling by threading espp::QosBand (and optional DSCP) through the RTPS transport, adding per-endpoint dedicated unicast ports (announced via standard SEDP PID_UNICAST_LOCATOR), and providing a shared-port fallback via deferred banded dispatch for user callbacks.

Changes:

• Add per-channel QosBand scheduling to RTPS transport receive ports, with metatraffic defaulting to High.
• Implement per-endpoint dedicated unicast ports (rationed) and route inbound dedicated-port traffic to the owning participant.
• Add deferred banded dispatch for banded shared-port readers / service handlers / client replies, plus Python/docs/tests/interop coverage.

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
pc/tests/rtps_sedp_dedicated_locator.cpp	New engine-level test validating dedicated-port allocation + SEDP locator encoding + ration + release.
pc/tests/rtps_interop_sub.cpp	Extend interop subscriber tool to accept a band argument and pass it into the reader config.
pc/tests/rtps_banded_ration.cpp	New end-to-end test proving ration exhaustion falls back while still delivering to both readers.
pc/tests/rtps_banded_pubsub.cpp	New loopback test proving traffic flows via the dedicated port announced in SEDP.
pc/tests/rtps_banded_deferred.cpp	New loopback test proving deferred shared-port dispatch preserves per-reader ordering.
lib/python_bindings/rtps_bindings.cpp	Expose participant config fields + band/dscp on endpoint creation APIs in Python.
doc/en/protocols/rtps.rst	Document banded channels, dedicated endpoint ports, and deferred dispatch behavior.
components/rtps/src/rtps_participant.cpp	Wire participant config into rtps::DomainConfig, pass endpoint options, and implement deferred dispatch helper + usage.
components/rtps/src/entities/Domain.cpp	Add DomainConfig, per-channel banding, dedicated-port allocation/release, and dedicated-port receive routing.
components/rtps/src/communication/EsppTransport.cpp	Add per-channel ChannelOptions (band/dscp) and band-aware submit() to the transport worker pool.
components/rtps/README.md	Add a “Priority scheduling” section describing bands, dedicated ports, DSCP, and deferred dispatch.
components/rtps/interop/run_interop.sh	Build and run new priority/dedicated-port tests; add ROS2→banded-espp interop scenario.
components/rtps/include/rtps/entities/Domain.hpp	Define DomainConfig + EndpointOptions, extend constructors/create APIs, expose getTransport().
components/rtps/include/rtps/discovery/TopicData.hpp	Add local-only band/dscp/hasDedicatedPort attributes (not serialized).
components/rtps/include/rtps/communication/EsppTransport.hpp	Define ChannelOptions; extend transport APIs for band/DSCP-aware channels and submissions.
components/rtps/include/rtps_service.hpp	Extend typed service facade configs to carry band/dscp into underlying endpoints.
components/rtps/include/rtps_pubsub.hpp	Extend typed publisher/subscriber facade configs to carry band/dscp.
components/rtps/include/rtps_participant.hpp	Extend public configs (participant/writer/reader/service/action) and define deferred dispatch helper.
components/rtps/include/rtps_action.hpp	Extend typed action facade configs to carry band/dscp into underlying endpoints.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[finger563]

The CI interop hang is fixed in c573c39..ef323f4 — the 30-minute timeout kill turned out to be a real shutdown deadlock, and reproducing it on Linux (gdb watchdog inside the interop image) surfaced three distinct concurrency bugs, two of them pre-existing in the engine and merely exposed by runtime endpoint deletion:

1. Use-after-free wedging stop() (the CI signature): releaseReceivePort() destroyed the UdpSocket right after the non-blocking reactor remove() while an in-flight dispatch still referenced it — the handler blocked forever on the freed object's mutex, and SocketReactor::stop() spun on "Still waiting for 1 in-flight handler(s)". Fixed by retiring released sockets (fd stays open, so stale select() readiness can't alias a reused fd) and closing them in stop() after the reactor/pool quiesce.
2. Pre-existing Participant↔SEDPAgent ABBA deadlock (exposed by runtime deletion): endpoint add/delete held Participant::m_mutex while calling into the agent; agent receive handlers take the locks in the opposite order. The SEDP call now happens outside Participant::m_mutex (global order SEDP→Participant), and deletion loops match by pointer identity (also fixing a null-slot deref).
3. Pre-existing unlocked proxy-pool iteration (reproduced as a SIGSEGV under churn): StatefulWriter::sendHeartBeat and several reader/writer accessors iterated m_proxies without the designated mutex while SEDP workers mutated it — all guarded now.

Defense-in-depth: the reactor's UDP receive is bounded with a 1 s SO_RCVTIMEO (Linux-documented spurious readiness + unbounded recvfrom could wedge stop even without the UAF), and the dedicated-port registry got its own mutex off the Domain::m_mutex hot path.

Evidence: new rtps_banded_churn stress test (create/receive/delete churn under flood + stop-under-deferred-load) reproduced hang@iter5 / deadlock@iter8 / segv@iter6 before the fixes and now passes 100/100 on Linux; the four banded suites pass 30/30 each on Linux; the full docker interop matrix is 33/33 PASS (including a new banded_churn entry); local sweep 75/75; esp32 rtps + socket examples build. Note for review: the engine lock-order and proxy-guard fixes touch shared code paths used by all traffic — the interop matrix and goldens are the regression evidence.

[Copilot]

Pull request overview

Copilot reviewed 27 out of 27 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

pc/tests/rtps_banded_deferred.cpp:76

• The comment says each sample is sent until it is "acknowledged by observation" via paced resends of the head, but the loop actually sends each sequence number at most once (next_to_send is always incremented on a successful publish and never retried). Either adjust the comment to match the current send strategy, or implement the described resend behavior.

  // Wait for discovery/matching, then send the numbered sequence. Reliable
  // writers retransmit on the wire, but a sample must reach the reader at least
  // once for the deferred queue to see it - send each one until acknowledged by
  // observation (simple paced resend of the not-yet-seen head).

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 4 comments.

Suppressed comments (3)

Previously missed (2) — in code that hasn't changed since the last review.

components/rtps/src/rtps_participant.cpp:673

• These two endpoint creations are not transactional. If the writer succeeds (and consumes a dedicated-port slot) but the reader pool is exhausted, the method returns false while leaving the writer announced and its dedicated socket allocated; the inverse case similarly leaves a reader behind. Roll back whichever endpoint was created before returning failure so failed service registration cannot permanently consume endpoint and fd budgets.

This issue also appears on line 1172 of the same file.

      domain_->createWriter(*participant_, rep_topic.c_str(), rep_type.c_str(), /*reliable=*/true,
                            /*enforceUnicast=*/false, endpoint_options);
  rtps::Reader *request_reader =
      domain_->createReader(*participant_, req_topic.c_str(), req_type.c_str(), /*reliable=*/true,
                            /*mcastaddress=*/{0, 0, 0, 0}, endpoint_options);

components/rtps/src/entities/Domain.cpp:413

• A failed probe window never advances m_nextDedicatedPortOffset; it is updated only on a successful bind. If offsets 0–15 are occupied while offset 16 is free, every later endpoint retries the same 16 ports and dedicated allocation remains permanently disabled despite the documented 100–249 range. Advance past the failed window (bounded by the domain block) before returning fallback.

  for (uint16_t probe = 0; probe < DEDICATED_PORT_PROBE_LIMIT; ++probe) {
    const uint16_t offset = m_nextDedicatedPortOffset + probe;
    if (DEDICATED_PORT_OFFSET + offset > 249) {
      break; // stay inside this domain's 250-port block

components/rtps/src/rtps_participant.cpp:1176

• As in the server path, partial creation leaks the successful endpoint. For example, a successful dedicated reply reader followed by request-writer exhaustion makes add_service_client() fail while retaining the reader/socket and ration slot. Delete whichever endpoint succeeded before returning failure.

      domain_->createReader(*participant_, rep_topic.c_str(), rep_type.c_str(), /*reliable=*/true,
                            /*mcastaddress=*/{0, 0, 0, 0}, endpoint_options);
  rtps::Writer *request_writer =
      domain_->createWriter(*participant_, req_topic.c_str(), req_type.c_str(), /*reliable=*/true,
                            /*enforceUnicast=*/false, endpoint_options);

[finger563]

Also addressed the two "suppressed" Copilot comments from the latest review (no threads to resolve):

• Non-transactional service/action creation (rtps_participant.cpp:673 / 1176): fixed in 2663af3 — the ROS service server (writer→reader) and client (reader→writer) paths now delete the surviving endpoint on partial failure (and both on callback-registration failure), native service pairs roll back via new remove_writer/remove_reader helpers, and all four action composites (ROS/native × server/client) unwind every endpoint created past a container mark when a later step fails — so a failed registration can no longer leak an SEDP announcement, a dedicated-port ration slot, or an endpoint pool slot. Proven by the new rtps_service_rollback suite: after an induced partial failure, the rolled-back banded endpoint's dedicated port becomes externally bindable (fd + ration released), and an action failed partway through releases exactly its consumed writer-budget slots (verified by refilling them).
• Stuck probe window (Domain.cpp:413): fixed in 46a01fd — a fully-failed probe window now advances m_nextDedicatedPortOffset past the probed range (bounded by the domain block) before falling back, so an occupied window no longer permanently disables dedicated allocation; proven by the occupied-window scope in rtps_sedp_dedicated_locator (16 external squatters → first allocation falls back, second succeeds past the window).

Full regression drill on the final binaries: rtps_banded_churn 100/100 on Linux docker (now with the live-sample requirement), the five banded/rollback suites 30/30 each, local sweep 78/78, docker interop matrix 34/34 PASS (now includes service_rollback), esp32 rtps + socket examples build, cppcheck clean.

[Copilot]

Pull request overview

Copilot reviewed 30 out of 30 changed files in this pull request and generated 7 comments.

[Copilot]

Pull request overview

Copilot reviewed 33 out of 33 changed files in this pull request and generated 4 comments.

Suppressed comments (4)

components/rtps/src/rtps_participant.cpp:310

• close() is irreversible, but it runs before deleteReader() is known to have succeeded. If SEDP deletion returns false, this method keeps the still-active reader/context for retry while permanently dropping all future deferred deliveries. Delete the engine reader first, then close the dispatcher only after deletion succeeds.

      (*it)->deferred.close();
      if (!domain_->deleteReader(*participant_, (*it)->reader)) {
        return false;
      }

components/rtps/src/rtps_participant.cpp:1068

• This removes the client's owning registry entry before checking whether its reply reader was actually deleted. On a deletion failure, that live reader still invokes service_reply_trampoline with impl_.get(), but rollback soon destroys the client/Impl, leaving a dangling callback. Keep the client registered until both endpoint deletions succeed, with partial-deletion state retained for retry.

  {
    std::lock_guard<std::mutex> lock(mutex_);
    std::erase(service_clients_, client);
  }
  client->impl_->deferred.close();

components/rtps/src/rtps_participant.cpp:1587

• Erasing this context before remove_reader() succeeds can leave the facade reader alive with its callback lambda still capturing the raw NativeServiceServerContext* created at line 1639. When the helper returns and the last shared pointer is released, later requests use freed memory. Only erase the context after the reader and writer removals have succeeded, retaining it on failure.

  {
    // Remove exactly THIS handle - a concurrently added server is untouched.
    std::lock_guard<std::mutex> lock(mutex_);
    std::erase(native_service_servers_, server);
  }

components/rtps/src/rtps_participant.cpp:1603

• The reply-reader callback captures NativeServiceClient::Impl* by raw pointer, so erasing the owning client before remove_reader() is confirmed creates a use-after-free if that removal fails. Retain the client in native_service_clients_ until both endpoint removals succeed and preserve partial-cleanup state for retry.

  {
    std::lock_guard<std::mutex> lock(mutex_);
    std::erase(native_service_clients_, client);
  }
  remove_reader(client->impl_->reply_topic);

[finger563]

Also addressed the four "suppressed" Copilot comments from the latest review — all part of the same engine-first removal-ordering fix in 0b63854 (no threads to resolve):

• rtps_participant.cpp:310 (remove_reader closed the dispatcher before deleteReader confirmed): now deletes the engine reader FIRST and calls deferred.close() (irreversible) only after deletion succeeds; a failed deletion leaves reader + dispatcher fully functional for retry.
• rtps_participant.cpp:1068 (ROS service client — registry erase before reply-reader deletion): reply reader deleted first, impl_->reply_reader nulled, registry entry erased only on confirmed deletion of both endpoints.
• rtps_participant.cpp:1587 (native service server — context erased before remove_reader success): remove_reader/remove_writer run first (they themselves only mutate facade state on confirmed engine deletion), registry erased only after both succeed, with request_removed/reply_removed markers for retry.
• rtps_participant.cpp:1603 (native service client — same for Impl*): reply reader removed and confirmed before the client is unregistered.

The invariant is now uniform across every removal/rollback path: delete the engine endpoint, then mutate facade state only on confirmed success; on failure every handle and the callbacks/dispatchers it anchors stay alive for retry. Verified: host sweep 75/75 (25 binaries x3), the two new tests 10/10 each, cppcheck clean, esp32 rtps example builds, and the full docker interop matrix 35/35 (all service/action paths exercise these removal orderings).