diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/blockchain_db/blockchain_db.h | 16 | ||||
| -rw-r--r-- | src/blockchain_db/lmdb/db_lmdb.cpp | 86 | ||||
| -rw-r--r-- | src/cryptonote_core/cryptonote_core.cpp | 51 | ||||
| -rw-r--r-- | src/cryptonote_core/cryptonote_core.h | 7 | ||||
| -rw-r--r-- | src/cryptonote_core/tx_pool.cpp | 10 | ||||
| -rw-r--r-- | src/cryptonote_core/tx_pool.h | 4 |
6 files changed, 122 insertions, 52 deletions
diff --git a/src/blockchain_db/blockchain_db.h b/src/blockchain_db/blockchain_db.h index 263948fa2..a163ef98c 100644 --- a/src/blockchain_db/blockchain_db.h +++ b/src/blockchain_db/blockchain_db.h @@ -1883,16 +1883,18 @@ public: } virtual ~db_txn_guard() { - if (active) - stop(); + stop(); } void stop() { - if (readonly) - db->block_rtxn_stop(); - else - db->block_wtxn_stop(); - active = false; + if (active) + { + if (readonly) + db->block_rtxn_stop(); + else + db->block_wtxn_stop(); + active = false; + } } void abort() { diff --git a/src/blockchain_db/lmdb/db_lmdb.cpp b/src/blockchain_db/lmdb/db_lmdb.cpp index db7fa6c7c..f80013d02 100644 --- a/src/blockchain_db/lmdb/db_lmdb.cpp +++ b/src/blockchain_db/lmdb/db_lmdb.cpp @@ -465,6 +465,32 @@ void mdb_txn_safe::increment_txns(int i) num_active_txns += i; } +#define TXN_PREFIX(flags); \ + mdb_txn_safe auto_txn; \ + mdb_txn_safe* txn_ptr = &auto_txn; \ + if (m_batch_active) \ + txn_ptr = m_write_txn; \ + else \ + { \ + if (auto mdb_res = lmdb_txn_begin(m_env, NULL, flags, auto_txn)) \ + throw0(DB_ERROR(lmdb_error(std::string("Failed to create a transaction for the db in ")+__FUNCTION__+": ", mdb_res).c_str())); \ + } \ + +#define TXN_PREFIX_RDONLY() \ + MDB_txn *m_txn; \ + mdb_txn_cursors *m_cursors; \ + mdb_txn_safe auto_txn; \ + bool my_rtxn = block_rtxn_start(&m_txn, &m_cursors); \ + if (my_rtxn) auto_txn.m_tinfo = m_tinfo.get(); \ + else auto_txn.uncheck() +#define TXN_POSTFIX_RDONLY() + +#define TXN_POSTFIX_SUCCESS() \ + do { \ + if (! m_batch_active) \ + auto_txn.commit(); \ + } while(0) + void lmdb_resized(MDB_env *env, int isactive) { mdb_txn_safe::prevent_new_txns(); @@ -713,21 +739,20 @@ uint64_t BlockchainLMDB::get_estimated_batch_size(uint64_t batch_num_blocks, uin } else { - MDB_txn *rtxn; - mdb_txn_cursors *rcurs; - bool my_rtxn = block_rtxn_start(&rtxn, &rcurs); - for (uint64_t block_num = block_start; block_num <= block_stop; ++block_num) { - // we have access to block weight, which will be greater or equal to block size, - // so use this as a proxy. If it's too much off, we might have to check actual size, - // which involves reading more data, so is not really wanted - size_t block_weight = get_block_weight(block_num); - total_block_size += block_weight; - // Track number of blocks being totalled here instead of assuming, in case - // some blocks were to be skipped for being outliers. - ++num_blocks_used; + TXN_PREFIX_RDONLY(); + for (uint64_t block_num = block_start; block_num <= block_stop; ++block_num) + { + // we have access to block weight, which will be greater or equal to block size, + // so use this as a proxy. If it's too much off, we might have to check actual size, + // which involves reading more data, so is not really wanted + size_t block_weight = get_block_weight(block_num); + total_block_size += block_weight; + // Track number of blocks being totalled here instead of assuming, in case + // some blocks were to be skipped for being outliers. + ++num_blocks_used; + } } - if (my_rtxn) block_rtxn_stop(); avg_block_size = total_block_size / (num_blocks_used ? num_blocks_used : 1); MDEBUG("average block size across recent " << num_blocks_used << " blocks: " << avg_block_size); } @@ -1678,32 +1703,6 @@ void BlockchainLMDB::unlock() check_open(); } -#define TXN_PREFIX(flags); \ - mdb_txn_safe auto_txn; \ - mdb_txn_safe* txn_ptr = &auto_txn; \ - if (m_batch_active) \ - txn_ptr = m_write_txn; \ - else \ - { \ - if (auto mdb_res = lmdb_txn_begin(m_env, NULL, flags, auto_txn)) \ - throw0(DB_ERROR(lmdb_error(std::string("Failed to create a transaction for the db in ")+__FUNCTION__+": ", mdb_res).c_str())); \ - } \ - -#define TXN_PREFIX_RDONLY() \ - MDB_txn *m_txn; \ - mdb_txn_cursors *m_cursors; \ - mdb_txn_safe auto_txn; \ - bool my_rtxn = block_rtxn_start(&m_txn, &m_cursors); \ - if (my_rtxn) auto_txn.m_tinfo = m_tinfo.get(); \ - else auto_txn.uncheck() -#define TXN_POSTFIX_RDONLY() - -#define TXN_POSTFIX_SUCCESS() \ - do { \ - if (! m_batch_active) \ - auto_txn.commit(); \ - } while(0) - // The below two macros are for DB access within block add/remove, whether // regular batch txn is in use or not. m_write_txn is used as a batch txn, even @@ -3923,13 +3922,20 @@ void BlockchainLMDB::block_rtxn_stop() const LOG_PRINT_L3("BlockchainLMDB::" << __func__); mdb_txn_reset(m_tinfo->m_ti_rtxn); memset(&m_tinfo->m_ti_rflags, 0, sizeof(m_tinfo->m_ti_rflags)); + /* cancel out the increment from rtxn_start */ + mdb_txn_safe::increment_txns(-1); } bool BlockchainLMDB::block_rtxn_start() const { MDB_txn *mtxn; mdb_txn_cursors *mcur; - return block_rtxn_start(&mtxn, &mcur); + /* auto_txn is only used for the create gate */ + mdb_txn_safe auto_txn; + bool ret = block_rtxn_start(&mtxn, &mcur); + if (ret) + auto_txn.increment_txns(1); /* remember there is an active readtxn */ + return ret; } void BlockchainLMDB::block_wtxn_start() diff --git a/src/cryptonote_core/cryptonote_core.cpp b/src/cryptonote_core/cryptonote_core.cpp index a78f5d673..95cd1c83b 100644 --- a/src/cryptonote_core/cryptonote_core.cpp +++ b/src/cryptonote_core/cryptonote_core.cpp @@ -1406,21 +1406,66 @@ namespace cryptonote return true; } //----------------------------------------------------------------------------------------------- + bool core::notify_txpool_event(const epee::span<const cryptonote::blobdata> tx_blobs, epee::span<const crypto::hash> tx_hashes, epee::span<const cryptonote::transaction> txs, const std::vector<bool> &just_broadcasted) const + { + if (!m_zmq_pub) + return true; + + if (tx_blobs.size() != tx_hashes.size() || tx_blobs.size() != txs.size() || tx_blobs.size() != just_broadcasted.size()) + return false; + + /* Publish txs via ZMQ that are "just broadcasted" by the daemon. This is + done here in addition to `handle_incoming_txs` in order to guarantee txs + are pub'd via ZMQ when we know the daemon has/will broadcast to other + nodes & *after* the tx is visible in the pool. This should get called + when the user submits a tx to a daemon in the "fluff" epoch relaying txs + via a public network. */ + if (std::count(just_broadcasted.begin(), just_broadcasted.end(), true) == 0) + return true; + + std::vector<txpool_event> results{}; + results.resize(tx_blobs.size()); + for (std::size_t i = 0; i < results.size(); ++i) + { + results[i].tx = std::move(txs[i]); + results[i].hash = std::move(tx_hashes[i]); + results[i].blob_size = tx_blobs[i].size(); + results[i].weight = results[i].tx.pruned ? get_pruned_transaction_weight(results[i].tx) : get_transaction_weight(results[i].tx, results[i].blob_size); + results[i].res = just_broadcasted[i]; + } + + m_zmq_pub(std::move(results)); + + return true; + } + //----------------------------------------------------------------------------------------------- void core::on_transactions_relayed(const epee::span<const cryptonote::blobdata> tx_blobs, const relay_method tx_relay) { + // lock ensures duplicate txs aren't pub'd via zmq + CRITICAL_REGION_LOCAL(m_incoming_tx_lock); + std::vector<crypto::hash> tx_hashes{}; tx_hashes.resize(tx_blobs.size()); + std::vector<cryptonote::transaction> txs{}; + txs.resize(tx_blobs.size()); + for (std::size_t i = 0; i < tx_blobs.size(); ++i) { - cryptonote::transaction tx{}; - if (!parse_and_validate_tx_from_blob(tx_blobs[i], tx, tx_hashes[i])) + if (!parse_and_validate_tx_from_blob(tx_blobs[i], txs[i], tx_hashes[i])) { LOG_ERROR("Failed to parse relayed transaction"); return; } } - m_mempool.set_relayed(epee::to_span(tx_hashes), tx_relay); + + std::vector<bool> just_broadcasted{}; + just_broadcasted.reserve(tx_hashes.size()); + + m_mempool.set_relayed(epee::to_span(tx_hashes), tx_relay, just_broadcasted); + + if (m_zmq_pub && matches_category(tx_relay, relay_category::legacy)) + notify_txpool_event(tx_blobs, epee::to_span(tx_hashes), epee::to_span(txs), just_broadcasted); } //----------------------------------------------------------------------------------------------- bool core::get_block_template(block& b, const account_public_address& adr, difficulty_type& diffic, uint64_t& height, uint64_t& expected_reward, const blobdata& ex_nonce, uint64_t &seed_height, crypto::hash &seed_hash) diff --git a/src/cryptonote_core/cryptonote_core.h b/src/cryptonote_core/cryptonote_core.h index 0b36730b6..5f134a999 100644 --- a/src/cryptonote_core/cryptonote_core.h +++ b/src/cryptonote_core/cryptonote_core.h @@ -1036,6 +1036,13 @@ namespace cryptonote bool relay_txpool_transactions(); /** + * @brief sends notification of txpool events to subscribers + * + * @return true on success, false otherwise + */ + bool notify_txpool_event(const epee::span<const cryptonote::blobdata> tx_blobs, epee::span<const crypto::hash> tx_hashes, epee::span<const cryptonote::transaction> txs, const std::vector<bool> &just_broadcasted) const; + + /** * @brief checks DNS versions * * @return true on success, false otherwise diff --git a/src/cryptonote_core/tx_pool.cpp b/src/cryptonote_core/tx_pool.cpp index a68da0e62..0c18b2a34 100644 --- a/src/cryptonote_core/tx_pool.cpp +++ b/src/cryptonote_core/tx_pool.cpp @@ -820,8 +820,10 @@ namespace cryptonote return true; } //--------------------------------------------------------------------------------- - void tx_memory_pool::set_relayed(const epee::span<const crypto::hash> hashes, const relay_method method) + void tx_memory_pool::set_relayed(const epee::span<const crypto::hash> hashes, const relay_method method, std::vector<bool> &just_broadcasted) { + just_broadcasted.clear(); + crypto::random_poisson_seconds embargo_duration{dandelionpp_embargo_average}; const auto now = std::chrono::system_clock::now(); uint64_t next_relay = uint64_t{std::numeric_limits<time_t>::max()}; @@ -831,12 +833,14 @@ namespace cryptonote LockedTXN lock(m_blockchain.get_db()); for (const auto& hash : hashes) { + bool was_just_broadcasted = false; try { txpool_tx_meta_t meta; if (m_blockchain.get_txpool_tx_meta(hash, meta)) { // txes can be received as "stem" or "fluff" in either order + const bool already_broadcasted = meta.matches(relay_category::broadcasted); meta.upgrade_relay_method(method); meta.relayed = true; @@ -849,6 +853,9 @@ namespace cryptonote meta.last_relayed_time = std::chrono::system_clock::to_time_t(now); m_blockchain.update_txpool_tx(hash, meta); + + // wait until db update succeeds to ensure tx is visible in the pool + was_just_broadcasted = !already_broadcasted && meta.matches(relay_category::broadcasted); } } catch (const std::exception &e) @@ -856,6 +863,7 @@ namespace cryptonote MERROR("Failed to update txpool transaction metadata: " << e.what()); // continue } + just_broadcasted.emplace_back(was_just_broadcasted); } lock.commit(); set_if_less(m_next_check, time_t(next_relay)); diff --git a/src/cryptonote_core/tx_pool.h b/src/cryptonote_core/tx_pool.h index 62bef6c06..65c39f87c 100644 --- a/src/cryptonote_core/tx_pool.h +++ b/src/cryptonote_core/tx_pool.h @@ -353,8 +353,10 @@ namespace cryptonote * * @param hashes list of tx hashes that are about to be relayed * @param tx_relay update how the tx left this node + * @param just_broadcasted true if a tx was just broadcasted + * */ - void set_relayed(epee::span<const crypto::hash> hashes, relay_method tx_relay); + void set_relayed(epee::span<const crypto::hash> hashes, relay_method tx_relay, std::vector<bool> &just_broadcasted); /** * @brief get the total number of transactions in the pool |