diff --git a/src/cryptonote_protocol/levin_notify.cpp b/src/cryptonote_protocol/levin_notify.cpp index 4f0cd77a92f..83913f096fc 100644 --- a/src/cryptonote_protocol/levin_notify.cpp +++ b/src/cryptonote_protocol/levin_notify.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -567,15 +568,16 @@ namespace levin { std::shared_ptr zone_; i_core_events* core_; + std::shared_ptr tx_queue_; std::vector txs_; std::vector tx_hashes_; boost::uuids::uuid source_; - relay_method tx_relay; + const relay_method tx_relay; //! \pre Called in `zone_->strand` void operator()() { - if (!zone_ || !core_ || txs_.empty()) + if (!zone_ || !core_ || !tx_queue_ || txs_.empty()) return; if (!zone_->fluffing || tx_relay == relay_method::local) @@ -589,6 +591,8 @@ namespace levin /* Source is intentionally omitted in debug log for privacy - a nil uuid indicates source is that node. */ MDEBUG("Sent " << txs_.size() << " transaction(s) to " << destination << " using Dandelion++ stem"); + if (!tx_queue_->dequeue(tx_hashes_, tx_relay)) + MWARNING("Some expected tx(s) weren't in the notify queue"); return; } @@ -600,6 +604,8 @@ namespace levin } core_->on_transactions_relayed(epee::to_span(txs_), relay_method::fluff); + if (!tx_queue_->dequeue(tx_hashes_, tx_relay)) + MWARNING("Some expected tx(s) weren't in the notify queue"); fluff_notify::run(std::move(zone_), epee::to_span(txs_), epee::to_span(tx_hashes_), source_); } }; @@ -740,9 +746,94 @@ namespace levin }; } // anonymous + bool notify_tx_queue::enqueue(const relay_method tx_relay, std::vector &txs, std::vector &tx_hashes) + { + std::lock_guard lock(m_mutex); + + CHECK_AND_ASSERT_MES(txs.size() == tx_hashes.size(), false, "Expected txs == tx_hashes"); + if (txs.empty()) + return false; + + std::vector notify_txs; + std::vector notify_tx_hashes; + notify_txs.reserve(txs.size()); + notify_tx_hashes.reserve(tx_hashes.size()); + + auto it_txs = txs.begin(); + auto it_hashes = tx_hashes.begin(); + while (it_txs != txs.end() && it_hashes != tx_hashes.end()) + { + if (this->enqueue(*it_hashes, tx_relay)) + { + notify_txs.emplace_back(std::move(*it_txs)); + notify_tx_hashes.emplace_back(std::move(*it_hashes)); + } + ++it_txs; + ++it_hashes; + } + + txs = std::move(notify_txs); + tx_hashes = std::move(notify_tx_hashes); + return txs.size() > 0; + } + + bool notify_tx_queue::dequeue(const std::vector &tx_hashes, const relay_method tx_relay) + { + std::lock_guard lock(m_mutex); + if (tx_hashes.empty()) + return true; + + bool all_dequeued = true; + for (auto it = tx_hashes.begin(); it != tx_hashes.end(); ++it) + all_dequeued = this->dequeue(*it, tx_relay) && all_dequeued; + return all_dequeued; + } + + bool notify_tx_queue::enqueue(const crypto::hash &tx, const relay_method tx_relay) + { + std::lock_guard lock(m_mutex); + + auto it = m_queue.find(tx); + if (it == m_queue.end()) + { + m_queue.emplace(tx, std::vector{tx_relay}); + return true; + } + + const auto vec_it = std::find(it->second.begin(), it->second.end(), tx_relay); + if (vec_it == it->second.end()) + { + it->second.push_back(tx_relay); + return true; + } + + // Tx is already in the queue + return false; + } + + bool notify_tx_queue::dequeue(const crypto::hash &tx, const relay_method tx_relay) + { + std::lock_guard lock(m_mutex); + + auto it = m_queue.find(tx); + if (it == m_queue.end()) + return false; + + const auto vec_it = std::find(it->second.begin(), it->second.end(), tx_relay); + if (vec_it == it->second.end()) + return false; + + it->second.erase(vec_it); + if (it->second.empty()) + m_queue.erase(it); + + return true; + } + notify::notify(boost::asio::io_context& service, std::shared_ptr p2p, epee::byte_slice noise, epee::net_utils::zone zone, const bool pad_txs, i_core_events& core) : zone_(std::make_shared(service, std::move(p2p), std::move(noise), zone, pad_txs)) , core_(std::addressof(core)) + , tx_queue_(std::make_shared()) { if (!zone_->p2p) throw std::logic_error{"cryptonote::levin::notify cannot have nullptr p2p argument"}; @@ -847,6 +938,8 @@ namespace levin if (!zone_) return false; + if (!tx_queue_) + return false; CHECK_AND_ASSERT_MES(txs.size() == tx_hashes.size(), false, "Mismatch size of txs <> tx_hashes in send_txs"); @@ -908,10 +1001,15 @@ namespace levin case relay_method::local: if (zone_->nzone == epee::net_utils::zone::public_) { + if (!tx_queue_->enqueue(tx_relay, txs, tx_hashes)) + { + MDEBUG("Tx(s) already in the notify queue"); + return true; + } // this will change a local/forward tx to stem or fluff ... boost::asio::dispatch( zone_->strand, - dandelionpp_notify{zone_, core_, std::move(txs), std::move(tx_hashes), source, tx_relay} + dandelionpp_notify{zone_, core_, tx_queue_, std::move(txs), std::move(tx_hashes), source, tx_relay} ); break; } diff --git a/src/cryptonote_protocol/levin_notify.h b/src/cryptonote_protocol/levin_notify.h index f3c85d6a6bc..928f65d3b89 100644 --- a/src/cryptonote_protocol/levin_notify.h +++ b/src/cryptonote_protocol/levin_notify.h @@ -31,6 +31,8 @@ #include #include #include +#include +#include #include #include "byte_slice.h" @@ -66,6 +68,28 @@ namespace levin using connections = epee::levin::async_protocol_handler_config; + //! Prevents unnecessary duplicate tx notifies + class notify_tx_queue + { + public: + // Enqueues txs not already in the queue. Returns true iff any tx(s) were added to the queue. + // Note: this modifies the passed in txs and tx_hashes, removing any txs from those containers + // that are already in the queue. Any remaining txs should be relayed. + bool enqueue(const relay_method tx_relay, std::vector &txs, std::vector &tx_hashes); + + // Dequeues txs from the queue. Returns true if all given txs were already in the queue. + bool dequeue(const std::vector &tx_hashes, const relay_method tx_relay); + private: + // Returns true if the tx is added to the queue, false if it was already in the queue. + bool enqueue(const crypto::hash &tx, const relay_method tx_relay); + + // Returns true if the tx was already in the in queue and gets removed. + bool dequeue(const crypto::hash &tx, const relay_method tx_relay); + private: + std::unordered_map> m_queue; + std::recursive_mutex m_mutex; + }; + //! Provides tx notification privacy class notify { @@ -84,6 +108,7 @@ namespace levin notify() noexcept : zone_(nullptr) , core_(nullptr) + , tx_queue_(nullptr) {} //! Construct an instance with available notification `zones`. @@ -132,6 +157,9 @@ namespace levin \return True iff the notification is queued for sending. */ bool send_txs(std::vector txs, std::vector &&tx_hashes, const boost::uuids::uuid& source, relay_method tx_relay); + + private: + std::shared_ptr tx_queue_; }; } // levin } // net