Merge pull request #55 from michael-poehnl/iox-#51-runtime_cleanup_shm_resources

iox-#51: capability to destroy sender + receiver ports

Author: budrus

iceoryx_posh/include/iceoryx_posh/internal/popo/base_port.hpp

@@ -14,8 +14,8 @@
 
 #pragma once
 
-#include "iceoryx_posh/internal/popo/base_port_data.hpp"
 #include "iceoryx_posh/iceoryx_posh_types.hpp"
+#include "iceoryx_posh/internal/popo/base_port_data.hpp"
 
 namespace iox
 {
@@ -57,31 +57,32 @@ class BasePort
     operator bool() const;
 
     /// @brief Reads Type of actual CaPro Port (sender/receiver...)
-    /// @param              Nothing
     /// @return m_portType  Type of Port in struct BasePortType
     BasePortType getPortType() const noexcept;
 
     /// @brief Reads Type of actual CaPro Port (sender/receiver...)
-    /// @param              Nothing
     /// @return m_portType  Type of Port in struct BasePortType
     capro::ServiceDescription getCaProServiceDescription() const noexcept;
 
     /// @brief Gets Application Name for the active port
-    /// @param              Nothing
     /// @return             Application name as String
     cxx::CString100 getApplicationName() const noexcept;
 
-
     /// @brief Gets Interface Name for the active port
-    /// @param              Nothing
     /// @return             Interface name as enum value
     Interfaces getInterface() const noexcept;
 
     /// @brief Gets Id of thethe active port
-    /// @param              Nothing
     /// @return             UniqueId name as Integer
     uint64_t getUniqueID() const noexcept;
 
+    /// @brief Indicate that this port can be destroyed
+    void destroy() noexcept;
+
+    /// @brief Checks whether port can be destroyed
+    /// @return             true if it shall be destroyed, false if not
+    bool toBeDestroyed() const noexcept;
+
   protected:
     const MemberType_t* getMembers() const noexcept;
     MemberType_t* getMembers() noexcept;

iceoryx_posh/include/iceoryx_posh/internal/popo/base_port_data.hpp

@@ -76,6 +76,7 @@ struct BasePortData
 
     static std::atomic<uint64_t> s_uniqueIdCounter;
     std::atomic<uint64_t> m_uniqueId{0};
+    std::atomic_bool m_toBeDestroyed{false};
 
     iox::relative_ptr<runtime::RunnableData> m_runnable;
 };

iceoryx_posh/include/iceoryx_posh/internal/roudi/shared_memory_manager.hpp

@@ -192,6 +192,10 @@ class SharedMemoryManager
     void deletePortsOfProcess(std::string processName);
     void deleteRunnableAndItsPorts(std::string runnableName);
 
+    void destroySenderPort(SenderPortType::MemberType_t* const senderPortData);
+
+    void destroyReceiverPort(ReceiverPortType::MemberType_t* const receiverPortData);
+
     void printmempool();
     std::string GetShmAddrString();
     uint64_t getShmSizeInBytes() const;

iceoryx_posh/include/iceoryx_posh/popo/publisher.hpp

@@ -25,7 +25,6 @@ namespace iox
 {
 namespace popo
 {
-
 class Publisher
 {
   public:
@@ -40,13 +39,14 @@ class Publisher
     Publisher& operator=(Publisher&&) = default;
     Publisher(Publisher&& other) = default;
 
-    virtual ~Publisher() noexcept = default;
+    virtual ~Publisher() noexcept;
 
     /// @brief Allocate memory for the chunk to be sent
     /// @param[in] payloadSize size of shared memory to be allocated
     /// @param[in] useDynamicPayloadSizes bool value of using dynamic payload size
     /// @return Information about the chunk reserved
-    virtual mepoo::ChunkHeader* allocateChunkWithHeader(uint32_t payloadSize, bool useDynamicPayloadSizes = false) noexcept;
+    virtual mepoo::ChunkHeader* allocateChunkWithHeader(uint32_t payloadSize,
+                                                        bool useDynamicPayloadSizes = false) noexcept;
 
     /// @brief Allocate memory for chunk to be sent
     /// @param[in] payloadSize size of shared memory to be allocated
@@ -66,7 +66,8 @@ class Publisher
     /// @param[in] chunkHeader Information of the chunk to be removed.
     virtual void freeChunk(mepoo::ChunkHeader* const chunkHeader) noexcept;
 
-    /// @brief Function for converting payload information to ChunkHeader , deleting particular chunk from chunkcontainer
+    /// @brief Function for converting payload information to ChunkHeader , deleting particular chunk from
+    /// chunkcontainer
     /// @param[in] payload payload of the chunk to be removed.
     virtual void freeChunk(void* const payload) noexcept;
 

iceoryx_posh/include/iceoryx_posh/popo/subscriber.hpp

@@ -127,6 +127,9 @@ class Subscriber
     /// @return true if the references are set otherwise false
     bool isChunkReceiveSemaphoreSet() noexcept;
 
+    /// @brief Unset the semaphore if one is set
+    void unsetChunkReceiveSemaphore() noexcept;
+
   protected:
     // needed for unit testing
     Subscriber() noexcept;

iceoryx_posh/source/popo/base_port.cpp

@@ -74,5 +74,15 @@ BasePort::operator bool() const
     return m_basePortDataPtr != nullptr;
 }
 
+void BasePort::destroy() noexcept
+{
+    getMembers()->m_toBeDestroyed.store(true, std::memory_order_relaxed);
+}
+
+bool BasePort::toBeDestroyed() const noexcept
+{
+    return getMembers()->m_toBeDestroyed.load(std::memory_order_relaxed);
+}
+
 } // namespace popo
 } // namespace iox

iceoryx_posh/source/popo/publisher.cpp

@@ -29,6 +29,11 @@ Publisher::Publisher(const capro::ServiceDescription& service, const cxx::CStrin
 {
 }
 
+Publisher::~Publisher() noexcept
+{
+    m_sender.destroy();
+}
+
 const void* Publisher::getLastChunk() const noexcept
 {
     assert(false && "Not yet supported");

iceoryx_posh/source/popo/subscriber.cpp

@@ -30,8 +30,7 @@ Subscriber::Subscriber() noexcept
 }
 
 Subscriber::Subscriber(const capro::ServiceDescription& service, const cxx::CString100& runnableName) noexcept
-    : m_receiver(
-        runtime::PoshRuntime::getInstance().getMiddlewareReceiver(service, Interfaces::INTERNAL, runnableName))
+    : m_receiver(runtime::PoshRuntime::getInstance().getMiddlewareReceiver(service, Interfaces::INTERNAL, runnableName))
 {
 }
 
@@ -40,6 +39,7 @@ Subscriber::~Subscriber() noexcept
     unsetReceiveHandler();
     // TODO: Find an alternative like an RAII receive handler which
     //          is called in the dtor. You cannot expect the user to call it before destruction
+    m_receiver.destroy();
 }
 
 void Subscriber::subscribe(const uint32_t cacheSize) noexcept
@@ -220,6 +220,12 @@ bool Subscriber::isChunkReceiveSemaphoreSet() noexcept
     return m_receiver.AreCallbackReferencesSet();
 }
 
+void Subscriber::unsetChunkReceiveSemaphore() noexcept
+{
+    m_receiver.UnsetCallbackReferences();
+}
+
+
 void Subscriber::eventCallbackMain() noexcept
 {
     while (m_callbackRunFlag)

iceoryx_posh/source/roudi/shared_memory_manager.cpp

@@ -117,6 +117,11 @@ void SharedMemoryManager::handleSenderPorts()
             // forward to interfaces
             sendToAllMatchingInterfacePorts(caproMessage, l_senderPort.getInterface());
         }
+        // check if we have to destroy this sender port
+        if (l_senderPort.toBeDestroyed())
+        {
+            destroySenderPort(l_senderPortData);
+        }
     }
 }
 
@@ -141,6 +146,11 @@ void SharedMemoryManager::handleReceiverPorts()
                 l_receiverPort.dispatchCaProMessage(nackMessage);
             }
         }
+        // check if we have to destroy this sender port
+        if (l_receiverPort.toBeDestroyed())
+        {
+            destroyReceiverPort(l_receiverPortData);
+        }
     }
 }
 
@@ -344,21 +354,7 @@ void SharedMemoryManager::deletePortsOfProcess(std::string f_processName)
         SenderPortType l_sender(port);
         if (f_processName == l_sender.getApplicationName())
         {
-            const auto& serviceDescription = l_sender.getCaProServiceDescription();
-            removeEntryFromServiceRegistry(serviceDescription.getServiceIDString(),
-                                           serviceDescription.getInstanceIDString());
-            l_sender.cleanup();
-
-            capro::CaproMessage message(capro::CaproMessageType::STOP_OFFER, serviceDescription);
-            m_portIntrospection.reportMessage(message);
-
-            sendToAllMatchingReceiverPorts(message, l_sender);
-
-            m_portIntrospection.removeSender(f_processName, serviceDescription);
-
-            // delete sender impl from list after StopOffer was processed
-            l_shm->m_senderPortMembers.erase(port);
-            DEBUG_PRINTF("Deleted SenderPortImpl of application %s\n", f_processName.c_str());
+            destroySenderPort(port);
         }
     }
 
@@ -367,21 +363,7 @@ void SharedMemoryManager::deletePortsOfProcess(std::string f_processName)
         ReceiverPortType l_receiver(port);
         if (f_processName == l_receiver.getApplicationName())
         {
-            // do the complete cleanup for the receiver port for being able to erase it
-            l_receiver.cleanup();
-
-            const auto& serviceDescription = l_receiver.getCaProServiceDescription();
-            capro::CaproMessage message(capro::CaproMessageType::UNSUB, serviceDescription);
-            message.m_requestPort = port;
-            m_portIntrospection.reportMessage(message);
-
-            sendToAllMatchingSenderPorts(message, l_receiver);
-
-            m_portIntrospection.removeReceiver(f_processName, serviceDescription);
-
-            // delete receiver impl from list after unsubscribe was processed
-            l_shm->m_receiverPortMembers.erase(port);
-            DEBUG_PRINTF("Deleted ReceiverPortImpl of application %s\n", f_processName.c_str());
+            destroyReceiverPort(port);
         }
     }
 
@@ -435,6 +417,49 @@ void SharedMemoryManager::deleteRunnableAndItsPorts(std::string f_runnableName)
     }
 }
 
+void SharedMemoryManager::destroySenderPort(SenderPortType::MemberType_t* const senderPortData)
+{
+    SenderPortType senderPort(senderPortData);
+
+    const auto& serviceDescription = senderPort.getCaProServiceDescription();
+    removeEntryFromServiceRegistry(serviceDescription.getServiceIDString(), serviceDescription.getInstanceIDString());
+    senderPort.cleanup();
+
+    const capro::CaproMessage message(capro::CaproMessageType::STOP_OFFER, serviceDescription);
+    m_portIntrospection.reportMessage(message);
+
+    sendToAllMatchingReceiverPorts(message, senderPort);
+    sendToAllMatchingInterfacePorts(message, senderPort.getInterface());
+
+    m_portIntrospection.removeSender(senderPort.getApplicationName(), serviceDescription);
+
+    // delete sender impl from list after StopOffer was processed
+    const auto shm = m_ShmInterface.getShmInterface();
+    shm->m_senderPortMembers.erase(senderPortData);
+    DEBUG_PRINTF("Destroyed SenderPortImpl\n");
+}
+
+void SharedMemoryManager::destroyReceiverPort(ReceiverPortType::MemberType_t* const receiverPortData)
+{
+    ReceiverPortType receiverPort(receiverPortData);
+
+    receiverPort.cleanup();
+
+    const auto& serviceDescription = receiverPort.getCaProServiceDescription();
+    capro::CaproMessage message(capro::CaproMessageType::UNSUB, serviceDescription);
+    message.m_requestPort = receiverPortData;
+    m_portIntrospection.reportMessage(message);
+
+    sendToAllMatchingSenderPorts(message, receiverPort);
+
+    m_portIntrospection.removeReceiver(receiverPort.getApplicationName(), serviceDescription);
+
+    // delete receiver impl from list after unsubscribe was processed
+    const auto shm = m_ShmInterface.getShmInterface();
+    shm->m_receiverPortMembers.erase(receiverPortData);
+    DEBUG_PRINTF("Destroyed ReceiverPortImpl\n");
+}
+
 std::string SharedMemoryManager::GetShmAddrString()
 {
     return m_ShmInterface.getBaseAddrString();