Remove boost threads (#269)

* remove boost::chrono

* make script validation checkqueue threads use internal pool instead of global scheduler pool bitcoin#15632

* Revert "remove boost::chrono"

This reverts commit 434d8d7.

* remove boost::thread and use std::thread

* add cmath to fix build

* fix build - add missing headers, change boost::chrono to std chrono

* remove static from all critical section code to fix linking error

* remove CheckLastCritical from leave critical section

* replace boost::interuption point with std:: yield

* remove scheduling from bench assemble block

* fix hanging benchmark test

* fix test - stop scheduler threads separately.

* remove  this_thread::yield

* review feedback - fix unused boost::thread interrupt and  comments

* fix CI failure in CheckQueueControl. Replace c style interface with RAII class and unique_ptr

* change scheduler thread  in test_tapyrus to use the same syntax as tapyrusd

* fix initialization of scheduler  service thread
initialize m_request_stop variable in the constructor and handle  empty queue length separately

* port commits 0682003 to 95ad70a from bitcoin  to fix test hanging

* ifixing build - use move operations  in all vchecks

* review feedback - remove unused header

src/addrman.cpp

@@ -8,6 +8,7 @@
 #include <hash.h>
 #include <serialize.h>
 #include <streams.h>
+#include <cmath>
 
 int CAddrInfo::GetTriedBucket(const uint256& nKey) const
 {

src/bench/block_assemble.cpp

@@ -86,15 +86,15 @@ static void AssembleBlock(benchmark::State& state)
     InitSignatureCache();
     InitScriptExecutionCache();
 
-    boost::thread_group thread_group;
     CScheduler scheduler;
+    std::thread thread(std::bind(&CScheduler::serviceQueue, &scheduler));
     {
         ::pblocktree.reset(new CBlockTreeDB(1 << 20, true));
         ::pcoinsdbview.reset(new CCoinsViewDB(1 << 23, true));
         ::pcoinsTip.reset(new CCoinsViewCache(pcoinsdbview.get()));
 
-        thread_group.create_thread(std::bind(&CScheduler::serviceQueue, &scheduler));
         GetMainSignals().RegisterBackgroundSignalScheduler(scheduler);
+
         LoadGenesisBlock();
         CValidationState state;
         ActivateBestChain(state);
@@ -130,8 +130,9 @@ static void AssembleBlock(benchmark::State& state)
         PrepareBlock(SCRIPT_PUB);
     }
 
-    thread_group.interrupt_all();
-    thread_group.join_all();
+    scheduler.stop();
+    if(thread.joinable())
+        thread.join();
     GetMainSignals().FlushBackgroundCallbacks();
     GetMainSignals().UnregisterBackgroundSignalScheduler();
 }

src/bench/checkqueue.cpp

@@ -8,7 +8,6 @@
 #include <checkqueue.h>
 #include <prevector.h>
 #include <vector>
-#include <boost/thread/thread.hpp>
 #include <random.h>
 
 
@@ -36,11 +35,8 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::State& state)
         }
         void swap(PrevectorJob& x){p.swap(x.p);};
     };
-    CCheckQueue<PrevectorJob> queue {QUEUE_BATCH_SIZE};
-    boost::thread_group tg;
-    for (auto x = 0; x < std::max(MIN_CORES, GetNumCores()); ++x) {
-       tg.create_thread([&]{queue.Thread();});
-    }
+    int num_threads= GetNumCores() - 1;
+    CCheckQueue<PrevectorJob> queue {QUEUE_BATCH_SIZE, num_threads};
     while (state.KeepRunning()) {
         // Make insecure_rand here so that each iteration is identical.
         FastRandomContext insecure_rand(true);
@@ -50,13 +46,11 @@ static void CCheckQueueSpeedPrevectorJob(benchmark::State& state)
             vChecks.reserve(BATCH_SIZE);
             for (size_t x = 0; x < BATCH_SIZE; ++x)
                 vChecks.emplace_back(insecure_rand);
-            control.Add(vChecks);
+            control.Add(std::move(vChecks));
         }
         // control waits for completion by RAII, but
         // it is done explicitly here for clarity
         control.Wait();
     }
-    tg.interrupt_all();
-    tg.join_all();
 }
 BENCHMARK(CCheckQueueSpeedPrevectorJob, 1400);

src/checkqueue.h

@@ -10,8 +10,6 @@
 #include <algorithm>
 #include <vector>
 
-#include <boost/thread/condition_variable.hpp>
-#include <boost/thread/mutex.hpp>
 
 template <typename T>
 class CCheckQueueControl;
@@ -31,13 +29,13 @@ class CCheckQueue
 {
 private:
     //! Mutex to protect the inner state
-    boost::mutex mutex;
+    std::mutex mutex;
 
     //! Worker threads block on this when out of work
-    boost::condition_variable condWorker;
+    std::condition_variable condWorker;
 
     //! Master thread blocks on this when out of work
-    boost::condition_variable condMaster;
+    std::condition_variable condMaster;
 
     //! The queue of elements to be processed.
     //! As the order of booleans doesn't matter, it is used as a LIFO (stack)
@@ -62,17 +60,20 @@ class CCheckQueue
     //! The maximum number of elements to be processed in one batch
     unsigned int nBatchSize;
 
+    std::vector<std::thread> m_worker_threads;
+    bool m_request_stop;
+
     /** Internal function that does bulk of the verification work. */
     bool Loop(bool fMaster = false)
     {
-        boost::condition_variable& cond = fMaster ? condMaster : condWorker;
+        std::condition_variable& cond = fMaster ? condMaster : condWorker;
         std::vector<T> vChecks;
         vChecks.reserve(nBatchSize);
         unsigned int nNow = 0;
         bool fOk = true;
         do {
             {
-                boost::unique_lock<boost::mutex> lock(mutex);
+                WaitableLock lock(mutex);
                 // first do the clean-up of the previous loop run (allowing us to do it in the same critsect)
                 if (nNow) {
                     fAllOk &= fOk;
@@ -85,7 +86,7 @@ class CCheckQueue
                     nTotal++;
                 }
                 // logically, the do loop starts here
-                while (queue.empty()) {
+                while (queue.empty() && !m_request_stop) {
                     if (fMaster && nTodo == 0) {
                         nTotal--;
                         bool fRet = fAllOk;
@@ -99,19 +100,18 @@ class CCheckQueue
                     cond.wait(lock); // wait
                     nIdle--;
                 }
+                if (m_request_stop) {
+                    return false;
+                }
                 // Decide how many work units to process now.
                 // * Do not try to do everything at once, but aim for increasingly smaller batches so
                 //   all workers finish approximately simultaneously.
                 // * Try to account for idle jobs which will instantly start helping.
                 // * Don't do batches smaller than 1 (duh), or larger than nBatchSize.
                 nNow = std::max(1U, std::min(nBatchSize, (unsigned int)queue.size() / (nTotal + nIdle + 1)));
-                vChecks.resize(nNow);
-                for (unsigned int i = 0; i < nNow; i++) {
-                    // We want the lock on the mutex to be as short as possible, so swap jobs from the global
-                    // queue to the local batch vector instead of copying.
-                    vChecks[i].swap(queue.back());
-                    queue.pop_back();
-                }
+                auto start_it = queue.end() - nNow;
+                vChecks.assign(std::make_move_iterator(start_it), std::make_move_iterator(queue.end()));
+                queue.erase(start_it, queue.end());
                 // Check whether we need to do work at all
                 fOk = fAllOk;
             }
@@ -125,40 +125,66 @@ class CCheckQueue
 
 public:
     //! Mutex to ensure only one concurrent CCheckQueueControl
-    boost::mutex ControlMutex;
+    std::mutex ControlMutex;
 
     //! Create a new check queue
-    explicit CCheckQueue(unsigned int nBatchSizeIn) : nIdle(0), nTotal(0), fAllOk(true), nTodo(0), nBatchSize(nBatchSizeIn) {}
-
-    //! Worker thread
-    void Thread()
+    explicit CCheckQueue(unsigned int batch_size, int worker_threads_num) : nIdle(0), nTotal(0), fAllOk(true), nTodo(0), nBatchSize(batch_size), m_request_stop(false)
     {
-        Loop();
+        {
+            WaitableLock loc(mutex);
+            nIdle = 0;
+            nTotal = 0;
+            fAllOk = true;
+        }
+        assert(m_worker_threads.empty());
+        for (int n = 0; n < worker_threads_num; ++n) {
+            m_worker_threads.emplace_back([this, n]() {
+                RenameThread(strprintf("scriptch.%i", n));
+                Loop(false /* worker thread */);
+        });
+        }
     }
 
+    // Since this class manages its own resources, which is a thread
+    // pool `m_worker_threads`, copy and move operations are not appropriate.
+    CCheckQueue(const CCheckQueue&) = delete;
+    CCheckQueue& operator=(const CCheckQueue&) = delete;
+    CCheckQueue(CCheckQueue&&) = delete;
+    CCheckQueue& operator=(CCheckQueue&&) = delete;
+
     //! Wait until execution finishes, and return whether all evaluations were successful.
     bool Wait()
     {
         return Loop(true);
     }
 
     //! Add a batch of checks to the queue
-    void Add(std::vector<T>& vChecks)
+    void Add(std::vector<T>&& vChecks)
     {
-        boost::unique_lock<boost::mutex> lock(mutex);
-        for (T& check : vChecks) {
-            queue.push_back(T());
-            check.swap(queue.back());
+        if (vChecks.empty()) {
+            return;
+        }
+
+        {
+            WaitableLock lock(mutex);
+            queue.insert(queue.end(), std::make_move_iterator(vChecks.begin()), std::make_move_iterator(vChecks.end()));
+            nTodo += vChecks.size();
         }
-        nTodo += vChecks.size();
+
         if (vChecks.size() == 1)
             condWorker.notify_one();
         else if (vChecks.size() > 1)
             condWorker.notify_all();
     }
 
+    //! Stop all of the worker threads.
     ~CCheckQueue()
     {
+        m_request_stop = true;
+        condWorker.notify_all();
+        for (std::thread& t : m_worker_threads) {
+            t.join();
+        }
     }
 
 };
@@ -195,10 +221,10 @@ class CCheckQueueControl
         return fRet;
     }
 
-    void Add(std::vector<T>& vChecks)
+    void Add(std::vector<T>&& vChecks)
     {
         if (pqueue != nullptr)
-            pqueue->Add(vChecks);
+            pqueue->Add(std::move(vChecks));
     }
 
     ~CCheckQueueControl()

src/index/base.cpp

@@ -263,7 +263,7 @@ void BaseIndex::Start()
         return;
     }
 
-    m_thread_sync = std::thread(&TraceThread<std::function<void()>>, GetName(),
+    m_thread_sync = std::thread(&TraceThread, GetName(),
                                 std::bind(&BaseIndex::ThreadSync, this));
 }
 

src/index/txindex.cpp

@@ -152,7 +152,6 @@ bool TxIndex::DB::MigrateData(CBlockTreeDB& block_tree_db, const CBlockLocator&
     bool interrupted = false;
     std::unique_ptr<CDBIterator> cursor(block_tree_db.NewIterator());
     for (cursor->Seek(begin_key); cursor->Valid(); cursor->Next()) {
-        boost::this_thread::interruption_point();
         if (ShutdownRequested()) {
             interrupted = true;
             break;

src/init.cpp

@@ -60,7 +60,8 @@
 #include <boost/algorithm/string/replace.hpp>
 #include <boost/algorithm/string/split.hpp>
 #include <boost/interprocess/sync/file_lock.hpp>
-#include <boost/thread.hpp>
+
+#include <thread>
 
 #if ENABLE_ZMQ
 #include <zmq/zmqnotificationinterface.h>
@@ -166,7 +167,6 @@ class CCoinsViewErrorCatcher final : public CCoinsViewBacked
 static std::unique_ptr<CCoinsViewErrorCatcher> pcoinscatcher;
 static std::unique_ptr<ECCVerifyHandle> globalVerifyHandle;
 
-static boost::thread_group threadGroup;
 static CScheduler scheduler;
 
 void Interrupt()
@@ -214,10 +214,9 @@ void Shutdown()
 
     StopTorControl();
 
-    // After everything has been shut down, but before things get flushed, stop the
-    // CScheduler/checkqueue threadGroup
-    threadGroup.interrupt_all();
-    threadGroup.join_all();
+    // After everything has been shut down, but before things get flushed,
+    //stop scheduler and load block threads.
+    scheduler.stop();
 
     // After the threads that potentially access these pointers have been stopped,
     // destruct and reset all to nullptr.
@@ -1230,13 +1229,12 @@ bool AppInitMain()
 
     LogPrintf("Using %u threads for script verification\n", nScriptCheckThreads);
     if (nScriptCheckThreads) {
-        for (int i=0; i<nScriptCheckThreads-1; i++)
-            threadGroup.create_thread(&ThreadScriptCheck);
+        StartScriptCheckWorkerThreads(nScriptCheckThreads);
     }
 
     // Start the lightweight task scheduler thread
     CScheduler::Function serviceLoop = std::bind(&CScheduler::serviceQueue, &scheduler);
-    threadGroup.create_thread(std::bind(&TraceThread<CScheduler::Function>, "scheduler", serviceLoop));
+    scheduler.m_service_thread = std::thread(&TraceThread, "scheduler", serviceLoop);
 
     // Gather some entropy once per minute.
     scheduler.scheduleEvery([]{
@@ -1634,7 +1632,7 @@ bool AppInitMain()
         vImportFiles.push_back(strFile);
     }
 
-    threadGroup.create_thread(std::bind(&ThreadImport, vImportFiles, fReloadxfield));
+    scheduler.m_load_block = std::thread(std::bind(&ThreadImport, vImportFiles, fReloadxfield));
 
     // Wait for genesis block to be processed
     {

src/interfaces/node.cpp

@@ -41,7 +41,6 @@
 #endif
 
 #include <atomic>
-#include <boost/thread/thread.hpp>
 #include <univalue.h>
 
 namespace interfaces {

src/net.cpp

@@ -1564,7 +1564,7 @@ void StartMapPort()
 {
     if (!g_upnp_thread.joinable()) {
         assert(!g_upnp_interrupt);
-        g_upnp_thread = std::thread((std::bind(&TraceThread<void (*)()>, "upnp", &ThreadMapPort)));
+        g_upnp_thread = std::thread((std::bind(&TraceThread, "upnp", &ThreadMapPort)));
     }
 }
 
@@ -2354,15 +2354,15 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
     }
 
     // Send and receive from sockets, accept connections
-    threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));
+    threadSocketHandler = std::thread(&TraceThread, "net", std::bind(&CConnman::ThreadSocketHandler, this));
 
     if (!gArgs.GetBoolArg("-dnsseed", true))
         LogPrintf("DNS seeding disabled\n");
     else
-        threadDNSAddressSeed = std::thread(&TraceThread<std::function<void()> >, "dnsseed", std::function<void()>(std::bind(&CConnman::ThreadDNSAddressSeed, this)));
+        threadDNSAddressSeed = std::thread(&TraceThread, "dnsseed", std::bind(&CConnman::ThreadDNSAddressSeed, this));
 
     // Initiate outbound connections from -addnode
-    threadOpenAddedConnections = std::thread(&TraceThread<std::function<void()> >, "addcon", std::function<void()>(std::bind(&CConnman::ThreadOpenAddedConnections, this)));
+    threadOpenAddedConnections = std::thread(&TraceThread, "addcon", std::bind(&CConnman::ThreadOpenAddedConnections, this));
 
     if (connOptions.m_use_addrman_outgoing && !connOptions.m_specified_outgoing.empty()) {
         if (clientInterface) {
@@ -2373,10 +2373,10 @@ bool CConnman::Start(CScheduler& scheduler, const Options& connOptions)
         return false;
     }
     if (connOptions.m_use_addrman_outgoing || !connOptions.m_specified_outgoing.empty())
-        threadOpenConnections = std::thread(&TraceThread<std::function<void()> >, "opencon", std::function<void()>(std::bind(&CConnman::ThreadOpenConnections, this, connOptions.m_specified_outgoing)));
+        threadOpenConnections = std::thread(&TraceThread, "opencon", std::bind(&CConnman::ThreadOpenConnections, this, connOptions.m_specified_outgoing));
 
     // Process messages
-    threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this)));
+    threadMessageHandler = std::thread(&TraceThread, "msghand", std::bind(&CConnman::ThreadMessageHandler, this));
 
     // Dump network addresses
     scheduler.scheduleEvery(std::bind(&CConnman::DumpData, this), DUMP_ADDRESSES_INTERVAL * 1000);

src/rpc/blockchain.cpp

@@ -40,7 +40,6 @@
 #include <univalue.h>
 
 #include <boost/algorithm/string.hpp>
-#include <boost/thread/thread.hpp> // boost::thread::interrupt
 
 #include <memory>
 #include <mutex>
@@ -839,7 +838,6 @@ static bool GetUTXOStats(CCoinsView *view, CCoinsStats &stats)
     uint256 prevkey;
     std::map<uint32_t, Coin> outputs;
     while (pcursor->Valid()) {
-        boost::this_thread::interruption_point();
         COutPoint key;
         Coin coin;
         if (pcursor->GetKey(key) && pcursor->GetValue(coin)) {
@@ -1832,7 +1830,6 @@ bool FindScriptPubKey(std::atomic<int>& scan_progress, const std::atomic<bool>&
         Coin coin;
         if (!cursor->GetKey(key) || !cursor->GetValue(coin)) return false;
         if (++count % 8192 == 0) {
-            boost::this_thread::interruption_point();
             if (should_abort) {
                 // allow to abort the scan via the abort reference
                 return false;