threadpool: allow leaf functions to run concurrently

Decrease the number of worker threads by one to account
for the fact the calling thread acts as a worker thread now

src/common/threadpool.cpp

@@ -36,6 +36,7 @@
 #include "common/util.h"
 
 static __thread int depth = 0;
+static __thread bool is_leaf = false;
 
 namespace tools
 {
@@ -43,9 +44,9 @@ threadpool::threadpool(unsigned int max_threads) : running(true), active(0) {
   boost::thread::attributes attrs;
   attrs.set_stack_size(THREAD_STACK_SIZE);
   max = max_threads ? max_threads : tools::get_max_concurrency();
-  unsigned int i = max;
+  size_t i = max ? max - 1 : 0;
   while(i--) {
-    threads.push_back(boost::thread(attrs, boost::bind(&threadpool::run, this)));
+    threads.push_back(boost::thread(attrs, boost::bind(&threadpool::run, this, false)));
   }
 }
 
@@ -60,20 +61,25 @@ threadpool::~threadpool() {
   }
 }
 
-void threadpool::submit(waiter *obj, std::function<void()> f) {
-  entry e = {obj, f};
+void threadpool::submit(waiter *obj, std::function<void()> f, bool leaf) {
+  CHECK_AND_ASSERT_THROW_MES(!is_leaf, "A leaf routine is using a thread pool");
   boost::unique_lock<boost::mutex> lock(mutex);
-  if ((active == max && !queue.empty()) || depth > 0) {
+  if (!leaf && ((active == max && !queue.empty()) || depth > 0)) {
     // if all available threads are already running
     // and there's work waiting, just run in current thread
     lock.unlock();
     ++depth;
+    is_leaf = leaf;
     f();
     --depth;
+    is_leaf = false;
   } else {
     if (obj)
       obj->inc();
-    queue.push_back(e);
+    if (leaf)
+      queue.push_front({obj, f, leaf});
+    else
+      queue.push_back({obj, f, leaf});
     has_work.notify_one();
   }
 }
@@ -91,7 +97,7 @@ threadpool::waiter::~waiter()
   }
   try
   {
-    wait();
+    wait(NULL);
   }
   catch (const std::exception &e)
   {
@@ -99,9 +105,12 @@ threadpool::waiter::~waiter()
   }
 }
 
-void threadpool::waiter::wait() {
+void threadpool::waiter::wait(threadpool *tpool) {
+  if (tpool)
+    tpool->run(true);
   boost::unique_lock<boost::mutex> lock(mt);
-  while(num) cv.wait(lock);
+  while(num)
+    cv.wait(lock);
 }
 
 void threadpool::waiter::inc() {
@@ -113,15 +122,19 @@ void threadpool::waiter::dec() {
   const boost::unique_lock<boost::mutex> lock(mt);
   num--;
   if (!num)
-    cv.notify_one();
+    cv.notify_all();
 }
 
-void threadpool::run() {
+void threadpool::run(bool flush) {
   boost::unique_lock<boost::mutex> lock(mutex);
   while (running) {
     entry e;
     while(queue.empty() && running)
+    {
+      if (flush)
+        return;
       has_work.wait(lock);
+    }
     if (!running) break;
 
     active++;
@@ -129,8 +142,10 @@ void threadpool::run() {
     queue.pop_front();
     lock.unlock();
     ++depth;
+    is_leaf = e.leaf;
     e.f();
     --depth;
+    is_leaf = false;
 
     if (e.wo)
       e.wo->dec();

src/common/threadpool.h

@@ -59,7 +59,7 @@ public:
     public:
     void inc();
     void dec();
-    void wait();  //! Wait for a set of tasks to finish.
+    void wait(threadpool *tpool);  //! Wait for a set of tasks to finish.
     waiter() : num(0){}
     ~waiter();
   };
@@ -67,7 +67,7 @@ public:
   // Submit a task to the pool. The waiter pointer may be
   // NULL if the caller doesn't care to wait for the
   // task to finish.
-  void submit(waiter *waiter, std::function<void()> f);
+  void submit(waiter *waiter, std::function<void()> f, bool leaf = false);
 
   unsigned int get_max_concurrency() const;
 
@@ -78,6 +78,7 @@ public:
     typedef struct entry {
       waiter *wo;
       std::function<void()> f;
+      bool leaf;
     } entry;
     std::deque<entry> queue;
     boost::condition_variable has_work;
@@ -86,7 +87,7 @@ public:
     unsigned int active;
     unsigned int max;
     bool running;
-    void run();
+    void run(bool flush = false);
 };
 
 }

src/cryptonote_core/blockchain.cpp

@@ -2818,7 +2818,7 @@ bool Blockchain::check_tx_inputs(transaction& tx, tx_verification_context &tvc,
       {
         // ND: Speedup
         // 1. Thread ring signature verification if possible.
-        tpool.submit(&waiter, boost::bind(&Blockchain::check_ring_signature, this, std::cref(tx_prefix_hash), std::cref(in_to_key.k_image), std::cref(pubkeys[sig_index]), std::cref(tx.signatures[sig_index]), std::ref(results[sig_index])));
+        tpool.submit(&waiter, boost::bind(&Blockchain::check_ring_signature, this, std::cref(tx_prefix_hash), std::cref(in_to_key.k_image), std::cref(pubkeys[sig_index]), std::cref(tx.signatures[sig_index]), std::ref(results[sig_index])), true);
       }
       else
       {
@@ -2842,7 +2842,7 @@ bool Blockchain::check_tx_inputs(transaction& tx, tx_verification_context &tvc,
     sig_index++;
   }
   if (tx.version == 1 && threads > 1)
-    waiter.wait();
+    waiter.wait(&tpool);
 
   if (tx.version == 1)
   {
@@ -4128,11 +4128,11 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector<block_complete
       tools::threadpool::waiter waiter;
       for (uint64_t i = 0; i < threads; i++)
       {
-        tpool.submit(&waiter, boost::bind(&Blockchain::block_longhash_worker, this, thread_height, std::cref(blocks[i]), std::ref(maps[i])));
+        tpool.submit(&waiter, boost::bind(&Blockchain::block_longhash_worker, this, thread_height, std::cref(blocks[i]), std::ref(maps[i])), true);
         thread_height += blocks[i].size();
       }
 
-      waiter.wait();
+      waiter.wait(&tpool);
 
       if (m_cancel)
          return false;
@@ -4267,9 +4267,9 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector<block_complete
     for (size_t i = 0; i < amounts.size(); i++)
     {
       uint64_t amount = amounts[i];
-      tpool.submit(&waiter, boost::bind(&Blockchain::output_scan_worker, this, amount, std::cref(offset_map[amount]), std::ref(tx_map[amount]), std::ref(transactions[i])));
+      tpool.submit(&waiter, boost::bind(&Blockchain::output_scan_worker, this, amount, std::cref(offset_map[amount]), std::ref(tx_map[amount]), std::ref(transactions[i])), true);
     }
-    waiter.wait();
+    waiter.wait(&tpool);
   }
   else
   {

src/cryptonote_core/cryptonote_core.cpp

@@ -697,7 +697,7 @@ namespace cryptonote
         }
       });
     }
-    waiter.wait();
+    waiter.wait(&tpool);
     it = tx_blobs.begin();
     for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
       if (!results[i].res)
@@ -725,7 +725,7 @@ namespace cryptonote
         });
       }
     }
-    waiter.wait();
+    waiter.wait(&tpool);
 
     bool ok = true;
     it = tx_blobs.begin();

src/ringct/rctSigs.cpp

@@ -862,9 +862,9 @@ namespace rct {
                   results[i] = verBulletproof(rv.p.bulletproofs[i]);
                 else
                   results[i] = verRange(rv.outPk[i].mask, rv.p.rangeSigs[i]);
-              });
+              }, true);
             }
-            waiter.wait();
+            waiter.wait(&tpool);
 
             for (size_t i = 0; i < rv.outPk.size(); ++i) {
               if (!results[i]) {
@@ -970,9 +970,9 @@ namespace rct {
                 results[i] = verBulletproof(rv.p.bulletproofs[i]);
               else
                 results[i] = verRange(rv.outPk[i].mask, rv.p.rangeSigs[i]);
-            });
+            }, true);
           }
-          waiter.wait();
+          waiter.wait(&tpool);
 
           for (size_t i = 0; i < results.size(); ++i) {
             if (!results[i]) {
@@ -989,9 +989,9 @@ namespace rct {
           for (size_t i = 0 ; i < rv.mixRing.size() ; i++) {
             tpool.submit(&waiter, [&, i] {
                 results[i] = verRctMGSimple(message, rv.p.MGs[i], rv.mixRing[i], pseudoOuts[i]);
-            });
+            }, true);
           }
-          waiter.wait();
+          waiter.wait(&tpool);
 
           for (size_t i = 0; i < results.size(); ++i) {
             if (!results[i]) {

src/wallet/wallet2.cpp

@@ -1254,9 +1254,9 @@ void wallet2::process_new_transaction(const crypto::hash &txid, const cryptonote
         for (size_t i = 1; i < tx.vout.size(); ++i)
         {
           tpool.submit(&waiter, boost::bind(&wallet2::check_acc_out_precomp, this, std::cref(tx.vout[i]), std::cref(derivation), std::cref(additional_derivations), i,
-            std::cref(is_out_data_ptr), std::ref(tx_scan_info[i])));
+            std::cref(is_out_data_ptr), std::ref(tx_scan_info[i])), true);
         }
-        waiter.wait();
+        waiter.wait(&tpool);
         // then scan all outputs from 0
         hw::device &hwdev = m_account.get_device();
         boost::unique_lock<hw::device> hwdev_lock (hwdev);
@@ -1277,9 +1277,9 @@ void wallet2::process_new_transaction(const crypto::hash &txid, const cryptonote
       for (size_t i = 0; i < tx.vout.size(); ++i)
       {
         tpool.submit(&waiter, boost::bind(&wallet2::check_acc_out_precomp, this, std::cref(tx.vout[i]), std::cref(derivation), std::cref(additional_derivations), i,
-            std::cref(is_out_data_ptr), std::ref(tx_scan_info[i])));
+            std::cref(is_out_data_ptr), std::ref(tx_scan_info[i])), true);
       }
-      waiter.wait();
+      waiter.wait(&tpool);
 
       hw::device &hwdev = m_account.get_device();
       boost::unique_lock<hw::device> hwdev_lock (hwdev);
@@ -1822,7 +1822,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector<cry
     }
   }
   THROW_WALLET_EXCEPTION_IF(txidx != num_txes, error::wallet_internal_error, "txidx does not match tx_cache_data size");
-  waiter.wait();
+  waiter.wait(&tpool);
 
   hw::device &hwdev =  m_account.get_device();
   hw::reset_mode rst(hwdev);
@@ -1842,11 +1842,11 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector<cry
   for (auto &slot: tx_cache_data)
   {
     for (auto &iod: slot.primary)
-      tpool.submit(&waiter, [&gender, &iod]() { gender(iod); });
+      tpool.submit(&waiter, [&gender, &iod]() { gender(iod); }, true);
     for (auto &iod: slot.additional)
-      tpool.submit(&waiter, [&gender, &iod]() { gender(iod); });
+      tpool.submit(&waiter, [&gender, &iod]() { gender(iod); }, true);
   }
-  waiter.wait();
+  waiter.wait(&tpool);
 
   auto geniod = [&](const cryptonote::transaction &tx, size_t n_vouts, size_t txidx) {
     for (size_t k = 0; k < n_vouts; ++k)
@@ -1876,18 +1876,18 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector<cry
     {
       THROW_WALLET_EXCEPTION_IF(txidx >= tx_cache_data.size(), error::wallet_internal_error, "txidx out of range");
       const size_t n_vouts = m_refresh_type == RefreshType::RefreshOptimizeCoinbase ? 1 : parsed_blocks[i].block.miner_tx.vout.size();
-      tpool.submit(&waiter, [&, i, txidx](){ geniod(parsed_blocks[i].block.miner_tx, n_vouts, txidx); });
+      tpool.submit(&waiter, [&, i, txidx](){ geniod(parsed_blocks[i].block.miner_tx, n_vouts, txidx); }, true);
     }
     ++txidx;
     for (size_t j = 0; j < parsed_blocks[i].txes.size(); ++j)
     {
       THROW_WALLET_EXCEPTION_IF(txidx >= tx_cache_data.size(), error::wallet_internal_error, "txidx out of range");
-      tpool.submit(&waiter, [&, i, j, txidx](){ geniod(parsed_blocks[i].txes[j], parsed_blocks[i].txes[j].vout.size(), txidx); });
+      tpool.submit(&waiter, [&, i, j, txidx](){ geniod(parsed_blocks[i].txes[j], parsed_blocks[i].txes[j].vout.size(), txidx); }, true);
       ++txidx;
     }
   }
   THROW_WALLET_EXCEPTION_IF(txidx != tx_cache_data.size(), error::wallet_internal_error, "txidx did not reach expected value");
-  waiter.wait();
+  waiter.wait(&tpool);
   hwdev.set_mode(hw::device::NONE);
 
   size_t tx_cache_data_offset = 0;
@@ -1960,9 +1960,9 @@ void wallet2::pull_and_parse_next_blocks(uint64_t start_height, uint64_t &blocks
     for (size_t i = 0; i < blocks.size(); ++i)
     {
       tpool.submit(&waiter, boost::bind(&wallet2::parse_block_round, this, std::cref(blocks[i].block),
-        std::ref(parsed_blocks[i].block), std::ref(parsed_blocks[i].hash), std::ref(parsed_blocks[i].error)));
+        std::ref(parsed_blocks[i].block), std::ref(parsed_blocks[i].hash), std::ref(parsed_blocks[i].error)), true);
     }
-    waiter.wait();
+    waiter.wait(&tpool);
     for (size_t i = 0; i < blocks.size(); ++i)
     {
       if (parsed_blocks[i].error)
@@ -1985,10 +1985,10 @@ void wallet2::pull_and_parse_next_blocks(uint64_t start_height, uint64_t &blocks
             boost::unique_lock<boost::mutex> lock(error_lock);
             error = true;
           }
-        });
+        }, true);
       }
     }
-    waiter.wait();
+    waiter.wait(&tpool);
   }
   catch(...)
   {
@@ -2423,7 +2423,7 @@ void wallet2::refresh(uint64_t start_height, uint64_t & blocks_fetched, bool& re
         process_parsed_blocks(blocks_start_height, blocks, parsed_blocks, added_blocks);
         blocks_fetched += added_blocks;
       }
-      waiter.wait();
+      waiter.wait(&tpool);
       if(!first && blocks_start_height == next_blocks_start_height)
       {
         m_node_rpc_proxy.set_height(m_blockchain.size());
@@ -2446,7 +2446,7 @@ void wallet2::refresh(uint64_t start_height, uint64_t & blocks_fetched, bool& re
     catch (const std::exception&)
     {
       blocks_fetched += added_blocks;
-      waiter.wait();
+      waiter.wait(&tpool);
       if(try_count < 3)
       {
         LOG_PRINT_L1("Another try pull_blocks (try_count=" << try_count << ")...");

tests/unit_tests/threadpool.cpp

@@ -35,7 +35,7 @@ TEST(threadpool, wait_nothing)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests());
   tools::threadpool::waiter waiter;
-  waiter.wait();
+  waiter.wait(tpool.get());
 }
 
 TEST(threadpool, wait_waits)
@@ -45,7 +45,7 @@ TEST(threadpool, wait_waits)
   std::atomic<bool> b(false);
   tpool->submit(&waiter, [&b](){ epee::misc_utils::sleep_no_w(1000); b = true; });
   ASSERT_FALSE(b);
-  waiter.wait();
+  waiter.wait(tpool.get());
   ASSERT_TRUE(b);
 }
 
@@ -59,7 +59,7 @@ TEST(threadpool, one_thread)
   {
     tpool->submit(&waiter, [&counter](){++counter;});
   }
-  waiter.wait();
+  waiter.wait(tpool.get());
   ASSERT_EQ(counter, 4096);
 }
 
@@ -73,7 +73,7 @@ TEST(threadpool, many_threads)
   {
     tpool->submit(&waiter, [&counter](){++counter;});
   }
-  waiter.wait();
+  waiter.wait(tpool.get());
   ASSERT_EQ(counter, 4096);
 }
 
@@ -85,7 +85,7 @@ static uint64_t fibonacci(std::shared_ptr<tools::threadpool> tpool, uint64_t n)
   tools::threadpool::waiter waiter;
   tpool->submit(&waiter, [&tpool, &f1, n](){ f1 = fibonacci(tpool, n-1); });
   tpool->submit(&waiter, [&tpool, &f2, n](){ f2 = fibonacci(tpool, n-2); });
-  waiter.wait();
+  waiter.wait(tpool.get());
   return f1 + f2;
 }
 
@@ -95,7 +95,52 @@ TEST(threadpool, reentrency)
   tools::threadpool::waiter waiter;
 
   uint64_t f = fibonacci(tpool, 13);
-  waiter.wait();
+  waiter.wait(tpool.get());
   ASSERT_EQ(f, 233);
 }
 
+TEST(threadpool, reentrancy)
+{
+  std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(4));
+  tools::threadpool::waiter waiter;
+
+  uint64_t f = fibonacci(tpool, 13);
+  waiter.wait(tpool.get());
+  ASSERT_EQ(f, 233);
+}
+
+TEST(threadpool, leaf_throws)
+{
+  std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests());
+  tools::threadpool::waiter waiter;
+
+  bool thrown = false, executed = false;
+  tpool->submit(&waiter, [&](){
+    try { tpool->submit(&waiter, [&](){ executed = true; }); }
+    catch(const std::exception &e) { thrown = true; }
+  }, true);
+  waiter.wait(tpool.get());
+  ASSERT_TRUE(thrown);
+  ASSERT_FALSE(executed);
+}
+
+TEST(threadpool, leaf_reentrancy)
+{
+  std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(4));
+  tools::threadpool::waiter waiter;
+
+  std::atomic<int> counter(0);
+  for (int i = 0; i < 1000; ++i)
+  {
+    tpool->submit(&waiter, [&](){
+      tools::threadpool::waiter waiter;
+      for (int j = 0; j < 500; ++j)
+      {
+        tpool->submit(&waiter, [&](){ ++counter; }, true);
+      }
+      waiter.wait(tpool.get());
+    });
+  }
+  waiter.wait(tpool.get());
+  ASSERT_EQ(counter, 500000);
+}