iox-#33 Wait until NamedPipe is initialized

Signed-off-by: Christian Eltzschig <[email protected]>

Author: elfenpiff

iceoryx_hoofs/include/iceoryx_hoofs/posix_wrapper/named_pipe.hpp

@@ -122,17 +122,35 @@ class NamedPipe : public DesignPattern::Creation<NamedPipe, IpcChannelError>
   private:
     cxx::optional<SharedMemoryObject> m_sharedMemory;
 
-    struct NamedPipeData
+    class NamedPipeData
     {
+      public:
         NamedPipeData(bool& isInitialized, IpcChannelError& error, const uint64_t maxMsgNumber) noexcept;
+        NamedPipeData(const NamedPipeData&) = delete;
+        NamedPipeData(NamedPipeData&& rhs) = delete;
+        ~NamedPipeData() noexcept;
+
+        NamedPipeData& operator=(const NamedPipeData&) = delete;
+        NamedPipeData& operator=(NamedPipeData&& rhs) = delete;
 
         Semaphore& sendSemaphore() noexcept;
         Semaphore& receiveSemaphore() noexcept;
 
+        bool waitForInitialization() const noexcept;
+        bool hasValidState() const noexcept;
+
+        MessageQueue_t messages;
+
+      private:
         static constexpr uint64_t SEND_SEMAPHORE = 0U;
         static constexpr uint64_t RECEIVE_SEMAPHORE = 1U;
 
-        MessageQueue_t messages;
+        static constexpr uint64_t INVALID_DATA = 0xBAADF00DAFFEDEAD;
+        static constexpr uint64_t VALID_DATA = 0xBAD0FF1CEBEEFBEE;
+        static constexpr units::Duration WAIT_FOR_INIT_TIMEOUT = units::Duration::fromSeconds(1);
+        static constexpr units::Duration WAIT_FOR_INIT_SLEEP_TIME = units::Duration::fromMilliseconds(1);
+
+        std::atomic<uint64_t> initializationGuard{INVALID_DATA};
         using semaphoreMemory_t = uint8_t[sizeof(Semaphore)];
         alignas(alignof(Semaphore)) semaphoreMemory_t semaphores[2U];
     };

iceoryx_hoofs/source/posix_wrapper/named_pipe.cpp

@@ -28,6 +28,7 @@ constexpr const char NamedPipe::NAMED_PIPE_PREFIX[];
 constexpr const char NamedPipe::SEND_SEMAPHORE_PREFIX[];
 constexpr const char NamedPipe::RECEIVE_SEMAPHORE_PREFIX[];
 constexpr units::Duration NamedPipe::CYCLE_TIME;
+constexpr units::Duration NamedPipe::NamedPipeData::WAIT_FOR_INIT_SLEEP_TIME;
 
 NamedPipe::NamedPipe() noexcept
 {
@@ -113,6 +114,14 @@ NamedPipe::NamedPipe(const IpcChannelName_t& name,
     {
         new (m_data) NamedPipeData(m_isInitialized, m_errorValue, maxMsgNumber);
     }
+    else
+    {
+        m_isInitialized = m_data->waitForInitialization();
+        if (m_isInitialized == false)
+        {
+            m_errorValue = IpcChannelError::INTERNAL_LOGIC_ERROR;
+        }
+    }
 }
 
 NamedPipe::NamedPipe(NamedPipe&& rhs) noexcept
@@ -326,8 +335,29 @@ NamedPipe::NamedPipeData::NamedPipeData(bool& isInitialized,
         &semaphores[SEND_SEMAPHORE], CreateUnnamedSharedMemorySemaphore, static_cast<unsigned int>(maxMsgNumber))
         .or_else([&](auto) { signalError("send"); });
 
+    if (!isInitialized)
+    {
+        return;
+    }
+
     Semaphore::placementCreate(&semaphores[RECEIVE_SEMAPHORE], CreateUnnamedSharedMemorySemaphore, 0U)
         .or_else([&](auto) { signalError("receive"); });
+
+    if (!isInitialized)
+    {
+        return;
+    }
+
+    initializationGuard.store(VALID_DATA);
+}
+
+NamedPipe::NamedPipeData::~NamedPipeData() noexcept
+{
+    if (hasValidState())
+    {
+        sendSemaphore().~Semaphore();
+        receiveSemaphore().~Semaphore();
+    }
 }
 
 Semaphore& NamedPipe::NamedPipeData::sendSemaphore() noexcept
@@ -340,5 +370,32 @@ Semaphore& NamedPipe::NamedPipeData::receiveSemaphore() noexcept
     return reinterpret_cast<Semaphore&>(semaphores[RECEIVE_SEMAPHORE]);
 }
 
+bool NamedPipe::NamedPipeData::waitForInitialization() const noexcept
+{
+    if (hasValidState())
+    {
+        return true;
+    }
+
+    units::Duration remainingTime = WAIT_FOR_INIT_TIMEOUT;
+
+    while (remainingTime.toNanoseconds() > 0U)
+    {
+        std::this_thread::sleep_for(std::chrono::nanoseconds(WAIT_FOR_INIT_SLEEP_TIME.toNanoseconds()));
+        remainingTime = remainingTime - WAIT_FOR_INIT_SLEEP_TIME;
+        if (hasValidState())
+        {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+bool NamedPipe::NamedPipeData::hasValidState() const noexcept
+{
+    return initializationGuard.load() == VALID_DATA;
+}
+
 } // namespace posix
 } // namespace iox