Don't copy contiguous bytes on reception (#343)

* Don't copy contiguous bytes on reception

This uses the slices iterator API of zenoh-cpp to avoid unecessarily
copying bytes into a vecotr, if and only if the bytes is made up of
exactly one slice.

* Don't use auto type specifiers

* Remove unused `<vector>` includes

* Explain lifetime of `Contiguous::slice`

* Move `Payload` into `zenoh_utils`

Signed-off-by: Mahmoud Mazouz <mazouz.mahmoud@outlook.com>
Co-authored-by: Chris Lalancette <clalancette@gmail.com>

Author: 2 people

rmw_zenoh_cpp/src/detail/rmw_subscription_data.cpp

@@ -24,7 +24,6 @@
 #include <string>
 #include <utility>
 #include <variant>
-#include <vector>
 
 #include "attachment_helpers.hpp"
 #include "cdr.hpp"
@@ -44,10 +43,10 @@ namespace rmw_zenoh_cpp
 {
 ///=============================================================================
 SubscriptionData::Message::Message(
-  std::vector<uint8_t> && p,
+  const zenoh::Bytes & p,
   uint64_t recv_ts,
   AttachmentData && attachment_)
-: payload(std::move(p)), recv_timestamp(recv_ts), attachment(std::move(attachment_))
+: payload(p), recv_timestamp(recv_ts), attachment(std::move(attachment_))
 {
 }
 
@@ -225,7 +224,7 @@ bool SubscriptionData::init()
 
         sub_data->add_new_message(
           std::make_unique<SubscriptionData::Message>(
-            sample.get_payload().as_vector(),
+            sample.get_payload(),
             std::chrono::system_clock::now().time_since_epoch().count(),
             std::move(attachment_data)),
           std::string(sample.get_keyexpr().as_string_view()));
@@ -303,13 +302,12 @@ bool SubscriptionData::init()
             "Unable to obtain attachment")
           return;
         }
-        auto payload = sample.get_payload().clone();
         auto attachment_value = attachment.value();
 
         AttachmentData attachment_data(attachment_value);
         sub_data->add_new_message(
           std::make_unique<SubscriptionData::Message>(
-            payload.as_vector(),
+            sample.get_payload(),
             std::chrono::system_clock::now().time_since_epoch().count(),
             std::move(attachment_data)),
           std::string(sample.get_keyexpr().as_string_view()));

rmw_zenoh_cpp/src/detail/rmw_subscription_data.hpp

@@ -25,7 +25,6 @@
 #include <string>
 #include <unordered_map>
 #include <variant>
-#include <vector>
 
 #include <zenoh.hxx>
 
@@ -51,13 +50,13 @@ class SubscriptionData final : public std::enable_shared_from_this<SubscriptionD
   struct Message
   {
     explicit Message(
-      std::vector<uint8_t> && p,
+      const zenoh::Bytes & bytes,
       uint64_t recv_ts,
       AttachmentData && attachment);
 
     ~Message();
 
-    std::vector<uint8_t> payload;
+    Payload payload;
     uint64_t recv_timestamp;
     AttachmentData attachment;
   };

rmw_zenoh_cpp/src/detail/zenoh_utils.cpp

@@ -26,18 +26,6 @@
 
 namespace rmw_zenoh_cpp
 {
-///=============================================================================
-zenoh::Bytes create_map_and_set_sequence_num(
-  int64_t sequence_number, std::array<uint8_t, RMW_GID_STORAGE_SIZE> gid)
-{
-  auto now = std::chrono::system_clock::now().time_since_epoch();
-  auto now_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(now);
-  int64_t source_timestamp = now_ns.count();
-
-  rmw_zenoh_cpp::AttachmentData data(sequence_number, source_timestamp, gid);
-  return data.serialize_to_zbytes();
-}
-
 ///=============================================================================
 ZenohQuery::ZenohQuery(
   const zenoh::Query & query,
@@ -82,4 +70,55 @@ std::chrono::nanoseconds::rep ZenohReply::get_received_timestamp() const
 {
   return received_timestamp_;
 }
+
+Payload::Payload(const zenoh::Bytes & bytes)
+{
+  // NOTE(fuzzypixelz): `zenoh::Bytes` is an list of reference-couted buffers. When the list of
+  // buffers contains exactly one element, it is not necessary to concatenate the list of buffers.
+  // In this case, we store a clone of the bytes object to maintain a non-zero reference-count on
+  // the buffer. This ensures that the slice into said buffer stays valid until we drop our copy
+  // of the bytes object (at the very least). This case corresponds to the `Contiguous`
+  // alternative of the `bytes_` variant and aims to optimize away a memcpy during "session-local"
+  // communication.
+
+  zenoh::Bytes::SliceIterator slices = bytes.slice_iter();
+  std::optional<zenoh::Slice> slice = slices.next();
+  if (!slice.has_value()) {
+    bytes_ = nullptr;
+  } else {
+    if (!slices.next().has_value()) {
+      bytes_ = Contiguous {slice.value(), bytes.clone()};
+    } else {
+      bytes_ = bytes.as_vector();
+    }
+  }
+}
+
+const uint8_t * Payload::data() const
+{
+  if (std::holds_alternative<Empty>(bytes_)) {
+    return nullptr;
+  } else if (std::holds_alternative<NonContiguous>(bytes_)) {
+    return std::get<NonContiguous>(bytes_).data();
+  } else {
+    return std::get<Contiguous>(bytes_).slice.data;
+  }
+}
+
+size_t Payload::size() const
+{
+  if (std::holds_alternative<Empty>(bytes_)) {
+    return 0;
+  } else if (std::holds_alternative<NonContiguous>(bytes_)) {
+    return std::get<NonContiguous>(bytes_).size();
+  } else {
+    return std::get<Contiguous>(bytes_).slice.len;
+  }
+}
+
+bool Payload::empty() const
+{
+  return std::holds_alternative<Empty>(bytes_);
+}
+
 }  // namespace rmw_zenoh_cpp

rmw_zenoh_cpp/src/detail/zenoh_utils.hpp

@@ -21,15 +21,14 @@
 #include <chrono>
 #include <functional>
 #include <optional>
+#include <utility>
+#include <variant>
+#include <vector>
 
 #include "rmw/types.h"
 
 namespace rmw_zenoh_cpp
 {
-///=============================================================================
-zenoh::Bytes create_map_and_set_sequence_num(
-  int64_t sequence_number, std::array<uint8_t, RMW_GID_STORAGE_SIZE> gid);
-
 ///=============================================================================
 // A class to store the replies to service requests.
 class ZenohReply final
@@ -65,6 +64,32 @@ class ZenohQuery final
   zenoh::Query query_;
   std::chrono::nanoseconds::rep received_timestamp_;
 };
+
+class Payload
+{
+public:
+  explicit Payload(const zenoh::Bytes & bytes);
+
+  ~Payload() = default;
+
+  const uint8_t * data() const;
+
+  size_t size() const;
+
+  bool empty() const;
+
+private:
+  struct Contiguous
+  {
+    zenoh::Slice slice;
+    zenoh::Bytes bytes;
+  };
+  using NonContiguous = std::vector<uint8_t>;
+  using Empty = std::nullptr_t;
+  // Is `std::vector<uint8_t>` in case of a non-contiguous payload
+  // and `zenoh::Slice` plus a `zenoh::Bytes` otherwise.
+  std::variant<NonContiguous, Contiguous, Empty> bytes_;
+};
 }  // namespace rmw_zenoh_cpp
 
 #endif  // DETAIL__ZENOH_UTILS_HPP_