Consistent Mutex Usage (#19)

This commit replaces calls to sys_mutex_* and replaces them with direct calls to FreeRTOS API. The reason behind this change is that some lwip abstraction layers use recursive mutexes, and others do not. embeddedRTPS hangs if non-recursive mutexes are used. This change ensures that only recursive mutexes are used.

Co-authored-by: Alexandru Kampmann <kampmann@embedded.rwth-aachen.de>

Author: akampmann

include/rtps/discovery/SEDPAgent.h

@@ -59,7 +59,7 @@ class SEDPAgent {
 
 private:
   Participant *m_part;
-  sys_mutex_t m_mutex;
+  SemaphoreHandle_t m_mutex;
   uint8_t m_buffer[600]; // TODO check size, currently changed from 300 to 600
                          // (FastDDS gives too many options)
   BuiltInEndpoints m_endpoints;

include/rtps/discovery/SPDPAgent.h

@@ -52,7 +52,6 @@ class ReaderCacheChange;
 
 class SPDPAgent {
 public:
-  ~SPDPAgent();
   void init(Participant &participant, BuiltInEndpoints &endpoints);
   void start();
   void stop();
@@ -67,7 +66,7 @@ class SPDPAgent {
   ucdrBuffer m_microbuffer{};
   uint8_t m_cycleHB = 0;
 
-  sys_mutex_t m_mutex;
+  SemaphoreHandle_t m_mutex;
   bool initialized = false;
   static void receiveCallback(void *callee,
                               const ReaderCacheChange &cacheChange);

include/rtps/entities/Domain.h

@@ -84,7 +84,7 @@ class Domain {
   }
 
   bool m_initComplete = false;
-  sys_mutex_t m_mutex;
+  SemaphoreHandle_t m_mutex;
 
   void receiveCallback(const PacketInfo &packet);
   GuidPrefix_t generateGuidPrefix(ParticipantId_t id) const;

include/rtps/entities/Participant.h

@@ -115,7 +115,7 @@ class Participant {
   std::array<Reader *, Config::NUM_READERS_PER_PARTICIPANT> m_readers = {
       nullptr};
 
-  sys_mutex_t m_mutex;
+  SemaphoreHandle_t m_mutex;
   MemoryPool<ParticipantProxyData, Config::SPDP_MAX_NUMBER_FOUND_PARTICIPANTS>
       m_remoteParticipants;
 

include/rtps/entities/Reader.h

@@ -31,6 +31,7 @@ Author: i11 - Embedded Software, RWTH Aachen University
 #include "rtps/entities/WriterProxy.h"
 #include "rtps/storages/MemoryPool.h"
 #include "rtps/storages/PBufWrapper.h"
+#include "semphr.h"
 #include <cstring>
 
 namespace rtps {
@@ -135,10 +136,10 @@ class Reader {
   std::array<callbackElement_t, Config::MAX_NUM_READER_CALLBACKS> m_callbacks;
 
   // Guards manipulation of the proxies array
-  sys_mutex_t m_proxies_mutex = nullptr;
+  SemaphoreHandle_t m_proxies_mutex = nullptr;
 
   // Guards manipulation of callback array
-  sys_mutex_t m_callback_mutex = nullptr;
+  SemaphoreHandle_t m_callback_mutex = nullptr;
 };
 } // namespace rtps
 

include/rtps/entities/StatefulReader.tpp

@@ -66,18 +66,16 @@ void StatefulReaderT<NetworkDriver>::newChange(
   if (m_callback_count == 0 || !m_is_initialized_) {
     return;
   }
-  sys_mutex_lock(&m_proxies_mutex);
+  Lock{m_proxies_mutex};
   for (auto &proxy : m_proxies) {
     if (proxy.remoteWriterGuid == cacheChange.writerGuid) {
       if (proxy.expectedSN == cacheChange.sn) {
-        sys_mutex_unlock(&m_proxies_mutex);
         executeCallbacks(cacheChange);
         ++proxy.expectedSN;
         return;
       }
     }
   }
-  sys_mutex_unlock(&m_proxies_mutex);
 }
 
 template <class NetworkDriver>

include/rtps/entities/StatelessWriter.tpp

@@ -67,7 +67,7 @@ bool StatelessWriterT<NetworkDriver>::init(TopicData attributes,
   m_attributes = attributes;
 
   if (m_mutex == nullptr) {
-    if (sys_mutex_new(&m_mutex) != ERR_OK) {
+    if (!createMutex(&m_mutex)) {
 #if SLW_VERBOSE
       SLW_LOG("Failed to create mutex \n");
 #endif

include/rtps/entities/Writer.h

@@ -82,7 +82,7 @@ class Writer {
 protected:
   SequenceNumber_t m_sedp_sequence_number;
 
-  sys_mutex_t m_mutex = nullptr;
+  SemaphoreHandle_t m_mutex = nullptr;
   ThreadPool *mp_threadPool = nullptr;
 
   Ip4Port_t m_srcPort;

include/rtps/storages/ThreadSafeCircularBuffer.h

@@ -26,6 +26,7 @@ Author: i11 - Embedded Software, RWTH Aachen University
 #define RTPS_THREADSAFEQUEUE_H
 
 #include "lwip/sys.h"
+#include "semphr.h"
 
 #include <array>
 #include <limits>
@@ -37,8 +38,6 @@ template <typename T, uint16_t SIZE> class ThreadSafeCircularBuffer {
 public:
   bool init();
 
-  ~ThreadSafeCircularBuffer();
-
   bool moveElementIntoBuffer(T &&elem);
 
   /**
@@ -57,7 +56,7 @@ template <typename T, uint16_t SIZE> class ThreadSafeCircularBuffer {
   static_assert(SIZE + 1 < std::numeric_limits<decltype(m_head)>::max(),
                 "Iterator is large enough for given size");
 
-  sys_mutex_t m_mutex;
+  SemaphoreHandle_t m_mutex;
   bool m_initialized = false;
 
   inline bool isFull();

include/rtps/storages/ThreadSafeCircularBuffer.tpp

@@ -23,7 +23,7 @@ bool ThreadSafeCircularBuffer<T, SIZE>::init() {
   if (m_initialized) {
     return true;
   }
-  if (sys_mutex_new(&m_mutex) != ERR_OK) {
+  if (!createMutex(&m_mutex)) {
     TSCB_LOG("Failed to create mutex \n");
     return false;
   } else {
@@ -34,13 +34,6 @@ bool ThreadSafeCircularBuffer<T, SIZE>::init() {
   }
 }
 
-template <typename T, uint16_t SIZE>
-ThreadSafeCircularBuffer<T, SIZE>::~ThreadSafeCircularBuffer() {
-  if (m_initialized) {
-    sys_mutex_free(&m_mutex);
-  }
-}
-
 template <typename T, uint16_t SIZE>
 bool ThreadSafeCircularBuffer<T, SIZE>::moveElementIntoBuffer(T &&elem) {
   Lock lock(m_mutex);