Add direct ROS 2 CycloneDDS Subspace overlay

c_client/CMakeLists.txt

@@ -8,16 +8,31 @@ add_library(subspace_c_client STATIC
     subspace.h
 )
 
+add_library(subspace_c_client_shared SHARED
+    subspace.cc
+    subspace.h
+)
+set_target_properties(subspace_c_client_shared PROPERTIES
+    OUTPUT_NAME subspace_c_client
+)
+
 target_include_directories(subspace_c_client PUBLIC
     ${CMAKE_CURRENT_SOURCE_DIR}
 )
+target_include_directories(subspace_c_client_shared PUBLIC
+    ${CMAKE_CURRENT_SOURCE_DIR}
+)
 #
 # Link against other internal libraries or external dependencies if needed.
 # Note: C libraries might have different linking requirements than C++ ones.
 target_link_libraries(subspace_c_client PUBLIC
     subspace_proto
     subspace_client
 )
+target_link_libraries(subspace_c_client_shared PUBLIC
+    subspace_proto
+    subspace_client
+)
 
 add_executable(c_client_test client_test.cc)
 add_test(NAME c_client_test COMMAND c_client_test)

c_client/subspace.cc

@@ -98,19 +98,17 @@ SubspaceMessage ToCMessage(const subspace::Message &msg,
           .checksum_error = msg.checksum_error};
 }
 
-SubspaceMessage TakeCMessage(subspace::Message &&msg) {
-  subspace::Message *stored = new subspace::Message(std::move(msg));
-  auto *handle = new std::shared_ptr<subspace::Message>(stored);
-  return ToCMessage(*stored, handle);
+SubspaceMessage OwnCMessage(subspace::Message &&msg) {
+  auto *handle = new subspace::Message(std::move(msg));
+  return ToCMessage(*handle, handle);
 }
 
 void FreeMessageStorage(SubspaceMessage *message) {
   if (message == nullptr || message->message == nullptr) {
     return;
   }
-  auto msg_ptr =
-      reinterpret_cast<std::shared_ptr<subspace::Message> *>(message->message);
-  delete msg_ptr;
+  auto *msg = reinterpret_cast<subspace::Message *>(message->message);
+  delete msg;
   *message = EmptyMessage();
 }
 
@@ -624,7 +622,7 @@ SubspaceMessage subspace_read_message_with_mode(SubspaceSubscriber subscriber,
   if (status_or_msg->length == 0) {
     return message;
   }
-  return TakeCMessage(std::move(*status_or_msg));
+  return OwnCMessage(std::move(*status_or_msg));
 }
 
 SubspaceMessage subspace_read_message(SubspaceSubscriber subscriber) {
@@ -648,7 +646,7 @@ SubspaceMessage subspace_find_message(SubspaceSubscriber subscriber,
   if (status_or_msg->length == 0) {
     return EmptyMessage();
   }
-  return TakeCMessage(std::move(*status_or_msg));
+  return OwnCMessage(std::move(*status_or_msg));
 }
 
 bool subspace_get_all_messages(SubspaceSubscriber subscriber,
@@ -682,7 +680,7 @@ bool subspace_get_all_messages(SubspaceSubscriber subscriber,
   cache.messages.reserve(status_or_messages->size());
   for (subspace::Message &message : *status_or_messages) {
     if (message.length != 0) {
-      cache.messages.push_back(TakeCMessage(std::move(message)));
+      cache.messages.push_back(OwnCMessage(std::move(message)));
     }
   }
   *messages = cache.messages.data();
@@ -730,9 +728,9 @@ bool subspace_register_subscriber_callback(SubspaceSubscriber subscriber,
   absl::Status status = (*sub_ptr)->RegisterMessageCallback(
       [subscriber, callback](const subspace::Subscriber *,
                              subspace::Message msg) {
-        // This takes ownership of the message.  It must be freed by calline
-        // subspace_free_message. by the callback function.
-        callback(subscriber, TakeCMessage(std::move(msg)));
+        // This takes ownership of the message. The callback must release it
+        // with subspace_free_message.
+        callback(subscriber, OwnCMessage(std::move(msg)));
       });
   if (!status.ok()) {
     subspace_set_error(status.ToString().c_str());
@@ -842,12 +840,11 @@ bool subspace_invoke_subscriber_callback(SubspaceSubscriber subscriber,
   }
   auto sub_ptr = reinterpret_cast<std::shared_ptr<subspace::Subscriber> *>(
       subscriber.subscriber);
-  auto msg_ptr =
-      reinterpret_cast<std::shared_ptr<subspace::Message> *>(message.message);
+  auto *msg = reinterpret_cast<subspace::Message *>(message.message);
   // InvokeMessageCallback takes the Message by value; the copy increments the
   // active-message refcount so the caller's SubspaceMessage retains ownership
   // and is still its responsibility to subspace_free_message.
-  (*sub_ptr)->InvokeMessageCallback(**msg_ptr);
+  (*sub_ptr)->InvokeMessageCallback(*msg);
   return true;
 }
 

c_client/subspace.h

@@ -151,8 +151,8 @@ typedef struct {
   void *user_data;
 } SubspaceSplitBufferCallbacks;
 
-// This is a received message.  The 'message' member is a pointer to a smart
-// message object that is used to manage the message.  The 'length' member is
+// This is a received message.  The 'message' member is an opaque owned message
+// handle that is used to manage the message.  The 'length' member is
 // the length of the message data in bytes and 'buffer' points to the start
 // address of the message in shared memory.  If there is no message read, the
 // length member will be zero and buffer will be nullptr.
@@ -165,7 +165,7 @@ typedef struct {
 // message, the subscriber will run out of slots and you will be unable to read
 // any more messages.
 typedef struct {
-  void *message;      // Smart message pointer
+  void *message;      // Opaque owned message handle.
   size_t length;      // Length of the message
   const void *buffer; // Address of the message payload
   uint64_t ordinal;   // Monotonic number of the message

ros2/README.md

@@ -0,0 +1,62 @@
+# ROS 2 Jazzy Subspace Local IPC Overlay
+
+This directory contains the ROS 2 overlay pieces for using Subspace directly
+inside CycloneDDS for same-host shared-memory delivery.
+
+The approach keeps `rmw_cyclonedds_cpp` as the ROS RMW. CycloneDDS still owns
+DDS discovery, QoS matching, graph behavior, services, actions, and remote
+traffic. The CycloneDDS patch in `ros2/patches/cyclonedds-direct-subspace.patch`
+replaces the Iceoryx-backed SHM implementation with a direct Subspace backend
+that links `subspace_client`.
+
+## Workspace Layout
+
+From a Jazzy workspace:
+
+```bash
+vcs import src < /path/to/subspace/ros2/ros2_subspace_jazzy.repos
+ln -s /path/to/subspace/ros2/subspace_vendor src/subspace_vendor
+git -C src/eclipse-cyclonedds/cyclonedds apply /path/to/subspace/ros2/patches/cyclonedds-direct-subspace.patch
+git -C src/ros2/rmw_cyclonedds apply /path/to/subspace/ros2/patches/rmw-cyclonedds-direct-subspace.patch
+SUBSPACE_SOURCE_DIR=/path/to/subspace colcon build --packages-up-to rmw_cyclonedds_cpp
+```
+
+Run `subspace_server` before starting ROS nodes that enable CycloneDDS shared
+memory. CycloneDDS discovers local readers and writers through DDS as usual; the
+shared-memory publication and subscription calls are served by Subspace.
+
+## Subspace Shared-Memory Sizing
+
+Subspace sizing is configured in CycloneDDS XML under
+`Domain/SharedMemory/Subspace`:
+
+```xml
+<Domain Id="any">
+  <SharedMemory>
+    <Enable>true</Enable>
+    <Subspace>
+      <Socket>/tmp/subspace</Socket>
+      <SlotSize>256</SlotSize>
+      <Slots>132</Slots>
+      <MaxActiveMessages>64</MaxActiveMessages>
+      <ExpectedSubscribers>1</ExpectedSubscribers>
+      <FixedSize>false</FixedSize>
+      <Reliable>false</Reliable>
+      <LogDroppedMessages>false</LogDroppedMessages>
+      <DetectDroppedMessages>false</DetectDroppedMessages>
+    </Subspace>
+  </SharedMemory>
+</Domain>
+```
+
+If `Slots` or `MaxActiveMessages` is `0`, CycloneDDS derives conservative
+values from DDS history depth and `ExpectedSubscribers`. For explicit sizing,
+`Slots` must include Subspace's publisher/subscriber bookkeeping headroom in
+addition to active messages; `132` is the smallest tested slot count that cleanly
+supports `MaxActiveMessages=64` for the single-subscriber local benchmark.
+`Reliable=false` is the default until the direct backend's reliable-mode
+interaction with DDS discovery and startup ordering has been tuned. The
+benchmark configs set `DetectDroppedMessages=false` because DDS/performance_test
+already counts lost samples from sample IDs. The
+performance matrix runner writes per-scenario XML files so message size,
+fan-out, and queue depth benchmarks exercise the intended Subspace sizing.

ros2/cyclonedds_subspace.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<CycloneDDS xmlns="https://cdds.io/config"
+            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+            xsi:schemaLocation="https://cdds.io/config https://raw.githubusercontent.com/eclipse-cyclonedds/cyclonedds/releases/0.10.x/etc/cyclonedds.xsd">
+  <Domain Id="any">
+    <SharedMemory>
+      <Enable>true</Enable>
+      <LogLevel>info</LogLevel>
+      <Subspace>
+        <Socket>/tmp/subspace</Socket>
+        <SlotSize>256</SlotSize>
+        <Slots>132</Slots>
+        <MaxActiveMessages>64</MaxActiveMessages>
+        <ExpectedSubscribers>1</ExpectedSubscribers>
+        <FixedSize>false</FixedSize>
+        <Reliable>false</Reliable>
+        <LogDroppedMessages>false</LogDroppedMessages>
+        <DetectDroppedMessages>false</DetectDroppedMessages>
+      </Subspace>
+    </SharedMemory>
+  </Domain>
+</CycloneDDS>