Merge pull request #13 from n7space/n7s-aur#274-fix-queue-issue

fixed queue issues and did small refactor

Author: igorJarek

src/Queue.h

@@ -29,13 +29,10 @@
  */
 
 #include <condition_variable>
-#include <cstddef>
 #include <iostream>
 #include <mutex>
 #include <queue>
 
-#include <sched.h>
-
 #include "Request.h"
 
 namespace taste {
@@ -109,11 +106,14 @@ class Queue final
      */
     bool is_empty() const;
 
+  private:
+    bool check_for_message_loss() const;
+
   private:
     const size_t m_max_elements;
     const char* m_queue_name;
     mutable std::mutex m_mutex;
-    mutable std::condition_variable m_cv;
+    mutable std::condition_variable m_condition_variable;
     std::queue<Request<PARAMETER_SIZE>> m_queue;
 };
 
@@ -128,58 +128,48 @@ template<size_t PARAMETER_SIZE>
 void
 Queue<PARAMETER_SIZE>::put(const Request<PARAMETER_SIZE>& request)
 {
-    m_mutex.lock();
-    if(m_queue.size() >= m_max_elements) {
-        std::cerr << "Message loss in " << m_queue_name << " - queue is full, " << m_max_elements << " allowed"
-                  << std::endl;
-        m_mutex.unlock();
-    } else {
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+
+        if(check_for_message_loss()) {
+            return;
+        }
+
         m_queue.push(request);
-        m_mutex.unlock();
-        m_cv.notify_one();
     }
+    
+    m_condition_variable.notify_one();
 }
 
 template<size_t PARAMETER_SIZE>
 void
 Queue<PARAMETER_SIZE>::put(const asn1SccPID sender_pid, const uint8_t* data, size_t length)
 {
-    m_mutex.lock();
-    static Request<PARAMETER_SIZE> request;
-
-    if(length > PARAMETER_SIZE) {
-        std::cerr << "Internal error in " << m_queue_name << " - queue accepts messages with size " << PARAMETER_SIZE
-                  << std::endl;
-        m_mutex.unlock();
-    }
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
 
-    memcpy(request.data(), data, length);
-    request.set_length(length);
-    request.set_sender_pid(sender_pid);
+        if(check_for_message_loss()) {
+            return;
+        }
 
-    if(m_queue.size() >= m_max_elements) {
-        std::cerr << "Message loss in " << m_queue_name << " - queue is full, " << m_max_elements << " allowed"
-                  << std::endl;
-        m_mutex.unlock();
-    } else {
-        m_queue.push(request);
-        m_mutex.unlock();
-        m_cv.notify_one();
+        m_queue.emplace(sender_pid, data, length);
     }
+
+    m_condition_variable.notify_one();
 }
 
 template<size_t PARAMETER_SIZE>
 void
 Queue<PARAMETER_SIZE>::get(Request<PARAMETER_SIZE>& request)
 {
     std::unique_lock<std::mutex> lock(m_mutex);
+
     while(true) {
         if(m_queue.empty()) {
-            m_cv.wait(lock);
+            m_condition_variable.wait(lock);
         } else {
             request = m_queue.front();
             m_queue.pop();
-            lock.unlock();
             return;
         }
     }
@@ -192,6 +182,21 @@ Queue<PARAMETER_SIZE>::is_empty() const
     std::unique_lock<std::mutex> lock(m_mutex);
     return m_queue.empty();
 }
+
+template<size_t PARAMETER_SIZE>
+bool
+Queue<PARAMETER_SIZE>::check_for_message_loss() const
+{
+    if(m_queue.size() >= m_max_elements) {
+        std::cerr << "Message loss in '" << m_queue_name << "' - queue is full, " << m_max_elements
+                  << " elements are allowed" << std::endl;
+
+        return true;
+    }
+
+    return false;
+}
+
 } // namespace taste
 
 #endif

src/Request.h

@@ -31,6 +31,7 @@
 #include <cstdint>
 #include <cstring>
 #include <array>
+#include <iostream>
 #include "dataview-uniq.h"
 
 namespace taste {
@@ -49,6 +50,13 @@ struct Request final
      */
     Request();
 
+    /**
+     * @brief Constructor
+     *
+     * Constructs Request with provided arguments
+     */
+    Request(const asn1SccPID sender_pid, const uint8_t* const data, const size_t length);
+
     /// @brief default copy constructor
     Request(const Request& other) = default;
 
@@ -65,16 +73,16 @@ struct Request final
      *
      * @return the actual length of data
      */
-    uint32_t length() const;
+    size_t length() const;
 
     /**
      * @brief set the request length
      *
-     * The value shall be between 0 and PARAMETER_SIZE
+     * The length shall be between 0 and PARAMETER_SIZE
      *
-     * @param value    new length value
+     * @param length   new length value
      */
-    void set_length(uint32_t value);
+    void set_length(size_t length);
 
     /**
      * @brief get the request data
@@ -105,31 +113,45 @@ struct Request final
     void set_sender_pid(asn1SccPID sender_pid);
 
   private:
-    uint32_t m_length;
-    std::array<uint8_t, PARAMETER_SIZE> m_data;
+    void check_length(size_t length) const;
+
+  private:
+    size_t m_length;
     asn1SccPID m_sender_pid;
+    std::array<uint8_t, PARAMETER_SIZE> m_data;
 };
 
 template<size_t PARAMETER_SIZE>
 Request<PARAMETER_SIZE>::Request()
     : m_length(0)
+    , m_sender_pid(PID_env)
 {
 }
 
 template<size_t PARAMETER_SIZE>
-uint32_t
+Request<PARAMETER_SIZE>::Request(const asn1SccPID sender_pid, const uint8_t* const data, const size_t length)
+    : m_length(length)
+    , m_sender_pid(sender_pid)
+{
+    check_length(length);
+
+    memcpy(m_data.data(), data, length);
+}
+
+template<size_t PARAMETER_SIZE>
+size_t
 Request<PARAMETER_SIZE>::length() const
 {
     return m_length;
 }
 
 template<size_t PARAMETER_SIZE>
 void
-Request<PARAMETER_SIZE>::set_length(uint32_t value)
+Request<PARAMETER_SIZE>::set_length(size_t length)
 {
-    if(value <= PARAMETER_SIZE && value >= 0) {
-        m_length = value;
-    }
+    check_length(length);
+
+    m_length = length;
 }
 
 template<size_t PARAMETER_SIZE>
@@ -159,6 +181,19 @@ Request<PARAMETER_SIZE>::set_sender_pid(asn1SccPID sender_pid)
 {
     m_sender_pid = sender_pid;
 }
+
+template<size_t PARAMETER_SIZE>
+void
+Request<PARAMETER_SIZE>::check_length(size_t length) const
+{
+    if(length > PARAMETER_SIZE) {
+        std::cerr << "Request data size shall be <= " << PARAMETER_SIZE << " - new length value ("
+                  << length << ") is greater than " << PARAMETER_SIZE << std::endl;
+
+        exit(EXIT_FAILURE);
+    }
+}
+
 } // namespace taste
 
 #endif

src/StartBarrier.cc

@@ -21,19 +21,30 @@
  */
 
 #include "StartBarrier.h"
+#include <iostream>
 
 namespace taste {
 void
 StartBarrier::initialize(size_t number, InitCallback init_callback)
 {
     m_init_callback = init_callback;
-    pthread_barrier_init(&m_init_barrier, nullptr, number);
+
+    const int error_code = pthread_barrier_init(&m_init_barrier, nullptr, number);
+    if(error_code != 0) {
+        std::cerr << "Barrier has not been created properly. Error code : " << error_code << std::endl;
+        exit(EXIT_FAILURE);
+    }
 }
 
 void
 StartBarrier::wait()
 {
-    pthread_barrier_wait(&m_init_barrier);
+    const int error_code = pthread_barrier_wait(&m_init_barrier);
+    if(error_code != PTHREAD_BARRIER_SERIAL_THREAD && error_code != 0) {
+        std::cerr << "Barrier Wait has been failed. Error code : " << error_code << std::endl;
+        exit(EXIT_FAILURE);
+    }
+
     std::call_once(m_init_callback_flag, m_init_callback);
 }