Removing mutex from Track hot-path (#261)

* Reimplement Track with incredibly fine grane

Signed-off-by: jparisu <javierparis@eprosima.com>

* Add documentation forthcoming

Signed-off-by: jparisu <javierparis@eprosima.com>

* apply suggestion

Signed-off-by: jparisu <javierparis@eprosima.com>

* uncrustify

Signed-off-by: jparisu <javierparis@eprosima.com>

Signed-off-by: jparisu <javierparis@eprosima.com>

Author: jparisu

ddsrouter_core/src/cpp/communication/Track.cpp

@@ -135,7 +135,7 @@ void Track::disable() noexcept
 
 bool Track::should_transmit_() noexcept
 {
-    return !exit_ && enabled_ && this->is_data_available_();
+    return !exit_ && enabled_;
 }
 
 void Track::data_available_() noexcept
@@ -145,27 +145,19 @@ void Track::data_available_() noexcept
     {
         logDebug(DDSROUTER_TRACK, "Track " << *this << " has data ready to be sent.");
 
-        // 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_);
+        // Get previous status and set current one to >=2 (it it was already >=2 it will keep being >2)
+        unsigned int previous_status = data_available_status_.fetch_add(DataAvailableStatus::new_data_arrived);
 
-        // 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)
+        if (previous_status == DataAvailableStatus::no_more_data)
         {
-            // Only send the callback to thread pool if it was not running
+            // no_more_data was set as current status, so no thread was running
+            // (and will not start as 2 is set as new current status)
             thread_pool_->emit(transmit_task_id_);
             logDebug(DDSROUTER_TRACK, "Track " << *this << " send callback to queue.");
         }
     }
 }
 
-bool Track::is_data_available_() const noexcept
-{
-    return data_available_status_ == DataAvailableStatus::new_data_arrived ||
-           data_available_status_ == DataAvailableStatus::transmitting_data;
-}
-
 void Track::transmit_() noexcept
 {
     // Loop that ends if it should stop transmitting (should_transmit_nts_).
@@ -180,29 +172,29 @@ void Track::transmit_() noexcept
     while (should_transmit_())
     {
         // It starts transmitting, so it sets the data available status as transmitting
-        data_available_status_ = DataAvailableStatus::transmitting_data;
+        // This will erase every previous value added in on_data_available and set 1
+        data_available_status_.store(DataAvailableStatus::transmitting_data);
 
         // Get data received
         std::unique_ptr<DataReceived> data = std::make_unique<DataReceived>();
         utils::ReturnCode ret = reader_->take(data);
 
         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
-            DataAvailableStatus current_status = data_available_status_.exchange(DataAvailableStatus::no_more_data);
-            if (current_status == DataAvailableStatus::new_data_arrived)
+            // There is no more data, so reduce in 1 the status
+            unsigned int previous_status = data_available_status_.fetch_sub(DataAvailableStatus::transmitting_data);
+            if (previous_status == DataAvailableStatus::transmitting_data)
             {
-                // New data has arrived while setting no_more_data, so it should continues
-                data_available_status_.store(DataAvailableStatus::transmitting_data);
-                continue;
+                // Previous Status = 1 (transmitting => no on_data_available callback has been called)
+                // Current Status  = 0 (new callbacks will emit the task)
+                // => close this thread and keeps status as 0 so new data available emit task
+                break;
             }
             else
             {
-                // no_more_data has been set, so if any other data arrives it will send a callback to thread pool
-                break;
+                // New data has arrived while setting no_more_data, so it should continue
+                // While setting status to 1 again, the value is still >=1 so no other thread will start
+                continue;
             }
         }
         else if (!ret)
@@ -237,8 +229,6 @@ 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

@@ -102,75 +102,44 @@ class Track
      * WORKAROUND:
      * A problem has been found in the use of Track within FastDDS Readers:
      * the on_data_available callback is called with the Reader mutex taken, so it may occur a deadlock while
-     * reading a data and receiving it at the same time from different threads, and this is a scenario that
-     * could happen with this design.
+     * reading a data and receiving it at the same time from different threads if on_data_available and read
+     * methods share a mutex.
      *
      * In order to avoid this deadlock, there is a DataAvailableStatus enumeration setting the status
      * of the data taking into account the Listener(listen) update and the Track(read) update.
+     * This enumeration works as numbers and not as enumeration (could be seen as a collection of constexpr)
      *
-     * The main point is to not have any mutex taken while take method is called in the Reader, but a mutex could
-     * be used to guard the access to the actual Track data available status.
+     * The main point is to not have to tak any mutex in on_data_available neither in read.
      */
     //! Status of the data available in the Track's Reader
-    enum class DataAvailableStatus
+    enum DataAvailableStatus
     {
-        new_data_arrived,   //! Listener has announced that new data has arrived
-        transmitting_data,  //! Track is taking data from the Reader, so it could or could not be data
-        no_more_data,       //! Track has announced that Reader has no more data, and Listener has not notified new data
+        no_more_data = 0,               //! Track has announced that Reader has no more data
+        transmitting_data = 1,          //! Track is taking data from the Reader, so it could or could not be data
+        new_data_arrived = 2 /* >2 */,  //! Listener has announced that new data has arrived
     };
 
     /**
      * Callback that will be called by the reader in case there is available data to be forwarded.
      *
-     * This method is sent to the Reader so it could call it when there is new data.
+     * This method is registered in the Reader so it could call it when there is new data.
      *
-     * This method will set the variable \c data_available_status_ to \c new_data_arrived and awake the transmit thread.
-     * If Track is disabled, the callback will be lost.
+     * This method will add the variable \c data_available_status_ in \c new_data_arrived .
+     * It will emit a task to execute transmit in a different thread if there was no previous thread before.
      */
     void data_available_() noexcept;
 
     /**
-     * @brief Whether there is data waiting to be taken in the Reader
-     *
-     * The times there is data is when \c data_available_status_ is set as \c new_data_arrived or \c transmitting_data
-     *
-     * @return true if there is available data
-     * @return false otherwise
-     */
-    bool is_data_available_() const noexcept;
-
-    /**
-     * Callback that will be called when there is no more data available to be forwarded.
-     *
-     * @note: this method is called from the Track after receiving a NO_DATA from Reader. But during the time to
-     * set \c data_available_status_ the Listener could notify new data (it is not possible to guard this
-     * behaviour as no shared mutex could be locked in transmit and listen because of FastDDS Reader mutex taken
-     * while \c on_data_available callback). If this happens, it should not be set as NO_DATA, but as new data.
-     * If this happens, the transmit thread will stop transmit loop, it will arrive to wait and it will automatically
-     * exit it as there is actual data to be sent, so there is no case where it gets stopped with new data available.
-     */
-    void no_more_data_available_() noexcept;
-
-    /**
-     * Whether this Track is enabled
+     * Whether this Track is enabled and should not exit.
      *
      * This method does not lock a mutex as it only acces atomic values to read them.
      */
     bool should_transmit_() noexcept;
 
-    /**
-     * Main function of Track.
-     * It waits for data to be available
-     * Once there is data available, call \c transmit_ till there is no data to be forwarded, and turn back to read
-     *
-     * Transmission is not executed in case track must be terminated or is not enabled.
-     */
-    void transmit_thread_function_() noexcept;
-
     /**
      * Take data from the Reader \c source and send this data through every writer in \c targets .
      *
-     * When no more data is available, call \c no_more_data_available_ and exit.
+     * When no more data is available, set \c data_available_status_ as \c no_more_data .
      *
      * It could exit without having finished transmitting all the data if track should terminate or track becomes
      * disabled.
@@ -226,18 +195,16 @@ class Track
      * Current status of the data available
      *
      * There are 3 states:
-     * \c DataAvailableStatus::new_data_arrived  : Reader Listener has notified that there are new data
-     * \c DataAvailableStatus::transmitting_data : Track is currently taking data, so there may or may not be data available
-     * \c DataAvailableStatus::no_more_data      : Track has received a NO_DATA from Reader
+     * \c 0 no_more_data      : Track has received a NO_DATA from Reader and no data has been set from on_data_available
+     * \c 1 transmitting_data : Track is currently taking data, so there may or may not be data available
+     * \c >1 new_data_arrived  : Reader Listener has notified that there are new data
      *
-     * This variable is protected by \c data_available_mutex_
+     * This variable does not need to be protected as it is atomic.
      */
-    std::atomic<DataAvailableStatus> data_available_status_;
-
-    std::mutex data_available_status_mutex_;
+    std::atomic<unsigned int> data_available_status_;
 
     /**
-     * Mutex to guard while the Track is sending a message.
+     * Mutex to guard while the Track is sending a message so it could not be disabled.
      */
     std::mutex on_transmission_mutex_;
 

docs/rst/notes/forthcoming_version.rst

@@ -27,3 +27,7 @@ Next release will include the following **major changes**:
 
 * ``wan`` Participant Kind uses now Initial Peers Discovery Protocol, while Discovery Server
   requires a new Participant Kind ``wan-discovery-server``.
+
+Next release will include the following **performance improvements**:
+
+* No locking in the :code:`Track` hot-path.