SHM Buffer recovery mechanishm (#1159)

* Refs #8219. Change BufferNode to pre-allocated pool with fixed addresses for the nodes during entire life-cycle.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. SHM Buffer invalidation implementation.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. logWarning -> logInfo when segment overflow.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. Optimization.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. Style changes.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. buffer_recover test.

Signed-off-by: AdolfoMartinez <[email protected]>

* Refs #8219. 'error:' string removed from log msg.

Signed-off-by: AdolfoMartinez <[email protected]>

Author: adolfomarver

src/cpp/rtps/transport/shared_mem/SharedMemGlobal.hpp

@@ -72,10 +72,9 @@ class SharedMemGlobal
      */
     struct BufferDescriptor
     {
-        alignas(8) uint32_t flags;
-        uint64_t enqueue_tick;
         SharedMemSegment::Id source_segment_id;
         SharedMemSegment::Offset buffer_node_offset;
+        uint32_t validity_id;
     };
 
     typedef MultiProducerConsumerRingBuffer<BufferDescriptor>::Listener Listener;
@@ -358,7 +357,7 @@ class SharedMemGlobal
                                 (*port_it)->node->is_port_ok = false;
 
                                 logWarning(RTPS_TRANSPORT_SHM, "Port " << (*port_it)->node->port_id
-                                    << " error: " << e.what());
+                                    << ": " << e.what());
 
                                 // Remove the port from watch
                                 port_it = watched_ports_.erase(port_it);
@@ -791,7 +790,7 @@ class SharedMemGlobal
             }
 
             try
-            {                
+            {
                 port->healthy_check();
 
                 if ( (port_node->is_opened_read_exclusive && open_mode != Port::OpenMode::Write) ||
@@ -867,6 +866,16 @@ class SharedMemGlobal
         return port;
     }
 
+    /**
+     * Remove a port from the system.
+     */
+    void remove_port(
+            uint32_t port_id)
+    {
+        auto port_segment_name = domain_name_ + "_port" + std::to_string(port_id);
+        SharedMemSegment::remove(port_segment_name.c_str());
+    }
+
 private:
 
     std::string domain_name_;