Zenoh 1.0.0 Porting

.github/workflows/style.yaml

@@ -19,6 +19,6 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - name: uncrustify
-      run: /ros_entrypoint.sh ament_uncrustify --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp rmw_zenoh_cpp/
+      run: /ros_entrypoint.sh ament_uncrustify rmw_zenoh_cpp/ --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp
     - name: cpplint
-      run: /ros_entrypoint.sh ament_cpplint --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp rmw_zenoh_cpp/
+      run: /ros_entrypoint.sh ament_cpplint rmw_zenoh_cpp/ --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp

rmw_zenoh_cpp/CMakeLists.txt

@@ -22,10 +22,6 @@ find_package(rosidl_typesupport_fastrtps_c REQUIRED)
 find_package(rosidl_typesupport_fastrtps_cpp REQUIRED)
 find_package(rmw REQUIRED)
 find_package(zenoh_c_vendor REQUIRED)
-find_package(zenohc_debug QUIET)
-if(NOT zenohc_debug_FOUND)
-  find_package(zenohc REQUIRED)
-endif()
 
 add_library(rmw_zenoh_cpp SHARED
   src/detail/attachment_helpers.cpp

rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_ROUTER_CONFIG.json5

@@ -29,10 +29,10 @@
   },
   /// Configure the scouting mechanisms and their behaviours
   scouting: {
-    /// In client mode, the period dedicated to scouting for a router before failing
+    /// In client mode, the period in milliseconds dedicated to scouting for a router before failing.
     timeout: 3000,
-    /// In peer mode, the period dedicated to scouting remote peers before attempting other operations
-    delay: 1,
+    /// In peer mode, the maximum period in milliseconds dedicated to scouting remote peers before attempting other operations.
+    delay: 500,
     /// The multicast scouting configuration.
     multicast: {
       /// Whether multicast scouting is enabled or not
@@ -44,7 +44,7 @@
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on UDP multicast.
       /// Accepts a single value or different values for router, peer and client.
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
-      autoconnect: { router: "", peer: "router|peer" },
+      autoconnect: { router: [], peer: ["router", "peer"] },
       /// Whether or not to listen for scout messages on UDP multicast and reply to them.
       listen: true,
     },
@@ -61,7 +61,7 @@
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on gossip.
       /// Accepts a single value or different values for router, peer and client.
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
-      autoconnect: { router: "", peer: "router|peer" },
+      autoconnect: { router: [], peer: ["router", "peer"] },
     },
   },
 
@@ -183,9 +183,16 @@
             data_low: 4,
             background: 4,
           },
-          /// The initial exponential backoff time in nanoseconds to allow the batching to eventually progress.
-          /// Higher values lead to a more aggressive batching but it will introduce additional latency.
-          backoff: 100,
+          /// Perform batching of messages if they are smaller of the batch_size
+          batching: {
+            /// Perform adaptive batching of messages if they are smaller of the batch_size.
+            /// When the network is detected to not be fast enough to transmit every message individually, many small messages may be
+            /// batched together and sent all at once on the wire reducing the overall network overhead. This is typically of a high-throughput
+            /// scenario mainly composed of small messages. In other words, batching is activated by the network back-pressure.
+            enabled: true,
+            /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
+            time_limit: 1,
+          }
         },
         // Number of threads dedicated to transmission
         // By default, the number of threads is calculated as follows: 1 + ((#cores - 1) / 4)

rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_SESSION_CONFIG.json5

@@ -32,10 +32,10 @@
   },
   /// Configure the scouting mechanisms and their behaviours
   scouting: {
-    /// In client mode, the period dedicated to scouting for a router before failing
+    /// In client mode, the period in milliseconds dedicated to scouting for a router before failing.
     timeout: 3000,
-    /// In peer mode, the period dedicated to scouting remote peers before attempting other operations
-    delay: 1,
+    /// In peer mode, the maximum period in milliseconds dedicated to scouting remote peers before attempting other operations.
+    delay: 500,
     /// The multicast scouting configuration.
     multicast: {
       /// Whether multicast scouting is enabled or not
@@ -47,7 +47,7 @@
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on UDP multicast.
       /// Accepts a single value or different values for router, peer and client.
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
-      autoconnect: { router: "", peer: "router|peer" },
+      autoconnect: { router: [], peer: ["router", "peer"] },
       /// Whether or not to listen for scout messages on UDP multicast and reply to them.
       listen: true,
     },
@@ -64,7 +64,7 @@
       /// Which type of Zenoh instances to automatically establish sessions with upon discovery on gossip.
       /// Accepts a single value or different values for router, peer and client.
       /// Each value is bit-or-like combinations of "peer", "router" and "client".
-      autoconnect: { router: "", peer: "router|peer" },
+      autoconnect: { router: [], peer: ["router", "peer"] },
     },
   },
 
@@ -186,9 +186,16 @@
             data_low: 4,
             background: 4,
           },
-          /// The initial exponential backoff time in nanoseconds to allow the batching to eventually progress.
-          /// Higher values lead to a more aggressive batching but it will introduce additional latency.
-          backoff: 100,
+          /// Perform batching of messages if they are smaller of the batch_size
+          batching: {
+            /// Perform adaptive batching of messages if they are smaller of the batch_size.
+            /// When the network is detected to not be fast enough to transmit every message individually, many small messages may be
+            /// batched together and sent all at once on the wire reducing the overall network overhead. This is typically of a high-throughput
+            /// scenario mainly composed of small messages. In other words, batching is activated by the network back-pressure.
+            enabled: true,
+            /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
+            time_limit: 1,
+          }
         },
         // Number of threads dedicated to transmission
         // By default, the number of threads is calculated as follows: 1 + ((#cores - 1) / 4)

rmw_zenoh_cpp/src/detail/attachment_helpers.cpp

@@ -17,80 +17,166 @@
 #include <cstdlib>
 #include <cstring>
 #include <string>
+#include <utility>
 
 #include "rmw/types.h"
 
+#include "logging_macros.hpp"
+
 #include "attachment_helpers.hpp"
 
 namespace rmw_zenoh_cpp
 {
-bool get_gid_from_attachment(
-  const z_attachment_t * const attachment, uint8_t gid[RMW_GID_STORAGE_SIZE])
+
+attachement_context_t::attachement_context_t(std::unique_ptr<attachement_data_t> && _data)
+: data(std::move(_data)) {}
+
+bool create_attachment_iter(z_owned_bytes_t * kv_pair, void * context)
 {
-  if (!z_check(*attachment)) {
+  attachement_context_t * ctx = reinterpret_cast<attachement_context_t *>(context);
+  z_owned_bytes_t k, v;
+
+  if (ctx->idx == 0) {
+    z_bytes_serialize_from_str(&k, "sequence_number");
+    z_bytes_serialize_from_int64(&v, ctx->data->sequence_number);
+  } else if (ctx->idx == 1) {
+    z_bytes_serialize_from_str(&k, "source_timestamp");
+    z_bytes_serialize_from_int64(&v, ctx->data->source_timestamp);
+  } else if (ctx->idx == 2) {
+    z_bytes_serialize_from_str(&k, "source_gid");
+    z_bytes_serialize_from_buf(
+      &v, ctx->data->source_gid,
+      RMW_GID_STORAGE_SIZE);
+  } else {
     return false;
   }
 
-  z_bytes_t index = z_attachment_get(*attachment, z_bytes_new("source_gid"));
-  if (!z_check(index)) {
+  z_bytes_from_pair(kv_pair, z_move(k), z_move(v));
+  ctx->idx += 1;
+  return true;
+}
+
+attachement_data_t::attachement_data_t(
+  const int64_t _sequence_number,
+  const int64_t _source_timestamp,
+  const uint8_t _source_gid[RMW_GID_STORAGE_SIZE])
+{
+  sequence_number = _sequence_number;
+  source_timestamp = _source_timestamp;
+  memcpy(source_gid, _source_gid, RMW_GID_STORAGE_SIZE);
+}
+
+z_result_t attachement_data_t::serialize_to_zbytes(z_owned_bytes_t * attachment)
+{
+  attachement_context_t context =
+    attachement_context_t(std::make_unique<attachement_data_t>(*this));
+  return z_bytes_from_iter(
+    attachment, create_attachment_iter,
+    reinterpret_cast<void *>(&context));
+}
+
+
+bool get_attachment(
+  const z_loaned_bytes_t * const attachment,
+  const std::string & key, z_owned_bytes_t * val)
+{
+  if (attachment == NULL || z_bytes_is_empty(attachment)) {
     return false;
   }
 
-  if (index.len != RMW_GID_STORAGE_SIZE) {
+  z_bytes_iterator_t iter = z_bytes_get_iterator(attachment);
+  z_owned_bytes_t pair, key_;
+  bool found = false;
+
+  while (z_bytes_iterator_next(&iter, &pair)) {
+    z_bytes_deserialize_into_pair(z_loan(pair), &key_, val);
+    z_owned_string_t key_string;
+    z_bytes_deserialize_into_string(z_loan(key_), &key_string);
+
+    const char * key_string_ptr = z_string_data(z_loan(key_string));
+    size_t key_string_len = z_string_len(z_loan(key_string));
+    if (key_string_len == key.length() && strncmp(key_string_ptr, key.c_str(), key.length()) == 0) {
+      found = true;
+    }
+
+    z_drop(z_move(pair));
+    z_drop(z_move(key_));
+    z_drop(z_move(key_string));
+
+    if (found) {
+      break;
+    } else {
+      z_drop(z_move(*val));
+    }
+  }
+
+  if (!found) {
     return false;
   }
 
-  memcpy(gid, index.start, index.len);
+  if (z_bytes_is_empty(z_loan(*val))) {
+    return false;
+  }
 
   return true;
 }
 
-int64_t get_int64_from_attachment(
-  const z_attachment_t * const attachment, const std::string & name)
+bool get_gid_from_attachment(
+  const z_loaned_bytes_t * const attachment,
+  uint8_t gid[RMW_GID_STORAGE_SIZE])
 {
-  if (!z_check(*attachment)) {
-    // A valid request must have had an attachment
-    return -1;
+  if (attachment == NULL || z_bytes_is_empty(attachment)) {
+    return false;
   }
 
-  z_bytes_t index = z_attachment_get(*attachment, z_bytes_new(name.c_str()));
-  if (!z_check(index)) {
-    return -1;
+  z_owned_bytes_t val;
+  if (!get_attachment(attachment, "source_gid", &val)) {
+    return false;
   }
 
-  if (index.len < 1) {
-    return -1;
+  z_owned_slice_t slice;
+  z_bytes_deserialize_into_slice(z_loan(val), &slice);
+  if (z_slice_len(z_loan(slice)) != RMW_GID_STORAGE_SIZE) {
+    RMW_ZENOH_LOG_ERROR_NAMED("rmw_zenoh_cpp", "GID length mismatched.")
+    return false;
   }
+  memcpy(gid, z_slice_data(z_loan(slice)), z_slice_len(z_loan(slice)));
+
+  z_drop(z_move(val));
+  z_drop(z_move(slice));
+
+  return true;
+}
 
-  if (index.len > 19) {
-    // The number was larger than we expected
+int64_t get_int64_from_attachment(
+  const z_loaned_bytes_t * const attachment,
+  const std::string & name)
+{
+  // A valid request must have had an attachment
+  if (attachment == NULL || z_bytes_is_empty(attachment)) {
     return -1;
   }
 
-  // The largest possible int64_t number is INT64_MAX, i.e. 9223372036854775807.
-  // That is 19 characters long, plus one for the trailing \0, means we need 20 bytes.
-  char int64_str[20];
+  // TODO(yuyuan): This key should be specific
+  z_owned_bytes_t val;
+  if (!get_attachment(attachment, name, &val)) {
+    RMW_ZENOH_LOG_ERROR_NAMED(
+      "rmw_zenoh_cpp", "Failed to deserialize int64 from the attachment.")
+    return false;
+  }
 
-  memcpy(int64_str, index.start, index.len);
-  int64_str[index.len] = '\0';
+  int64_t num;
+  if (z_bytes_deserialize_into_int64(z_loan(val), &num) != Z_OK) {
+    return -1;
+  }
 
-  errno = 0;
-  char * endptr;
-  int64_t num = strtol(int64_str, &endptr, 10);
   if (num == 0) {
     // This is an error regardless; the client should never send this
     return -1;
-  } else if (endptr == int64_str) {
-    // No values were converted, this is an error
-    return -1;
-  } else if (*endptr != '\0') {
-    // There was junk after the number
-    return -1;
-  } else if (errno != 0) {
-    // Some other error occurred, which may include overflow or underflow
-    return -1;
   }
 
+  z_drop(z_move(val));
+
   return num;
 }
 }  // namespace rmw_zenoh_cpp

rmw_zenoh_cpp/src/detail/attachment_helpers.hpp

@@ -18,16 +18,45 @@
 #include <zenoh.h>
 
 #include <string>
+#include <memory>
 
 #include "rmw/types.h"
 
 namespace rmw_zenoh_cpp
 {
+
+class attachement_data_t final
+{
+public:
+  int64_t sequence_number;
+  int64_t source_timestamp;
+  uint8_t source_gid[RMW_GID_STORAGE_SIZE];
+  explicit attachement_data_t(
+    const int64_t _sequence_number,
+    const int64_t _source_timestamp,
+    const uint8_t _source_gid[RMW_GID_STORAGE_SIZE]);
+  z_result_t serialize_to_zbytes(z_owned_bytes_t *);
+};
+
+class attachement_context_t final
+{
+public:
+  std::unique_ptr<attachement_data_t> data;
+  int length = 3;
+  int idx = 0;
+
+  attachement_context_t(std::unique_ptr<attachement_data_t> && _data);
+};
+
+bool get_attachment(
+  const z_loaned_bytes_t * const attachment,
+  const std::string & key, z_owned_bytes_t * val);
+
 bool get_gid_from_attachment(
-  const z_attachment_t * const attachment, uint8_t gid[RMW_GID_STORAGE_SIZE]);
+  const z_loaned_bytes_t * const attachment, uint8_t gid[RMW_GID_STORAGE_SIZE]);
 
 int64_t get_int64_from_attachment(
-  const z_attachment_t * const attachment, const std::string & name);
+  const z_loaned_bytes_t * const attachment, const std::string & name);
 }  // namespace rmw_zenoh_cpp
 
 #endif  // DETAIL__ATTACHMENT_HELPERS_HPP_