diff options
Diffstat (limited to 'src/cryptonote_protocol')
| -rw-r--r-- | src/cryptonote_protocol/cryptonote_protocol_handler.h | 2 | ||||
| -rw-r--r-- | src/cryptonote_protocol/cryptonote_protocol_handler.inl | 43 | ||||
| -rw-r--r-- | src/cryptonote_protocol/cryptonote_protocol_handler_common.h | 4 | ||||
| -rw-r--r-- | src/cryptonote_protocol/enums.h | 43 | ||||
| -rw-r--r-- | src/cryptonote_protocol/fwd.h | 37 | ||||
| -rw-r--r-- | src/cryptonote_protocol/levin_notify.cpp | 208 | ||||
| -rw-r--r-- | src/cryptonote_protocol/levin_notify.h | 18 |
7 files changed, 279 insertions, 76 deletions
diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.h b/src/cryptonote_protocol/cryptonote_protocol_handler.h index 3b456e324..ddbd45a61 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.h +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.h @@ -130,7 +130,7 @@ namespace cryptonote //----------------- i_bc_protocol_layout --------------------------------------- virtual bool relay_block(NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& exclude_context); - virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, cryptonote_connection_context& exclude_context); + virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone); //---------------------------------------------------------------------------------- //bool get_payload_sync_data(HANDSHAKE_DATA::request& hshd, cryptonote_connection_context& context); bool should_drop_connection(cryptonote_connection_context& context, uint32_t next_stripe); diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler.inl b/src/cryptonote_protocol/cryptonote_protocol_handler.inl index 74ceeb41d..a7bf0c283 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler.inl +++ b/src/cryptonote_protocol/cryptonote_protocol_handler.inl @@ -188,7 +188,7 @@ namespace cryptonote auto connection_time = time(NULL) - cntxt.m_started; ss << std::setw(30) << std::left << std::string(cntxt.m_is_income ? " [INC]":"[OUT]") + cntxt.m_remote_address.str() - << std::setw(20) << std::hex << peer_id + << std::setw(20) << nodetool::peerid_to_string(peer_id) << std::setw(20) << std::hex << support_flags << std::setw(30) << std::to_string(cntxt.m_recv_cnt)+ "(" + std::to_string(time(NULL) - cntxt.m_last_recv) + ")" + "/" + std::to_string(cntxt.m_send_cnt) + "(" + std::to_string(time(NULL) - cntxt.m_last_send) + ")" << std::setw(25) << get_protocol_state_string(cntxt.m_state) @@ -248,9 +248,7 @@ namespace cryptonote cnx.rpc_port = cntxt.m_rpc_port; cnx.rpc_credits_per_hash = cntxt.m_rpc_credits_per_hash; - std::stringstream peer_id_str; - peer_id_str << std::hex << std::setw(16) << peer_id; - peer_id_str >> cnx.peer_id; + cnx.peer_id = nodetool::peerid_to_string(peer_id); cnx.support_flags = support_flags; @@ -455,7 +453,7 @@ namespace cryptonote for(auto tx_blob_it = arg.b.txs.begin(); tx_blob_it!=arg.b.txs.end();tx_blob_it++) { cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc); - m_core.handle_incoming_tx(*tx_blob_it, tvc, true, true, false); + m_core.handle_incoming_tx(*tx_blob_it, tvc, relay_method::block, true); if(tvc.m_verifivation_failed) { LOG_PRINT_CCONTEXT_L1("Block verification failed: transaction verification failed, dropping connection"); @@ -619,7 +617,7 @@ namespace cryptonote { MDEBUG("Incoming tx " << tx_hash << " not in pool, adding"); cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc); - if(!m_core.handle_incoming_tx(tx_blob, tvc, true, true, false) || tvc.m_verifivation_failed) + if(!m_core.handle_incoming_tx(tx_blob, tvc, relay_method::block, true) || tvc.m_verifivation_failed) { LOG_PRINT_CCONTEXT_L1("Block verification failed: transaction verification failed, dropping connection"); drop_connection(context, false, false); @@ -667,13 +665,13 @@ namespace cryptonote drop_connection(context, false, false); m_core.resume_mine(); return 1; - } - + } + size_t tx_idx = 0; for(auto& tx_hash: new_block.tx_hashes) { cryptonote::blobdata txblob; - if(m_core.get_pool_transaction(tx_hash, txblob)) + if(m_core.get_pool_transaction(tx_hash, txblob, relay_category::broadcasted)) { have_tx.push_back({txblob, crypto::null_hash}); } @@ -702,7 +700,7 @@ namespace cryptonote need_tx_indices.push_back(tx_idx); } } - + ++tx_idx; } @@ -909,8 +907,8 @@ namespace cryptonote newtxs.reserve(arg.txs.size()); for (size_t i = 0; i < arg.txs.size(); ++i) { - cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc); - m_core.handle_incoming_tx({arg.txs[i], crypto::null_hash}, tvc, false, true, false); + cryptonote::tx_verification_context tvc{}; + m_core.handle_incoming_tx({arg.txs[i], crypto::null_hash}, tvc, relay_method::fluff, true); if(tvc.m_verifivation_failed) { LOG_PRINT_CCONTEXT_L1("Tx verification failed, dropping connection"); @@ -925,7 +923,7 @@ namespace cryptonote if(arg.txs.size()) { //TODO: add announce usage here - relay_transactions(arg, context); + relay_transactions(arg, context.m_connection_id, context.m_remote_address.get_zone()); } return 1; @@ -1316,7 +1314,7 @@ namespace cryptonote TIME_MEASURE_START(transactions_process_time); num_txs += block_entry.txs.size(); std::vector<tx_verification_context> tvc; - m_core.handle_incoming_txs(block_entry.txs, tvc, true, true, false); + m_core.handle_incoming_txs(block_entry.txs, tvc, relay_method::block, true); if (tvc.size() != block_entry.txs.size()) { LOG_ERROR_CCONTEXT("Internal error: tvc.size() != block_entry.txs.size()"); @@ -2181,7 +2179,8 @@ skip: MGINFO_YELLOW(ENDL << "**********************************************************************" << ENDL << "You are now synchronized with the network. You may now start monero-wallet-cli." << ENDL << ENDL - << "Use the \"help\" command to see the list of available commands." << ENDL + << "Use the \"help\" command to see a simplified list of available commands." << ENDL + << "Use the \"help_advanced\" command to see an advanced list of available commands." << ENDL << "**********************************************************************"); m_sync_timer.pause(); if (ELPP->vRegistry()->allowed(el::Level::Info, "sync-info")) @@ -2344,14 +2343,14 @@ skip: } //------------------------------------------------------------------------------------------------------------------------ template<class t_core> - bool t_cryptonote_protocol_handler<t_core>::relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, cryptonote_connection_context& exclude_context) + bool t_cryptonote_protocol_handler<t_core>::relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone) { - for(auto& tx_blob : arg.txs) - m_core.on_transaction_relayed(tx_blob); - - // no check for success, so tell core they're relayed unconditionally - m_p2p->send_txs(std::move(arg.txs), exclude_context.m_remote_address.get_zone(), exclude_context.m_connection_id, m_core.pad_transactions()); - return true; + /* Push all outgoing transactions to this function. The behavior needs to + identify how the transaction is going to be relayed, and then update the + local mempool before doing the relay. The code was already updating the + DB twice on received transactions - it is difficult to workaround this + due to the internal design. */ + return m_p2p->send_txs(std::move(arg.txs), zone, source, m_core) != epee::net_utils::zone::invalid; } //------------------------------------------------------------------------------------------------------------------------ template<class t_core> diff --git a/src/cryptonote_protocol/cryptonote_protocol_handler_common.h b/src/cryptonote_protocol/cryptonote_protocol_handler_common.h index a67178c52..978a9ebf3 100644 --- a/src/cryptonote_protocol/cryptonote_protocol_handler_common.h +++ b/src/cryptonote_protocol/cryptonote_protocol_handler_common.h @@ -41,7 +41,7 @@ namespace cryptonote struct i_cryptonote_protocol { virtual bool relay_block(NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& exclude_context)=0; - virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, cryptonote_connection_context& exclude_context)=0; + virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone)=0; //virtual bool request_objects(NOTIFY_REQUEST_GET_OBJECTS::request& arg, cryptonote_connection_context& context)=0; }; @@ -54,7 +54,7 @@ namespace cryptonote { return false; } - virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, cryptonote_connection_context& exclude_context) + virtual bool relay_transactions(NOTIFY_NEW_TRANSACTIONS::request& arg, const boost::uuids::uuid& source, epee::net_utils::zone zone) { return false; } diff --git a/src/cryptonote_protocol/enums.h b/src/cryptonote_protocol/enums.h new file mode 100644 index 000000000..2ec622d94 --- /dev/null +++ b/src/cryptonote_protocol/enums.h @@ -0,0 +1,43 @@ +// Copyright (c) 2019, The Monero Project +// +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#pragma once + +#include <cstdint> + +namespace cryptonote +{ + //! Methods tracking how a tx was received and relayed + enum class relay_method : std::uint8_t + { + none = 0, //!< Received via RPC with `do_not_relay` set + local, //!< Received via RPC; trying to send over i2p/tor, etc. + block, //!< Received in block, takes precedence over others + fluff //!< Received/sent over public networks + }; +} diff --git a/src/cryptonote_protocol/fwd.h b/src/cryptonote_protocol/fwd.h new file mode 100644 index 000000000..616b48be3 --- /dev/null +++ b/src/cryptonote_protocol/fwd.h @@ -0,0 +1,37 @@ +// Copyright (c) 2019, The Monero Project +// +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without modification, are +// permitted provided that the following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of +// conditions and the following disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list +// of conditions and the following disclaimer in the documentation and/or other +// materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be +// used to endorse or promote products derived from this software without specific +// prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY +// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL +// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF +// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +#pragma once + +namespace cryptonote +{ + class core; + struct cryptonote_connection_context; + struct i_core_events; +} + diff --git a/src/cryptonote_protocol/levin_notify.cpp b/src/cryptonote_protocol/levin_notify.cpp index 4b41b5bfc..e45c34e02 100644 --- a/src/cryptonote_protocol/levin_notify.cpp +++ b/src/cryptonote_protocol/levin_notify.cpp @@ -33,6 +33,7 @@ #include <chrono> #include <deque> #include <stdexcept> +#include <utility> #include "common/expect.h" #include "common/varint.h" @@ -43,6 +44,14 @@ #include "net/dandelionpp.h" #include "p2p/net_node.h" +namespace +{ + int get_command_from_message(const cryptonote::blobdata &msg) + { + return msg.size() >= sizeof(epee::levin::bucket_head2) ? SWAP32LE(((epee::levin::bucket_head2*)msg.data())->m_command) : 0; + } +} + namespace cryptonote { namespace levin @@ -57,6 +66,37 @@ namespace levin constexpr const std::chrono::seconds noise_min_delay{CRYPTONOTE_NOISE_MIN_DELAY}; constexpr const std::chrono::seconds noise_delay_range{CRYPTONOTE_NOISE_DELAY_RANGE}; + /* A custom duration is used for the poisson distribution because of the + variance. If 5 seconds is given to `std::poisson_distribution`, 95% of + the values fall between 1-9s in 1s increments (not granular enough). If + 5000 milliseconds is given, 95% of the values fall between 4859ms-5141ms + in 1ms increments (not enough time variance). Providing 20 quarter + seconds yields 95% of the values between 3s-7.25s in 1/4s increments. */ + using fluff_stepsize = std::chrono::duration<std::chrono::milliseconds::rep, std::ratio<1, 4>>; + constexpr const std::chrono::seconds fluff_average_in{CRYPTONOTE_DANDELIONPP_FLUSH_AVERAGE}; + + /*! Bitcoin Core is using 1/2 average seconds for outgoing connections + compared to incoming. The thinking is that the user controls outgoing + connections (Dandelion++ makes similar assumptions in its stem + algorithm). The randomization yields 95% values between 1s-4s in + 1/4s increments. */ + constexpr const fluff_stepsize fluff_average_out{fluff_stepsize{fluff_average_in} / 2}; + + class random_poisson + { + std::poisson_distribution<fluff_stepsize::rep> dist; + public: + explicit random_poisson(fluff_stepsize average) + : dist(average.count() < 0 ? 0 : average.count()) + {} + + fluff_stepsize operator()() + { + crypto::random_device rand{}; + return fluff_stepsize{dist(rand)}; + } + }; + /*! Select a randomized duration from 0 to `range`. The precision will be to the systems `steady_clock`. As an example, supplying 3 seconds to this function will select a duration from [0, 3] seconds, and the increments @@ -129,6 +169,16 @@ namespace levin return fullBlob; } + bool make_payload_send_txs(connections& p2p, std::vector<blobdata>&& txs, const boost::uuids::uuid& destination, const bool pad) + { + const cryptonote::blobdata blob = make_tx_payload(std::move(txs), pad); + p2p.for_connection(destination, [&blob](detail::p2p_context& context) { + on_levin_traffic(context, true, true, false, blob.size(), get_command_from_message(blob)); + return true; + }); + return p2p.notify(NOTIFY_NEW_TRANSACTIONS::ID, epee::strspan<std::uint8_t>(blob), destination); + } + /* The current design uses `asio::strand`s. The documentation isn't as clear as it should be - a `strand` has an internal `mutex` and `bool`. The `mutex` synchronizes thread access and the `bool` is set when a thread is @@ -187,15 +237,18 @@ namespace levin { struct zone { - explicit zone(boost::asio::io_service& io_service, std::shared_ptr<connections> p2p, epee::byte_slice noise_in, bool is_public) + explicit zone(boost::asio::io_service& io_service, std::shared_ptr<connections> p2p, epee::byte_slice noise_in, bool is_public, bool pad_txs) : p2p(std::move(p2p)), noise(std::move(noise_in)), next_epoch(io_service), + flush_txs(io_service), strand(io_service), map(), channels(), + flush_time(std::chrono::steady_clock::time_point::max()), connection_count(0), - is_public(is_public) + is_public(is_public), + pad_txs(pad_txs) { for (std::size_t count = 0; !noise.empty() && count < CRYPTONOTE_NOISE_CHANNELS; ++count) channels.emplace_back(io_service); @@ -204,11 +257,14 @@ namespace levin const std::shared_ptr<connections> p2p; const epee::byte_slice noise; //!< `!empty()` means zone is using noise channels boost::asio::steady_timer next_epoch; + boost::asio::steady_timer flush_txs; boost::asio::io_service::strand strand; net::dandelionpp::connection_map map;//!< Tracks outgoing uuid's for noise channels or Dandelion++ stems std::deque<noise_channel> channels; //!< Never touch after init; only update elements on `noise_channel.strand` + std::chrono::steady_clock::time_point flush_time; //!< Next expected Dandelion++ fluff flush std::atomic<std::size_t> connection_count; //!< Only update in strand, can be read at any time const bool is_public; //!< Zone is public ipv4/ipv6 connections + const bool pad_txs; //!< Pad txs to the next boundary for privacy }; } // detail @@ -245,49 +301,112 @@ namespace levin } }; - //! Sends a message to every active connection - class flood_notify + //! Sends txs on connections with expired timers, and queues callback for next timer expiration (if any). + struct fluff_flush { std::shared_ptr<detail::zone> zone_; - epee::byte_slice message_; // Requires manual copy - boost::uuids::uuid source_; + std::chrono::steady_clock::time_point flush_time_; - public: - explicit flood_notify(std::shared_ptr<detail::zone> zone, epee::byte_slice message, const boost::uuids::uuid& source) - : zone_(std::move(zone)), message_(message.clone()), source_(source) - {} + static void queue(std::shared_ptr<detail::zone> zone, const std::chrono::steady_clock::time_point flush_time) + { + assert(zone != nullptr); + assert(zone->strand.running_in_this_thread()); - flood_notify(flood_notify&&) = default; - flood_notify(const flood_notify& source) - : zone_(source.zone_), message_(source.message_.clone()), source_(source.source_) - {} + detail::zone& this_zone = *zone; + this_zone.flush_time = flush_time; + this_zone.flush_txs.expires_at(flush_time); + this_zone.flush_txs.async_wait(this_zone.strand.wrap(fluff_flush{std::move(zone), flush_time})); + } - void operator()() const + void operator()(const boost::system::error_code error) { if (!zone_ || !zone_->p2p) return; assert(zone_->strand.running_in_this_thread()); - /* The foreach should be quick, but then it iterates and acquires the - same lock for every connection. So do in a strand because two threads - will ping-pong each other with cacheline invalidations. Revisit if - algorithm changes or the locking strategy within the levin config - class changes. */ - - std::vector<boost::uuids::uuid> connections; - connections.reserve(connection_id_reserve_size); - zone_->p2p->foreach_connection([this, &connections] (detail::p2p_context& context) { - /* Only send to outgoing connections when "flooding" over i2p/tor. - Otherwise this makes the tx linkable to a hidden service address, - making things linkable across connections. */ + const bool timer_error = bool(error); + if (timer_error) + { + if (error != boost::system::errc::operation_canceled) + throw boost::system::system_error{error, "fluff_flush timer failed"}; + + // new timer canceled this one set in future + if (zone_->flush_time < flush_time_) + return; + } + + const auto now = std::chrono::steady_clock::now(); + auto next_flush = std::chrono::steady_clock::time_point::max(); + std::vector<std::pair<std::vector<blobdata>, boost::uuids::uuid>> connections{}; + zone_->p2p->foreach_connection([timer_error, now, &next_flush, &connections] (detail::p2p_context& context) + { + if (!context.fluff_txs.empty()) + { + if (context.flush_time <= now || timer_error) // flush on canceled timer + { + context.flush_time = std::chrono::steady_clock::time_point::max(); + connections.emplace_back(std::move(context.fluff_txs), context.m_connection_id); + context.fluff_txs.clear(); + } + else // not flushing yet + next_flush = std::min(next_flush, context.flush_time); + } + else // nothing to flush + context.flush_time = std::chrono::steady_clock::time_point::max(); + return true; + }); + + for (auto& connection : connections) + make_payload_send_txs(*zone_->p2p, std::move(connection.first), connection.second, zone_->pad_txs); + + if (next_flush != std::chrono::steady_clock::time_point::max()) + fluff_flush::queue(std::move(zone_), next_flush); + else + zone_->flush_time = next_flush; // signal that no timer is set + } + }; + + /*! The "fluff" portion of the Dandelion++ algorithm. Every tx is queued + per-connection and flushed with a randomized poisson timer. This + implementation only has one system timer per-zone, and instead tracks + the lowest flush time. */ + struct fluff_notify + { + std::shared_ptr<detail::zone> zone_; + std::vector<blobdata> txs_; + boost::uuids::uuid source_; + + void operator()() + { + if (!zone_ || !zone_->p2p || txs_.empty()) + return; + + assert(zone_->strand.running_in_this_thread()); + + const auto now = std::chrono::steady_clock::now(); + auto next_flush = std::chrono::steady_clock::time_point::max(); + + random_poisson in_duration(fluff_average_in); + random_poisson out_duration(fluff_average_out); + + zone_->p2p->foreach_connection([this, now, &in_duration, &out_duration, &next_flush] (detail::p2p_context& context) + { if (this->source_ != context.m_connection_id && (this->zone_->is_public || !context.m_is_income)) - connections.emplace_back(context.m_connection_id); + { + if (context.fluff_txs.empty()) + context.flush_time = now + (context.m_is_income ? in_duration() : out_duration()); + + next_flush = std::min(next_flush, context.flush_time); + context.fluff_txs.reserve(context.fluff_txs.size() + this->txs_.size()); + for (const blobdata& tx : this->txs_) + context.fluff_txs.push_back(tx); // must copy instead of move (multiple conns) + } return true; }); - for (const boost::uuids::uuid& connection : connections) - zone_->p2p->send(message_.clone(), connection); + if (next_flush < zone_->flush_time) + fluff_flush::queue(std::move(zone_), next_flush); } }; @@ -432,6 +551,10 @@ namespace levin else message = zone_->noise.clone(); + zone_->p2p->for_connection(channel.connection, [&](detail::p2p_context& context) { + on_levin_traffic(context, true, true, false, message.size(), "noise"); + return true; + }); if (zone_->p2p->send(std::move(message), channel.connection)) { if (!channel.queue.empty() && channel.active.empty()) @@ -451,7 +574,7 @@ namespace levin } }; - //! Prepares connections for new channel epoch and sets timer for next epoch + //! Prepares connections for new channel/dandelionpp epoch and sets timer for next epoch struct start_epoch { // Variables allow for Dandelion++ extension @@ -481,8 +604,8 @@ namespace levin }; } // anonymous - notify::notify(boost::asio::io_service& service, std::shared_ptr<connections> p2p, epee::byte_slice noise, bool is_public) - : zone_(std::make_shared<detail::zone>(service, std::move(p2p), std::move(noise), is_public)) + notify::notify(boost::asio::io_service& service, std::shared_ptr<connections> p2p, epee::byte_slice noise, const bool is_public, const bool pad_txs) + : zone_(std::make_shared<detail::zone>(service, std::move(p2p), std::move(noise), is_public, pad_txs)) { if (!zone_->p2p) throw std::logic_error{"cryptonote::levin::notify cannot have nullptr p2p argument"}; @@ -533,9 +656,19 @@ namespace levin channel.next_noise.cancel(); } - bool notify::send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source, const bool pad_txs) + void notify::run_fluff() { if (!zone_) + return; + zone_->flush_txs.cancel(); + } + + bool notify::send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source) + { + if (txs.empty()) + return true; + + if (!zone_) return false; if (!zone_->noise.empty() && !zone_->channels.empty()) @@ -565,12 +698,7 @@ namespace levin } else { - const std::string payload = make_tx_payload(std::move(txs), pad_txs); - epee::byte_slice message = - epee::levin::make_notify(NOTIFY_NEW_TRANSACTIONS::ID, epee::strspan<std::uint8_t>(payload)); - - // traditional monero send technique - zone_->strand.dispatch(flood_notify{zone_, std::move(message), source}); + zone_->strand.dispatch(fluff_notify{zone_, std::move(txs), source}); } return true; diff --git a/src/cryptonote_protocol/levin_notify.h b/src/cryptonote_protocol/levin_notify.h index 484243af5..ce652d933 100644 --- a/src/cryptonote_protocol/levin_notify.h +++ b/src/cryptonote_protocol/levin_notify.h @@ -35,6 +35,7 @@ #include "byte_slice.h" #include "cryptonote_basic/blobdatatype.h" +#include "cryptonote_protocol/fwd.h" #include "net/enums.h" #include "span.h" @@ -53,11 +54,6 @@ namespace nodetool namespace cryptonote { - struct cryptonote_connection_context; -} - -namespace cryptonote -{ namespace levin { namespace detail @@ -86,7 +82,7 @@ namespace levin {} //! Construct an instance with available notification `zones`. - explicit notify(boost::asio::io_service& service, std::shared_ptr<connections> p2p, epee::byte_slice noise, bool is_public); + explicit notify(boost::asio::io_service& service, std::shared_ptr<connections> p2p, epee::byte_slice noise, bool is_public, bool pad_txs); notify(const notify&) = delete; notify(notify&&) = default; @@ -108,11 +104,14 @@ namespace levin //! Run the logic for the next stem timeout imemdiately. Only use in testing. void run_stems(); + //! Run the logic for flushing all Dandelion++ fluff queued txs. Only use in testing. + void run_fluff(); + /*! Send txs using `cryptonote_protocol_defs.h` payload format wrapped in a levin header. The message will be sent in a "discreet" manner if enabled - if `!noise.empty()` then the `command`/`payload` will be queued to send at the next available noise interval. Otherwise, a - standard Monero flood notification will be used. + Dandelion++ fluff algorithm will be used. \note Eventually Dandelion++ stem sending will be used here when enabled. @@ -121,12 +120,9 @@ namespace levin \param source The source of the notification. `is_nil()` indicates this node is the source. Dandelion++ will use this to map a source to a particular stem. - \param pad_txs A request to pad txs to help conceal origin via - statistical analysis. Ignored if noise was enabled during - construction. \return True iff the notification is queued for sending. */ - bool send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source, bool pad_txs); + bool send_txs(std::vector<blobdata> txs, const boost::uuids::uuid& source); }; } // levin } // net |