github-wownero/wownero
1
0
Fork 0
mirror of https://github.com/wownero/wownero.git synced 2026-10-17 15:41:44 +02:00
Code Issues Packages Projects Releases Wiki Activity

Change p2p connection map from raw pointers to weak_ptrs

Browse Source
This commit is contained in:
Lee Clagett
2026-08-05 15:00:34 -04:00
committed by Lee *!* Clagett
parent 1bf1dff4be
commit b8028e31ad
11 changed files with 460 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/epee/include/net/abstract_tcp_server2.h
+12 -10
View File
@@ -92,10 +92,10 @@ namespace net_utils
/// Represents a single connection from a client.
template<class t_protocol_handler>
class connection
: public boost::enable_shared_from_this<connection<t_protocol_handler> >,
private boost::noncopyable,
public i_service_endpoint,
public connection_basic
: public std::enable_shared_from_this<connection<t_protocol_handler>>,
private boost::noncopyable,
public connection_basic, // shared_state shared_ptr must be destroyed after service_endpoint
public service_endpoint<t_protocol_handler>
{
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<t_protocol_handler> > connection_ptr;
typedef std::shared_ptr<connection<t_protocol_handler>> 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.
contrib/epee/include/net/abstract_tcp_server2.inl
+22 -44
View File
@@ -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<char *>(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<shared_state&>(connection_basic::get_state()).after_init_connection(connection<T>::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<T>(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<typename T>
@@ -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<std::mutex> guard(m_state.lock);
--m_state.protocol.wait_callback;
@@ -1235,28 +1235,6 @@ namespace net_utils
return m_io_context;
}
template<typename T>
bool connection<T>::add_ref()
{
auto self = connection<T>::weak_from_this().lock();
if (!self)
return false;
std::lock_guard<std::mutex> guard(m_state.lock);
this->self = std::move(self);
++m_state.protocol.reference_counter;
return true;
}
template<typename T>
bool connection<T>::release()
{
connection_ptr self;
std::lock_guard<std::mutex> guard(m_state.lock);
if (!(--m_state.protocol.reference_counter))
self = std::move(this->self);
return true;
}
template<typename T>
void connection<T>::setRpcStation()
{
contrib/epee/include/net/http_protocol_handler.h
+10
View File
@@ -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<typename T>
static bool after_init_connection(const std::shared_ptr<T>& 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);
contrib/epee/include/net/levin_protocol_handler_async.h
+157 -219
View File
@@ -28,7 +28,6 @@
#include <boost/asio/steady_timer.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/unordered_map.hpp>
#include <boost/smart_ptr/make_shared.hpp>
#include <atomic>
@@ -75,18 +74,26 @@ namespace levin
template<class t_connection_context>
class async_protocol_handler;
template<typename T>
struct get_handler
{
using type = async_protocol_handler<T>;
};
template<class t_connection_context>
class async_protocol_handler_config
{
typedef boost::unordered_map<boost::uuids::uuid, async_protocol_handler<t_connection_context>* > connections_map;
using derived_handler = typename get_handler<t_connection_context>::type;
typedef net_utils::service_endpoint<derived_handler> levin_endpoint;
typedef boost::unordered_map<boost::uuids::uuid, std::weak_ptr<levin_endpoint>> connections_map;
critical_section m_connects_lock;
connections_map m_connects;
std::atomic<std::size_t> m_incoming_count;
std::atomic<std::size_t> m_outgoing_count;
void add_connection(async_protocol_handler<t_connection_context>* pc);
void del_connection(async_protocol_handler<t_connection_context>* pc);
async_protocol_handler<t_connection_context>* find_connection(boost::uuids::uuid connection_id) const;
int find_and_lock_connection(boost::uuids::uuid connection_id, async_protocol_handler<t_connection_context>*& aph);
std::shared_ptr<levin_endpoint> find_and_lock_connection(const boost::uuids::uuid& connection_id);
friend class async_protocol_handler<t_connection_context>;
@@ -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<class callback_t>
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<t_connection_context>* handler, void (*destroy)(levin_commands_handler<t_connection_context>*) = NULL);
bool after_init_connection(const std::shared_ptr<levin_endpoint>& 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<t_connection_context>::type;
typedef t_connection_context connection_context;
typedef async_protocol_handler_config<t_connection_context> 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<const uint8_t> 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 <class callback_t>
struct anvoke_handler: invoke_response_handler_base
class anvoke_handler final : public invoke_response_handler_base, public std::enable_shared_from_this<anvoke_handler<callback_t>>
{
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<const uint8_t> 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<net_utils::service_endpoint<derived_handler>> 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<const uint8_t> buff, typename async_protocol_handler::connection_context& context)
void failure(const int rc)
{
std::shared_ptr<net_utils::service_endpoint<derived_handler>> 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<net_utils::service_endpoint<derived_handler>> con, int command)
: invoke_response_handler_base(),
std::enable_shared_from_this<anvoke_handler<callback_t>>(),
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<const uint8_t> 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<net_utils::service_endpoint<derived_handler>> 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<const uint8_t> 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<anvoke_handler> 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<const uint8_t> 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<boost::shared_ptr<invoke_response_handler_base> > m_invoke_response_handlers;
std::list<std::weak_ptr<invoke_response_handler_base>> m_invoke_response_handlers;
template<class callback_t>
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<net_utils::service_endpoint<derived_handler>> 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<invoke_response_handler_base> handler(boost::make_shared<anvoke_handler<callback_t>>(cb, timeout, con, command));
m_invoke_response_handlers.push_back(handler);
return handler->is_timer_started();
std::shared_ptr<invoke_response_handler_base> handler(std::make_shared<anvoke_handler<callback_t>>(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<class callback_t> friend struct anvoke_handler;
template<class callback_t> 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<invoke_response_handler_base>& 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<invoke_response_handler_base> response_handler = m_invoke_response_handlers.front();
response_handler->reset_timer();
const std::shared_ptr<invoke_response_handler_base> 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<invoke_response_handler_base> 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<invoke_response_handler_base> 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<class callback_t>
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<net_utils::service_endpoint<derived_handler>> 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<t_connection_context>::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<class t_connection_context>
void async_protocol_handler_config<t_connection_context>::delete_connections(size_t count, bool incoming)
{
std::vector<typename connections_map::mapped_type> connections;
const scope_guard scope_exit_handler([&connections]{
for (auto &aph: connections)
aph->finish_outer_call();
});
std::vector<std::shared_ptr<levin_endpoint>> 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<t_connection_context>::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<t_connection_context>::del_in_connections(siz
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
void async_protocol_handler_config<t_connection_context>::add_connection(async_protocol_handler<t_connection_context>* pconn)
bool async_protocol_handler_config<t_connection_context>::after_init_connection(const std::shared_ptr<levin_endpoint>& 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<class t_connection_context>
async_protocol_handler<t_connection_context>* async_protocol_handler_config<t_connection_context>::find_connection(boost::uuids::uuid connection_id) const
{
auto it = m_connects.find(connection_id);
return it == m_connects.end() ? 0 : it->second;
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
int async_protocol_handler_config<t_connection_context>::find_and_lock_connection(boost::uuids::uuid connection_id, async_protocol_handler<t_connection_context>*& aph)
std::shared_ptr<net_utils::service_endpoint<typename get_handler<t_connection_context>::type>> async_protocol_handler_config<t_connection_context>::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<class t_connection_context> template<class callback_t>
int async_protocol_handler_config<t_connection_context>::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<t_connection_context>* 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<levin_endpoint> 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<class t_connection_context> template<class callback_t>
bool async_protocol_handler_config<t_connection_context>::foreach_connection(const callback_t &cb)
{
std::vector<typename connections_map::mapped_type> conn;
const scope_guard scope_exit_handler([&conn]{
for (auto &aph: conn)
aph->finish_outer_call();
});
std::vector<std::shared_ptr<levin_endpoint>> 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<t_connection_context>::foreach_connection(con
template<class t_connection_context> template<class callback_t>
bool async_protocol_handler_config<t_connection_context>::for_connection(const boost::uuids::uuid &connection_id, const callback_t &cb)
{
async_protocol_handler<t_connection_context>* 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<t_connection_context>::finish_outer_call, aph));
if(!cb(aph->get_context_ref()))
return false;
return true;
const std::shared_ptr<levin_endpoint> aph = find_and_lock_connection(connection_id);
return aph && cb(aph->context);
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
@@ -815,23 +780,13 @@ size_t async_protocol_handler_config<t_connection_context>::get_connections_coun
template<class t_connection_context>
size_t async_protocol_handler_config<t_connection_context>::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<class t_connection_context>
size_t async_protocol_handler_config<t_connection_context>::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<class t_connection_context>
@@ -846,20 +801,15 @@ void async_protocol_handler_config<t_connection_context>::set_handler(levin_comm
template<class t_connection_context>
int async_protocol_handler_config<t_connection_context>::send(byte_slice message, const boost::uuids::uuid& connection_id)
{
async_protocol_handler<t_connection_context>* aph;
int r = find_and_lock_connection(connection_id, aph);
return LEVIN_OK == r ? aph->send(std::move(message)) : 0;
const std::shared_ptr<levin_endpoint> aph = find_and_lock_connection(connection_id);
return aph ? aph->m_protocol_handler.send(std::move(message)) : 0;
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
bool async_protocol_handler_config<t_connection_context>::close(boost::uuids::uuid connection_id, const bool wait_for_shutdown)
{
async_protocol_handler<t_connection_context>* 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<t_connection_context>::finish_outer_call, aph));
if (!aph->close(wait_for_shutdown))
const std::shared_ptr<levin_endpoint> 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<t_connection_context>::close(boost::uuids::uu
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
bool async_protocol_handler_config<t_connection_context>::update_connection_context(const t_connection_context& contxt)
{
CRITICAL_REGION_LOCAL(m_connects_lock);
async_protocol_handler<t_connection_context>* aph = find_connection(contxt.m_connection_id);
if(0 == aph)
return false;
aph->update_connection_context(contxt);
return true;
}
//------------------------------------------------------------------------------------------
template<class t_connection_context>
bool async_protocol_handler_config<t_connection_context>::request_callback(boost::uuids::uuid connection_id)
{
async_protocol_handler<t_connection_context>* aph;
int r = find_and_lock_connection(connection_id, aph);
if(LEVIN_OK == r)
const std::shared_ptr<levin_endpoint> con = find_and_lock_connection(connection_id);
if(con)
{
aph->request_callback();
con->request_callback();
return true;
}
else
contrib/epee/include/net/net_utils_base.h
+21 -3
View File
@@ -34,6 +34,7 @@
#include <boost/asio/ip/address_v6.hpp>
#include <boost/optional/optional.hpp>
#include <cstddef>
#include <memory>
#include <stdexcept>
#include <typeinfo>
#include <type_traits>
@@ -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<typename t_protocol_handler>
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
tests/fuzz/levin.cpp
+11 -10
View File
@@ -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 <boost/uuid/random_generator.hpp>
#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<test_levin_protocol_handler>, public std::enable_shared_from_this<test_connection>
{
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<test_levin_protocol_handler>(protocol_config)
, std::enable_shared_from_this<test_connection>()
, 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<base_type&>(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> test_connection_ptr;
typedef std::shared_ptr<test_connection> 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<test_levin_connection_context> *handler) { delete handler; });
std::unique_ptr<test_connection> conn(new test_connection(io_service, m_handler_config));
std::shared_ptr<test_connection> 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);
tests/unit_tests/epee_boosted_tcp_server.cpp
+69 -32
View File
@@ -56,6 +56,11 @@ namespace
struct test_protocol_handler_config
{
template<typename T>
static constexpr bool after_init_connection(const std::shared_ptr<T>&) 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<context_t>;
using connection_t = epee::net_utils::connection<handler_t>;
using connection_ptr = boost::shared_ptr<connection_t>;
using connection_ptr = std::shared_ptr<connection_t>;
using shared_state_t = typename connection_t::shared_state;
using shared_state_ptr = std::shared_ptr<shared_state_t>;
using shared_states_t = std::vector<shared_state_ptr>;
@@ -185,7 +186,7 @@ TEST(test_epee_connection, test_lifetime)
using server_t = epee::net_utils::boosted_tcp_server<handler_t>;
using lock_t = std::mutex;
using lock_guard_t = std::lock_guard<lock_t>;
using connection_weak_ptr = boost::weak_ptr<connection_t>;
using connection_weak_ptr = std::weak_ptr<connection_t>;
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<lock_t>;
struct config_t {
using condition_t = std::condition_variable_any;
using lock_guard_t = std::lock_guard<lock_t>;
using lock_guard_t = std::lock_guard<std::mutex>;
void notify_success()
{
lock_guard_t guard(lock);
success = true;
condition.notify_all();
}
lock_t lock;
template<typename T>
static bool after_init_connection(const std::shared_ptr<T>& 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<lock_t>;
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<context_t> {
namespace epee { namespace levin
{
template<>
struct get_handler<shutdown_context_t> {
using type = shutdown_handler_t;
};
}}
namespace
{
struct shutdown_config_t : epee::levin::async_protocol_handler_config<shutdown_context_t> {
void received_handshake() { handshake_received.raise(); }
epee::simple_event handshake_received;
};
struct command_handler_t: epee::levin::levin_commands_handler<context_t> {
struct shutdown_command_handler_t: epee::levin::levin_commands_handler<shutdown_context_t> {
using context_t = shutdown_context_t;
virtual int invoke(int, const epee::span<const uint8_t>, epee::byte_stream&, context_t&) override { return {}; }
virtual int notify(int, const epee::span<const uint8_t>, 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<context_t>* ptr) { delete ptr; }
};
struct handler_t : epee::levin::async_protocol_handler<context_t> {
using config_type = config_t;
using connection_context = context_t;
using epee::levin::async_protocol_handler<context_t>::async_protocol_handler;
struct shutdown_handler_t : epee::levin::async_protocol_handler<shutdown_context_t> {
using config_type = shutdown_config_t;
using connection_context = shutdown_context_t;
using epee::levin::async_protocol_handler<connection_context>::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<config_t*>(&m_config);
config_type* config = dynamic_cast<config_type*>(&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<handler_t> 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<epee::net_utils::connection_basic>&) 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<connection_t> out_connection;
std::shared_ptr<connection_t> out_connection;
const auto shared = std::make_shared<shared_t>();
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<connection_t>(
out_connection = std::make_shared<connection_t>(
context,
std::move(out_socket),
shared,
tests/unit_tests/epee_levin_protocol_handler_async.cpp
+9 -11
View File
@@ -30,6 +30,7 @@
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <boost/uuid/random_generator.hpp>
#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<test_levin_protocol_handler>, public std::enable_shared_from_this<test_connection>
{
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<test_levin_protocol_handler>(protocol_config)
, std::enable_shared_from_this<test_connection>()
, 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<base_type&>(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> test_connection_ptr;
typedef std::shared_ptr<test_connection> test_connection_ptr;
async_protocol_handler_test():
m_pcommands_handler(new test_levin_commands_handler()),
tests/unit_tests/http.cpp
-2
View File
@@ -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;
tests/unit_tests/levin.cpp
+147 -122
View File
@@ -27,6 +27,7 @@
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <algorithm>
#include <boost/iterator/indirect_iterator.hpp>
#include <boost/uuid/nil_generator.hpp>
#include <boost/uuid/random_generator.hpp>
#include <boost/uuid/uuid.hpp>
@@ -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<epee::levin::async_protocol_handler<cryptonote::levin::detail::p2p_context>>
{
test_endpoint endpoint_;
cryptonote::levin::detail::p2p_context context_;
epee::levin::async_protocol_handler<cryptonote::levin::detail::p2p_context> handler_;
boost::asio::io_context& io_service_;
std::deque<epee::byte_slice> 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<epee::levin::async_protocol_handler<cryptonote::levin::detail::p2p_context>>(connections),
io_service_(io_service),
send_queue_()
{
using base_type = epee::net_utils::connection_context_base;
static_cast<base_type&>(context_) = base_type{random_generator(), {}, is_incoming, false};
context_.m_state = cryptonote::cryptonote_connection_context::state_normal;
handler_.after_init_connection();
static_cast<base_type&>(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<test_connection>(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<test_connection> contexts_;
std::deque<std::shared_ptr<test_connection>> 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<boost::uuids::uuid, boost::uuids::uuid> 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());
}
tests/unit_tests/node_server.cpp
+2 -2
View File
@@ -750,7 +750,7 @@ TEST(cryptonote_protocol_handler, race_condition)
using context_t = contexts::p2p;
using handler_t = epee::levin::async_protocol_handler<context_t>;
using connection_t = epee::net_utils::connection<handler_t>;
using connection_ptr = boost::shared_ptr<connection_t>;
using connection_ptr = std::shared_ptr<connection_t>;
using connections_t = std::vector<connection_ptr>;
using shared_state_t = typename connection_t::shared_state;
using shared_state_ptr = std::shared_ptr<shared_state_t>;
@@ -1448,7 +1448,7 @@ TEST(node_server, race_condition)
};
using handler_t = epee::levin::async_protocol_handler<context_t>;
using connection_t = epee::net_utils::connection<handler_t>;
using connection_ptr = boost::shared_ptr<connection_t>;
using connection_ptr = std::shared_ptr<connection_t>;
using shared_state_t = typename connection_t::shared_state;
using shared_state_ptr = std::shared_ptr<shared_state_t>;
using io_context_t = boost::asio::io_context;