Add wait_all_consumed method (#62)

* Add string to Timestamp converter

Signed-off-by: Juan López Fernández <juanlopez@eprosima.com>

* Add wait_all_consumed method

Signed-off-by: Juan López Fernández <juanlopez@eprosima.com>

* Notify all when threshold reached

Signed-off-by: Juan López Fernández <juanlopez@eprosima.com>

* Wait on a different condition variable

Signed-off-by: Juan López Fernández <juanlopez@eprosima.com>

---------

Signed-off-by: Juan López Fernández <juanlopez@eprosima.com>

Author: juanlofer-eprosima

cpp_utils/include/cpp_utils/thread_pool/pool/SlotThreadPool.hpp

@@ -112,6 +112,18 @@ class SlotThreadPool
             const TaskId& task_id,
             Task&& task);
 
+    /**
+     * @brief Wait until all queued tasks are executed.
+     *
+     * This method will wait until all scheduled tasks are executed.
+     * In case there is no programmed task at the moment of calling this method, it returns immediately.
+     *
+     * @param timeout maximum time to wait in milliseconds. If 0, not time limit. [default 0].
+     * @return AwakeReason Whether the method returned due to timeout or because all tasks were executed.
+     */
+    CPP_UTILS_DllAPI utils::event::AwakeReason wait_all_consumed(
+            const utils::Duration_ms& timeout = 0);
+
 protected:
 
     /**

cpp_utils/include/cpp_utils/wait/ConsumerWaitHandler.hpp

@@ -108,6 +108,21 @@ class ConsumerWaitHandler : protected CounterWaitHandler
     T consume(
             const utils::Duration_ms& timeout = 0);
 
+    /////
+    // Synchronization methods
+
+    /**
+     * @brief Wait until all elements in the internal collection are consumed.
+     *
+     * This method will wait until there is no more data to be consumed in the internal collection.
+     * In case there is no data at the moment of calling this method, it returns immediately.
+     *
+     * @param timeout maximum time to wait in milliseconds. If 0, not time limit. [default 0].
+     * @return AwakeReason Whether the method returned due to timeout or because all elements were consumed.
+     */
+    AwakeReason wait_all_consumed(
+            const utils::Duration_ms& timeout = 0);
+
 protected:
 
     /**

cpp_utils/include/cpp_utils/wait/CounterWaitHandler.hpp

@@ -99,6 +99,16 @@ class CounterWaitHandler : protected WaitHandler<CounterType>
     CPP_UTILS_DllAPI AwakeReason wait_and_decrement(
             const utils::Duration_ms& timeout = 0) noexcept;
 
+    /**
+     * @brief Wait current thread until counter reaches \c threshold.
+     *
+     * @param timeout maximum time in milliseconds that should wait until awaking for timeout
+     *
+     * @return reason why thread was awaken
+     */
+    CPP_UTILS_DllAPI AwakeReason wait_threshold_reached(
+            const utils::Duration_ms& timeout = 0) noexcept;
+
     /////
     // Value methods
 
@@ -121,6 +131,8 @@ class CounterWaitHandler : protected WaitHandler<CounterType>
     void decrease_1_nts_();
 
     const CounterType threshold_;
+
+    std::condition_variable threshold_reached_cv_;
 };
 
 } /* namespace event */

cpp_utils/include/cpp_utils/wait/impl/ConsumerWaitHandler.ipp

@@ -78,6 +78,13 @@ T ConsumerWaitHandler<T>::consume(
     }
 }
 
+template <typename T>
+AwakeReason ConsumerWaitHandler<T>::wait_all_consumed(
+        const utils::Duration_ms& timeout /* = 0 */)
+{
+    return wait_threshold_reached(timeout);
+}
+
 } /* namespace event */
 } /* namespace utils */
 } /* namespace eprosima */

cpp_utils/src/cpp/event/FileWatcherHandler.cpp

@@ -35,7 +35,7 @@ FileWatcherHandler::FileWatcherHandler(
     , filewatcher_started_(false)
 {
     logDebug(
-        DDSROUTER_PERIODICHANDLER,
+        UTILS_PERIODICHANDLER,
         "FileWatcher Event Handler created with file path " << file_path_ << " .");
 }
 

cpp_utils/src/cpp/event/PeriodicEventHandler.cpp

@@ -38,7 +38,7 @@ PeriodicEventHandler::PeriodicEventHandler(
     }
 
     logDebug(
-        DDSROUTER_PERIODICHANDLER,
+        UTILS_PERIODICHANDLER,
         "Periodic Event Handler created with period time " << period_time_ << " .");
 }
 
@@ -93,7 +93,7 @@ void PeriodicEventHandler::start_period_thread_nts_() noexcept
         &PeriodicEventHandler::period_thread_routine_, this);
 
     logDebug(
-        DDSROUTER_PERIODICHANDLER,
+        UTILS_PERIODICHANDLER,
         "Periodic Event Handler thread starts with period time " << period_time_ << " .");
 }
 
@@ -110,7 +110,7 @@ void PeriodicEventHandler::stop_period_thread_nts_() noexcept
     period_thread_.join();
 
     logDebug(
-        DDSROUTER_PERIODICHANDLER,
+        UTILS_PERIODICHANDLER,
         "Periodic Event Handler thread stops.");
 }
 

cpp_utils/src/cpp/thread_pool/pool/SlotThreadPool.cpp

@@ -114,6 +114,12 @@ void SlotThreadPool::slot(
     }
 }
 
+utils::event::AwakeReason SlotThreadPool::wait_all_consumed(
+        const utils::Duration_ms& timeout /* = 0 */)
+{
+    return task_queue_.wait_all_consumed(timeout);
+}
+
 void SlotThreadPool::thread_routine_()
 {
     logDebug(UTILS_THREAD_POOL, "Starting thread routine: " << std::this_thread::get_id() << ".");

cpp_utils/src/cpp/wait/CounterWaitHandler.cpp

@@ -62,6 +62,62 @@ AwakeReason CounterWaitHandler::wait_and_decrement(
     return result;
 }
 
+AwakeReason CounterWaitHandler::wait_threshold_reached(
+        const utils::Duration_ms& timeout /* = 0 */) noexcept
+{
+    // Do wait with mutex taken
+    std::unique_lock<std::mutex> lock(wait_condition_variable_mutex_);
+
+    // Check if it is disabled and exit
+    if (!enabled())
+    {
+        return AwakeReason::disabled;
+    }
+
+    // Increment number of threads waiting
+    // WARNING: mutex must be taken
+    threads_waiting_++;
+
+    utils::Timestamp time_to_wait_until;
+
+    // If timeout is 0, use wait, if not use wait for timeout
+    if (timeout > 0)
+    {
+        time_to_wait_until = utils::now() + utils::duration_to_ms(timeout);
+    }
+    else
+    {
+        time_to_wait_until = utils::the_end_of_time();
+    }
+
+    bool finished_for_condition_met = threshold_reached_cv_.wait_until(
+        lock,
+        time_to_wait_until,
+        [this]
+        {
+            // Exit if threshold reached or if this has been disabled
+            return !enabled_.load() || (value_ == threshold_);
+        });
+
+    // Decrement number of threads waiting
+    // NOTE: mutex is still taken
+    threads_waiting_--;
+
+    // Check awake reason. Mutex is taken so it can not change while checking
+    if (!enabled_.load())
+    {
+        return AwakeReason::disabled;
+    }
+    else if (finished_for_condition_met)
+    {
+        return AwakeReason::condition_met;
+    }
+    else
+    {
+        return AwakeReason::timeout;
+    }
+}
+
 CounterWaitHandler& CounterWaitHandler::operator ++()
 {
     // NOTE: This operation could be done using the WaitHandler methods, but it will be less efficient than
@@ -91,6 +147,11 @@ void CounterWaitHandler::decrease_1_nts_()
     {
         wait_condition_variable_.notify_one();
     }
+    // If the threshold is reached, notify threads waiting for this event
+    else if (value_ == threshold_)
+    {
+        threshold_reached_cv_.notify_all();
+    }
 }
 
 } /* namespace event */