4rkal/monero
1
0
Fork 0
mirror of https://github.com/monero-project/monero.git synced 2024-12-13 11:56:31 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

threadpool: guard against exceptions in jobs, and armour plating

Browse Source 
Those would, if uncaught, exit run and leave the waiter to wait
indefinitely for the number of active jobs to reach 0
This commit is contained in:
moneromooo-monero 2020-08-12 22:13:29 +00:00
parent 5d850dde99
commit 6a37da837e
No known key found for this signature in database
GPG Key ID: 686F07454D6CEFC3
8 changed files with 72 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/common/dns_utils.cpp
View File

@ -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

11
src/common/threadpool.cpp
View File

@ -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;

8
src/common/threadpool.h
View File

@ -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();
};

16
src/cryptonote_core/blockchain.cpp
View File

@ -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
{

8
src/cryptonote_core/cryptonote_core.cpp
View File

@ -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());

15
src/ringct/rctSigs.cpp
View File

@ -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]) {

30
src/wallet/wallet2.cpp
View File

@ -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 << ")...");

40
tests/unit_tests/threadpool.cpp
View File

@ -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);
}