Introduce a generic thread-safe buffer pool

src/masternodes/threadpool.cpp

@@ -31,16 +31,17 @@ void ShutdownDfTxGlobalTaskPool() {
 
 
 void TaskGroup::AddTask() { 
-    tasks.fetch_add(1, std::memory_order_relaxed);
+    tasks.fetch_add(1, std::memory_order_release);
 }
 
 void TaskGroup::RemoveTask() {
-    if (tasks.fetch_sub(1, std::memory_order_seq_cst) == 1) {
-        cv.notify_one();
+    if (tasks.fetch_sub(1, std::memory_order_acq_rel) == 1) {
+        cv.notify_all();
     }
 }
 
-void TaskGroup::WaitForCompletion() {
+void TaskGroup::WaitForCompletion(bool checkForPrematureCompletion) {
+    if (checkForPrematureCompletion && tasks.load() == 0) return;
     std::unique_lock<std::mutex> l(cv_m);
     cv.wait(l, [&] { return tasks.load() == 0; });
 }

src/masternodes/threadpool.h

@@ -2,7 +2,9 @@
 #define DEFI_MASTERNODES_THREADPOOL_H
 
 #include <boost/asio.hpp>
+#include <atomic>
 #include <condition_variable>
+#include <sync.h>
 
 static const int DEFAULT_DFTX_WORKERS=0;
 
@@ -33,11 +35,44 @@ class TaskGroup {
     public:
         void AddTask();
         void RemoveTask();
-        void WaitForCompletion();
+        void WaitForCompletion(bool checkForPrematureCompletion = true);
+        void MarkCancellation() { is_cancelled.store(true); }
+        bool IsCancelled() { return is_cancelled.load(); }
+
     private:
         std::atomic<uint64_t> tasks{0};
         std::mutex cv_m;
         std::condition_variable cv;
+        std::atomic_bool is_cancelled{false};
+};
+
+template <typename T>
+class BufferPool {
+    public:
+    explicit BufferPool(size_t size) {
+        pool.reserve(size);
+        for (size_t i = 0; i < size; i++) {
+            pool.push_back(std::make_shared<T>());
+        }
+    }
+
+    std::shared_ptr<T> Acquire() {
+        std::unique_lock l{m};
+        auto res = pool.back();
+        pool.pop_back();
+        return res;
+    }
+
+    void Release(std::shared_ptr<T> res) {
+        std::unique_lock l{m};
+        pool.push_back(res);
+    }
+
+    std::vector<std::shared_ptr<T>> &GetBuffer() { return pool; }
+
+    private:
+    AtomicMutex m{};
+    std::vector<std::shared_ptr<T>> pool;
 };
 
 extern std::unique_ptr<TaskPool> DfTxTaskPool;

src/sync.h

@@ -331,17 +331,81 @@ struct SCOPED_LOCKABLE LockAssertion
     ~LockAssertion() UNLOCK_FUNCTION() {}
 };
 
+class AtomicMutex {
+private:
+    std::atomic<bool> flag{false};
+    int64_t spins;
+    int64_t yields;
+
+public:
+    AtomicMutex(int64_t spins = 10, int64_t yields = 16): 
+        spins(spins), yields(yields) {}
+
+    void lock() {
+        // Note: The loop here addresses both, spurious failures as well
+        // as to suspend or spin wait until it's set
+        // Additional:
+        // - We use this a lock for external critical section, so we use
+        // seq ordering, to ensure it provides the right ordering guarantees
+        // for the others
+        // On failure of CAS, we don't care about the existing value, we just
+        // discard it, so relaxed ordering is sufficient.
+        bool expected = false;
+        auto i = 0;
+        while (std::atomic_compare_exchange_weak_explicit(
+                   &flag,
+                   &expected, true,
+                   std::memory_order_seq_cst,
+                   std::memory_order_relaxed) == false) {
+            // Could have been a spurious failure or another thread could have taken the
+            // lock in-between since we're now out of the atomic ops. 
+            // Reset expected to start from scratch again, since we only want
+            // a singular atomic false -> true transition.
+            expected = false;
+            if (i > spins) {
+                if (i > spins + yields) {
+                    // Use larger sleep, in line with the largest quantum, which is 
+                    // Windows with 16ms
+                    std::this_thread::sleep_for(std::chrono::milliseconds(16));
+                } else {
+                    std::this_thread::yield();
+                }
+            }
+            i++;
+        }
+    }
+
+    void unlock() {
+        flag.store(false, std::memory_order_seq_cst);
+    }
+
+    bool try_lock() noexcept {
+        // We locked it if and only if it was a false -> true transition.
+        // Otherwise, we just re-wrote an already existing value as true which is harmless 
+        // We could theoritically use CAS here to prevent the additional write, but
+        // but requires loop on weak, or using strong. Simpler to just use an exchange for
+        // for now, since all ops are seq_cst anyway.
+        return !flag.exchange(true, std::memory_order_seq_cst);
+    }
+};
+
 class CLockFreeGuard
 {
     std::atomic_bool& lock;
 public:
     CLockFreeGuard(std::atomic_bool& lock) : lock(lock)
     {
-        bool desired = false;
-        while (!lock.compare_exchange_weak(desired, true,
-                                           std::memory_order_release,
-                                           std::memory_order_relaxed)) {
-            desired = false;
+        bool expected = false;
+        while (std::atomic_compare_exchange_weak_explicit(
+                   &lock,
+                   &expected, true,
+                   std::memory_order_seq_cst,
+                   std::memory_order_relaxed) == false) {
+            // Could have been a spurious failure or another thread could have taken the
+            // lock in-between since we're now out of the atomic ops. 
+            // Reset expected to start from scratch again, since we only want
+            // a singular atomic false -> true transition.
+            expected = false;
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
     }