threadpool: guard against exceptions in jobs, and armour plating

Those would, if uncaught, exit run and leave the waiter to wait
indefinitely for the number of active jobs to reach 0

src/common/dns_utils.cpp

@@ -521,14 +521,14 @@ bool load_txt_records_from_dns(std::vector<std::string> &good_records, const std
   // send all requests in parallel
   std::deque<bool> avail(dns_urls.size(), false), valid(dns_urls.size(), false);
   tools::threadpool& tpool = tools::threadpool::getInstance();
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(tpool);
   for (size_t n = 0; n < dns_urls.size(); ++n)
   {
     tpool.submit(&waiter,[n, dns_urls, &records, &avail, &valid](){
       records[n] = tools::DNSResolver::instance().get_txt_record(dns_urls[n], avail[n], valid[n]); 
     });
   }
-  waiter.wait(&tpool);
+  waiter.wait();
 
   size_t cur_index = first_index;
   do

src/common/threadpool.cpp

@@ -120,7 +120,7 @@ threadpool::waiter::~waiter()
   catch (...) { /* ignore */ }
   try
   {
-    wait(NULL);
+    wait();
   }
   catch (const std::exception &e)
   {
@@ -128,12 +128,12 @@ threadpool::waiter::~waiter()
   }
 }
 
-void threadpool::waiter::wait(threadpool *tpool) {
-  if (tpool)
-    tpool->run(true);
+bool threadpool::waiter::wait() {
+  pool.run(true);
   boost::unique_lock<boost::mutex> lock(mt);
   while(num)
     cv.wait(lock);
+  return !error();
 }
 
 void threadpool::waiter::inc() {
@@ -166,7 +166,8 @@ void threadpool::run(bool flush) {
     lock.unlock();
     ++depth;
     is_leaf = e.leaf;
-    e.f();
+    try { e.f(); }
+    catch (const std::exception &ex) { e.wo->set_error(); try { MERROR("Exception in threadpool job: " << ex.what()); } catch (...) {} }
     --depth;
     is_leaf = false;
 

src/common/threadpool.h

@@ -55,12 +55,16 @@ public:
   class waiter {
     boost::mutex mt;
     boost::condition_variable cv;
+    threadpool &pool;
     int num;
+    bool error_flag;
     public:
     void inc();
     void dec();
-    void wait(threadpool *tpool);  //! Wait for a set of tasks to finish.
-    waiter() : num(0){}
+    bool wait();  //! Wait for a set of tasks to finish, returns false iff any error
+    void set_error() noexcept { error_flag = true; }
+    bool error() const noexcept { return error_flag; }
+    waiter(threadpool &pool) : pool(pool), num(0), error_flag(false) {}
     ~waiter();
   };
 

src/cryptonote_core/blockchain.cpp

@@ -3250,8 +3250,7 @@ bool Blockchain::check_tx_inputs(transaction& tx, tx_verification_context &tvc,
   results.resize(tx.vin.size(), 0);
 
   tools::threadpool& tpool = tools::threadpool::getInstance();
-  tools::threadpool::waiter waiter;
-  const auto waiter_guard = epee::misc_utils::create_scope_leave_handler([&]() { waiter.wait(&tpool); });
+  tools::threadpool::waiter waiter(tpool);
   int threads = tpool.get_max_concurrency();
 
   uint64_t max_used_block_height = 0;
@@ -3321,7 +3320,8 @@ bool Blockchain::check_tx_inputs(transaction& tx, tx_verification_context &tvc,
     sig_index++;
   }
   if (tx.version == 1 && threads > 1)
-    waiter.wait(&tpool);
+    if (!waiter.wait())
+      return false;
 
   // enforce min output age
   if (hf_version >= HF_VERSION_ENFORCE_MIN_AGE)
@@ -4845,7 +4845,7 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector<block_complete
     {
       m_blocks_longhash_table.clear();
       uint64_t thread_height = height;
-      tools::threadpool::waiter waiter;
+      tools::threadpool::waiter waiter(tpool);
       m_prepare_height = height;
       m_prepare_nblocks = blocks_entry.size();
       m_prepare_blocks = &blocks;
@@ -4858,7 +4858,8 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector<block_complete
         thread_height += nblocks;
       }
 
-      waiter.wait(&tpool);
+      if (!waiter.wait())
+        return false;
       m_prepare_height = 0;
 
       if (m_cancel)
@@ -4992,14 +4993,15 @@ bool Blockchain::prepare_handle_incoming_blocks(const std::vector<block_complete
 
   if (threads > 1 && amounts.size() > 1)
   {
-    tools::threadpool::waiter waiter;
+    tools::threadpool::waiter waiter(tpool);
 
     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])), true);
     }
-    waiter.wait(&tpool);
+    if (!waiter.wait())
+      return false;
   }
   else
   {

src/cryptonote_core/cryptonote_core.cpp

@@ -963,7 +963,7 @@ namespace cryptonote
     CRITICAL_REGION_LOCAL(m_incoming_tx_lock);
 
     tools::threadpool& tpool = tools::threadpool::getInstance();
-    tools::threadpool::waiter waiter;
+    tools::threadpool::waiter waiter(tpool);
     epee::span<tx_blob_entry>::const_iterator it = tx_blobs.begin();
     for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
       tpool.submit(&waiter, [&, i, it] {
@@ -979,7 +979,8 @@ namespace cryptonote
         }
       });
     }
-    waiter.wait(&tpool);
+    if (!waiter.wait())
+      return false;
     it = tx_blobs.begin();
     std::vector<bool> already_have(tx_blobs.size(), false);
     for (size_t i = 0; i < tx_blobs.size(); i++, ++it) {
@@ -1011,7 +1012,8 @@ namespace cryptonote
         });
       }
     }
-    waiter.wait(&tpool);
+    if (!waiter.wait())
+      return false;
 
     std::vector<tx_verification_batch_info> tx_info;
     tx_info.reserve(tx_blobs.size());

src/ringct/rctSigs.cpp

@@ -937,12 +937,13 @@ namespace rct {
         {
           if (semantics) {
             tools::threadpool& tpool = tools::threadpool::getInstance();
-            tools::threadpool::waiter waiter;
+            tools::threadpool::waiter waiter(tpool);
             std::deque<bool> results(rv.outPk.size(), false);
             DP("range proofs verified?");
             for (size_t i = 0; i < rv.outPk.size(); i++)
               tpool.submit(&waiter, [&, i] { results[i] = verRange(rv.outPk[i].mask, rv.p.rangeSigs[i]); });
-            waiter.wait(&tpool);
+            if (!waiter.wait())
+              return false;
 
             for (size_t i = 0; i < results.size(); ++i) {
               if (!results[i]) {
@@ -986,7 +987,7 @@ namespace rct {
         PERF_TIMER(verRctSemanticsSimple);
 
         tools::threadpool& tpool = tools::threadpool::getInstance();
-        tools::threadpool::waiter waiter;
+        tools::threadpool::waiter waiter(tpool);
         std::deque<bool> results;
         std::vector<const Bulletproof*> proofs;
         size_t max_non_bp_proofs = 0, offset = 0;
@@ -1060,7 +1061,8 @@ namespace rct {
           return false;
         }
 
-        waiter.wait(&tpool);
+        if (!waiter.wait())
+          return false;
         for (size_t i = 0; i < results.size(); ++i) {
           if (!results[i]) {
             LOG_PRINT_L1("Range proof verified failed for proof " << i);
@@ -1108,7 +1110,7 @@ namespace rct {
 
         std::deque<bool> results(threads);
         tools::threadpool& tpool = tools::threadpool::getInstance();
-        tools::threadpool::waiter waiter;
+        tools::threadpool::waiter waiter(tpool);
 
         const keyV &pseudoOuts = bulletproof ? rv.p.pseudoOuts : rv.pseudoOuts;
 
@@ -1121,7 +1123,8 @@ namespace rct {
               results[i] = verRctMGSimple(message, rv.p.MGs[i], rv.mixRing[i], pseudoOuts[i]);
           });
         }
-        waiter.wait(&tpool);
+        if (!waiter.wait())
+          return false;
 
         for (size_t i = 0; i < results.size(); ++i) {
           if (!results[i]) {

src/wallet/wallet2.cpp

@@ -1936,7 +1936,7 @@ void wallet2::process_new_transaction(const crypto::hash &txid, const cryptonote
     int num_vouts_received = 0;
     tx_pub_key = pub_key_field.pub_key;
     tools::threadpool& tpool = tools::threadpool::getInstance();
-    tools::threadpool::waiter waiter;
+    tools::threadpool::waiter waiter(tpool);
     const cryptonote::account_keys& keys = m_account.get_keys();
     crypto::key_derivation derivation;
 
@@ -2009,7 +2009,7 @@ void wallet2::process_new_transaction(const crypto::hash &txid, const cryptonote
           tpool.submit(&waiter, boost::bind(&wallet2::check_acc_out_precomp_once, 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::ref(output_found[i])), true);
         }
-        waiter.wait(&tpool);
+        THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
         // then scan all outputs from 0
         hw::device &hwdev = m_account.get_device();
         boost::unique_lock<hw::device> hwdev_lock (hwdev);
@@ -2036,7 +2036,7 @@ void wallet2::process_new_transaction(const crypto::hash &txid, const cryptonote
         tpool.submit(&waiter, boost::bind(&wallet2::check_acc_out_precomp_once, 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::ref(output_found[i])), true);
       }
-      waiter.wait(&tpool);
+      THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
 
       hw::device &hwdev = m_account.get_device();
       boost::unique_lock<hw::device> hwdev_lock (hwdev);
@@ -2674,7 +2674,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector<cry
   THROW_WALLET_EXCEPTION_IF(!m_blockchain.is_in_bounds(current_index), error::out_of_hashchain_bounds_error);
 
   tools::threadpool& tpool = tools::threadpool::getInstance();
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(tpool);
 
   size_t num_txes = 0;
   std::vector<tx_cache_data> tx_cache_data;
@@ -2701,7 +2701,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(&tpool);
+  THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
 
   hw::device &hwdev =  m_account.get_device();
   hw::reset_mode rst(hwdev);
@@ -2730,7 +2730,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector<cry
         gender(iod);
     }, true);
   }
-  waiter.wait(&tpool);
+  THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
 
   auto geniod = [&](const cryptonote::transaction &tx, size_t n_vouts, size_t txidx) {
     for (size_t k = 0; k < n_vouts; ++k)
@@ -2778,7 +2778,7 @@ 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 did not reach expected value");
-  waiter.wait(&tpool);
+  THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
   hwdev.set_mode(hw::device::NONE);
 
   size_t tx_cache_data_offset = 0;
@@ -2850,14 +2850,14 @@ void wallet2::pull_and_parse_next_blocks(uint64_t start_height, uint64_t &blocks
     THROW_WALLET_EXCEPTION_IF(blocks.size() != o_indices.size(), error::wallet_internal_error, "Mismatched sizes of blocks and o_indices");
 
     tools::threadpool& tpool = tools::threadpool::getInstance();
-    tools::threadpool::waiter waiter;
+    tools::threadpool::waiter waiter(tpool);
     parsed_blocks.resize(blocks.size());
     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)), true);
     }
-    waiter.wait(&tpool);
+    THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
     for (size_t i = 0; i < blocks.size(); ++i)
     {
       if (parsed_blocks[i].error)
@@ -2883,7 +2883,7 @@ void wallet2::pull_and_parse_next_blocks(uint64_t start_height, uint64_t &blocks
         }, true);
       }
     }
-    waiter.wait(&tpool);
+    THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
     last = !blocks.empty() && cryptonote::get_block_height(parsed_blocks.back().block) + 1 == current_height;
   }
   catch(...)
@@ -3325,7 +3325,7 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo
   crypto::hash last_tx_hash_id = m_transfers.size() ? m_transfers.back().m_txid : null_hash;
   std::list<crypto::hash> short_chain_history;
   tools::threadpool& tpool = tools::threadpool::getInstance();
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(tpool);
   uint64_t blocks_start_height;
   std::vector<cryptonote::block_complete_entry> blocks;
   std::vector<parsed_block> parsed_blocks;
@@ -3433,7 +3433,7 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo
         }
         blocks_fetched += added_blocks;
       }
-      waiter.wait(&tpool);
+      THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
       if(!first && blocks_start_height == next_blocks_start_height)
       {
         m_node_rpc_proxy.set_height(m_blockchain.size());
@@ -3465,19 +3465,19 @@ void wallet2::refresh(bool trusted_daemon, uint64_t start_height, uint64_t & blo
     catch (const tools::error::password_needed&)
     {
       blocks_fetched += added_blocks;
-      waiter.wait(&tpool);
+      THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
       throw;
     }
     catch (const error::payment_required&)
     {
       // no point in trying again, it'd just eat up credits
-      waiter.wait(&tpool);
+      THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
       throw;
     }
     catch (const std::exception&)
     {
       blocks_fetched += added_blocks;
-      waiter.wait(&tpool);
+      THROW_WALLET_EXCEPTION_IF(!waiter.wait(), error::wallet_internal_error, "Exception in thread pool");
       if(try_count < 3)
       {
         LOG_PRINT_L1("Another try pull_blocks (try_count=" << try_count << ")...");

tests/unit_tests/threadpool.cpp

@@ -34,46 +34,46 @@
 TEST(threadpool, wait_nothing)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests());
-  tools::threadpool::waiter waiter;
-  waiter.wait(tpool.get());
+  tools::threadpool::waiter waiter(*tpool);;
+  waiter.wait();
 }
 
 TEST(threadpool, wait_waits)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests());
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
   std::atomic<bool> b(false);
   tpool->submit(&waiter, [&b](){ epee::misc_utils::sleep_no_w(1000); b = true; });
   ASSERT_FALSE(b);
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_TRUE(b);
 }
 
 TEST(threadpool, one_thread)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(1));
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   std::atomic<unsigned int> counter(0);
   for (size_t n = 0; n < 4096; ++n)
   {
     tpool->submit(&waiter, [&counter](){++counter;});
   }
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_EQ(counter, 4096);
 }
 
 TEST(threadpool, many_threads)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(256));
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   std::atomic<unsigned int> counter(0);
   for (size_t n = 0; n < 4096; ++n)
   {
     tpool->submit(&waiter, [&counter](){++counter;});
   }
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_EQ(counter, 4096);
 }
 
@@ -82,44 +82,44 @@ static uint64_t fibonacci(std::shared_ptr<tools::threadpool> tpool, uint64_t n)
   if (n <= 1)
     return n;
   uint64_t f1, f2;
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
   tpool->submit(&waiter, [&tpool, &f1, n](){ f1 = fibonacci(tpool, n-1); });
   tpool->submit(&waiter, [&tpool, &f2, n](){ f2 = fibonacci(tpool, n-2); });
-  waiter.wait(tpool.get());
+  waiter.wait();
   return f1 + f2;
 }
 
 TEST(threadpool, reentrency)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(4));
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   uint64_t f = fibonacci(tpool, 13);
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_EQ(f, 233);
 }
 
 TEST(threadpool, reentrancy)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(4));
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   uint64_t f = fibonacci(tpool, 13);
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_EQ(f, 233);
 }
 
 TEST(threadpool, leaf_throws)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests());
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   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());
+  waiter.wait();
   ASSERT_TRUE(thrown);
   ASSERT_FALSE(executed);
 }
@@ -127,20 +127,20 @@ TEST(threadpool, leaf_throws)
 TEST(threadpool, leaf_reentrancy)
 {
   std::shared_ptr<tools::threadpool> tpool(tools::threadpool::getNewForUnitTests(4));
-  tools::threadpool::waiter waiter;
+  tools::threadpool::waiter waiter(*tpool);
 
   std::atomic<int> counter(0);
   for (int i = 0; i < 1000; ++i)
   {
     tpool->submit(&waiter, [&](){
-      tools::threadpool::waiter waiter;
+      tools::threadpool::waiter waiter(*tpool);
       for (int j = 0; j < 500; ++j)
       {
         tpool->submit(&waiter, [&](){ ++counter; }, true);
       }
-      waiter.wait(tpool.get());
+      waiter.wait();
     });
   }
-  waiter.wait(tpool.get());
+  waiter.wait();
   ASSERT_EQ(counter, 500000);
 }