[#1599] addressed comments

diff --git a/src/lib/util/tests/multi_threading_mgr_unittest.cc b/src/lib/util/tests/multi_threading_mgr_unittest.cc
index ddaf779dcfd5aa78ec97f6a6e7de36dab21b180b..9131a0a16ffeba8d9fddc150427aaa717c8e0ce8 100644 (file)
--- a/src/lib/util/tests/multi_threading_mgr_unittest.cc
+++ b/src/lib/util/tests/multi_threading_mgr_unittest.cc
@@ -34,13 +34,17 @@ struct MultiThreadingMgrTest : ::testing::Test {
     /// @param count The thread queue size.
     /// @param running The running threads count.
     /// @param in_cs Flag which indicates if running inside critical section.
+    /// @param enabled Flag which indicates if thread pool is started and running.
+    /// @param paused Flag which indicates if thread pool is started and paused.
     void checkState(bool mode, size_t size, size_t count, size_t running,
-                    bool in_cs = false) {
+                    bool in_cs = false, bool enabled = true, bool paused = false) {
         EXPECT_EQ(MultiThreadingMgr::instance().getMode(), mode);
         EXPECT_EQ(MultiThreadingMgr::instance().getThreadPoolSize(), size);
         EXPECT_EQ(MultiThreadingMgr::instance().getPacketQueueSize(), count);
         EXPECT_EQ(MultiThreadingMgr::instance().getThreadPool().size(), running);
         EXPECT_EQ(MultiThreadingMgr::instance().isInCriticalSection(), in_cs);
+        EXPECT_EQ(MultiThreadingMgr::instance().getThreadPool().enabled(), enabled);
+        EXPECT_EQ(MultiThreadingMgr::instance().getThreadPool().paused(), paused);
     }
 };
 

diff --git a/src/lib/util/tests/thread_pool_unittest.cc b/src/lib/util/tests/thread_pool_unittest.cc
index b45c09b2b9db4e18869997004c2aca1fa3d86cd8..9c367c914c9f2124140b8f0775acdb8804023e34 100644 (file)
--- a/src/lib/util/tests/thread_pool_unittest.cc
+++ b/src/lib/util/tests/thread_pool_unittest.cc
@@ -559,10 +559,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
     uint32_t thread_count;
     CallBack call_back;
     ThreadPool<CallBack> thread_pool;
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     items_count = 4;
     thread_count = 4;
@@ -575,38 +572,23 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // calling resume should do nothing if not started
     EXPECT_NO_THROW(thread_pool.resume());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     // do it once again to check if it works
     EXPECT_NO_THROW(thread_pool.resume());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     // calling pause should do nothing if not started
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     // do it once again to check if it works
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     // calling resume should do nothing if not started
     EXPECT_NO_THROW(thread_pool.resume());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     // add items to stopped thread pool
     for (uint32_t i = 0; i < items_count; ++i) {
@@ -615,38 +597,23 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
         EXPECT_TRUE(ret);
     }
 
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, items_count, 0);
 
     // calling pause should do nothing if not started
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, items_count, 0);
 
     // calling resume should do nothing if not started
     EXPECT_NO_THROW(thread_pool.resume());
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, items_count, 0);
 
     // calling pause should do nothing if not started
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, items_count, 0);
 
     // calling start should create the threads and should keep the queued items
     EXPECT_NO_THROW(thread_pool.start(thread_count));
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, items_count, thread_count);
 
     // calling resume should resume threads
     EXPECT_NO_THROW(thread_pool.resume());
@@ -655,10 +622,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // wait for all items to be processed
     waitTasks(thread_count, items_count);
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
     // as each thread pool thread is still waiting on main to unblock, each
     // thread should have been registered in ids list
     checkIds(items_count);
@@ -673,17 +637,11 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // do it once again to check if it works
     EXPECT_NO_THROW(thread_pool.resume());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
 
     // calling stop should clear all threads and should keep queued items
     EXPECT_NO_THROW(thread_pool.stop());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 
     items_count = 64;
     thread_count = 16;
@@ -702,10 +660,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
         EXPECT_TRUE(ret);
     }
 
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, items_count, 0);
 
     // calling start should create the threads and should keep the queued items
     EXPECT_NO_THROW(thread_pool.start(thread_count));
@@ -714,10 +669,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // wait for all items to be processed
     waitTasks(thread_count, items_count);
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
     // all items should have been processed
     ASSERT_EQ(count(), items_count);
 
@@ -726,17 +678,11 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // calling pause should pause threads
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
 
     // do it once again to check if it works
     EXPECT_NO_THROW(thread_pool.pause());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
 
     // prepare setup
     reset(thread_count);
@@ -748,10 +694,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
         EXPECT_TRUE(ret);
     }
 
-    // the item count should match
-    ASSERT_EQ(thread_pool.count(), items_count);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, items_count, thread_count);
 
     // calling resume should resume threads
     EXPECT_NO_THROW(thread_pool.resume());
@@ -760,10 +703,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // wait for all items to be processed
     waitTasks(thread_count, items_count);
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should match
-    ASSERT_EQ(thread_pool.size(), thread_count);
+    checkState(thread_pool, 0, thread_count);
     // all items should have been processed
     ASSERT_EQ(count(), items_count);
 
@@ -772,10 +712,7 @@ TEST_F(ThreadPoolTest, pauseAndResume) {
 
     // calling stop should clear all threads and should keep queued items
     EXPECT_NO_THROW(thread_pool.stop());
-    // the item count should be 0
-    ASSERT_EQ(thread_pool.count(), 0);
-    // the thread count should be 0
-    ASSERT_EQ(thread_pool.size(), 0);
+    checkState(thread_pool, 0, 0);
 }
 
 /// @brief test ThreadPool max queue size

diff --git a/src/lib/util/thread_pool.h b/src/lib/util/thread_pool.h
index aecd261ca313af7018066050d27888172bba2cd8..1b741e97c98c8d195543fa501006c94ed7bed7de 100644 (file)
--- a/src/lib/util/thread_pool.h
+++ b/src/lib/util/thread_pool.h
@@ -393,8 +393,7 @@ private:
                 wait_cv_.notify_all();
             }
             // Wait for push or disable functions.
-            cv_.wait(lock, [&]() {return (!enabled_ || !queue_.empty());});
-            pause_cv_.wait(lock, [&]() {return (!enabled_ || !paused_);});
+            cv_.wait(lock, [&]() {return (!enabled_ || (!queue_.empty() && !paused_));});
             ++working_;
             if (!enabled_) {
                 return (Item());
@@ -463,7 +462,7 @@ private:
         void resume() {
             std::unique_lock<std::mutex> lock(mutex_);
             paused_ = false;
-            pause_cv_.notify_all();
+            cv_.notify_all();
         }
 
         /// @brief get queue length statistic
@@ -512,7 +511,6 @@ private:
                 enabled_ = false;
             }
             // Notify pop so that it can exit.
-            pause_cv_.notify_all();
             cv_.notify_all();
         }
 
@@ -550,9 +548,6 @@ private:
         /// @brief condition variable used to wait for all threads to be paused
         std::condition_variable wait_threads_cv_;
 
-        /// @brief condition variable used to pause threads
-        std::condition_variable pause_cv_;
-
         /// @brief the state of the queue
         /// The 'enabled' state corresponds to true value
         /// The 'disabled' state corresponds to false value