p2p robustness and scalability by gmartin82 · Pull Request #1222 · eclipse-zenoh/zenoh-pico

gmartin82 · 2026-05-14T13:12:28Z

Description

Work in progress toward consistently starting interconnected peer zenoh-pico nodes at larger scales.

What does this PR do?

Make the listen connection limit configurable via CMake
Add connect timeout support to the C11 z_pub / z_sub examples
Update the peer mesh run script to use configurable connect timeout
Speed up run script log scanning and report time to full observed connectivity

Why is this change needed?

The peer-to-peer stress script needs to exercise larger interconnected zenoh-pico meshes, ideally up to 50 nodes, without relying on hardcoded connection limits or slow log parsing. This PR starts improving that workflow by making the listen limit configurable, allowing examples to keep retrying peer connects, and making the script report the actual time until full observed connectivity.

Related Issues

🏷️ Label-Based Checklist

Based on the labels applied to this PR, please complete these additional requirements:

Labels: internal

🏠 Internal Change

This PR is marked as internal (not user-facing):

No API changes - Public APIs unchanged
No behavior changes - External behavior identical
Refactoring/maintenance - Code improvements only
Tests still pass - All existing tests pass without modification

Lighter review: Internal changes may have lighter review requirements.

Instructions:

Check off items as you complete them (change - [ ] to - [x])
The PR checklist CI will verify these are completed

This checklist updates automatically when labels change, but preserves your checked boxes.

Work in progress toward consistently starting interconnected peer zenoh-pico nodes at larger scales. - Make the listen connection limit configurable via CMake - Add connect timeout support to the C11 z_pub/z_sub examples - Update the peer mesh run script to use configurable connect timeout - Speed up run script log scanning and report time to full observed connectivity

@@ -0,0 +1,79 @@
+nb=${1:-5}


+connect_timeout=${3:--1}
+
+echo "" > log
+for i in $(seq -f "%03g" 1 1 $nb 2>/dev/null)


+  do
+    connect="$connect -e tcp/127.0.0.1:8$j -e tcp/127.0.0.1:9$j"
+  done
+  stdbuf -o0 ./build/examples/z_pub -m peer -l tcp/127.0.0.1:8$i -t $connect_timeout $connect -k demo/example/$i | while read line; do echo "[pub $i][$(date +%s.%N)] $line"; done >> log &


+  do
+    connect="$connect -e tcp/127.0.0.1:8$j -e tcp/127.0.0.1:9$j"
+  done
+  stdbuf -o0 ./build/examples/z_pub -m peer -l tcp/127.0.0.1:8$i -t $connect_timeout $connect -k demo/example/$i | while read line; do echo "[pub $i][$(date +%s.%N)] $line"; done >> log &


+    connect="$connect -e tcp/127.0.0.1:8$j -e tcp/127.0.0.1:9$j"
+  done
+  stdbuf -o0 ./build/examples/z_pub -m peer -l tcp/127.0.0.1:8$i -t $connect_timeout $connect -k demo/example/$i | while read line; do echo "[pub $i][$(date +%s.%N)] $line"; done >> log &
+  stdbuf -o0 ./build/examples/z_sub -m peer -l tcp/127.0.0.1:9$i -t $connect_timeout -e tcp/127.0.0.1:8$i $connect | while read line; do echo "[sub $i][$(date +%s.%N)] $line"; done >> log &


+    connect="$connect -e tcp/127.0.0.1:8$j -e tcp/127.0.0.1:9$j"
+  done
+  stdbuf -o0 ./build/examples/z_pub -m peer -l tcp/127.0.0.1:8$i -t $connect_timeout $connect -k demo/example/$i | while read line; do echo "[pub $i][$(date +%s.%N)] $line"; done >> log &
+  stdbuf -o0 ./build/examples/z_sub -m peer -l tcp/127.0.0.1:9$i -t $connect_timeout -e tcp/127.0.0.1:8$i $connect | while read line; do echo "[sub $i][$(date +%s.%N)] $line"; done >> log &


+  stdbuf -o0 ./build/examples/z_sub -m peer -l tcp/127.0.0.1:9$i -t $connect_timeout -e tcp/127.0.0.1:8$i $connect | while read line; do echo "[sub $i][$(date +%s.%N)] $line"; done >> log &
+done
+
+sleep $duration


+
+if [[ $stop != "" ]]
+then
+  echo OK $(($stop - $start)) seconds


+then
+  echo OK $(($stop - $start)) seconds
+else
+  echo KO $failure


                      bool *add_matching_listener) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:a")) != -1) {
+    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:at:")) != -1) {


 static int parse_args(int argc, char **argv, z_owned_config_t *config, char **ke, int *n) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:e:m:l:n:")) != -1) {
+    while ((opt = getopt(argc, argv, "k:e:m:l:n:t:")) != -1) {


                      bool *add_matching_listener) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:a")) != -1) {
+    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:at:")) != -1) {


                      bool *add_matching_listener) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:a")) != -1) {
+    while ((opt = getopt(argc, argv, "k:v:e:m:l:n:at:")) != -1) {


                break;
+            case 't':
+#if defined(Z_FEATURE_UNSTABLE_API)
+                zp_config_insert(z_loan_mut(*config), Z_CONFIG_CONNECT_TIMEOUT_KEY, optarg);


            case '?':
                if (optopt == 'k' || optopt == 'v' || optopt == 'e' || optopt == 'm' || optopt == 'l' ||
-                    optopt == 'n') {
+                    optopt == 'n' || optopt == 't') {


 static int parse_args(int argc, char **argv, z_owned_config_t *config, char **ke, int *n) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:e:m:l:n:")) != -1) {
+    while ((opt = getopt(argc, argv, "k:e:m:l:n:t:")) != -1) {


 static int parse_args(int argc, char **argv, z_owned_config_t *config, char **ke, int *n) {
    int opt;
-    while ((opt = getopt(argc, argv, "k:e:m:l:n:")) != -1) {
+    while ((opt = getopt(argc, argv, "k:e:m:l:n:t:")) != -1) {


                break;
+            case 't':
+#if defined(Z_FEATURE_UNSTABLE_API)
+                zp_config_insert(z_loan_mut(*config), Z_CONFIG_CONNECT_TIMEOUT_KEY, optarg);


+                break;
            case '?':
-                if (optopt == 'k' || optopt == 'e' || optopt == 'm' || optopt == 'l' || optopt == 'n') {
+                if (optopt == 'k' || optopt == 'e' || optopt == 'm' || optopt == 'l' || optopt == 'n' ||


            case '?':
-                if (optopt == 'k' || optopt == 'e' || optopt == 'm' || optopt == 'l' || optopt == 'n') {
+                if (optopt == 'k' || optopt == 'e' || optopt == 'm' || optopt == 'l' || optopt == 'n' ||
+                    optopt == 't') {


-                                    const _z_config_t *config) {
-    if (!_z_pending_peers_has_pending(pending_peers)) {
+                                    const _z_config_t *config, size_t max_attempts) {
+    size_t pending_count = _z_pending_peers_count_pending(pending_peers);


+
+    for (size_t attempt = 0; attempt < loop_count; attempt++) {
+        size_t i = 0;
+        if (!_z_pending_peers_next_pending_idx(pending_peers, &i)) {


In single-thread mode, spin the async-opened session from only one test thread while waiting for peer discovery, avoiding concurrent zp_spin_once() calls on the same session.

Limit the background add-peers task to a small number of peer attempts per executor tick, rotating through pending locators before applying backoff. This prevents peer retry work from monopolizing single-thread runtimes while still retrying the full pending set before sleeping.

 }

 #if Z_FEATURE_UNICAST_PEER == 1
+#define _Z_ADD_PEERS_ALL_PENDING 0



 #if Z_FEATURE_UNICAST_PEER == 1
+#define _Z_ADD_PEERS_ALL_PENDING 0
+#define _Z_ADD_PEERS_TASK_MAX_ATTEMPTS 1


-            // Non-retryable error
+        bool has_pending = _z_pending_peers_has_pending(pending_peers);
+
+        if (result._last_non_retryable_ret != _Z_RES_OK) {


+
+        if (result._last_non_retryable_ret != _Z_RES_OK) {
+            // Non-retryable error.
            if (exit_on_failure) {


+            _z_pending_peers_clear(pending_peers);
+            return _Z_RES_OK;
+        }
+


+    if (pending_peers->_remaining_attempts == 0) {
+        pending_peers->_remaining_attempts = _z_pending_peers_count_pending(pending_peers);
+    }
+


+    if (pending_peers->_remaining_attempts > 0) {
+        return _z_fut_fn_result_continue();
+    }
+


gmartin82 added the internal Changes not included in the changelog label May 14, 2026

github-advanced-security Bot found potential problems May 14, 2026

View reviewed changes

gmartin82 force-pushed the ZEN-937 branch from 274e10c to 222a95d Compare May 14, 2026 15:49

github-advanced-security Bot found potential problems May 14, 2026

View reviewed changes

Comment thread src/transport/manager.c Fixed

Comment thread src/transport/manager.c Outdated

gmartin82 force-pushed the ZEN-937 branch from 6dc6f0f to 253f8a0 Compare May 14, 2026 17:47

gmartin82 added 2 commits May 15, 2026 11:25

Fix z_open single-thread async progress test

f85a1d1

In single-thread mode, spin the async-opened session from only one test thread while waiting for peer discovery, avoiding concurrent zp_spin_once() calls on the same session.

github-advanced-security Bot found potential problems May 15, 2026

View reviewed changes

Navigation Menu

Uh oh!

p2p robustness and scalability - #1222

p2p robustness and scalability#1222
gmartin82 wants to merge 3 commits into
eclipse-zenoh:mainfrom
gmartin82:ZEN-937

gmartin82 commented May 14, 2026 •

edited by github-actions Bot

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

gmartin82 commented May 14, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

What does this PR do?

Why is this change needed?

Related Issues

🏷️ Label-Based Checklist

🏠 Internal Change

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

gmartin82 commented May 14, 2026 •

edited by github-actions Bot

Loading