From 2771a18e85bd91dd881f89f6c14d26cba35a1844 Mon Sep 17 00:00:00 2001 From: moneromooo-monero Date: Thu, 26 Apr 2018 11:44:47 +0100 Subject: [PATCH] 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 | 37 +++++++++++----- src/common/threadpool.h | 7 +-- src/cryptonote_core/blockchain.cpp | 12 +++--- src/cryptonote_core/cryptonote_core.cpp | 4 +- src/ringct/rctSigs.cpp | 12 +++--- src/wallet/wallet2.cpp | 34 +++++++-------- tests/unit_tests/threadpool.cpp | 57 ++++++++++++++++++++++--- 7 files changed, 112 insertions(+), 51 deletions(-) diff --git a/src/common/threadpool.cpp b/src/common/threadpool.cpp index fb238dca7..6b69e2a12 100644 --- a/src/common/threadpool.cpp +++ b/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 f) { - entry e = {obj, f}; +void threadpool::submit(waiter *obj, std::function f, bool leaf) { + CHECK_AND_ASSERT_THROW_MES(!is_leaf, "A leaf routine is using a thread pool"); boost::unique_lock 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 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 lock(mt); num--; if (!num) - cv.notify_one(); + cv.notify_all(); } -void threadpool::run() { +void threadpool::run(bool flush) { boost::unique_lock 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(); diff --git a/src/common/threadpool.h b/src/common/threadpool.h index bf80a87f6..a43e38a76 100644 --- a/src/common/threadpool.h +++ b/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 f); + void submit(waiter *waiter, std::function f, bool leaf = false); unsigned int get_max_concurrency() const; @@ -78,6 +78,7 @@ public: typedef struct entry { waiter *wo; std::function f; + bool leaf; } entry; std::deque 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); }; } diff --git a/src/cryptonote_core/blockchain.cpp b/src/cryptonote_core/blockchain.cpp index 208ea1efa..854570eb5 100644 --- a/src/cryptonote_core/blockchain.cpp +++ b/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 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 hwdev_lock (hwdev); @@ -1822,7 +1822,7 @@ void wallet2::process_parsed_blocks(uint64_t start_height, const std::vector= 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 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 << ")..."); diff --git a/tests/unit_tests/threadpool.cpp b/tests/unit_tests/threadpool.cpp index 34be1417a..1307cd738 100644 --- a/tests/unit_tests/threadpool.cpp +++ b/tests/unit_tests/threadpool.cpp @@ -35,7 +35,7 @@ TEST(threadpool, wait_nothing) { std::shared_ptr 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 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 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 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 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 tpool(tools::threadpool::getNewForUnitTests(4)); + tools::threadpool::waiter waiter; + + std::atomic 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); +}