From b8028e31ade5fb3bf58cfc49fab0dcc48ed54e2a Mon Sep 17 00:00:00 2001 From: Lee Clagett Date: Mon, 18 Jan 2021 00:08:19 +0000 Subject: [PATCH] Change p2p connection map from raw pointers to weak_ptrs --- .../epee/include/net/abstract_tcp_server2.h | 22 +- .../epee/include/net/abstract_tcp_server2.inl | 66 +-- .../epee/include/net/http_protocol_handler.h | 10 + .../net/levin_protocol_handler_async.h | 376 ++++++++---------- contrib/epee/include/net/net_utils_base.h | 24 +- tests/fuzz/levin.cpp | 21 +- tests/unit_tests/epee_boosted_tcp_server.cpp | 101 +++-- .../epee_levin_protocol_handler_async.cpp | 20 +- tests/unit_tests/http.cpp | 2 - tests/unit_tests/levin.cpp | 269 +++++++------ tests/unit_tests/node_server.cpp | 4 +- 11 files changed, 460 insertions(+), 455 deletions(-) diff --git a/contrib/epee/include/net/abstract_tcp_server2.h b/contrib/epee/include/net/abstract_tcp_server2.h index f673ff59f..a41069921 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.h +++ b/contrib/epee/include/net/abstract_tcp_server2.h @@ -92,10 +92,10 @@ namespace net_utils /// Represents a single connection from a client. template class connection - : public boost::enable_shared_from_this >, - private boost::noncopyable, - public i_service_endpoint, - public connection_basic + : public std::enable_shared_from_this>, + private boost::noncopyable, + public connection_basic, // shared_state shared_ptr must be destroyed after service_endpoint + public service_endpoint { public: typedef typename t_protocol_handler::connection_context t_connection_context; @@ -260,14 +260,13 @@ namespace net_utils io_context_t &m_io_context; t_connection_type m_connection_type; - t_connection_context m_conn_context{}; strand_t m_strand; timers_t m_timers; connection_ptr self{}; bool m_local{}; std::string m_host{}; state_t m_state{}; - t_protocol_handler m_handler; + public: struct shared_state : connection_basic_shared_state, t_protocol_handler::config_type { @@ -310,7 +309,7 @@ namespace net_utils // `real_remote` is the actual endpoint (if connection is to proxy, etc.) bool start(bool is_income, bool is_multithreaded, network_address real_remote); - void get_context(t_connection_context& context_){context_ = m_conn_context;} + void get_context(t_connection_context& context_){context_ = get_context();} void call_back_starter(); @@ -331,9 +330,12 @@ namespace net_utils virtual bool call_run_once_service_io(); virtual bool request_callback(); virtual io_context_t& get_io_context(); - virtual bool add_ref(); - virtual bool release(); //------------------------------------------------------ + const t_connection_context& get_context() const noexcept { return this->context; } + t_connection_context& get_context() noexcept { return this->context; } + + const t_protocol_handler& get_protocol_handler() const noexcept { return this->m_protocol_handler; } + t_protocol_handler& get_protocol_handler() noexcept { return this->m_protocol_handler; } public: void setRpcStation(); }; @@ -354,7 +356,7 @@ namespace net_utils }; public: - typedef boost::shared_ptr > connection_ptr; + typedef std::shared_ptr> connection_ptr; typedef typename t_protocol_handler::connection_context t_connection_context; /// Construct the server to listen on the specified TCP address and port, and /// serve up files from the given directory. diff --git a/contrib/epee/include/net/abstract_tcp_server2.inl b/contrib/epee/include/net/abstract_tcp_server2.inl index 09d5a5e78..3473b1a37 100644 --- a/contrib/epee/include/net/abstract_tcp_server2.inl +++ b/contrib/epee/include/net/abstract_tcp_server2.inl @@ -370,9 +370,9 @@ namespace net_utils { m_state.stat.in.throttle.handle_trafic_exact(bytes_transferred); const auto speed = m_state.stat.in.throttle.get_current_speed(); - m_conn_context.m_current_speed_down = speed; - m_conn_context.m_max_speed_down = std::max( - m_conn_context.m_max_speed_down, + get_context().m_current_speed_down = speed; + get_context().m_max_speed_down = std::max( + get_context().m_max_speed_down, speed ); if (speed_limit_is_enabled()) { @@ -383,8 +383,8 @@ namespace net_utils ).handle_trafic_exact(bytes_transferred); } connection_basic::logger_handle_net_read(bytes_transferred); - m_conn_context.m_last_recv = time(NULL); - m_conn_context.m_recv_cnt += bytes_transferred; + get_context().m_last_recv = time(NULL); + get_context().m_recv_cnt += bytes_transferred; start_timer(get_timeout_from_bytes_read(bytes_transferred), true); } handle_read(bytes_transferred); @@ -428,7 +428,7 @@ namespace net_utils [this, self, bytes_transferred]{ bool success = false; TRY_ENTRY(); - success = m_handler.handle_recv( + success = get_protocol_handler().handle_recv( reinterpret_cast(m_state.data.read.buffer.data()), bytes_transferred ); @@ -546,9 +546,9 @@ namespace net_utils { m_state.stat.out.throttle.handle_trafic_exact(bytes_transferred); const auto speed = m_state.stat.out.throttle.get_current_speed(); - m_conn_context.m_current_speed_up = speed; - m_conn_context.m_max_speed_down = std::max( - m_conn_context.m_max_speed_down, + get_context().m_current_speed_up = speed; + get_context().m_max_speed_down = std::max( + get_context().m_max_speed_down, speed ); if (speed_limit_is_enabled()) { @@ -559,8 +559,8 @@ namespace net_utils ).handle_trafic_exact(bytes_transferred); } connection_basic::logger_handle_net_write(bytes_transferred); - m_conn_context.m_last_send = time(NULL); - m_conn_context.m_send_cnt += bytes_transferred; + get_context().m_last_send = time(NULL); + get_context().m_send_cnt += bytes_transferred; start_timer(get_default_timeout(), true); } @@ -678,7 +678,7 @@ namespace net_utils return; m_state.protocol.wait_release = true; m_state.lock.unlock(); - m_handler.release_protocol(); + get_protocol_handler().release_protocol(); m_state.lock.lock(); m_state.protocol.wait_release = false; m_state.protocol.released = true; @@ -854,7 +854,7 @@ namespace net_utils return true; if (m_connection_type == e_connection_type_P2P) { - MWARNING("Connection " << m_conn_context.m_connection_id << " tripped write limit, terminating"); + MWARNING("Connection " << get_context().m_connection_id << " tripped write limit, terminating"); terminate_async(); return false; } @@ -997,7 +997,7 @@ namespace net_utils if (ec.value()) return false; connection_basic::m_is_multithreaded = is_multithreaded; - m_conn_context.set_details( + get_context().set_details( boost::uuids::random_generator()(), *real_remote, is_income, @@ -1021,7 +1021,7 @@ namespace net_utils ); m_state.protocol.wait_init = true; guard.unlock(); - m_handler.after_init_connection(); + static_cast(connection_basic::get_state()).after_init_connection(connection::shared_from_this()); guard.lock(); m_state.protocol.wait_init = false; m_state.protocol.initialized = true; @@ -1065,13 +1065,13 @@ namespace net_utils t_connection_context&& initial ): connection_basic(io_context, std::move(socket), shared_state, ssl_support), - m_handler(this, *shared_state, m_conn_context), + service_endpoint(check_and_get(shared_state)), m_connection_type(connection_type), m_io_context{io_context}, - m_conn_context(std::move(initial)), m_strand{m_io_context}, m_timers{m_io_context} { + get_context() = std::move(initial); } template @@ -1126,7 +1126,7 @@ namespace net_utils " connection type " << std::to_string(m_connection_type) << " " << connection_basic::socket().local_endpoint().address().to_string() << ":" << connection_basic::socket().local_endpoint().port() << - " <--> " << m_conn_context.m_remote_address.str() << + " <--> " << get_context().m_remote_address.str() << " (via " << address << ":" << port << ")" ); } @@ -1175,7 +1175,7 @@ namespace net_utils // stopping the server, we don't want the io_context to stop before the shutdown sequence completes, since we // execute terminate inside m_strand. So we wait for the connection's shutdown sequence to complete before stopping // the io_context. - MDEBUG("Waiting for connection " << m_conn_context.m_connection_id << " to shutdown, current state: " << m_state.status); + MDEBUG("Waiting for connection " << get_context().m_connection_id << " to shutdown, current state: " << m_state.status); const bool shutdown = m_state.condition.wait_for( m_state.lock, std::chrono::seconds(5), @@ -1186,9 +1186,9 @@ namespace net_utils } ); if (shutdown) - MDEBUG("Shut down connection " << m_conn_context.m_connection_id); + MDEBUG("Shut down connection " << get_context().m_connection_id); else - MERROR("Connection " << m_conn_context.m_connection_id << " did not shut down"); + MERROR("Connection " << get_context().m_connection_id << " did not shut down"); return shutdown; } @@ -1217,7 +1217,7 @@ namespace net_utils ++m_state.protocol.wait_callback; boost::asio::post(connection_basic::strand_, [this, self]{ TRY_ENTRY(); - m_handler.handle_qued_callback(); + get_protocol_handler().handle_qued_callback(); CATCH_ENTRY_SWALLOW_EX("m_handler.handle_qued_callback"); std::lock_guard guard(m_state.lock); --m_state.protocol.wait_callback; @@ -1235,28 +1235,6 @@ namespace net_utils return m_io_context; } - template - bool connection::add_ref() - { - auto self = connection::weak_from_this().lock(); - if (!self) - return false; - std::lock_guard guard(m_state.lock); - this->self = std::move(self); - ++m_state.protocol.reference_counter; - return true; - } - - template - bool connection::release() - { - connection_ptr self; - std::lock_guard guard(m_state.lock); - if (!(--m_state.protocol.reference_counter)) - self = std::move(this->self); - return true; - } - template void connection::setRpcStation() { diff --git a/contrib/epee/include/net/http_protocol_handler.h b/contrib/epee/include/net/http_protocol_handler.h index 8f1ddd14c..844980240 100644 --- a/contrib/epee/include/net/http_protocol_handler.h +++ b/contrib/epee/include/net/http_protocol_handler.h @@ -64,6 +64,14 @@ namespace net_utils std::size_t m_max_private_ip_connections{25}; std::size_t m_max_connections{100}; critical_section m_lock; + + template + static bool after_init_connection(const std::shared_ptr& self) + { + if (!self) + return false; + return self->m_protocol_handler.after_init_connection(); + } }; // RPC limits groupable IPv6 clients by /64 to avoid per-address limit bypasses. @@ -96,7 +104,9 @@ namespace net_utils { return true; } + bool after_init_connection(); + virtual bool handle_recv(const void* ptr, size_t cb); virtual bool handle_request(const http::http_request_info& query_info, http_response_info& response); diff --git a/contrib/epee/include/net/levin_protocol_handler_async.h b/contrib/epee/include/net/levin_protocol_handler_async.h index 8ad40aef2..4a0e56514 100644 --- a/contrib/epee/include/net/levin_protocol_handler_async.h +++ b/contrib/epee/include/net/levin_protocol_handler_async.h @@ -28,7 +28,6 @@ #include #include #include -#include #include @@ -75,18 +74,26 @@ namespace levin template class async_protocol_handler; +template +struct get_handler +{ + using type = async_protocol_handler; +}; + template class async_protocol_handler_config { - typedef boost::unordered_map* > connections_map; + using derived_handler = typename get_handler::type; + typedef net_utils::service_endpoint levin_endpoint; + typedef boost::unordered_map> connections_map; critical_section m_connects_lock; connections_map m_connects; + std::atomic m_incoming_count; + std::atomic m_outgoing_count; - void add_connection(async_protocol_handler* pc); void del_connection(async_protocol_handler* pc); - async_protocol_handler* find_connection(boost::uuids::uuid connection_id) const; - int find_and_lock_connection(boost::uuids::uuid connection_id, async_protocol_handler*& aph); + std::shared_ptr find_and_lock_connection(const boost::uuids::uuid& connection_id); friend class async_protocol_handler; @@ -106,7 +113,6 @@ public: int send(epee::byte_slice message, const boost::uuids::uuid& connection_id); bool close(boost::uuids::uuid connection_id, const bool wait_for_shutdown); - bool update_connection_context(const t_connection_context& contxt); bool request_callback(boost::uuids::uuid connection_id); template bool foreach_connection(const callback_t &cb); @@ -116,8 +122,13 @@ public: size_t get_out_connections_count(); size_t get_in_connections_count(); void set_handler(levin_commands_handler* handler, void (*destroy)(levin_commands_handler*) = NULL); + bool after_init_connection(const std::shared_ptr& pconn); - async_protocol_handler_config():m_pcommands_handler(NULL), m_pcommands_handler_destroy(NULL), m_initial_max_packet_size(LEVIN_INITIAL_MAX_PACKET_SIZE), m_max_packet_size(LEVIN_DEFAULT_MAX_PACKET_SIZE), m_invoke_timeout(LEVIN_DEFAULT_TIMEOUT_PRECONFIGURED) + async_protocol_handler_config() + : m_incoming_count(0), m_outgoing_count(0), + m_pcommands_handler(NULL), m_pcommands_handler_destroy(NULL), + m_initial_max_packet_size(LEVIN_INITIAL_MAX_PACKET_SIZE), m_max_packet_size(LEVIN_DEFAULT_MAX_PACKET_SIZE), + m_invoke_timeout(LEVIN_DEFAULT_TIMEOUT_PRECONFIGURED) {} virtual ~async_protocol_handler_config() { set_handler(NULL, NULL); } void del_out_connections(size_t count); @@ -153,6 +164,7 @@ class async_protocol_handler } public: + using derived_handler = typename get_handler::type; typedef t_connection_context connection_context; typedef async_protocol_handler_config config_type; @@ -180,68 +192,73 @@ public: struct invoke_response_handler_base { + virtual ~invoke_response_handler_base() {} virtual bool handle(int res, const epee::span buff, connection_context& context)=0; - virtual bool is_timer_started() const=0; virtual void cancel()=0; virtual bool cancel_timer()=0; - virtual void reset_timer()=0; + virtual bool reset_timer(bool first)=0; }; template - struct anvoke_handler: invoke_response_handler_base + class anvoke_handler final : public invoke_response_handler_base, public std::enable_shared_from_this> { - anvoke_handler(const callback_t& cb, const std::chrono::milliseconds timeout, async_protocol_handler& con, int command) - :m_cb(cb), m_timeout(timeout), m_con(con), m_timer(con.m_pservice_endpoint->get_io_context()), m_timer_started(false), - m_cancel_timer_called(false), m_timer_cancelled(false), m_command(command) - { - if(m_con.start_outer_call()) - { - MDEBUG(con.get_context_ref() << "anvoke_handler, timeout: " << timeout.count()); - m_timer.expires_after(timeout); - m_timer.async_wait([&con, command, cb, timeout](const boost::system::error_code& ec) - { - if(ec == boost::asio::error::operation_aborted) - return; - MINFO(con.get_context_ref() << "Timeout on invoke operation happened, command: " << command << " timeout: " << timeout.count()); - epee::span fake; - cb(LEVIN_ERROR_CONNECTION_TIMEDOUT, fake, con.get_context_ref()); - con.close(false); - con.finish_outer_call(); - }); - m_timer_started = true; - } - } - virtual ~anvoke_handler() - {} - callback_t m_cb; - async_protocol_handler& m_con; + std::shared_ptr> m_con; boost::asio::steady_timer m_timer; - bool m_timer_started; + callback_t m_cb; + const std::chrono::milliseconds m_timeout; + const int m_command; bool m_cancel_timer_called; bool m_timer_cancelled; - const std::chrono::milliseconds m_timeout; - int m_command; - virtual bool handle(int res, const epee::span buff, typename async_protocol_handler::connection_context& context) + + void failure(const int rc) + { + std::shared_ptr> con; + m_con.swap(con); + if (!con) + return; + + MINFO(con->context << "Error (" << rc << ") on invoke operation happened, command: " << m_command << " timeout: " << m_timeout.count()); + m_cb(rc, nullptr, con->context); + con->close(false); + } + + public: + anvoke_handler(const callback_t& cb, const std::chrono::milliseconds timeout, std::shared_ptr> con, int command) + : invoke_response_handler_base(), + std::enable_shared_from_this>(), + m_con(con), + m_timer(con->get_io_context()), + m_cb(std::move(cb)), + m_timeout(timeout), + m_command(command), + m_cancel_timer_called(false), + m_timer_cancelled(false) + { + if (!m_con) + throw std::logic_error{"Unexpected nullptr connection"}; + } + + virtual ~anvoke_handler() override + { + failure(LEVIN_ERROR_CONNECTION_DESTROYED); + } + + virtual bool handle(int res, const epee::span buff, typename async_protocol_handler::connection_context& context) override final { if(!cancel_timer()) return false; - m_cb(res, buff, context); - m_con.finish_outer_call(); + + std::shared_ptr> con; + m_con.swap(con); + if (con) + m_cb(res, buff, context); return true; } - virtual bool is_timer_started() const - { - return m_timer_started; - } - virtual void cancel() + virtual void cancel() override final { if(cancel_timer()) - { - epee::span fake; - m_cb(LEVIN_ERROR_CONNECTION_DESTROYED, fake, m_con.get_context_ref()); - m_con.finish_outer_call(); - } + failure(LEVIN_ERROR_CONNECTION_DESTROYED); } - virtual bool cancel_timer() + virtual bool cancel_timer() override final { if(!m_cancel_timer_called) { @@ -250,33 +267,27 @@ public: } return m_timer_cancelled; } - virtual void reset_timer() + virtual bool reset_timer(bool first) override final { - if (!m_cancel_timer_called && m_timer.cancel() > 0) + std::shared_ptr self; + if (!m_cancel_timer_called && (self = this->weak_from_this().lock()) && (first || m_timer.cancel() > 0)) { - callback_t& cb = m_cb; - const auto timeout = m_timeout; - async_protocol_handler& con = m_con; - int command = m_command; m_timer.expires_after(m_timeout); - m_timer.async_wait([&con, cb, command, timeout](const boost::system::error_code& ec) + m_timer.async_wait([self = std::move(self)](const boost::system::error_code& ec) { - if(ec == boost::asio::error::operation_aborted) - return; - MINFO(con.get_context_ref() << "Timeout on invoke operation happened, command: " << command << " timeout: " << timeout.count()); - epee::span fake; - cb(LEVIN_ERROR_CONNECTION_TIMEDOUT, fake, con.get_context_ref()); - con.close(false); - con.finish_outer_call(); + if(ec != boost::asio::error::operation_aborted) + self->failure(LEVIN_ERROR_CONNECTION_TIMEDOUT); }); + return true; } + return false; } }; critical_section m_invoke_response_handlers_lock; - std::list > m_invoke_response_handlers; + std::list> m_invoke_response_handlers; template - bool add_invoke_response_handler(const callback_t &cb, const std::chrono::milliseconds timeout, async_protocol_handler& con, int command) + bool add_invoke_response_handler(const callback_t &cb, const std::chrono::milliseconds timeout, std::shared_ptr> con, int command) { CRITICAL_REGION_LOCAL(m_invoke_response_handlers_lock); if (m_protocol_released) @@ -284,17 +295,22 @@ public: MERROR("Adding response handler to a released object"); return false; } - boost::shared_ptr handler(boost::make_shared>(cb, timeout, con, command)); - m_invoke_response_handlers.push_back(handler); - return handler->is_timer_started(); + std::shared_ptr handler(std::make_shared>(cb, timeout, std::move(con), command)); + if (handler->reset_timer(true)) + { + m_invoke_response_handlers.push_back(std::move(handler)); + return true; + } + return false; } - template friend struct anvoke_handler; + template friend class anvoke_handler; public: async_protocol_handler(net_utils::i_service_endpoint* psnd_hndlr, config_type& config, t_connection_context& conn_context): + m_wait_count(0), m_current_head(bucket_head2()), - m_pservice_endpoint(psnd_hndlr), + m_pservice_endpoint(psnd_hndlr), m_config(config), m_connection_context(conn_context), m_max_packet_size(config.m_initial_max_packet_size), @@ -303,7 +319,6 @@ public: { m_close_called = 0; m_protocol_released = false; - m_wait_count = 0; m_oponent_protocol_ver = 0; m_connection_initialized = false; } @@ -317,37 +332,12 @@ public: m_config.del_connection(this); } - for (size_t i = 0; i < 60 * 1000 / 100 && 0 != m_wait_count; ++i) - { - misc_utils::sleep_no_w(100); - } - CHECK_AND_ASSERT_MES_NO_RET(0 == m_wait_count, "Failed to wait for operation completion. m_wait_count = " << m_wait_count.load()); - MTRACE(m_connection_context << "~async_protocol_handler()"); } catch (...) { /* ignore */ } } - bool start_outer_call() - { - MTRACE(m_connection_context << "[levin_protocol] -->> start_outer_call"); - if(!m_pservice_endpoint->add_ref()) - { - MERROR(m_connection_context << "[levin_protocol] -->> start_outer_call failed"); - return false; - } - ++m_wait_count; - return true; - } - bool finish_outer_call() - { - MTRACE(m_connection_context << "[levin_protocol] <<-- finish_outer_call"); - --m_wait_count; - m_pservice_endpoint->release(); - return true; - } - bool release_protocol() { decltype(m_invoke_response_handlers) local_invoke_response_handlers; @@ -358,9 +348,12 @@ public: // Never call callback inside critical section, that can cause deadlock. Callback can be called when // invoke_response_handler_base is cancelled - std::for_each(local_invoke_response_handlers.begin(), local_invoke_response_handlers.end(), [](const boost::shared_ptr& pinv_resp_hndlr) { - pinv_resp_hndlr->cancel(); - }); + for (const auto& weak : local_invoke_response_handlers) + { + const auto strong = weak.lock(); + if (strong) + strong->cancel(); + } return true; } @@ -373,15 +366,8 @@ public: return true; } - void update_connection_context(const connection_context& contxt) - { - m_connection_context = contxt; - } - void request_callback() { - const scope_guard scope_exit_handler(boost::bind(&async_protocol_handler::finish_outer_call, this)); - m_pservice_endpoint->request_callback(); } @@ -432,8 +418,10 @@ public: if (!m_invoke_response_handlers.empty()) { //async call scenario - boost::shared_ptr response_handler = m_invoke_response_handlers.front(); - response_handler->reset_timer(); + const std::shared_ptr response_handler = + m_invoke_response_handlers.front().lock(); + if (!response_handler || !response_handler->reset_timer(false)) + return false; // close should've been called before this, but just in case MDEBUG(m_connection_context << "LEVIN_PACKET partial msg received. len=" << cb << ", current total " << m_cache_in_buffer.size() << "/" << m_current_head.m_cb << " (" << (100.0f * m_cache_in_buffer.size() / (m_current_head.m_cb ? m_current_head.m_cb : 1)) << "%)"); } } @@ -502,11 +490,11 @@ public: boost::unique_lock invoke_response_handlers_guard(m_invoke_response_handlers_lock); if(!m_invoke_response_handlers.empty()) {//async call scenario - boost::shared_ptr response_handler = m_invoke_response_handlers.front(); - bool timer_cancelled = response_handler->cancel_timer(); - // Don't pop handler, to avoid destroying it - if(timer_cancelled) - m_invoke_response_handlers.pop_front(); + const std::shared_ptr response_handler = m_invoke_response_handlers.front().lock(); + bool timer_cancelled = false; + if (response_handler) + timer_cancelled = response_handler->cancel_timer(); + m_invoke_response_handlers.pop_front(); invoke_response_handlers_guard.unlock(); if(timer_cancelled) @@ -605,20 +593,10 @@ public: return true; } - bool after_init_connection() - { - if (!m_connection_initialized) - { - m_connection_initialized = true; - m_config.add_connection(this); - } - return true; - } - template - bool async_invoke(int command, message_writer in_msg, const callback_t &cb, std::chrono::milliseconds timeout = LEVIN_DEFAULT_TIMEOUT_PRECONFIGURED) + bool async_invoke(std::shared_ptr> self, int command, message_writer in_msg, const callback_t &cb, std::chrono::milliseconds timeout = LEVIN_DEFAULT_TIMEOUT_PRECONFIGURED) { - const scope_guard scope_exit_handler(boost::bind(&async_protocol_handler::finish_outer_call, this)); + assert(self && this == std::addressof(self->m_protocol_handler)); if(timeout == LEVIN_DEFAULT_TIMEOUT_PRECONFIGURED) timeout = m_config.m_invoke_timeout; @@ -638,7 +616,7 @@ public: break; } - if(!add_invoke_response_handler(cb, timeout, *this, command)) + if(!add_invoke_response_handler(cb, timeout, std::move(self), command)) { err_code = LEVIN_ERROR_CONNECTION_DESTROYED; break; @@ -665,8 +643,6 @@ public: \return 1 on success */ int send(byte_slice message) { - const scope_guard scope_exit_handler(boost::bind(&async_protocol_handler::finish_outer_call, this)); - if (!send_message(std::move(message))) { LOG_ERROR_CC(m_connection_context, "Failed to send message, dropping it"); @@ -685,26 +661,25 @@ void async_protocol_handler_config::del_connection(async_p { CRITICAL_REGION_BEGIN(m_connects_lock); m_connects.erase(pconn->get_connection_id()); + if (pconn->get_context_ref().m_is_income) + --m_incoming_count; + else + --m_outgoing_count; CRITICAL_REGION_END(); - m_pcommands_handler->on_connection_close(pconn->m_connection_context); + if (m_pcommands_handler) + m_pcommands_handler->on_connection_close(pconn->get_context_ref()); } //------------------------------------------------------------------------------------------ template void async_protocol_handler_config::delete_connections(size_t count, bool incoming) { - std::vector connections; - - const scope_guard scope_exit_handler([&connections]{ - for (auto &aph: connections) - aph->finish_outer_call(); - }); - + std::vector> connections; CRITICAL_REGION_BEGIN(m_connects_lock); for (auto& c: m_connects) { - if (c.second->m_connection_context.m_is_income == incoming) - if (c.second->start_outer_call()) - connections.push_back(c.second); + auto locked = c.second.lock(); + if (locked && locked->context.m_is_income == incoming) + connections.push_back(std::move(locked)); } // close random connections from the provided set @@ -712,7 +687,7 @@ void async_protocol_handler_config::delete_connections(siz unsigned seed = std::chrono::system_clock::now().time_since_epoch().count(); shuffle(connections.begin(), connections.end(), std::default_random_engine(seed)); for (size_t i = 0; i < connections.size() && i < count; ++i) - m_connects.erase(connections[i]->get_connection_id()); + m_connects.erase(connections[i]->context.m_connection_id); CRITICAL_REGION_END(); @@ -733,60 +708,56 @@ void async_protocol_handler_config::del_in_connections(siz } //------------------------------------------------------------------------------------------ template -void async_protocol_handler_config::add_connection(async_protocol_handler* pconn) +bool async_protocol_handler_config::after_init_connection(const std::shared_ptr& pconn) { + if (!pconn || pconn->m_protocol_handler.m_connection_initialized) + return false; + CRITICAL_REGION_BEGIN(m_connects_lock); - m_connects[pconn->get_connection_id()] = pconn; + if (!m_connects.emplace(pconn->context.m_connection_id, pconn).second) + return false; + + pconn->m_protocol_handler.m_connection_initialized = true; + if (pconn->context.m_is_income) + ++m_incoming_count; + else + ++m_outgoing_count; CRITICAL_REGION_END(); - m_pcommands_handler->on_connection_new(pconn->m_connection_context); + if (m_pcommands_handler) + m_pcommands_handler->on_connection_new(pconn->context); + return true; } //------------------------------------------------------------------------------------------ template -async_protocol_handler* async_protocol_handler_config::find_connection(boost::uuids::uuid connection_id) const -{ - auto it = m_connects.find(connection_id); - return it == m_connects.end() ? 0 : it->second; -} -//------------------------------------------------------------------------------------------ -template -int async_protocol_handler_config::find_and_lock_connection(boost::uuids::uuid connection_id, async_protocol_handler*& aph) +std::shared_ptr::type>> async_protocol_handler_config::find_and_lock_connection(const boost::uuids::uuid& connection_id) { CRITICAL_REGION_LOCAL(m_connects_lock); - aph = find_connection(connection_id); - if(0 == aph) - return LEVIN_ERROR_CONNECTION_NOT_FOUND; - if(!aph->start_outer_call()) - return LEVIN_ERROR_CONNECTION_DESTROYED; - return LEVIN_OK; + const auto aph = m_connects.find(connection_id); + return aph == m_connects.end() ? nullptr : aph->second.lock(); } //------------------------------------------------------------------------------------------ template template int async_protocol_handler_config::invoke_async(int command, message_writer in_msg, boost::uuids::uuid connection_id, const callback_t &cb, const std::chrono::milliseconds timeout) { - async_protocol_handler* aph; - int r = find_and_lock_connection(connection_id, aph); - return LEVIN_OK == r ? aph->async_invoke(command, std::move(in_msg), cb, timeout) : r; + std::shared_ptr con = find_and_lock_connection(connection_id); + if (!con) + return LEVIN_ERROR_CONNECTION_NOT_FOUND; + levin_endpoint& ref = *con; + return ref.m_protocol_handler.async_invoke(std::move(con), command, std::move(in_msg), cb, timeout); } //------------------------------------------------------------------------------------------ template template bool async_protocol_handler_config::foreach_connection(const callback_t &cb) { - std::vector conn; - - const scope_guard scope_exit_handler([&conn]{ - for (auto &aph: conn) - aph->finish_outer_call(); - }); - + std::vector> conn; CRITICAL_REGION_BEGIN(m_connects_lock); conn.reserve(m_connects.size()); for (auto &e: m_connects) - if (e.second->start_outer_call()) - conn.push_back(e.second); - CRITICAL_REGION_END() + conn.push_back(e.second.lock()); + CRITICAL_REGION_END(); - for (auto &aph: conn) - if (!cb(aph->get_context_ref())) + for (auto &c: conn) + if (c && !cb(c->context)) return false; return true; @@ -795,14 +766,8 @@ bool async_protocol_handler_config::foreach_connection(con template template bool async_protocol_handler_config::for_connection(const boost::uuids::uuid &connection_id, const callback_t &cb) { - async_protocol_handler* aph = nullptr; - if (find_and_lock_connection(connection_id, aph) != LEVIN_OK) - return false; - const scope_guard scope_exit_handler( - boost::bind(&async_protocol_handler::finish_outer_call, aph)); - if(!cb(aph->get_context_ref())) - return false; - return true; + const std::shared_ptr aph = find_and_lock_connection(connection_id); + return aph && cb(aph->context); } //------------------------------------------------------------------------------------------ template @@ -815,23 +780,13 @@ size_t async_protocol_handler_config::get_connections_coun template size_t async_protocol_handler_config::get_out_connections_count() { - CRITICAL_REGION_LOCAL(m_connects_lock); - size_t count = 0; - for (const auto &c: m_connects) - if (!c.second->m_connection_context.m_is_income) - ++count; - return count; + return m_outgoing_count; } //------------------------------------------------------------------------------------------ template size_t async_protocol_handler_config::get_in_connections_count() { - CRITICAL_REGION_LOCAL(m_connects_lock); - size_t count = 0; - for (const auto &c: m_connects) - if (c.second->m_connection_context.m_is_income) - ++count; - return count; + return m_incoming_count; } //------------------------------------------------------------------------------------------ template @@ -846,20 +801,15 @@ void async_protocol_handler_config::set_handler(levin_comm template int async_protocol_handler_config::send(byte_slice message, const boost::uuids::uuid& connection_id) { - async_protocol_handler* aph; - int r = find_and_lock_connection(connection_id, aph); - return LEVIN_OK == r ? aph->send(std::move(message)) : 0; + const std::shared_ptr aph = find_and_lock_connection(connection_id); + return aph ? aph->m_protocol_handler.send(std::move(message)) : 0; } //------------------------------------------------------------------------------------------ template bool async_protocol_handler_config::close(boost::uuids::uuid connection_id, const bool wait_for_shutdown) { - async_protocol_handler* aph = nullptr; - if (find_and_lock_connection(connection_id, aph) != LEVIN_OK) - return false; - const scope_guard scope_exit_handler( - boost::bind(&async_protocol_handler::finish_outer_call, aph)); - if (!aph->close(wait_for_shutdown)) + const std::shared_ptr aph = find_and_lock_connection(connection_id); + if (!aph || !aph->m_protocol_handler.close(wait_for_shutdown)) return false; CRITICAL_REGION_LOCAL(m_connects_lock); m_connects.erase(connection_id); @@ -867,24 +817,12 @@ bool async_protocol_handler_config::close(boost::uuids::uu } //------------------------------------------------------------------------------------------ template -bool async_protocol_handler_config::update_connection_context(const t_connection_context& contxt) -{ - CRITICAL_REGION_LOCAL(m_connects_lock); - async_protocol_handler* aph = find_connection(contxt.m_connection_id); - if(0 == aph) - return false; - aph->update_connection_context(contxt); - return true; -} -//------------------------------------------------------------------------------------------ -template bool async_protocol_handler_config::request_callback(boost::uuids::uuid connection_id) { - async_protocol_handler* aph; - int r = find_and_lock_connection(connection_id, aph); - if(LEVIN_OK == r) + const std::shared_ptr con = find_and_lock_connection(connection_id); + if(con) { - aph->request_callback(); + con->request_callback(); return true; } else diff --git a/contrib/epee/include/net/net_utils_base.h b/contrib/epee/include/net/net_utils_base.h index 8dfdc5e1a..69525b0cc 100644 --- a/contrib/epee/include/net/net_utils_base.h +++ b/contrib/epee/include/net/net_utils_base.h @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -460,13 +461,30 @@ namespace net_utils virtual bool call_run_once_service_io()=0; virtual bool request_callback()=0; virtual boost::asio::io_context& get_io_context()=0; - //protect from deletion connection object(with protocol instance) during external call "invoke" - virtual bool add_ref()=0; - virtual bool release()=0; protected: virtual ~i_service_endpoint() noexcept(false) {} }; + template + struct service_endpoint : i_service_endpoint + { + typedef typename t_protocol_handler::connection_context t_connection_context; + + service_endpoint(typename t_protocol_handler::config_type& config) + : i_service_endpoint(), context(), m_protocol_handler(this, config, context) + {} + + t_connection_context context; + + // TODO what do they mean about wait on destructor?? --rfree : + //this should be the last one, because it could be wait on destructor, while other activities possible on other threads + t_protocol_handler m_protocol_handler; + + protected: + virtual ~service_endpoint() noexcept(false) + {} + }; + //some helpers diff --git a/tests/fuzz/levin.cpp b/tests/fuzz/levin.cpp index e1469a090..2cf870267 100644 --- a/tests/fuzz/levin.cpp +++ b/tests/fuzz/levin.cpp @@ -26,6 +26,8 @@ // 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. +#include + #include "include_base_utils.h" #include "file_io_utils.h" #include "net/net_utils_base.h" @@ -135,19 +137,22 @@ namespace std::string m_last_in_buf; }; - class test_connection : public epee::net_utils::i_service_endpoint + class test_connection : public epee::net_utils::service_endpoint, public std::enable_shared_from_this { public: test_connection(boost::asio::io_context& io_service, test_levin_protocol_handler_config& protocol_config) - : m_io_service(io_service) - , m_protocol_handler(this, protocol_config, m_context) + : epee::net_utils::service_endpoint(protocol_config) + , std::enable_shared_from_this() + , m_io_service(io_service) , m_send_return(true) { } void start() { - m_protocol_handler.after_init_connection(); + using base_type = epee::net_utils::connection_context_base; + static_cast(context) = base_type{boost::uuids::random_generator{}(), {}, true, false}; + m_protocol_handler.m_config.after_init_connection(shared_from_this()); } // Implement epee::net_utils::i_service_endpoint interface @@ -175,10 +180,6 @@ namespace bool send_return() const { return m_send_return; } void send_return(bool v) { m_send_return = v; } - public: - test_levin_connection_context m_context; - test_levin_protocol_handler m_protocol_handler; - private: boost::asio::io_context& m_io_service; @@ -197,7 +198,7 @@ namespace const static uint64_t invoke_timeout = 5 * 1000; const static size_t max_packet_size = 10 * 1024 * 1024; - typedef std::unique_ptr test_connection_ptr; + typedef std::shared_ptr test_connection_ptr; async_protocol_handler_test(): m_pcommands_handler(new test_levin_commands_handler()), @@ -304,7 +305,7 @@ BEGIN_SIMPLE_FUZZER() test_levin_protocol_handler_config m_handler_config; test_levin_commands_handler *m_pcommands_handler = new test_levin_commands_handler(); m_handler_config.set_handler(m_pcommands_handler, [](epee::levin::levin_commands_handler *handler) { delete handler; }); - std::unique_ptr conn(new test_connection(io_service, m_handler_config)); + std::shared_ptr conn(new test_connection(io_service, m_handler_config)); conn->start(); //m_commands_handler.invoke_out_buf(expected_out_data); //m_commands_handler.return_code(expected_return_code); diff --git a/tests/unit_tests/epee_boosted_tcp_server.cpp b/tests/unit_tests/epee_boosted_tcp_server.cpp index 3cbdfa7a1..2c721b045 100644 --- a/tests/unit_tests/epee_boosted_tcp_server.cpp +++ b/tests/unit_tests/epee_boosted_tcp_server.cpp @@ -56,6 +56,11 @@ namespace struct test_protocol_handler_config { + template + static constexpr bool after_init_connection(const std::shared_ptr&) noexcept + { + return true; + } }; struct test_protocol_handler @@ -67,10 +72,6 @@ namespace { } - void after_init_connection() - { - } - void handle_qued_callback() { } @@ -171,7 +172,7 @@ TEST(test_epee_connection, test_lifetime) using handler_t = epee::levin::async_protocol_handler; using connection_t = epee::net_utils::connection; - using connection_ptr = boost::shared_ptr; + using connection_ptr = std::shared_ptr; using shared_state_t = typename connection_t::shared_state; using shared_state_ptr = std::shared_ptr; using shared_states_t = std::vector; @@ -185,7 +186,7 @@ TEST(test_epee_connection, test_lifetime) using server_t = epee::net_utils::boosted_tcp_server; using lock_t = std::mutex; using lock_guard_t = std::lock_guard; - using connection_weak_ptr = boost::weak_ptr; + using connection_weak_ptr = std::weak_ptr; struct shared_conn_t { lock_t lock; connection_weak_ptr conn; @@ -239,7 +240,7 @@ TEST(test_epee_connection, test_lifetime) auto create_connection = [&io_context, &endpoint, &shared_state] { connection_ptr conn(new connection_t(io_context, shared_state, {}, {})); conn->socket().connect(endpoint); - conn->start({}, {}); + EXPECT_TRUE(conn->start({}, {})); context_t context; conn->get_context(context); auto tag = context.m_connection_id; @@ -604,26 +605,38 @@ TEST(test_epee_connection, ssl_handshake) workers.back().join(); } - -TEST(boosted_tcp_server, strand_deadlock) +namespace { - using context_t = epee::net_utils::connection_context_base; - using lock_t = std::mutex; - using unique_lock_t = std::unique_lock; - struct config_t { using condition_t = std::condition_variable_any; - using lock_guard_t = std::lock_guard; + using lock_guard_t = std::lock_guard; void notify_success() { lock_guard_t guard(lock); success = true; condition.notify_all(); } - lock_t lock; + + template + static bool after_init_connection(const std::shared_ptr& conn) + { + if (!conn) + return false; + conn->m_protocol_handler.after_init_connection(); + return true; + } + + std::mutex lock; condition_t condition; bool success; }; +} + +TEST(boosted_tcp_server, strand_deadlock) +{ + using context_t = epee::net_utils::connection_context_base; + using lock_t = std::mutex; + using unique_lock_t = std::unique_lock; struct handler_t { using config_type = config_t; @@ -735,45 +748,66 @@ TEST(boosted_tcp_server, strand_deadlock) server.deinit_server(); } -TEST(boosted_tcp_server, shutdown) +namespace { - struct context_t: epee::net_utils::connection_context_base { + struct shutdown_handler_t; + struct shutdown_context_t: epee::net_utils::connection_context_base { static constexpr size_t get_max_bytes(int) noexcept { return -1; } static constexpr int handshake_command() noexcept { return 1001; } static constexpr bool handshake_complete() noexcept { return true; } }; +} - struct config_t : epee::levin::async_protocol_handler_config { +namespace epee { namespace levin +{ + template<> + struct get_handler { + using type = shutdown_handler_t; + }; +}} + +namespace +{ + struct shutdown_config_t : epee::levin::async_protocol_handler_config { void received_handshake() { handshake_received.raise(); } epee::simple_event handshake_received; }; - struct command_handler_t: epee::levin::levin_commands_handler { + struct shutdown_command_handler_t: epee::levin::levin_commands_handler { + using context_t = shutdown_context_t; virtual int invoke(int, const epee::span, epee::byte_stream&, context_t&) override { return {}; } virtual int notify(int, const epee::span, context_t&) override { return {}; } virtual void callback(context_t&) override {} virtual void on_connection_new(context_t&) override {} virtual void on_connection_close(context_t&) override { } - virtual ~command_handler_t() override {} + virtual ~shutdown_command_handler_t() override {} static void destroy(epee::levin::levin_commands_handler* ptr) { delete ptr; } }; - struct handler_t : epee::levin::async_protocol_handler { - using config_type = config_t; - using connection_context = context_t; - using epee::levin::async_protocol_handler::async_protocol_handler; + struct shutdown_handler_t : epee::levin::async_protocol_handler { + using config_type = shutdown_config_t; + using connection_context = shutdown_context_t; + using epee::levin::async_protocol_handler::async_protocol_handler; - bool handle_recv(const void *data, size_t bytes_transferred) + bool handle_recv(const void *data, size_t bytes_transferred) override { // We don't respond to the handshake (the async_invoke_remote_command2 is waiting for a response) MINFO("handle_recv just came in"); - config_t* config = dynamic_cast(&m_config); + config_type* config = dynamic_cast(&m_config); if (config == nullptr) throw std::runtime_error("m_config must be of type config_t"); config->received_handshake(); return true; } }; +} + + +TEST(boosted_tcp_server, shutdown) +{ + using context_t = shutdown_context_t; + using command_handler_t = shutdown_command_handler_t; + using handler_t = shutdown_handler_t; boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::make_address("127.0.0.1"), 5262); epee::net_utils::boosted_tcp_server server(epee::net_utils::e_connection_type_P2P); @@ -848,7 +882,12 @@ TEST(boosted_tcp_server, write_failure) { using context_t = epee::net_utils::connection_context_base; - struct config_t {}; + struct config_t { + static constexpr bool after_init_connection(const std::shared_ptr&) noexcept + { + return true; + } + }; struct handler_t { using config_type = config_t; @@ -858,9 +897,7 @@ TEST(boosted_tcp_server, write_failure) handler_t(socket_t *socket, config_t &config, context_t &): config(config) {} - void after_init_connection() - {} - + void handle_qued_callback() {} @@ -896,7 +933,7 @@ TEST(boosted_tcp_server, write_failure) socket_t in_socket{context}; - boost::shared_ptr out_connection; + std::shared_ptr out_connection; const auto shared = std::make_shared(); const auto make_connection = [&] { in_socket = socket_t{context}; @@ -908,7 +945,7 @@ TEST(boosted_tcp_server, write_failure) context.restart(); ASSERT_EQ(2u, context.run()); // connect and accept - out_connection = boost::make_shared( + out_connection = std::make_shared( context, std::move(out_socket), shared, diff --git a/tests/unit_tests/epee_levin_protocol_handler_async.cpp b/tests/unit_tests/epee_levin_protocol_handler_async.cpp index 5a398517a..d7146cee8 100644 --- a/tests/unit_tests/epee_levin_protocol_handler_async.cpp +++ b/tests/unit_tests/epee_levin_protocol_handler_async.cpp @@ -30,6 +30,7 @@ #include #include +#include #include "gtest/gtest.h" @@ -126,19 +127,22 @@ namespace std::string m_last_in_buf; }; - class test_connection : public epee::net_utils::i_service_endpoint + class test_connection : public epee::net_utils::service_endpoint, public std::enable_shared_from_this { public: test_connection(boost::asio::io_context& io_service, test_levin_protocol_handler_config& protocol_config) - : m_io_service(io_service) - , m_protocol_handler(this, protocol_config, m_context) + : epee::net_utils::service_endpoint(protocol_config) + , std::enable_shared_from_this() + , m_io_service(io_service) , m_send_return(true) { } void start() { - ASSERT_TRUE(m_protocol_handler.after_init_connection()); + using base_type = epee::net_utils::connection_context_base; + static_cast(context) = base_type{boost::uuids::random_generator{}(), {}, true, false}; + ASSERT_TRUE(m_protocol_handler.m_config.after_init_connection(shared_from_this())); } // Implement epee::net_utils::i_service_endpoint interface @@ -156,8 +160,6 @@ namespace virtual bool call_run_once_service_io() { std::cout << "test_connection::call_run_once_service_io()" << std::endl; return true; } virtual bool request_callback() { std::cout << "test_connection::request_callback()" << std::endl; return true; } virtual boost::asio::io_context& get_io_context() { std::cout << "test_connection::get_io_context()" << std::endl; return m_io_service; } - virtual bool add_ref() { std::cout << "test_connection::add_ref()" << std::endl; return true; } - virtual bool release() { std::cout << "test_connection::release()" << std::endl; return true; } size_t send_counter() const { return m_send_counter.get(); } @@ -167,12 +169,8 @@ namespace bool send_return() const { return m_send_return; } void send_return(bool v) { m_send_return = v; } - public: - test_levin_protocol_handler m_protocol_handler; - private: boost::asio::io_context& m_io_service; - test_levin_connection_context m_context; unit_test::call_counter m_send_counter; boost::mutex m_mutex; @@ -188,7 +186,7 @@ namespace constexpr const static std::chrono::seconds invoke_timeout{5}; const static size_t max_packet_size = 10 * 1024 * 1024; - typedef std::unique_ptr test_connection_ptr; + typedef std::shared_ptr test_connection_ptr; async_protocol_handler_test(): m_pcommands_handler(new test_levin_commands_handler()), diff --git a/tests/unit_tests/http.cpp b/tests/unit_tests/http.cpp index 4341fbb6d..298e2aa08 100644 --- a/tests/unit_tests/http.cpp +++ b/tests/unit_tests/http.cpp @@ -84,8 +84,6 @@ public: bool call_run_once_service_io() override { return true; } bool request_callback() override { return true; } boost::asio::io_context& get_io_context() override { return io_context; } - bool add_ref() override { return true; } - bool release() override { return true; } boost::asio::io_context io_context; std::string sent; diff --git a/tests/unit_tests/levin.cpp b/tests/unit_tests/levin.cpp index a0e8acca1..99c4aa9cf 100644 --- a/tests/unit_tests/levin.cpp +++ b/tests/unit_tests/levin.cpp @@ -27,6 +27,7 @@ // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #include +#include #include #include #include @@ -88,18 +89,6 @@ namespace return io_service_; } - virtual bool add_ref() override final - { - ++ref_count_; - return true; - } - - virtual bool release() override final - { - --ref_count_; - return true; - } - public: test_endpoint(boost::asio::io_context& io_service) : epee::net_utils::i_service_endpoint(), @@ -164,43 +153,78 @@ namespace } }; - class test_connection + class test_connection final : public epee::net_utils::service_endpoint> { - test_endpoint endpoint_; - cryptonote::levin::detail::p2p_context context_; - epee::levin::async_protocol_handler handler_; + boost::asio::io_context& io_service_; + std::deque send_queue_; + + virtual bool do_send(epee::byte_slice message) override final + { + send_queue_.push_back(std::move(message)); + return true; + } + + virtual bool close(bool) override final + { + return true; + } + + virtual bool send_done() override final + { + throw std::logic_error{"send_done not implemented"}; + } + + virtual bool call_run_once_service_io() override final + { + return io_service_.run_one(); + } + + virtual bool request_callback() override final + { + throw std::logic_error{"request_callback not implemented"}; + } + + virtual boost::asio::io_context& get_io_context() override final + { + return io_service_; + } public: test_connection(boost::asio::io_context& io_service, cryptonote::levin::connections& connections, boost::uuids::random_generator& random_generator, const bool is_incoming) - : endpoint_(io_service), - context_(), - handler_(std::addressof(endpoint_), connections, context_) + : epee::net_utils::service_endpoint>(connections), + io_service_(io_service), + send_queue_() { using base_type = epee::net_utils::connection_context_base; - static_cast(context_) = base_type{random_generator(), {}, is_incoming, false}; - context_.m_state = cryptonote::cryptonote_connection_context::state_normal; - handler_.after_init_connection(); + static_cast(context) = base_type{random_generator(), {}, is_incoming, false}; + context.m_state = cryptonote::cryptonote_connection_context::state_normal; } + virtual ~test_connection() noexcept override final + try {} + catch (...) + {} + //\return Number of messages processed std::size_t process_send_queue(const bool valid = true) { std::size_t count = 0; - for ( ; !endpoint_.send_queue_.empty(); ++count, endpoint_.send_queue_.pop_front()) + for ( ; !send_queue_.empty(); ++count, send_queue_.pop_front()) { - EXPECT_EQ(valid, handler_.handle_recv(endpoint_.send_queue_.front().data(), endpoint_.send_queue_.front().size())); + // invalid messages shouldn't be possible in this test + EXPECT_EQ(valid, m_protocol_handler.handle_recv(send_queue_.front().data(), send_queue_.front().size())); } return count; } const boost::uuids::uuid& get_id() const noexcept { - return context_.m_connection_id; + return context.m_connection_id; } bool is_incoming() const noexcept { - return context_.m_is_income; + return context.m_is_income; } }; @@ -346,8 +370,9 @@ namespace void add_connection(const bool is_incoming) { - contexts_.emplace_back(io_service_, *connections_, random_generator_, is_incoming); - EXPECT_TRUE(connection_ids_.emplace(contexts_.back().get_id()).second); + contexts_.emplace_back(std::make_shared(io_service_, *connections_, random_generator_, is_incoming)); + connections_->after_init_connection(contexts_.back()); + EXPECT_TRUE(connection_ids_.emplace(contexts_.back()->get_id()).second); EXPECT_EQ(connection_ids_.size(), connections_->get_connections_count()); } @@ -366,7 +391,7 @@ namespace boost::uuids::random_generator random_generator_; boost::asio::io_context io_service_; test_receiver receiver_; - std::deque contexts_; + std::deque> contexts_; test_core_events events_; }; } @@ -623,7 +648,7 @@ TEST_F(levin_notify, fluff_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::fluff)); io_service_.restart(); @@ -632,7 +657,7 @@ TEST_F(levin_notify, fluff_without_padding) ASSERT_LT(0u, io_service_.poll()); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) EXPECT_EQ(1u, context->process_send_queue()); EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::fluff)); @@ -677,7 +702,7 @@ TEST_F(levin_notify, stem_without_padding) bool has_fluffed = false; while (!has_stemmed || !has_fluffed) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -693,12 +718,12 @@ TEST_F(levin_notify, stem_without_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); } send_count += sent; } @@ -748,7 +773,7 @@ TEST_F(levin_notify, stem_no_outs_without_padding) ASSERT_EQ(10u, contexts_.size()); - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -764,7 +789,7 @@ TEST_F(levin_notify, stem_no_outs_without_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { send_count += context->process_send_queue(); } @@ -817,7 +842,7 @@ TEST_F(levin_notify, local_without_padding) while (!has_stemmed || !has_fluffed) { // run their "their" txes first - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(their_txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -833,12 +858,12 @@ TEST_F(levin_notify, local_without_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); } send_count += sent; } @@ -867,12 +892,12 @@ TEST_F(levin_notify, local_without_padding) send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); } send_count += sent; } @@ -919,7 +944,7 @@ TEST_F(levin_notify, forward_without_padding) bool has_fluffed = false; while (!has_stemmed || !has_fluffed) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::forward)); io_service_.restart(); @@ -935,12 +960,12 @@ TEST_F(levin_notify, forward_without_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); } send_count += sent; } @@ -987,7 +1012,7 @@ TEST_F(levin_notify, block_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::block)); io_service_.restart(); @@ -1018,7 +1043,7 @@ TEST_F(levin_notify, none_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::none)); io_service_.restart(); @@ -1049,7 +1074,7 @@ TEST_F(levin_notify, fluff_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::fluff)); io_service_.restart(); @@ -1060,7 +1085,7 @@ TEST_F(levin_notify, fluff_with_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::fluff)); std::sort(txs.begin(), txs.end()); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) EXPECT_EQ(1u, context->process_send_queue()); ASSERT_EQ(9u, receiver_.notified_size()); @@ -1100,7 +1125,7 @@ TEST_F(levin_notify, stem_with_padding) bool has_fluffed = false; while (!has_stemmed || !has_fluffed) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -1116,12 +1141,12 @@ TEST_F(levin_notify, stem_with_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); } send_count += sent; @@ -1169,7 +1194,7 @@ TEST_F(levin_notify, stem_no_outs_with_padding) ASSERT_EQ(10u, contexts_.size()); - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -1185,7 +1210,7 @@ TEST_F(levin_notify, stem_no_outs_with_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { send_count += context->process_send_queue(); } @@ -1232,7 +1257,7 @@ TEST_F(levin_notify, local_with_padding) while (!has_stemmed || !has_fluffed) { // run their "their" txes first - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(their_txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -1248,12 +1273,12 @@ TEST_F(levin_notify, local_with_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); } send_count += sent; @@ -1280,12 +1305,12 @@ TEST_F(levin_notify, local_with_padding) send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); } send_count += sent; } @@ -1329,7 +1354,7 @@ TEST_F(levin_notify, forward_with_padding) bool has_fluffed = false; while (!has_stemmed || !has_fluffed) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::forward)); io_service_.restart(); @@ -1345,12 +1370,12 @@ TEST_F(levin_notify, forward_with_padding) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent && is_stem) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); } send_count += sent; @@ -1395,7 +1420,7 @@ TEST_F(levin_notify, block_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::block)); io_service_.restart(); @@ -1426,7 +1451,7 @@ TEST_F(levin_notify, none_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::none)); io_service_.restart(); @@ -1457,7 +1482,7 @@ TEST_F(levin_notify, private_fluff_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::fluff)); io_service_.restart(); @@ -1469,9 +1494,9 @@ TEST_F(levin_notify, private_fluff_without_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::fluff)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1510,7 +1535,7 @@ TEST_F(levin_notify, private_stem_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -1522,9 +1547,9 @@ TEST_F(levin_notify, private_stem_without_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::stem)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1563,7 +1588,7 @@ TEST_F(levin_notify, private_local_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::local)); io_service_.restart(); @@ -1575,9 +1600,9 @@ TEST_F(levin_notify, private_local_without_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::local)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1616,7 +1641,7 @@ TEST_F(levin_notify, private_forward_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::forward)); io_service_.restart(); @@ -1628,9 +1653,9 @@ TEST_F(levin_notify, private_forward_without_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::forward)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1669,7 +1694,7 @@ TEST_F(levin_notify, private_block_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::block)); io_service_.restart(); @@ -1701,7 +1726,7 @@ TEST_F(levin_notify, private_none_without_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::none)); io_service_.restart(); @@ -1732,7 +1757,7 @@ TEST_F(levin_notify, private_fluff_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::fluff)); io_service_.restart(); @@ -1744,9 +1769,9 @@ TEST_F(levin_notify, private_fluff_with_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::fluff)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1784,7 +1809,7 @@ TEST_F(levin_notify, private_stem_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -1796,9 +1821,9 @@ TEST_F(levin_notify, private_stem_with_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::stem)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1836,7 +1861,7 @@ TEST_F(levin_notify, private_local_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::local)); io_service_.restart(); @@ -1848,9 +1873,9 @@ TEST_F(levin_notify, private_local_with_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::local)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1888,7 +1913,7 @@ TEST_F(levin_notify, private_forward_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::forward)); io_service_.restart(); @@ -1900,9 +1925,9 @@ TEST_F(levin_notify, private_forward_with_padding) EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::forward)); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { - const bool is_incoming = ((context - contexts_.begin()) % 2 == 0); + const bool is_incoming = ((context.base() - contexts_.begin()) % 2 == 0); EXPECT_EQ(is_incoming ? 0u : 1u, context->process_send_queue()); } @@ -1940,7 +1965,7 @@ TEST_F(levin_notify, private_block_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::block)); io_service_.restart(); @@ -1971,7 +1996,7 @@ TEST_F(levin_notify, private_none_with_padding) ASSERT_EQ(10u, contexts_.size()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_FALSE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::none)); io_service_.restart(); @@ -2005,7 +2030,7 @@ TEST_F(levin_notify, stem_mappings) ASSERT_EQ(test_connections_count, contexts_.size()); for (;;) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -2019,7 +2044,7 @@ TEST_F(levin_notify, stem_mappings) ASSERT_LT(0u, io_service_.poll()); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) EXPECT_EQ(1u, context->process_send_queue()); ASSERT_EQ(test_connections_count - 1, receiver_.notified_size()); @@ -2041,15 +2066,15 @@ TEST_F(levin_notify, stem_mappings) std::map mappings; { std::size_t send_count = 0; - for (auto context = contexts_.begin(); context != contexts_.end(); ++context) + for (auto context = boost::make_indirect_iterator(contexts_.begin()); context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); used.insert(context->get_id()); - mappings[contexts_.front().get_id()] = context->get_id(); + mappings[contexts_.front()->get_id()] = context->get_id(); } send_count += sent; } @@ -2068,23 +2093,23 @@ TEST_F(levin_notify, stem_mappings) for (unsigned i = 0; i < contexts_.size() * 2; i += 2) { auto& incoming = contexts_[i % contexts_.size()]; - EXPECT_TRUE(notifier.send_txs(txs, incoming.get_id(), cryptonote::relay_method::stem)); + EXPECT_TRUE(notifier.send_txs(txs, incoming->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); ASSERT_LT(0u, io_service_.poll()); EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::stem)); std::size_t send_count = 0; - for (auto context = contexts_.begin(); context != contexts_.end(); ++context) + for (auto context = boost::make_indirect_iterator(contexts_.begin()); context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); used.insert(context->get_id()); - auto inserted = mappings.emplace(incoming.get_id(), context->get_id()).first; + auto inserted = mappings.emplace(incoming->get_id(), context->get_id()).first; EXPECT_EQ(inserted->second, context->get_id()) << "incoming index " << i; } send_count += sent; @@ -2130,7 +2155,7 @@ TEST_F(levin_notify, fluff_multiple) ASSERT_EQ(test_connections_count, contexts_.size()); for (;;) { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); @@ -2142,12 +2167,12 @@ TEST_F(levin_notify, fluff_multiple) std::size_t send_count = 0; EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) { const std::size_t sent = context->process_send_queue(); if (sent) { - EXPECT_EQ(1u, (context - contexts_.begin()) % 2); + EXPECT_EQ(1u, (context.base() - contexts_.begin()) % 2); EXPECT_FALSE(context->is_incoming()); } send_count += sent; @@ -2172,9 +2197,9 @@ TEST_F(levin_notify, fluff_multiple) io_service_.restart(); ASSERT_LT(0u, io_service_.poll()); { - auto context = contexts_.begin(); + auto context = boost::make_indirect_iterator(contexts_.begin()); EXPECT_EQ(0u, context->process_send_queue()); - for (++context; context != contexts_.end(); ++context) + for (++context; context.base() != contexts_.end(); ++context) EXPECT_EQ(1u, context->process_send_queue()); ASSERT_EQ(contexts_.size() - 1, receiver_.notified_size()); @@ -2190,7 +2215,7 @@ TEST_F(levin_notify, fluff_multiple) for (unsigned i = 0; i < contexts_.size() * 2; i += 2) { auto& incoming = contexts_[i % contexts_.size()]; - EXPECT_TRUE(notifier.send_txs(txs, incoming.get_id(), cryptonote::relay_method::stem)); + EXPECT_TRUE(notifier.send_txs(txs, incoming->get_id(), cryptonote::relay_method::stem)); io_service_.restart(); ASSERT_LT(0u, io_service_.poll()); @@ -2203,9 +2228,9 @@ TEST_F(levin_notify, fluff_multiple) for (auto& context : contexts_) { if (std::addressof(incoming) == std::addressof(context)) - EXPECT_EQ(0u, context.process_send_queue()); + EXPECT_EQ(0u, context->process_send_queue()); else - EXPECT_EQ(1u, context.process_send_queue()); + EXPECT_EQ(1u, context->process_send_queue()); } ASSERT_EQ(contexts_.size() - 1, receiver_.notified_size()); @@ -2250,16 +2275,16 @@ TEST_F(levin_notify, fluff_with_duplicate) ASSERT_EQ(10u, contexts_.size()); { auto context = contexts_.begin(); - EXPECT_TRUE(notifier.send_txs(txs, context->get_id(), cryptonote::relay_method::fluff)); + EXPECT_TRUE(notifier.send_txs(txs, (*context)->get_id(), cryptonote::relay_method::fluff)); io_service_.restart(); ASSERT_LT(0u, io_service_.poll()); notifier.run_fluff(); ASSERT_LT(0u, io_service_.poll()); - EXPECT_EQ(0u, context->process_send_queue()); + EXPECT_EQ(0u, (*context)->process_send_queue()); for (++context; context != contexts_.end(); ++context) - EXPECT_EQ(1u, context->process_send_queue()); + EXPECT_EQ(1u, (*context)->process_send_queue()); EXPECT_EQ(txs, events_.take_relayed(cryptonote::relay_method::fluff)); std::sort(txs.begin(), txs.end()); @@ -2308,7 +2333,7 @@ TEST_F(levin_notify, noise) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); EXPECT_EQ(2u, sent); EXPECT_EQ(0u, receiver_.notified_size()); @@ -2323,7 +2348,7 @@ TEST_F(levin_notify, noise) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); ASSERT_EQ(2u, sent); while (sent--) @@ -2345,7 +2370,7 @@ TEST_F(levin_notify, noise) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); EXPECT_EQ(2u, sent); EXPECT_EQ(0u, receiver_.notified_size()); @@ -2357,7 +2382,7 @@ TEST_F(levin_notify, noise) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); ASSERT_EQ(2u, sent); while (sent--) @@ -2402,7 +2427,7 @@ TEST_F(levin_notify, noise_stem) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); EXPECT_EQ(2u, sent); EXPECT_EQ(0u, receiver_.notified_size()); @@ -2418,7 +2443,7 @@ TEST_F(levin_notify, noise_stem) { std::size_t sent = 0; for (auto& context : contexts_) - sent += context.process_send_queue(); + sent += context->process_send_queue(); ASSERT_EQ(2u, sent); while (sent--) @@ -2445,13 +2470,13 @@ TEST_F(levin_notify, command_max_bytes) bytes = dest.finalize_notify(ping_command); } - EXPECT_EQ(1, get_connections().send(bytes.clone(), contexts_.front().get_id())); - EXPECT_EQ(1u, contexts_.front().process_send_queue(true)); + EXPECT_EQ(1, get_connections().send(bytes.clone(), contexts_.front()->get_id())); + EXPECT_EQ(1u, contexts_.front()->process_send_queue(true)); EXPECT_EQ(1u, receiver_.notified_size()); const received_message msg = receiver_.get_raw_notification(); EXPECT_EQ(ping_command, msg.command); - EXPECT_EQ(contexts_.front().get_id(), msg.connection); + EXPECT_EQ(contexts_.front()->get_id(), msg.connection); EXPECT_EQ(payload, msg.payload); { @@ -2461,7 +2486,7 @@ TEST_F(levin_notify, command_max_bytes) bytes = dest.finalize_notify(ping_command); } - EXPECT_EQ(1, get_connections().send(std::move(bytes), contexts_.front().get_id())); - EXPECT_EQ(1u, contexts_.front().process_send_queue(false)); + EXPECT_EQ(1, get_connections().send(std::move(bytes), contexts_.front()->get_id())); + EXPECT_EQ(1u, contexts_.front()->process_send_queue(false)); EXPECT_EQ(0u, receiver_.notified_size()); } diff --git a/tests/unit_tests/node_server.cpp b/tests/unit_tests/node_server.cpp index 6ef280d90..a464b837d 100644 --- a/tests/unit_tests/node_server.cpp +++ b/tests/unit_tests/node_server.cpp @@ -750,7 +750,7 @@ TEST(cryptonote_protocol_handler, race_condition) using context_t = contexts::p2p; using handler_t = epee::levin::async_protocol_handler; using connection_t = epee::net_utils::connection; - using connection_ptr = boost::shared_ptr; + using connection_ptr = std::shared_ptr; using connections_t = std::vector; using shared_state_t = typename connection_t::shared_state; using shared_state_ptr = std::shared_ptr; @@ -1448,7 +1448,7 @@ TEST(node_server, race_condition) }; using handler_t = epee::levin::async_protocol_handler; using connection_t = epee::net_utils::connection; - using connection_ptr = boost::shared_ptr; + using connection_ptr = std::shared_ptr; using shared_state_t = typename connection_t::shared_state; using shared_state_ptr = std::shared_ptr; using io_context_t = boost::asio::io_context;