[FOLD] Have Workers use join(), also rename member variables

Author: scottschurr

Builds/CMake/RippledCore.cmake

@@ -796,7 +796,6 @@ if (tests)
     src/test/core/CryptoPRNG_test.cpp
     src/test/core/JobQueue_test.cpp
     src/test/core/SociDB_test.cpp
-    src/test/core/Workers_test.cpp
     #[===============================[
        test sources:
          subdir: csf

Builds/levelization/results/ordering.txt

@@ -121,7 +121,6 @@ test.core > ripple.basics
 test.core > ripple.beast
 test.core > ripple.core
 test.core > ripple.crypto
-test.core > ripple.json
 test.core > ripple.server
 test.core > test.jtx
 test.core > test.toplevel

src/ripple/core/impl/Workers.cpp

@@ -26,17 +26,18 @@
 namespace ripple {
 
 Workers::Workers(Callback& callback, std::string name, unsigned int count)
-    : m_callback(callback)
+    : callback_(callback)
 {
     assert(count != 0);
 
+    threads_.reserve(count);
     while (count--)
     {
-        std::thread t(
+        threads_.emplace_back(
             [this, name](unsigned int instance) {
                 auto const n = name + ":" + std::to_string(instance);
 
-                threads_++;
+                ++threadCount_;
 
                 while (!stopping_)
                 {
@@ -48,19 +49,19 @@ Workers::Workers(Callback& callback, std::string name, unsigned int count)
                     beast::setCurrentThreadName(n + " [zZz]");
 
                     // If there's nothing to do, go to sleep
-                    while (h == t && !stopping_.load(std::memory_order_relaxed))
+                    while (h == t && !stopping_.load())
                     {
-                        paused_++;
+                        ++paused_;
                         {
-                            std::unique_lock lock(m_mut);
-                            m_cv.wait_for(lock, std::chrono::seconds(5));
+                            std::unique_lock lock(mut_);
+                            cv_.wait_for(lock, std::chrono::seconds(5));
                         }
                         t = tail_.load();
                         h = head_.load();
-                        paused_--;
+                        --paused_;
                     }
 
-                    if (!stopping_.load(std::memory_order_relaxed) && h > t)
+                    if (!stopping_.load() && h > t)
                     {
                         // As long as we aren't stopping and there is a task
                         // that's waiting, we try to do work.
@@ -74,44 +75,44 @@ Workers::Workers(Callback& callback, std::string name, unsigned int count)
                             // just in case, if one does catch it.
                             try
                             {
-                                m_callback.processTask(instance);
+                                callback_.processTask(instance);
                             }
                             catch (...)
                             {
-                                m_callback.uncaughtException(
+                                callback_.uncaughtException(
                                     instance, std::current_exception());
                             }
                         }
                     }
                 }
 
                 // Track number of threads
-                threads_--;
+                --threadCount_;
             },
             count);
-        t.detach();
     };
 }
 
 Workers::~Workers()
 {
-    if (threads_.load() != 0)
+    if (threadCount_.load() != 0)
         stop();
 
-    assert(stopping_.load() && threads_.load() == 0);
+    assert(stopping_.load() && threadCount_.load() == 0);
 }
 
 void
 Workers::stop()
 {
-    if (threads_.load() != 0)
+    if (threadCount_.load() != 0)
     {
-        if (!stopping_.exchange(true) || (threads_.load() != 0))
-            m_cv.notify_all();
+        if (!stopping_.exchange(true) || (threadCount_.load() != 0))
+            cv_.notify_all();
     }
 
-    while (threads_.load() != 0)
-        std::this_thread::sleep_for(std::chrono::seconds(1));
+    for (auto& thread : threads_)
+        if (thread.joinable())
+            thread.join();
 }
 
 void
@@ -123,7 +124,7 @@ Workers::addTask()
     head_++;
 
     if (paused_ != 0)
-        m_cv.notify_one();
+        cv_.notify_one();
 }
 
 }  // namespace ripple

src/ripple/core/impl/Workers.h

@@ -25,6 +25,7 @@
 #include <exception>
 #include <mutex>
 #include <string>
+#include <vector>
 
 namespace ripple {
 
@@ -34,7 +35,7 @@ namespace ripple {
     the number of outstanding tasks, and dispatches work. When the pool detects
     that there is no work to be done, it puts threads to sleep.
 
-    The pool does not decide make decisions about which task to run; that is
+    The pool does not make decisions about which task to run; that is
     handled by a callback that the pool invokes.  This makes it possible to
     implement the work dispatch strategy (e.g. FIFO or priority queues) that
     makes sense without requiring changes in the thread pool itself.
@@ -93,8 +94,7 @@ class Workers
 
     /** Create a new thread pool.
 
-        The number of initial threads must be specified. The
-        default is to create one thread per CPU.
+        The number of initial threads must be specified.
 
         @param callback The task selection & execution algorithm.
         @param name The name for this pool (used to name threads)
@@ -108,7 +108,7 @@ class Workers
     unsigned int
     count() const noexcept
     {
-        return threads_.load();
+        return threadCount_.load();
     }
 
     /** Pause all threads and wait until they are paused.
@@ -134,19 +134,21 @@ class Workers
     addTask();
 
 private:
-    Callback& m_callback;
+    Callback& callback_;
+
+    std::vector<std::thread> threads_;
 
     /// Used to signal sleeping threads to wake up.
-    std::condition_variable m_cv;
+    std::condition_variable cv_;
 
     /// Used only to wait on the condition variable.
-    std::mutex m_mut;
+    std::mutex mut_;
 
     /// True if the worker threads should stop.
     std::atomic<bool> stopping_ = false;
 
     /// This represents the total number of threads in the thread pool:
-    std::atomic<unsigned int> threads_ = 0;
+    std::atomic<unsigned int> threadCount_ = 0;
 
     /// The number of threads that are paused at this time:
     std::atomic<unsigned int> paused_ = 0;