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

.github/workflows/host_sanitizers.yml

@@ -0,0 +1,119 @@
+name: Host sanitizers (TSan / ASan)
+
+# Build the host library + pc test suite under ThreadSanitizer and
+# AddressSanitizer and run the standalone (non-interop) tests. The RTPS stack
+# is heavily concurrent (transport worker pool, socket reactor, protocol
+# scheduler, deferred dispatchers), and most defects found in review have been
+# data races, deadlocks, and use-after-frees - exactly the classes these
+# sanitizers detect mechanically. Runs whenever RTPS or its concurrency
+# building blocks change.
+
+# Minimal token scope: build + run tests, never writes.
+permissions:
+  contents: read
+
+on:
+  pull_request:
+    paths:
+      - "components/rtps/**"
+      - "components/socket/**"
+      - "components/thread_pool/**"
+      - "components/task/**"
+      - "components/timer/**"
+      - "components/cdr/**"
+      - "lib/espp.cmake"
+      - "pc/tests/rtps_*"
+      - "pc/tests/socket_reactor.cpp"
+      - "pc/tests/thread_pool.cpp"
+      - ".github/workflows/host_sanitizers.yml"
+  workflow_dispatch:
+
+# Supersede in-progress runs (same rationale as rtps_interop.yml): keyed by
+# workflow + PR number; never runs on push to main, so cancel is always safe.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  sanitize:
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - sanitizer: thread
+            cflags: "-fsanitize=thread -fno-omit-frame-pointer"
+            ldflags: "-fsanitize=thread"
+          - sanitizer: address
+            cflags: "-fsanitize=address -fno-omit-frame-pointer"
+            ldflags: "-fsanitize=address"
+    name: ${{ matrix.sanitizer }}
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          submodules: "recursive"
+
+      - name: Build library (${{ matrix.sanitizer }})
+        run: |
+          cmake -S lib -B lib/build \
+            -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+            -DESPP_INSTALL=ON \
+            -DESPP_BUILD_PYTHON=OFF \
+            -DCMAKE_INSTALL_PREFIX="$PWD/install" \
+            -DCMAKE_C_FLAGS="${{ matrix.cflags }}" \
+            -DCMAKE_CXX_FLAGS="${{ matrix.cflags }}" \
+            -DCMAKE_EXE_LINKER_FLAGS="${{ matrix.ldflags }}" \
+            -DCMAKE_SHARED_LINKER_FLAGS="${{ matrix.ldflags }}"
+          cmake --build lib/build --target install --parallel 4
+
+      - name: Build pc tests (${{ matrix.sanitizer }})
+        run: |
+          cmake -S pc -B pc/build \
+            -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+            -DCMAKE_PREFIX_PATH="$PWD/install" \
+            -DCMAKE_CXX_FLAGS="${{ matrix.cflags }}" \
+            -DCMAKE_EXE_LINKER_FLAGS="${{ matrix.ldflags }}"
+          cmake --build pc/build --parallel 4
+
+      - name: Run standalone tests under ${{ matrix.sanitizer }} sanitizer
+        run: |
+          # halt_on_error=0: report every finding, fail via the exit code.
+          # detect_leaks=0: the engine deliberately holds pooled/static
+          # allocations for its lifetime; LSan end-of-process reports would be
+          # noise. ASan still catches use-after-free / overflow.
+          #
+          # The lock-order (deadlock) detector is ENABLED. Its first runs found
+          # four real ABBA cycles, all now fixed: checkAndResetHeartbeats
+          # agent->participant order, addBuiltInEndpoints lock scope,
+          # executeCallbacks snapshot-invoke, and StatefulReader delivering
+          # user callbacks under a leaf delivery mutex instead of the proxies
+          # mutex (which user callbacks could cycle back into via the
+          # facade/SEDP lock chain).
+          export TSAN_OPTIONS="second_deadlock_stack=1"
+          export ASAN_OPTIONS="detect_leaks=0"
+          fails=0
+          # The interop client/server binaries need a FastDDS/ROS 2 peer (the
+          # docker matrix covers them); everything else runs standalone. The
+          # generous per-test timeout absorbs the sanitizer's slowdown; a hang
+          # (e.g. a deadlock TSan cannot see) still fails the job.
+          for t in pc/build/rtps_* pc/build/socket_reactor pc/build/thread_pool; do
+            [ -x "$t" ] || continue
+            name=$(basename "$t")
+            case "$name" in *interop*) continue;; esac
+            echo "::group::$name"
+            if timeout 300 "$t"; then
+              echo "PASS: $name"
+            else
+              rc=$?
+              echo "FAIL: $name (exit $rc)"
+              fails=$((fails+1))
+            fi
+            echo "::endgroup::"
+          done
+          echo "==================== ${{ matrix.sanitizer }} summary ===================="
+          if [ "$fails" -ne 0 ]; then
+            echo "FAILED: $fails test(s) under ${{ matrix.sanitizer }} sanitizer"
+            exit 1
+          fi
+          echo "ALL PASS under ${{ matrix.sanitizer }} sanitizer"

.gitignore

@@ -64,3 +64,8 @@ docs/
 
 # Local example-build helper script (not part of the repo).
 build_examples.sh
+
+# local sanitizer / variant build trees (host TSan/ASan, static-storage validation)
+lib/build-*/
+pc/build-*/
+install-*/

components/rtps/CMakeLists.txt

@@ -20,7 +20,7 @@ idf_component_register(
   INCLUDE_DIRS
     "include"
   REQUIRES
-    base_component cdr task thread_pool socket
+    base_component cdr task thread_pool timer socket
 )
 
 # Select the RTPS static-limits profile from Kconfig (see Kconfig in this
@@ -40,6 +40,118 @@ endif()
 # limits profile never silently switches the MCU to heap-backed history. The
 # limits headers no longer define RTPS_STORAGE_DYNAMIC themselves; it is set here
 # on ESP and defaulted on in config.hpp for host/PC builds.
+# Per-limit capacity overrides (Kconfig "Custom capacity overrides"): a
+# nonzero value overrides that single cap of the selected profile via its
+# RTPS_CFG_* macro. PUBLIC so application translation units see the same
+# values as the engine (the pools are sized in this component's sources).
+set(RTPS_LIMIT_KNOBS
+    NUM_STATELESS_WRITERS
+    NUM_STATELESS_READERS
+    NUM_STATEFUL_WRITERS
+    NUM_STATEFUL_READERS
+    MAX_NUM_PARTICIPANTS
+    NUM_WRITERS_PER_PARTICIPANT
+    NUM_READERS_PER_PARTICIPANT
+    NUM_WRITER_PROXIES_PER_READER
+    NUM_READER_PROXIES_PER_WRITER
+    MAX_NUM_UNMATCHED_REMOTE_WRITERS
+    MAX_NUM_UNMATCHED_REMOTE_READERS
+    MAX_NUM_READER_CALLBACKS
+    HISTORY_SIZE_STATELESS
+    HISTORY_SIZE_STATEFUL
+    MAX_TYPENAME_LENGTH
+    MAX_TOPICNAME_LENGTH
+    MAX_NUM_UDP_CONNECTIONS
+)
+foreach(knob ${RTPS_LIMIT_KNOBS})
+  if(DEFINED CONFIG_RTPS_LIMIT_${knob} AND NOT "${CONFIG_RTPS_LIMIT_${knob}}" STREQUAL "" AND NOT "${CONFIG_RTPS_LIMIT_${knob}}" STREQUAL "0")
+    # Builtin-aware minima (0 stays the "use profile default" sentinel): the
+    # discovery endpoints draw from these pools, so a nonzero value below the
+    # reservation would crash startup (the second SEDP allocation returns null)
+    # or silently omit the builtins (per-participant caps < 3).
+    if((knob STREQUAL "NUM_STATEFUL_WRITERS" OR knob STREQUAL "NUM_STATEFUL_READERS")
+       AND CONFIG_RTPS_LIMIT_${knob} LESS 2)
+      message(FATAL_ERROR
+        "CONFIG_RTPS_LIMIT_${knob}=${CONFIG_RTPS_LIMIT_${knob}} is below the minimum of 2 "
+        "(2 slots are reserved for the builtin SEDP endpoints)")
+    endif()
+    if((knob STREQUAL "NUM_WRITERS_PER_PARTICIPANT" OR knob STREQUAL "NUM_READERS_PER_PARTICIPANT")
+       AND CONFIG_RTPS_LIMIT_${knob} LESS 3)
+      message(FATAL_ERROR
+        "CONFIG_RTPS_LIMIT_${knob}=${CONFIG_RTPS_LIMIT_${knob}} is below the minimum of 3 "
+        "(the 3 builtin discovery endpoints count against the per-participant cap)")
+    endif()
+    if(knob STREQUAL "MAX_NUM_UDP_CONNECTIONS" AND CONFIG_RTPS_LIMIT_${knob} LESS 4)
+      message(FATAL_ERROR
+        "CONFIG_RTPS_LIMIT_${knob}=${CONFIG_RTPS_LIMIT_${knob}} is below the minimum of 4 "
+        "(2 shared multicast + 2 unicast channels for a single participant)")
+    endif()
+    target_compile_definitions(${COMPONENT_LIB} PUBLIC "RTPS_CFG_${knob}=${CONFIG_RTPS_LIMIT_${knob}}")
+    set(RTPS_EFFECTIVE_${knob} ${CONFIG_RTPS_LIMIT_${knob}})
+  endif()
+endforeach()
+
+# Cross-limit capacity check (mirrors lib/espp.cmake): every participant
+# consumes 1 stateless writer/reader + 2 stateful writers/readers from the
+# GLOBAL pools for its builtin discovery endpoints. Effective value = the
+# Kconfig override if nonzero, else the selected profile's default.
+if(CONFIG_RTPS_LIMITS_PROFILE_HOST)
+  set(_rtps_defaults 8 16 16 32 32 24)
+elseif(CONFIG_RTPS_LIMITS_PROFILE_HOST_LARGE)
+  set(_rtps_defaults 32 64 64 128 128 72)
+else() # embedded (default)
+  set(_rtps_defaults 1 5 5 5 5 10)
+endif()
+list(GET _rtps_defaults 0 _rtps_d_participants)
+list(GET _rtps_defaults 1 _rtps_d_stateless_w)
+list(GET _rtps_defaults 2 _rtps_d_stateless_r)
+list(GET _rtps_defaults 3 _rtps_d_stateful_w)
+list(GET _rtps_defaults 4 _rtps_d_stateful_r)
+list(GET _rtps_defaults 5 _rtps_d_channels)
+foreach(pair
+    "MAX_NUM_PARTICIPANTS;_rtps_d_participants"
+    "NUM_STATELESS_WRITERS;_rtps_d_stateless_w"
+    "NUM_STATELESS_READERS;_rtps_d_stateless_r"
+    "NUM_STATEFUL_WRITERS;_rtps_d_stateful_w"
+    "NUM_STATEFUL_READERS;_rtps_d_stateful_r"
+    "MAX_NUM_UDP_CONNECTIONS;_rtps_d_channels")
+  list(GET pair 0 _rtps_k)
+  list(GET pair 1 _rtps_dvar)
+  if(NOT DEFINED RTPS_EFFECTIVE_${_rtps_k})
+    set(RTPS_EFFECTIVE_${_rtps_k} ${${_rtps_dvar}})
+  endif()
+endforeach()
+math(EXPR _rtps_need_stateful "2 * ${RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS}")
+if(RTPS_EFFECTIVE_NUM_STATELESS_WRITERS LESS RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS
+   OR RTPS_EFFECTIVE_NUM_STATELESS_READERS LESS RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS
+   OR RTPS_EFFECTIVE_NUM_STATEFUL_WRITERS LESS _rtps_need_stateful
+   OR RTPS_EFFECTIVE_NUM_STATEFUL_READERS LESS _rtps_need_stateful)
+  message(FATAL_ERROR
+    "RTPS limits cannot host the participant budget: MAX_NUM_PARTICIPANTS="
+    "${RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS} needs >= ${RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS} "
+    "stateless writers/readers (have ${RTPS_EFFECTIVE_NUM_STATELESS_WRITERS}/"
+    "${RTPS_EFFECTIVE_NUM_STATELESS_READERS}) and >= ${_rtps_need_stateful} stateful "
+    "writers/readers (have ${RTPS_EFFECTIVE_NUM_STATEFUL_WRITERS}/"
+    "${RTPS_EFFECTIVE_NUM_STATEFUL_READERS}) for the builtin discovery endpoints "
+    "(1 stateless W/R + 2 stateful W/R per participant). Raise the pool overrides "
+    "(menuconfig 'Custom capacity overrides') or lower MAX_NUM_PARTICIPANTS.")
+endif()
+# Channel-pool constraint (mirrors lib/espp.cmake): 2 shared multicast + 2
+# unicast channels per participant are permanently bound from the same
+# MAX_NUM_UDP_CONNECTIONS pool that runtime dedicated endpoint ports draw
+# from; without this check a combination can pass the endpoint-pool math yet
+# createParticipant() still fails on channels first.
+math(EXPR _rtps_need_channels "2 + 2 * ${RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS}")
+if(RTPS_EFFECTIVE_MAX_NUM_UDP_CONNECTIONS LESS _rtps_need_channels)
+  message(FATAL_ERROR
+    "RTPS limits cannot host the participant budget: MAX_NUM_PARTICIPANTS="
+    "${RTPS_EFFECTIVE_MAX_NUM_PARTICIPANTS} needs >= ${_rtps_need_channels} transport channels "
+    "(2 shared multicast + 2 unicast per participant) but MAX_NUM_UDP_CONNECTIONS is "
+    "${RTPS_EFFECTIVE_MAX_NUM_UDP_CONNECTIONS}. Raise MAX_NUM_UDP_CONNECTIONS via menuconfig "
+    "'Custom capacity overrides' (leave headroom for dedicated endpoint ports, which draw "
+    "from the same pool at runtime) or lower MAX_NUM_PARTICIPANTS.")
+endif()
+
 if(CONFIG_RTPS_STORAGE_DYNAMIC)
   target_compile_definitions(${COMPONENT_LIB} PUBLIC RTPS_STORAGE_DYNAMIC)
 endif()