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

Migrate test from UDP to TCP (core_test/request_aggregator.cpp) (#3920)

Browse source
This commit is contained in:
Thiago Silva 2022-08-26 13:28:42 -03:00 • committed by GitHub
commit 45017fbe3c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 53 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

94
nano/core_test/request_aggregator.cpp
View file

@ -1,5 +1,7 @@
#include <nano/lib/jsonconfig.hpp>
#include <nano/node/request_aggregator.hpp>
#include <nano/node/transport/inproc.hpp>
#include <nano/test_common/network.hpp>
#include <nano/test_common/system.hpp>
#include <nano/test_common/testutil.hpp>
@ -27,24 +29,25 @@ TEST (request_aggregator, one)
.build_shared ();
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send1->hash (), send1->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, node.aggregator.empty ());
// Not yet in the ledger
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_unknown));
ASSERT_EQ (nano::process_result::progress, node.ledger.process (node.store.tx_begin_write (), *send1).code);
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
// In the ledger but no vote generated yet
ASSERT_TIMELY (3s, 0 < node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
ASSERT_TRUE (node.aggregator.empty ());
node.aggregator.add (channel, request);
ASSERT_TIMELY (3s, node.aggregator.empty ());
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
// Already cached
ASSERT_TIMELY (3s, node.aggregator.empty ());
ASSERT_EQ (3, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
ASSERT_EQ (0, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_dropped));
ASSERT_TIMELY (3s, 3 == node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
ASSERT_TIMELY (3s, 0 == node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_dropped));
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_unknown));
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_cached_votes));
@ -95,18 +98,19 @@ TEST (request_aggregator, one_update)
ASSERT_EQ (nano::process_result::progress, node.ledger.process (node.store.tx_begin_write (), *receive1).code);
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send2->hash (), send2->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
request.clear ();
request.emplace_back (receive1->hash (), receive1->root ());
// Update the pool of requests with another hash
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
// In the ledger but no vote generated yet
ASSERT_TIMELY (3s, 0 < node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes))
ASSERT_TRUE (node.aggregator.empty ());
ASSERT_EQ (2, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
ASSERT_EQ (0, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_dropped));
ASSERT_TIMELY (3s, 2 == node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
ASSERT_TIMELY (3s, 0 == node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_dropped));
ASSERT_TIMELY (3s, 0 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_unknown));
ASSERT_TIMELY (3s, 2 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_hashes));
size_t count = 0;
@ -161,15 +165,16 @@ TEST (request_aggregator, two)
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send2->hash (), send2->root ());
request.emplace_back (receive1->hash (), receive1->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
// Process both blocks
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
// One vote should be generated for both blocks
ASSERT_TIMELY (3s, 0 < node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
ASSERT_TRUE (node.aggregator.empty ());
// The same request should now send the cached vote
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, node.aggregator.empty ());
ASSERT_EQ (2, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
@ -214,12 +219,12 @@ TEST (request_aggregator, two_endpoints)
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send1->hash (), send1->root ());
ASSERT_EQ (nano::process_result::progress, node1.ledger.process (node1.store.tx_begin_write (), *send1).code);
auto channel1 (node1.network.udp_channels.create (node1.network.endpoint ()));
auto channel2 (node2.network.udp_channels.create (node2.network.endpoint ()));
ASSERT_NE (nano::transport::map_endpoint_to_v6 (channel1->get_endpoint ()), nano::transport::map_endpoint_to_v6 (channel2->get_endpoint ()));
auto dummy_channel1 = std::make_shared<nano::transport::inproc::channel> (node1, node1);
auto dummy_channel2 = std::make_shared<nano::transport::inproc::channel> (node2, node2);
ASSERT_NE (nano::transport::map_endpoint_to_v6 (dummy_channel1->get_endpoint ()), nano::transport::map_endpoint_to_v6 (dummy_channel2->get_endpoint ()));
// Use the aggregator from node1 only, making requests from both nodes
node1.aggregator.add (channel1, request);
node1.aggregator.add (channel2, request);
node1.aggregator.add (dummy_channel1, request);
node1.aggregator.add (dummy_channel2, request);
ASSERT_EQ (2, node1.aggregator.size ());
// For the first request it generates the vote, for the second it uses the generated vote
ASSERT_TIMELY (3s, node1.aggregator.empty ());
@ -270,8 +275,9 @@ TEST (request_aggregator, split)
election->force_confirm ();
ASSERT_TIMELY (5s, max_vbh + 2 == node.ledger.cache.cemented_count);
ASSERT_EQ (max_vbh + 1, request.size ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
// In the ledger but no vote generated yet
ASSERT_TIMELY (3s, 2 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
@ -310,8 +316,9 @@ TEST (request_aggregator, channel_lifetime)
request.emplace_back (send1->hash (), send1->root ());
{
// The aggregator should extend the lifetime of the channel
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
}
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, 0 < node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
@ -340,12 +347,14 @@ TEST (request_aggregator, channel_update)
request.emplace_back (send1->hash (), send1->root ());
std::weak_ptr<nano::transport::channel> channel1_w;
{
auto channel1 (node.network.udp_channels.create (node.network.endpoint ()));
channel1_w = channel1;
node.aggregator.add (channel1, request);
auto channel2 (node.network.udp_channels.create (node.network.endpoint ()));
auto client1 = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel1 = std::make_shared<nano::transport::channel_tcp> (node, client1);
channel1_w = dummy_channel1;
node.aggregator.add (dummy_channel1, request);
auto client2 = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel2 = std::make_shared<nano::transport::channel_tcp> (node, client2);
// The aggregator then hold channel2 and drop channel1
node.aggregator.add (channel2, request);
node.aggregator.add (dummy_channel2, request);
}
// Both requests were for the same endpoint, so only one pool should exist
ASSERT_EQ (1, node.aggregator.size ());
@ -376,9 +385,10 @@ TEST (request_aggregator, channel_max_queue)
ASSERT_EQ (nano::process_result::progress, node.ledger.process (node.store.tx_begin_write (), *send1).code);
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send1->hash (), send1->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_dropped));
}
@ -403,11 +413,12 @@ TEST (request_aggregator, unique)
ASSERT_EQ (nano::process_result::progress, node.ledger.process (node.store.tx_begin_write (), *send1).code);
std::vector<std::pair<nano::block_hash, nano::root>> request;
request.emplace_back (send1->hash (), send1->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
node.aggregator.add (channel, request);
node.aggregator.add (channel, request);
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
node.aggregator.add (dummy_channel, request);
node.aggregator.add (dummy_channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_hashes));
ASSERT_TIMELY (3s, 1 == node.stats.count (nano::stat::type::requests, nano::stat::detail::requests_generated_votes));
}
@ -449,8 +460,9 @@ TEST (request_aggregator, cannot_vote)
request.emplace_back (send2->hash (), send2->root ());
// Incorrect hash, correct root
request.emplace_back (1, send2->root ());
auto channel (node.network.udp_channels.create (node.network.endpoint ()));
node.aggregator.add (channel, request);
auto client = std::make_shared<nano::client_socket> (node);
std::shared_ptr<nano::transport::channel> dummy_channel = std::make_shared<nano::transport::channel_tcp> (node, client);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, node.aggregator.empty ());
ASSERT_EQ (1, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
@ -463,7 +475,7 @@ TEST (request_aggregator, cannot_vote)
// With an ongoing election
node.block_confirm (send2);
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, node.aggregator.empty ());
ASSERT_EQ (2, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));
@ -480,7 +492,7 @@ TEST (request_aggregator, cannot_vote)
ASSERT_NE (nullptr, election);
election->force_confirm ();
ASSERT_TIMELY (3s, node.ledger.dependents_confirmed (node.store.tx_begin_read (), *send2));
node.aggregator.add (channel, request);
node.aggregator.add (dummy_channel, request);
ASSERT_EQ (1, node.aggregator.size ());
ASSERT_TIMELY (3s, node.aggregator.empty ());
ASSERT_EQ (3, node.stats.count (nano::stat::type::aggregator, nano::stat::detail::aggregator_accepted));