Minecon724/dncurrency
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions 1

Moving socket type entirely in to nano::socket class. This removes socket type tracking as parameters and on unrelated classes such as bootstrap_server. (#3359)

Browse source
This commit is contained in:
clemahieu 2021-07-01 18:39:40 +01:00 • committed by GitHub
commit def99da543
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 33 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

32
nano/node/bootstrap/bootstrap_server.cpp
View file

@ -112,11 +112,11 @@ nano::bootstrap_server::~bootstrap_server ()
{
node->logger.try_log ("Exiting incoming TCP/bootstrap server");
}
if (type == nano::socket::type_t::bootstrap)
if (socket->type () == nano::socket::type_t::bootstrap)
{
--node->bootstrap.bootstrap_count;
}
else if (type == nano::socket::type_t::realtime)
else if (socket->type () == nano::socket::type_t::realtime)
{
--node->bootstrap.realtime_count;
// Clear temporary channel
@ -525,7 +525,7 @@ void nano::bootstrap_server::receive_node_id_handshake_action (boost::system::er
auto request (std::make_unique<nano::node_id_handshake> (error, stream, header_a));
if (!error)
{
if (type == nano::socket::type_t::undefined && !node->flags.disable_tcp_realtime)
if (socket->type () == nano::socket::type_t::undefined && !node->flags.disable_tcp_realtime)
{
add_request (std::unique_ptr<nano::message> (request.release ()));
}
@ -616,19 +616,19 @@ public:
}
void keepalive (nano::keepalive const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::keepalive> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::keepalive> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void publish (nano::publish const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::publish> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::publish> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void confirm_req (nano::confirm_req const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::confirm_req> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::confirm_req> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void confirm_ack (nano::confirm_ack const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::confirm_ack> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::confirm_ack> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void bulk_pull (nano::bulk_pull const &) override
{
@ -652,11 +652,11 @@ public:
}
void telemetry_req (nano::telemetry_req const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::telemetry_req> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::telemetry_req> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void telemetry_ack (nano::telemetry_ack const & message_a) override
{
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::telemetry_ack> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::telemetry_ack> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
void node_id_handshake (nano::node_id_handshake const & message_a) override
{
@ -697,7 +697,7 @@ public:
if (!connection->node->network.syn_cookies.validate (nano::transport::map_tcp_to_endpoint (connection->remote_endpoint), node_id, message_a.response->second) && node_id != connection->node->node_id.pub)
{
connection->remote_node_id = node_id;
connection->type = nano::socket::type_t::realtime;
connection->socket->type_set (nano::socket::type_t::realtime);
++connection->node->bootstrap.realtime_count;
connection->finish_request_async ();
}
@ -712,9 +712,9 @@ public:
connection->finish_request_async ();
}
nano::account node_id (connection->remote_node_id);
nano::socket::type_t type (connection->type);
nano::socket::type_t type = connection->socket->type ();
debug_assert (node_id.is_zero () || type == nano::socket::type_t::realtime);
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::node_id_handshake> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket, connection->type });
connection->node->network.tcp_message_manager.put_message (nano::tcp_message_item{ std::make_shared<nano::node_id_handshake> (message_a), connection->remote_endpoint, connection->remote_node_id, connection->socket });
}
std::shared_ptr<nano::bootstrap_server> connection;
};
@ -754,15 +754,15 @@ void nano::bootstrap_server::run_next (nano::unique_lock<nano::mutex> & lock_a)
bool nano::bootstrap_server::is_bootstrap_connection ()
{
if (type == nano::socket::type_t::undefined && !node->flags.disable_bootstrap_listener && node->bootstrap.bootstrap_count < node->config.bootstrap_connections_max)
if (socket->type () == nano::socket::type_t::undefined && !node->flags.disable_bootstrap_listener && node->bootstrap.bootstrap_count < node->config.bootstrap_connections_max)
{
++node->bootstrap.bootstrap_count;
type = nano::socket::type_t::bootstrap;
socket->type_set (nano::socket::type_t::bootstrap);
}
return type == nano::socket::type_t::bootstrap;
return socket->type () == nano::socket::type_t::bootstrap;
}
bool nano::bootstrap_server::is_realtime_connection ()
{
return type == nano::socket::type_t::realtime || type == nano::socket::type_t::realtime_response_server;
return socket->type () == nano::socket::type_t::realtime || socket->type () == nano::socket::type_t::realtime_response_server;
}

1
nano/node/bootstrap/bootstrap_server.hpp
View file

@ -64,7 +64,6 @@ public:
nano::mutex mutex;
std::queue<std::unique_ptr<nano::message>> requests;
std::atomic<bool> stopped{ false };
std::atomic<nano::socket::type_t> type{ nano::socket::type_t::undefined };
// Remote enpoint used to remove response channel even after socket closing
nano::tcp_endpoint remote_endpoint{ boost::asio::ip::address_v6::any (), 0 };
nano::account remote_node_id{ 0 };

2
nano/node/network.cpp
View file

@ -925,7 +925,7 @@ nano::tcp_message_item nano::tcp_message_manager::get_message ()
}
else
{
result = nano::tcp_message_item{ std::make_shared<nano::keepalive> (), nano::tcp_endpoint (boost::asio::ip::address_v6::any (), 0), 0, nullptr, nano::socket::type_t::undefined };
result = nano::tcp_message_item{ nullptr, nano::tcp_endpoint (boost::asio::ip::address_v6::any (), 0), 0, nullptr };
}
lock.unlock ();
producer_condition.notify_one ();

11
nano/node/socket.hpp
View file

@ -68,6 +68,14 @@ public:
{
return queue_size >= queue_size_max * 2;
}
type_t type () const
{
return type_m;
};
void type_set (type_t type_a)
{
type_m = type_a;
}
protected:
/** Holds the buffer and callback for queued writes */
@ -99,6 +107,9 @@ protected:
void stop_timer ();
void checkup ();
private:
type_t type_m{ type_t::undefined };
public:
static size_t constexpr queue_size_max = 128;
};

7
nano/node/transport/tcp.cpp
View file

@ -283,13 +283,14 @@ void nano::transport::tcp_channels::process_messages ()
auto item (node.network.tcp_message_manager.get_message ());
if (item.message != nullptr)
{
process_message (*item.message, item.endpoint, item.node_id, item.socket, item.type);
process_message (*item.message, item.endpoint, item.node_id, item.socket);
}
}
}
void nano::transport::tcp_channels::process_message (nano::message const & message_a, nano::tcp_endpoint const & endpoint_a, nano::account const & node_id_a, std::shared_ptr<nano::socket> const & socket_a, nano::socket::type_t type_a)
void nano::transport::tcp_channels::process_message (nano::message const & message_a, nano::tcp_endpoint const & endpoint_a, nano::account const & node_id_a, std::shared_ptr<nano::socket> const & socket_a)
{
auto type_a = socket_a->type ();
if (!stopped && message_a.header.version_using >= protocol_constants ().protocol_version_min ())
{
auto channel (node.network.find_channel (nano::transport::map_tcp_to_endpoint (endpoint_a)));
@ -691,7 +692,7 @@ void nano::transport::tcp_channels::start_tcp_receive_node_id (std::shared_ptr<n
callback_a (channel_a);
}
// Listen for possible responses
response_server->type = nano::socket::type_t::realtime_response_server;
response_server->socket->type_set (nano::socket::type_t::realtime_response_server);
response_server->remote_node_id = channel_a->get_node_id ();
response_server->receive ();
node_l->network.tcp_channels.remove_node_id_handshake_socket (socket_l);

3
nano/node/transport/tcp.hpp
View file

@ -24,7 +24,6 @@ public:
nano::tcp_endpoint endpoint;
nano::account node_id;
std::shared_ptr<nano::socket> socket;
nano::socket::type_t type;
};
namespace transport
{
@ -92,7 +91,7 @@ namespace transport
void start ();
void stop ();
void process_messages ();
void process_message (nano::message const &, nano::tcp_endpoint const &, nano::account const &, std::shared_ptr<nano::socket> const &, nano::socket::type_t);
void process_message (nano::message const &, nano::tcp_endpoint const &, nano::account const &, std::shared_ptr<nano::socket> const &);
bool max_ip_connections (nano::tcp_endpoint const &);
// Should we reach out to this endpoint with a keepalive message
bool reachout (nano::endpoint const &);