Add concurrency to liquidated vault validation

src/masternodes/rpc_accounts.cpp

@@ -2009,54 +2009,23 @@ UniValue getburninfo(const JSONRPCRequest& request) {
         }
     }
 
-    class WorkerResultPool {
-      public:
-        explicit WorkerResultPool(size_t size) {
-            pool.reserve(size);
-            for (size_t i = 0; i < size; i++) {
-                pool.push_back(CGetBurnInfoResult{});
-            }
-        }
-
-        std::shared_ptr<CGetBurnInfoResult> Acquire() {
-            LOCK(syncFlag);
-            auto res = std::make_shared<CGetBurnInfoResult>(pool.back());
-            pool.pop_back();
-
-            return res;
-        }
-
-        void Release(std::shared_ptr<CGetBurnInfoResult> res) {
-            LOCK(syncFlag);
-            pool.push_back(*res);
-        }
-
-        std::vector<CGetBurnInfoResult> &GetContainer() {
-            return pool;
-        }
-
-      private:
-        CCriticalSection syncFlag;
-        std::vector<CGetBurnInfoResult> pool;
-    };
-
     auto nWorkers = DfTxTaskPool->GetAvailableThreads();
     if (static_cast<size_t>(height) < nWorkers) {
         nWorkers = height;
     }
 
-    const auto chunks = height / nWorkers;
+    const auto chunkSize = height / nWorkers;
 
     TaskGroup g;
-    WorkerResultPool resultsPool{nWorkers};
+    BufferPool<CGetBurnInfoResult> resultsPool{nWorkers};
 
     auto &pool = DfTxTaskPool->pool;
     auto processedHeight = initialResult.height;
     auto i               = 0;
     while (processedHeight < height)
     {
-        auto startHeight = initialResult.height + (chunks * (i + 1));
-        auto stopHeight  = initialResult.height + (chunks * (i));
+        auto startHeight = initialResult.height + (chunkSize * (i + 1));
+        auto stopHeight  = initialResult.height + (chunkSize * (i));
 
         g.AddTask();
         boost::asio::post(pool, [startHeight, stopHeight, &g, &resultsPool] {
@@ -2139,23 +2108,25 @@ UniValue getburninfo(const JSONRPCRequest& request) {
             g.RemoveTask();
         });
 
-        // perfect accuracy: processedHeight += (startHeight > height) ? chunksRemainder : chunks;
-        processedHeight += chunks;
+        // perfect accuracy: processedHeight += (startHeight > height) ? chunksRemainder : chunkSize;
+        processedHeight += chunkSize;
         i++;
     }
 
     g.WaitForCompletion();
 
-    for (const auto &r : resultsPool.GetContainer()) {
-        totalResult->burntDFI += r.burntDFI;
-        totalResult->burntFee += r.burntFee;
-        totalResult->auctionFee += r.auctionFee;
-        totalResult->burntTokens.AddBalances(r.burntTokens.balances);
-        totalResult->nonConsortiumTokens.AddBalances(r.nonConsortiumTokens.balances);
-        totalResult->dexfeeburn.AddBalances(r.dexfeeburn.balances);
-        totalResult->paybackFee.AddBalances(r.paybackFee.balances);
+    for (const auto &r : resultsPool.GetBuffer()) {
+        totalResult->burntDFI += r->burntDFI;
+        totalResult->burntFee += r->burntFee;
+        totalResult->auctionFee += r->auctionFee;
+        totalResult->burntTokens.AddBalances(r->burntTokens.balances);
+        totalResult->nonConsortiumTokens.AddBalances(r->nonConsortiumTokens.balances);
+        totalResult->dexfeeburn.AddBalances(r->dexfeeburn.balances);
+        totalResult->paybackFee.AddBalances(r->paybackFee.balances);
     }
 
+    GetMemoizedResultCache().Set(request, {height, hash, *totalResult});
+
     CDataStructureV0 liveKey = {AttributeTypes::Live, ParamIDs::Economy, EconomyKeys::ConsortiumMinted};
     auto balances = attributes->GetValue(liveKey, CConsortiumGlobalMinted{});
 
@@ -2189,7 +2160,6 @@ UniValue getburninfo(const JSONRPCRequest& request) {
     result.pushKV("dfip2203", AmountsToJSON(dfi2203Tokens.balances));
     result.pushKV("dfip2206f", AmountsToJSON(dfiToDUSDTokens.balances));
 
-    GetMemoizedResultCache().Set(request, {height, hash, *totalResult});
     return GetRPCResultCache()
         .Set(request, result);
 }

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;