Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

This converts the socket IO timer to a timer started/stopped with raii. #2965

Closed
wants to merge 5 commits into from
Closed

This converts the socket IO timer to a timer started/stopped with raii. #2965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fe0f654
This converts the socket IO timer to a timer started/stopped with raii.
clemahieu Sep 22, 2020
0edb69a
Adding direct tests to show RAII timers work correctly.
clemahieu Sep 25, 2020
d416b25
Merge branch 'develop' into raii_timer
clemahieu Mar 9, 2021
f7cef64
Merge branch 'develop' of https://github.com/nanocurrency/nano-node i…
Oct 26, 2021
15ac578
Clang-format fix
Oct 26, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
226 changes: 216 additions & 10 deletions nano/core_test/socket.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@

#include <gtest/gtest.h>

#include <boost/asio/read.hpp>

using namespace std::chrono_literals;

TEST (socket, max_connections)
Expand Down Expand Up @@ -45,13 +47,13 @@ TEST (socket, max_connections)

// start 3 clients, 2 will persist but 1 will be dropped

auto client1 = std::make_shared<nano::socket> (*node, boost::none);
auto client1 = std::make_shared<nano::socket> (*node);
client1->async_connect (dst_endpoint, connect_handler);

auto client2 = std::make_shared<nano::socket> (*node, boost::none);
auto client2 = std::make_shared<nano::socket> (*node);
client2->async_connect (dst_endpoint, connect_handler);

auto client3 = std::make_shared<nano::socket> (*node, boost::none);
auto client3 = std::make_shared<nano::socket> (*node);
client3->async_connect (dst_endpoint, connect_handler);

auto get_tcp_accept_failures = [&node] () {
Expand All @@ -70,10 +72,10 @@ TEST (socket, max_connections)

server_sockets[0].reset ();

auto client4 = std::make_shared<nano::socket> (*node, boost::none);
auto client4 = std::make_shared<nano::socket> (*node);
client4->async_connect (dst_endpoint, connect_handler);

auto client5 = std::make_shared<nano::socket> (*node, boost::none);
auto client5 = std::make_shared<nano::socket> (*node);
client5->async_connect (dst_endpoint, connect_handler);

ASSERT_TIMELY (5s, get_tcp_accept_failures () == 2);
Expand All @@ -87,13 +89,13 @@ TEST (socket, max_connections)
server_sockets[2].reset ();
ASSERT_EQ (server_sockets.size (), 3);

auto client6 = std::make_shared<nano::socket> (*node, boost::none);
auto client6 = std::make_shared<nano::socket> (*node);
client6->async_connect (dst_endpoint, connect_handler);

auto client7 = std::make_shared<nano::socket> (*node, boost::none);
auto client7 = std::make_shared<nano::socket> (*node);
client7->async_connect (dst_endpoint, connect_handler);

auto client8 = std::make_shared<nano::socket> (*node, boost::none);
auto client8 = std::make_shared<nano::socket> (*node);
client8->async_connect (dst_endpoint, connect_handler);

ASSERT_TIMELY (5s, get_tcp_accept_failures () == 3);
Expand Down Expand Up @@ -132,7 +134,7 @@ TEST (socket, drop_policy)
return true;
});

auto client = std::make_shared<nano::socket> (*node, boost::none);
auto client = std::make_shared<nano::socket> (*node);
nano::transport::channel_tcp channel{ *node, client };
nano::util::counted_completion write_completion (static_cast<unsigned> (total_message_count));

Expand Down Expand Up @@ -244,7 +246,7 @@ TEST (socket, concurrent_writes)
std::vector<std::shared_ptr<nano::socket>> clients;
for (unsigned i = 0; i < client_count; i++)
{
auto client = std::make_shared<nano::socket> (*node, boost::none);
auto client = std::make_shared<nano::socket> (*node);
clients.push_back (client);
client->async_connect (boost::asio::ip::tcp::endpoint (boost::asio::ip::address_v4::loopback (), 25000),
[&connection_count_completion] (boost::system::error_code const & ec_a) {
Expand Down Expand Up @@ -301,3 +303,207 @@ TEST (socket, concurrent_writes)
t.join ();
}
}

TEST (socket_timer, construct_destruct)
{
nano::system system (1);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
nano::socket::timer timer{ socket };
}

TEST (socket_timer, construct_release)
{
nano::system system (1);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
nano::socket::timer timer{ socket };
timer.release ();
}

TEST (socket_timer, construct_move_destruct)
{
nano::system system (1);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
nano::socket::timer timer0{ socket };
nano::socket::timer timer1{ std::move (timer0) };
}

TEST (socket_timer, construct_move_release)
{
nano::system system (1);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
nano::socket::timer timer0{ socket };
nano::socket::timer timer1{ std::move (timer0) };
timer1.release ();
}

TEST (socket_timeout, connect)
{
nano::system system{ 1 };
system.nodes[0]->config.tcp_io_timeout = std::chrono::seconds (1);
boost::asio::ip::tcp::endpoint endpoint{ boost::asio::ip::address_v6::loopback (), nano::get_available_port () };
boost::asio::ip::tcp::acceptor acceptor{ system.io_ctx };
acceptor.open (endpoint.protocol ());
acceptor.bind (endpoint);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
std::atomic<bool> done = false;
boost::system::error_code ec;
socket->async_connect (endpoint,
[&ec, &done] (boost::system::error_code const & ec_a) {
if (ec_a)
{
ec = ec_a;
done = true;
}
});
ASSERT_TIMELY (5s, done == true);
ASSERT_TRUE (ec);
}

TEST (socket_timeout, read)
{
nano::system system{ 1 };
system.nodes[0]->config.tcp_io_timeout = std::chrono::seconds (1);
boost::asio::ip::tcp::endpoint endpoint{ boost::asio::ip::address_v6::loopback (), nano::get_available_port () };
boost::asio::ip::tcp::acceptor acceptor{ system.io_ctx };
acceptor.open (endpoint.protocol ());
acceptor.bind (endpoint);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
auto buffer = std::make_shared<std::vector<uint8_t>> (1);
std::atomic<bool> done = false;
boost::system::error_code ec;
acceptor.listen (boost::asio::socket_base::max_listen_connections, ec);
boost::asio::ip::tcp::socket remote{ system.io_ctx };
acceptor.async_accept (remote, [] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
});
socket->async_connect (endpoint,
[&socket, buffer, &ec, &done] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
socket->async_read (buffer, 1, [&ec, &done] (boost::system::error_code const & ec_a, size_t size_a) {
if (ec_a)
{
ec = ec_a;
done = true;
}
});
});
ASSERT_TIMELY (50s, done == true);
ASSERT_TRUE (ec);
}

TEST (socket_timeout, write)
{
nano::system system{ 1 };
system.nodes[0]->config.tcp_io_timeout = std::chrono::seconds (1);
boost::asio::ip::tcp::endpoint endpoint{ boost::asio::ip::address_v6::loopback (), nano::get_available_port () };
boost::asio::ip::tcp::acceptor acceptor{ system.io_ctx };
acceptor.open (endpoint.protocol ());
acceptor.bind (endpoint);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
auto buffer = std::make_shared<std::vector<uint8_t>> (128 * 1024);
std::atomic<bool> done = false;
boost::system::error_code ec;
acceptor.listen (boost::asio::socket_base::max_listen_connections, ec);
boost::asio::ip::tcp::socket remote{ system.io_ctx };
acceptor.async_accept (remote, [] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
});
socket->async_connect (endpoint,
[&socket, buffer, &ec, &done] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
for (auto i = 0; i < 1024; ++i)
{
socket->async_write (nano::shared_const_buffer{ buffer }, [&ec, &done] (boost::system::error_code const & ec_a, size_t size_a) {
if (ec_a)
{
ec = ec_a;
done = true;
}
});
}
});
ASSERT_TIMELY (50s, done == true);
ASSERT_TRUE (ec);
}

TEST (socket_timeout, read_overlapped)
{
nano::system system{ 1 };
system.nodes[0]->config.tcp_io_timeout = std::chrono::seconds (1);
boost::asio::ip::tcp::endpoint endpoint{ boost::asio::ip::address_v6::loopback (), nano::get_available_port () };
boost::asio::ip::tcp::acceptor acceptor{ system.io_ctx };
acceptor.open (endpoint.protocol ());
acceptor.bind (endpoint);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
auto buffer = std::make_shared<std::vector<uint8_t>> (1);
std::atomic<bool> done = false;
boost::system::error_code ec;
acceptor.listen (boost::asio::socket_base::max_listen_connections, ec);
boost::asio::ip::tcp::socket remote{ system.io_ctx };
acceptor.async_accept (remote, [&remote, buffer] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
nano::async_write (remote, nano::shared_const_buffer{ buffer }, [] (boost::system::error_code const & ec_a, size_t size_a) {
debug_assert (!ec_a);
debug_assert (size_a == 1);
});
});
socket->async_connect (endpoint,
[&socket, buffer, &ec, &done] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
socket->async_read (buffer, 1, [] (boost::system::error_code const & ec_a, size_t size_a) {
debug_assert (size_a == 1);
});
socket->async_read (buffer, 1, [&ec, &done] (boost::system::error_code const & ec_a, size_t size_a) {
debug_assert (size_a == 0);
if (ec_a)
{
ec = ec_a;
done = true;
}
});
});
ASSERT_TIMELY (5s, done == true);
ASSERT_TRUE (ec);
}

TEST (socket_timeout, write_overlapped)
{
nano::system system{ 1 };
system.nodes[0]->config.tcp_io_timeout = std::chrono::seconds (1);
boost::asio::ip::tcp::endpoint endpoint{ boost::asio::ip::address_v6::loopback (), nano::get_available_port () };
boost::asio::ip::tcp::acceptor acceptor{ system.io_ctx };
acceptor.open (endpoint.protocol ());
acceptor.bind (endpoint);
auto socket = std::make_shared<nano::socket> (*system.nodes[0]);
auto buffer1 = std::make_shared<std::vector<uint8_t>> (1);
auto buffer2 = std::make_shared<std::vector<uint8_t>> (128 * 1024);
std::atomic<bool> done = false;
boost::system::error_code ec;
acceptor.listen (boost::asio::socket_base::max_listen_connections, ec);
boost::asio::ip::tcp::socket remote{ system.io_ctx };
acceptor.async_accept (remote, [&remote, buffer1] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
boost::asio::async_read (remote, boost::asio::buffer (buffer1->data (), buffer1->size ()), [] (boost::system::error_code const & ec_a, size_t size_a) {
debug_assert (size_a == 1);
});
});
socket->async_connect (endpoint,
[&socket, buffer1, buffer2, &ec, &done] (boost::system::error_code const & ec_a) {
debug_assert (!ec_a);
socket->async_write (nano::shared_const_buffer{ buffer1 }, [] (boost::system::error_code const & ec_a, size_t size_a) {
debug_assert (size_a == 1);
});
for (auto i = 0; i < 1024; ++i)
{
socket->async_write (nano::shared_const_buffer{ buffer2 }, [&ec, &done] (boost::system::error_code const & ec_a, size_t size_a) {
if (ec_a)
{
ec = ec_a;
done = true;
}
});
}
});
ASSERT_TIMELY (5s, done == true);
ASSERT_TRUE (ec);
}
1 change: 0 additions & 1 deletion nano/node/bootstrap/bootstrap_connections.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,6 @@ void nano::bootstrap_connections::pool_connection (std::shared_ptr<nano::bootstr
auto const & socket_l = client_a->socket;
if (!stopped && !client_a->pending_stop && !node.network.excluded_peers.check (client_a->channel->get_tcp_endpoint ()))
{
socket_l->start_timer (node.network_params.network.idle_timeout);
// Push into idle deque
if (!push_front)
{
Expand Down
4 changes: 0 additions & 4 deletions nano/node/bootstrap/bootstrap_server.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,17 +142,13 @@ void nano::bootstrap_server::stop ()

void nano::bootstrap_server::receive ()
{
// Increase timeout to receive TCP header (idle server socket)
socket->set_timeout (node->network_params.network.idle_timeout);
auto this_l (shared_from_this ());
socket->async_read (receive_buffer, 8, [this_l] (boost::system::error_code const & ec, std::size_t size_a) {
// Set remote_endpoint
if (this_l->remote_endpoint.port () == 0)
{
this_l->remote_endpoint = this_l->socket->remote_endpoint ();
}
// Decrease timeout to default
this_l->socket->set_timeout (this_l->node->config.tcp_io_timeout);
// Receive header
this_l->receive_header_action (ec, size_a);
});
Expand Down
Loading