Improve max_allocations calculation on SHM transport (#5659) (#5676)

* Improve `max_allocations` calculation on SHM transport (#5659)

* Refs #22841. Improve max_allocations calculation.

Signed-off-by: Miguel Company <[email protected]>

* Refs #22841. Consider a mean message size of half the max message size.

Signed-off-by: Miguel Company <[email protected]>

* Refs #22841. Move `shm_default_segment_size` to global scope.

Signed-off-by: Miguel Company <[email protected]>

* Refs #22841. Move `SharedMemTransportDescriptor` constants.

Signed-off-by: Miguel Company <[email protected]>

* Refs #22841. Change calculation of `mean_message_size`.

Signed-off-by: Miguel Company <[email protected]>

---------

Signed-off-by: Miguel Company <[email protected]>
(cherry picked from commit b54cb8e)

# Conflicts:
#	include/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h
#	src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp
#	src/cpp/rtps/transport/shared_mem/SharedMemTransportDescriptor.cpp
#	test/mock/rtps/SharedMemTransportDescriptor/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h

* Fix conflicts.
Signed-off-by: Miguel Company <[email protected]>

---------

Co-authored-by: Miguel Company <[email protected]>

Author: mergify[bot]

include/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h

@@ -39,6 +39,10 @@ class TransportInterface;
  */
 struct SharedMemTransportDescriptor : public TransportDescriptorInterface
 {
+    static constexpr uint32_t shm_default_segment_size = 0;
+    static constexpr uint32_t shm_default_port_queue_capacity = 512;
+    static constexpr uint32_t shm_default_healthy_check_timeout_ms = 1000;
+
     //! Destructor
     virtual ~SharedMemTransportDescriptor() = default;
 
@@ -132,10 +136,10 @@ struct SharedMemTransportDescriptor : public TransportDescriptorInterface
 
 private:
 
-    uint32_t segment_size_;
-    uint32_t port_queue_capacity_;
-    uint32_t healthy_check_timeout_ms_;
-    std::string rtps_dump_file_;
+    uint32_t segment_size_ = shm_default_segment_size;
+    uint32_t port_queue_capacity_ = shm_default_port_queue_capacity;
+    uint32_t healthy_check_timeout_ms_ = shm_default_healthy_check_timeout_ms;
+    std::string rtps_dump_file_ {""};
 
 };
 

src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include <algorithm>
+#include <cstdint>
 #include <cstring>
 #include <utility>
 
@@ -25,6 +26,7 @@
 #include <fastdds/rtps/common/Locator.h>
 #include <fastdds/rtps/transport/SenderResource.h>
 #include <fastdds/rtps/transport/TransportInterface.h>
+#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h>
 #include <fastrtps/rtps/messages/CDRMessage.h>
 #include <fastrtps/rtps/messages/MessageReceiver.h>
 
@@ -50,6 +52,9 @@ using LocatorSelectorEntry = fastrtps::rtps::LocatorSelectorEntry;
 using LocatorSelector = fastrtps::rtps::LocatorSelector;
 using PortParameters = fastrtps::rtps::PortParameters;
 
+// TODO(Adolfo): Calculate this value from UDP sockets buffers size.
+static constexpr uint32_t shm_default_segment_size = 512 * 1024;
+
 TransportInterface* SharedMemTransportDescriptor::create_transport() const
 {
     return new SharedMemTransport(*this);
@@ -241,9 +246,6 @@ bool SharedMemTransport::DoInputLocatorsMatch(
 bool SharedMemTransport::init(
         const fastrtps::rtps::PropertyPolicy*)
 {
-    // TODO(Adolfo): Calculate this value from UDP sockets buffers size.
-    static constexpr uint32_t shm_default_segment_size = 512 * 1024;
-
     if (configuration_.segment_size() == 0)
     {
         configuration_.segment_size(shm_default_segment_size);
@@ -271,8 +273,14 @@ bool SharedMemTransport::init(
         {
             return false;
         }
-        shared_mem_segment_ = shared_mem_manager_->create_segment(configuration_.segment_size(),
-                        configuration_.port_queue_capacity());
+        constexpr uint32_t mean_message_size =
+                shm_default_segment_size / SharedMemTransportDescriptor::shm_default_port_queue_capacity;
+        uint32_t max_allocations = configuration_.segment_size() / mean_message_size;
+        if (configuration_.port_queue_capacity() > max_allocations)
+        {
+            max_allocations = configuration_.port_queue_capacity();
+        }
+        shared_mem_segment_ = shared_mem_manager_->create_segment(configuration_.segment_size(), max_allocations);
 
         // Memset the whole segment to zero in order to force physical map of the buffer
         auto buffer = shared_mem_segment_->alloc_buffer(configuration_.segment_size(),

src/cpp/rtps/transport/shared_mem/SharedMemTransportDescriptor.cpp

@@ -17,27 +17,11 @@
 
 using namespace eprosima::fastdds::rtps;
 
-namespace eprosima {
-namespace fastdds {
-namespace rtps {
-
-static constexpr uint32_t shm_default_segment_size = 0;
-static constexpr uint32_t shm_default_port_queue_capacity = 512;
-static constexpr uint32_t shm_default_healthy_check_timeout_ms = 1000;
-
-} // rtps
-} // fastdds
-} // eprosima
-
 //*********************************************************
 // SharedMemTransportDescriptor
 //*********************************************************
 SharedMemTransportDescriptor::SharedMemTransportDescriptor()
     : TransportDescriptorInterface(shm_default_segment_size, s_maximumInitialPeersRange)
-    , segment_size_(shm_default_segment_size)
-    , port_queue_capacity_(shm_default_port_queue_capacity)
-    , healthy_check_timeout_ms_(shm_default_healthy_check_timeout_ms)
-    , rtps_dump_file_("")
 {
     maxMessageSize = s_maximumMessageSize;
 }

test/mock/rtps/SharedMemTransportDescriptor/fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h

@@ -28,8 +28,12 @@ class TransportInterface;
  *
  * @ingroup TRANSPORT_MODULE
  */
-typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterface
+struct SharedMemTransportDescriptor : public TransportDescriptorInterface
 {
+    static constexpr uint32_t shm_default_segment_size = 0;
+    static constexpr uint32_t shm_default_port_queue_capacity = 512;
+    static constexpr uint32_t shm_default_healthy_check_timeout_ms = 1000;
+
     virtual ~SharedMemTransportDescriptor()
     {
 
@@ -108,12 +112,12 @@ typedef struct SharedMemTransportDescriptor : public TransportDescriptorInterfac
 
 private:
 
-    uint32_t segment_size_;
-    uint32_t port_queue_capacity_;
-    uint32_t healthy_check_timeout_ms_;
+    uint32_t segment_size_ = shm_default_segment_size;
+    uint32_t port_queue_capacity_ = shm_default_port_queue_capacity;
+    uint32_t healthy_check_timeout_ms_ = shm_default_healthy_check_timeout_ms;
     std::string rtps_dump_file_;
 
-}SharedMemTransportDescriptor;
+};
 
 } // namespace rtps
 } // namespace fastdds