Slot ThreadPool [14767]

ddsrouter_core/include/ddsrouter_core/configuration/DDSRouterConfiguration.hpp

@@ -50,7 +50,8 @@ class DDSRouterConfiguration : public DDSRouterReloadConfiguration
             std::set<std::shared_ptr<types::FilterTopic>> allowlist,
             std::set<std::shared_ptr<types::FilterTopic>> blocklist,
             std::set<std::shared_ptr<types::RealTopic>> builtin_topics,
-            std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations);
+            std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations,
+            unsigned int number_of_threads = default_number_of_threads());
 
     /**
      * @brief Return a set with the different \c ParticipantConfigurations in the yaml
@@ -68,12 +69,20 @@ class DDSRouterConfiguration : public DDSRouterReloadConfiguration
     DDSROUTER_CORE_DllAPI void reload(
             const DDSRouterReloadConfiguration& new_configuration);
 
+    DDSROUTER_CORE_DllAPI unsigned int number_of_threads() const noexcept;
+
+    DDSROUTER_CORE_DllAPI static unsigned int default_number_of_threads() noexcept;
+
 protected:
 
     static bool check_correct_configuration_object_(
             const std::shared_ptr<ParticipantConfiguration> configuration);
 
     std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations_;
+
+    unsigned int number_of_threads_;
+
+    static const unsigned int DEFAULT_NUMBER_OF_THREADS_;
 };
 
 } /* namespace configuration */

ddsrouter_core/src/cpp/communication/Bridge.cpp

@@ -31,6 +31,7 @@ Bridge::Bridge(
         const RealTopic& topic,
         std::shared_ptr<ParticipantsDatabase> participants_database,
         std::shared_ptr<PayloadPool> payload_pool,
+        std::shared_ptr<utils::SlotThreadPool> thread_pool,
         bool enable /* = false */)
     : topic_(topic)
     , participants_(participants_database)
@@ -62,7 +63,13 @@ Bridge::Bridge(
         // This insert is required as there is no copy method for Track
         // Tracks are always created disabled and then enabled with Bridge enable() method
         tracks_[id] =
-                std::make_unique<Track>(topic_, id, readers_[id], std::move(writers_except_one), payload_pool_, false);
+                std::make_unique<Track>(
+            topic_,
+            id,
+            readers_[id], std::move(writers_except_one),
+            payload_pool_,
+            thread_pool,
+            false);
     }
 
     if (enable)

ddsrouter_core/src/cpp/communication/Bridge.hpp

@@ -25,6 +25,7 @@
 #include <participant/IParticipant.hpp>
 #include <core/ParticipantsDatabase.hpp>
 #include <ddsrouter_core/types/participant/ParticipantId.hpp>
+#include <ddsrouter_utils/thread_pool/pool/SlotThreadPool.hpp>
 
 namespace eprosima {
 namespace ddsrouter {
@@ -58,6 +59,7 @@ class Bridge
             const types::RealTopic& topic,
             std::shared_ptr<ParticipantsDatabase> participants_database,
             std::shared_ptr<PayloadPool> payload_pool,
+            std::shared_ptr<utils::SlotThreadPool> thread_pool,
             bool enable = false);
 
     /**

ddsrouter_core/src/cpp/communication/Track.cpp

@@ -19,6 +19,8 @@
 
 #include <ddsrouter_utils/exception/UnsupportedException.hpp>
 #include <ddsrouter_utils/Log.hpp>
+#include <ddsrouter_utils/thread_pool/pool/SlotThreadPool.hpp>
+#include <ddsrouter_utils/thread_pool/task/TaskId.hpp>
 
 #include <communication/Track.hpp>
 
@@ -28,12 +30,15 @@ namespace core {
 
 using namespace eprosima::ddsrouter::core::types;
 
+const unsigned int Track::MAX_MESSAGES_TRANSMIT_LOOP_ = 100;
+
 Track::Track(
         const RealTopic& topic,
         ParticipantId reader_participant_id,
         std::shared_ptr<IReader> reader,
         std::map<ParticipantId, std::shared_ptr<IWriter>>&& writers,
         std::shared_ptr<PayloadPool> payload_pool,
+        std::shared_ptr<utils::SlotThreadPool> thread_pool,
         bool enable /* = false */) noexcept
     : reader_participant_id_(reader_participant_id)
     , topic_(topic)
@@ -43,14 +48,18 @@ Track::Track(
     , enabled_(false)
     , exit_(false)
     , data_available_status_(DataAvailableStatus::no_more_data)
+    , thread_pool_(thread_pool)
+    , transmit_task_id_(utils::new_unique_task_id())
 {
     logDebug(DDSROUTER_TRACK, "Creating Track " << *this << ".");
 
     // Set this track to on_data_available lambda call
     reader_->set_on_data_available_callback(std::bind(&Track::data_available_, this));
 
-    // Activate transmit thread even without being enabled
-    transmit_thread_ = std::thread(&Track::transmit_thread_function_, this);
+    // Set slot in thread pool
+    thread_pool_->slot(
+        transmit_task_id_,
+        std::bind(&Track::transmit_, this));
 
     if (enable)
     {
@@ -71,21 +80,15 @@ Track::~Track()
     reader_->unset_on_data_available_callback();
 
     // It does need to guard the mutex to avoid notifying Track thread while it is checking variable condition
-    {
-        // Set exit status and call transmit thread to awake and terminate. Then wait for it.
-        std::lock_guard<std::mutex> lock(data_available_mutex_);
-        exit_.store(true);
-    }
-
-    data_available_condition_variable_.notify_all();
-    transmit_thread_.join();
+    // Set exit status and call transmit thread to awake and terminate. Then wait for it.
+    exit_.store(true);
 
     logDebug(DDSROUTER_TRACK, "Track " << *this << " destroyed.");
 }
 
 void Track::enable() noexcept
 {
-    std::lock_guard<std::recursive_mutex> lock(track_mutex_);
+    std::lock_guard<std::mutex> lock(track_mutex_);
 
     if (!enabled_)
     {
@@ -106,7 +109,7 @@ void Track::enable() noexcept
 
 void Track::disable() noexcept
 {
-    std::lock_guard<std::recursive_mutex> lock(track_mutex_);
+    std::lock_guard<std::mutex> lock(track_mutex_);
 
     if (enabled_)
     {
@@ -130,23 +133,6 @@ void Track::disable() noexcept
     }
 }
 
-void Track::no_more_data_available_() noexcept
-{
-    std::lock_guard<std::mutex> lock(data_available_mutex_);
-
-    // It may occur that within the process of set data_available_status, the actual status had changed
-    // Thus, it must take care that it is only set to no_data when it comes from transmitting data
-    if (data_available_status_ == DataAvailableStatus::transmitting_data)
-    {
-        logDebug(DDSROUTER_TRACK, "Track " << *this << " has no more data to send.");
-        data_available_status_.store(DataAvailableStatus::no_more_data);
-    }
-    // If it is new_data_arrived is that the Listener has notified new data AFTER Track has received a no_data
-    // from the Reader. Very unlikely timing, but possible.
-    // In this occasion, it must not be set as no_more_data because THERE IS data.
-    // If it is no_more_data it does not need to be changed (however it should never happen)
-}
-
 bool Track::should_transmit_() noexcept
 {
     return !exit_ && enabled_ && this->is_data_available_();
@@ -159,14 +145,18 @@ void Track::data_available_() noexcept
     {
         logDebug(DDSROUTER_TRACK, "Track " << *this << " has data ready to be sent.");
 
-        // It does need to guard the mutex to avoid notifying Track thread while it is checking variable condition
+        // Lock data_available_status_mutex_ to avoid changing the status while it is being checked
+        std::lock_guard<std::mutex> lock(data_available_status_mutex_);
+
+        // This method will always be called from the Reader thread, so it is safe to set the status
+        DataAvailableStatus current_status = data_available_status_.exchange(DataAvailableStatus::new_data_arrived);
+
+        if (current_status == DataAvailableStatus::no_more_data)
         {
-            // Set data available to true and notify transmit thread
-            std::lock_guard<std::mutex> lock(data_available_mutex_);
-            data_available_status_.store(DataAvailableStatus::new_data_arrived);
+            // Only send the callback to thread pool if it was not running
+            thread_pool_->emit(transmit_task_id_);
+            logDebug(DDSROUTER_TRACK, "Track " << *this << " send callback to queue.");
         }
-
-        data_available_condition_variable_.notify_one();
     }
 }
 
@@ -176,45 +166,19 @@ bool Track::is_data_available_() const noexcept
            data_available_status_ == DataAvailableStatus::transmitting_data;
 }
 
-void Track::transmit_thread_function_() noexcept
-{
-    while (!exit_)
-    {
-        // Wait in Condition Variable till there is data to send or it must exit
-        {
-            std::unique_lock<std::mutex> lock(data_available_mutex_);
-            data_available_condition_variable_.wait(
-                lock,
-                [this]
-                {
-                    return this->is_data_available_() || this->exit_;
-                });
-        }
-
-        // Once thread awakes, transmit without any mutex guarded
-        if (!this->exit_)
-        {
-            transmit_();
-        }
-    }
-}
-
 void Track::transmit_() noexcept
 {
     // Loop that ends if it should stop transmitting (should_transmit_nts_).
     // Called inside the loop so it is protected by a mutex that is freed in every iteration.
-    while (true)
-    {
-        // Lock Mutex on_transmition while a data is being transmitted
-        // This prevents the Track to be disabled (and disable writers and readers) while sending a data
-        std::unique_lock<std::mutex> lock(on_transmission_mutex_);
 
-        // If it must not keep transmitting, stop loop
-        if (!should_transmit_())
-        {
-            break;
-        }
+    // Lock Mutex on_transmition while a data is being transmitted
+    // This prevents the Track to be disabled (and disable writers and readers) while sending a data
+    // enabled_ will be set to false before taking the mutex, so the track will finish after current iteration
+    std::unique_lock<std::mutex> lock(on_transmission_mutex_);
 
+    // TODO: Count the times it loops to break it at some point if needed
+    while (should_transmit_())
+    {
         // It starts transmitting, so it sets the data available status as transmitting
         data_available_status_ = DataAvailableStatus::transmitting_data;
 
@@ -224,16 +188,22 @@ void Track::transmit_() noexcept
 
         if (ret == utils::ReturnCode::RETCODE_NO_DATA)
         {
+            // Lock data_available_status_mutex_ to avoid changing the status while it is being checked
+            std::lock_guard<std::mutex> lock(data_available_status_mutex_);
+
             // There is no more data, so finish loop and wait again for new data
-            no_more_data_available_();
-            break;
-        }
-        else if (ret == utils::ReturnCode::RETCODE_NOT_ENABLED)
-        {
-            // This may not happen because the Reader is only disabled from here, however
-            // it is better to cut it and set as no more data is available.
-            no_more_data_available_();
-            break;
+            DataAvailableStatus current_status = data_available_status_.exchange(DataAvailableStatus::no_more_data);
+            if (current_status == DataAvailableStatus::new_data_arrived)
+            {
+                // New data has arrived while setting no_more_data, so it should continues
+                data_available_status_.store(DataAvailableStatus::transmitting_data);
+                continue;
+            }
+            else
+            {
+                // no_more_data has been set, so if any other data arrives it will send a callback to thread pool
+                break;
+            }
         }
         else if (!ret)
         {
@@ -263,6 +233,8 @@ void Track::transmit_() noexcept
 
         payload_pool_->release_payload(data->payload);
     }
+
+    data_available_status_.store(DataAvailableStatus::no_more_data);
 }
 
 std::ostream& operator <<(

ddsrouter_core/src/cpp/communication/Track.hpp

@@ -27,6 +27,7 @@
 #include <participant/IParticipant.hpp>
 #include <reader/IReader.hpp>
 #include <writer/IWriter.hpp>
+#include <ddsrouter_utils/thread_pool/pool/SlotThreadPool.hpp>
 
 namespace eprosima {
 namespace ddsrouter {
@@ -56,6 +57,7 @@ class Track
             std::shared_ptr<IReader> reader,
             std::map<types::ParticipantId, std::shared_ptr<IWriter>>&& writers,
             std::shared_ptr<PayloadPool> payload_pool,
+            std::shared_ptr<utils::SlotThreadPool> thread_pool,
             bool enable = false) noexcept;
 
     /**
@@ -205,7 +207,7 @@ class Track
      * Mutex to prevent simultaneous calls to \c enable and/or \c disable .
      * It manages access to variable \c enabled_ .
      */
-    std::recursive_mutex track_mutex_;
+    std::mutex track_mutex_;
 
     /////
     // Transmit thread part
@@ -232,27 +234,19 @@ class Track
      */
     std::atomic<DataAvailableStatus> data_available_status_;
 
-    /**
-     * Condition variable to wait for new data available or track termination.
-     */
-    std::condition_variable data_available_condition_variable_;
-
-    /**
-     * Mutex to handle access to condition variable \c data_available_condition_variable_ .
-     * Mutex to manage access to variable \c data_available_status_ .
-     */
-    std::mutex data_available_mutex_;
-
-    /**
-     * Thread that will manage the transmission of the data
-     */
-    std::thread transmit_thread_;
+    std::mutex data_available_status_mutex_;
 
     /**
      * Mutex to guard while the Track is sending a message.
      */
     std::mutex on_transmission_mutex_;
 
+    utils::TaskId transmit_task_id_;
+
+    std::shared_ptr<utils::SlotThreadPool> thread_pool_;
+
+    static const unsigned int MAX_MESSAGES_TRANSMIT_LOOP_;
+
     // Allow operator << to use private variables
     friend std::ostream& operator <<(
             std::ostream&,

ddsrouter_core/src/cpp/configuration/DDSRouterConfiguration.cpp

@@ -33,13 +33,17 @@ namespace configuration {
 
 using namespace eprosima::ddsrouter::core::types;
 
+const unsigned int DDSRouterConfiguration::DEFAULT_NUMBER_OF_THREADS_ = 12;
+
 DDSRouterConfiguration::DDSRouterConfiguration(
         std::set<std::shared_ptr<FilterTopic>> allowlist,
         std::set<std::shared_ptr<FilterTopic>> blocklist,
         std::set<std::shared_ptr<RealTopic>> builtin_topics,
-        std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations)
+        std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations,
+        unsigned int number_of_threads /* = default_number_of_threads() */)
     : DDSRouterReloadConfiguration (allowlist, blocklist, builtin_topics)
     , participants_configurations_(participants_configurations)
+    , number_of_threads_(number_of_threads)
 {
 }
 
@@ -137,6 +141,16 @@ bool DDSRouterConfiguration::check_correct_configuration_object_(
     }
 }
 
+unsigned int DDSRouterConfiguration::number_of_threads() const noexcept
+{
+    return number_of_threads_;
+}
+
+unsigned int DDSRouterConfiguration::default_number_of_threads() noexcept
+{
+    return DEFAULT_NUMBER_OF_THREADS_;
+}
+
 } /* namespace configuration */
 } /* namespace core */
 } /* namespace ddsrouter */