fix: Remove tight loop in ETLng (#2398)

src/etlng/impl/TaskManager.cpp

@@ -155,6 +155,10 @@ TaskManager::spawnLoader(TaskQueue& queue)
                                  << " tps[" << txnCount / seconds << "], ops[" << objCount / seconds << "]";
 
                 monitor_.get().notifySequenceLoaded(data->seq);
+            } else {
+                // TODO (https://github.com/XRPLF/clio/issues/1852) this is probably better done with a timeout (on
+                // coroutine) so that the thread itself is not blocked
+                queue.awaitTask();
             }
         }
 

src/etlng/impl/TaskQueue.hpp

@@ -22,8 +22,12 @@
 #include "etlng/Models.hpp"
 #include "util/Mutex.hpp"
 
+#include <boost/atomic/atomic.hpp>
+
+#include <condition_variable>
 #include <cstddef>
 #include <cstdint>
+#include <mutex>
 #include <optional>
 #include <queue>
 #include <utility>
@@ -57,6 +61,8 @@ class TaskQueue {
     };
 
     util::Mutex<Data> data_;
+    std::condition_variable cv_;
+    boost::atomics::atomic_bool stopping_ = false;
 
 public:
     struct Settings {
@@ -75,6 +81,13 @@ public:
     {
     }
 
+    ~TaskQueue()
+    {
+        // unblock all waiters
+        stopping_ = true;
+        cv_.notify_all();
+    }
+
     /**
      * @brief Enqueue a new item onto the queue if space is available
      * @note This function blocks until the item is attempted to be added to the queue
@@ -89,6 +102,8 @@ public:
 
         if (limit_ == 0uz or lock->forwardLoadQueue.size() < limit_) {
             lock->forwardLoadQueue.push(std::move(item));
+            cv_.notify_all();
+
             return true;
         }
 
@@ -126,6 +141,17 @@ public:
     {
         return data_.lock()->forwardLoadQueue.empty();
     }
+
+    /**
+     * @brief Awaits for the queue to become non-empty
+     * @note This function blocks until there is a task or the queue is being destroyed
+     */
+    void
+    awaitTask()
+    {
+        auto lock = data_.lock<std::unique_lock>();
+        cv_.wait(lock, [&] { return stopping_ or not lock->forwardLoadQueue.empty(); });
+    }
 };
 
 }  // namespace etlng::impl

src/util/Mutex.hpp

@@ -76,6 +76,11 @@ public:
     {
         return &data_;
     }
+
+    operator LockType<MutexType>&()
+    {
+        return lock_;
+    }
     /** @endcond */
 
 private: