From 6a37da837e1c936200f8d71a142973cc60b17b32 Mon Sep 17 00:00:00 2001 From: moneromooo-monero Date: Wed, 12 Aug 2020 22:13:29 +0000 Subject: 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 | 4 ++-- src/common/threadpool.cpp | 11 ++++----- src/common/threadpool.h | 8 +++++-- src/cryptonote_core/blockchain.cpp | 16 +++++++------ src/cryptonote_core/cryptonote_core.cpp | 8 ++++--- src/ringct/rctSigs.cpp | 15 ++++++++----- src/wallet/wallet2.cpp | 30 ++++++++++++------------- tests/unit_tests/threadpool.cpp | 40 ++++++++++++++++----------------- 8 files changed, 72 insertions(+), 60 deletions(-) diff --git a/src/common/dns_utils.cpp b/src/common/dns_utils.cpp index c871177b1..4f4efcd81 100644 --- a/src/common/dns_utils.cpp +++ b/src/common/dns_utils.cpp @@ -521,14 +521,14 @@ bool load_txt_records_from_dns(std::vector &good_records, const std // send all requests in parallel std::deque 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 diff --git a/src/common/threadpool.cpp b/src/common/threadpool.cpp index a1737778c..edc87fc48 100644 --- a/src/common/threadpool.cpp +++ b/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 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; diff --git a/src/common/threadpool.h b/src/common/threadpool.h index 91f9fdf47..66b08fece 100644 --- a/src/common/threadpool.h +++ b/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(); }; diff --git a/src/cryptonote_core/blockchain.cpp b/src/cryptonote_core/blockchain.cpp index 7851b0f6a..636c6963c 100644 --- a/src/cryptonote_core/blockchain.cpp +++ b/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 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 { diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp index 141e54459..00ea4ebdd 100644 --- a/src/cryptonote_core/cryptonote_core.cpp +++ b/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::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 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_info; tx_info.reserve(tx_blobs.size()); diff --git a/src/ringct/rctSigs.cpp b/src/ringct/rctSigs.cpp index 2e3e7007e..5948cadc5 100644 --- a/src/ringct/rctSigs.cpp +++ b/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 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 results; std::vector 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 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]) { diff --git a/src/wallet/wallet2.cpp b/src/wallet/wallet2.cpp index d7ed3e999..d3f3d4880 100644 --- a/src/wallet/wallet2.cpp +++ b/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 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 hwdev_lock (hwdev); @@ -2674,7 +2674,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector tx_cache_data; @@ -2701,7 +2701,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector 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 blocks; std::vector 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 << ")..."); diff --git a/tests/unit_tests/threadpool.cpp b/tests/unit_tests/threadpool.cpp index 1307cd738..1017f04ff 100644 --- a/tests/unit_tests/threadpool.cpp +++ b/tests/unit_tests/threadpool.cpp @@ -34,46 +34,46 @@ TEST(threadpool, wait_nothing) { std::shared_ptr 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 tpool(tools::threadpool::getNewForUnitTests()); - tools::threadpool::waiter waiter; + tools::threadpool::waiter waiter(*tpool); std::atomic 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 tpool(tools::threadpool::getNewForUnitTests(1)); - tools::threadpool::waiter waiter; + tools::threadpool::waiter waiter(*tpool); std::atomic 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 tpool(tools::threadpool::getNewForUnitTests(256)); - tools::threadpool::waiter waiter; + tools::threadpool::waiter waiter(*tpool); std::atomic 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 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 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 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 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 tpool(tools::threadpool::getNewForUnitTests(4)); - tools::threadpool::waiter waiter; + tools::threadpool::waiter waiter(*tpool); std::atomic 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); } -- cgit v1.2.3