aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/common/dns_utils.cpp4
-rw-r--r--src/common/threadpool.cpp11
-rw-r--r--src/common/threadpool.h8
-rw-r--r--src/cryptonote_core/blockchain.cpp16
-rw-r--r--src/cryptonote_core/cryptonote_core.cpp8
-rw-r--r--src/ringct/rctSigs.cpp15
-rw-r--r--src/wallet/wallet2.cpp30
-rw-r--r--tests/unit_tests/threadpool.cpp40
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<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
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<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;
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 9d4c5a66c..66d223ecd 100644
--- a/src/cryptonote_core/blockchain.cpp
+++ b/src/cryptonote_core/blockchain.cpp
@@ -3298,8 +3298,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;
@@ -3369,7 +3368,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)
@@ -4908,7 +4908,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;
@@ -4921,7 +4921,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)
@@ -5055,14 +5056,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
{
diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp
index 474362ed0..894071924 100644
--- a/src/cryptonote_core/cryptonote_core.cpp
+++ b/src/cryptonote_core/cryptonote_core.cpp
@@ -985,7 +985,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] {
@@ -1001,7 +1001,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) {
@@ -1033,7 +1034,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());
diff --git a/src/ringct/rctSigs.cpp b/src/ringct/rctSigs.cpp
index 5fd7ac06d..93eb52d4e 100644
--- a/src/ringct/rctSigs.cpp
+++ b/src/ringct/rctSigs.cpp
@@ -1278,12 +1278,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]) {
@@ -1327,7 +1328,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;
@@ -1410,7 +1411,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);
@@ -1458,7 +1460,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;
@@ -1476,7 +1478,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 918b3fd41..3e2ccd1ff 100644
--- a/src/wallet/wallet2.cpp
+++ b/src/wallet/wallet2.cpp
@@ -1949,7 +1949,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;
@@ -2022,7 +2022,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);
@@ -2049,7 +2049,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);
@@ -2687,7 +2687,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;
@@ -2714,7 +2714,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);
@@ -2743,7 +2743,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)
@@ -2791,7 +2791,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;
@@ -2863,14 +2863,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)
@@ -2896,7 +2896,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(...)
@@ -3338,7 +3338,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;
@@ -3446,7 +3446,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());
@@ -3478,19 +3478,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<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);
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-25 05:51:14 +0000