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

Connecting the block deserializer in to the bulk_pull_client.

Browse source
This commit is contained in:
clemahieu 2022-07-12 15:05:58 +01:00
commit 946fc1c6a2
No known key found for this signature in database
GPG key ID: 43708520C8DFB938
3 changed files with 15 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

2
nano/core_test/bootstrap.cpp
View file

@ -1435,7 +1435,7 @@ TEST (bootstrap_processor, wallet_lazy_pending)
.work (*node0->work_generate_blocking (receive1->hash ())) .work (*node0->work_generate_blocking (receive1->hash ()))
.build_shared (); .build_shared ();
// Processing test c`hain // Processing test chain
node0->block_processor.add (send1); node0->block_processor.add (send1);
node0->block_processor.add (receive1); node0->block_processor.add (receive1);
node0->block_processor.add (send2); node0->block_processor.add (send2);

92
nano/node/bootstrap/bootstrap_bulk_pull.cpp
View file

@ -1,3 +1,4 @@
#include <nano/node/bootstrap/block_deserializer.hpp>
#include <nano/node/bootstrap/bootstrap.hpp> #include <nano/node/bootstrap/bootstrap.hpp>
#include <nano/node/bootstrap/bootstrap_bulk_pull.hpp> #include <nano/node/bootstrap/bootstrap_bulk_pull.hpp>
#include <nano/node/bootstrap/bootstrap_connections.hpp> #include <nano/node/bootstrap/bootstrap_connections.hpp>
@ -21,7 +22,8 @@ nano::pull_info::pull_info (nano::hash_or_account const & account_or_head_a, nan
nano::bulk_pull_client::bulk_pull_client (std::shared_ptr<nano::bootstrap_client> const & connection_a, std::shared_ptr<nano::bootstrap_attempt> const & attempt_a, nano::pull_info const & pull_a) : nano::bulk_pull_client::bulk_pull_client (std::shared_ptr<nano::bootstrap_client> const & connection_a, std::shared_ptr<nano::bootstrap_attempt> const & attempt_a, nano::pull_info const & pull_a) :
connection{ connection_a }, connection{ connection_a },
attempt{ attempt_a }, attempt{ attempt_a },
pull{ pull_a } pull{ pull_a },
block_deserializer{ std::make_shared<nano::bootstrap::block_deserializer> () }
{ {
attempt->condition.notify_all (); attempt->condition.notify_all ();
} }
@ -118,96 +120,18 @@ void nano::bulk_pull_client::throttled_receive_block ()
void nano::bulk_pull_client::receive_block () void nano::bulk_pull_client::receive_block ()
{ {
auto this_l (shared_from_this ()); block_deserializer->read (*connection->socket, [this_l = shared_from_this ()] (boost::system::error_code ec, std::shared_ptr<nano::block> block) {
connection->socket->async_read (connection->receive_buffer, 1, [this_l] (boost::system::error_code const & ec, std::size_t size_a) { this_l->received_block (ec, block);
if (!ec)
{
this_l->received_type ();
}
else
{
if (this_l->connection->node->config.logging.bulk_pull_logging ())
{
this_l->connection->node->logger.try_log (boost::str (boost::format ("Error receiving block type: %1%") % ec.message ()));
}
this_l->connection->node->stats.inc (nano::stat::type::bootstrap, nano::stat::detail::bulk_pull_receive_block_failure, nano::stat::dir::in);
this_l->network_error = true;
}
}); });
} }
void nano::bulk_pull_client::received_type () void nano::bulk_pull_client::received_block (boost::system::error_code ec, std::shared_ptr<nano::block> block)
{
auto this_l (shared_from_this ());
nano::block_type type (static_cast<nano::block_type> (connection->receive_buffer->data ()[0]));
auto const & socket_l = connection->socket;
switch (type)
{
case nano::block_type::send:
{
socket_l->async_read (connection->receive_buffer, nano::send_block::size, [this_l, type] (boost::system::error_code const & ec, std::size_t size_a) {
this_l->received_block (ec, size_a, type);
});
break;
}
case nano::block_type::receive:
{
socket_l->async_read (connection->receive_buffer, nano::receive_block::size, [this_l, type] (boost::system::error_code const & ec, std::size_t size_a) {
this_l->received_block (ec, size_a, type);
});
break;
}
case nano::block_type::open:
{
socket_l->async_read (connection->receive_buffer, nano::open_block::size, [this_l, type] (boost::system::error_code const & ec, std::size_t size_a) {
this_l->received_block (ec, size_a, type);
});
break;
}
case nano::block_type::change:
{
socket_l->async_read (connection->receive_buffer, nano::change_block::size, [this_l, type] (boost::system::error_code const & ec, std::size_t size_a) {
this_l->received_block (ec, size_a, type);
});
break;
}
case nano::block_type::state:
{
socket_l->async_read (connection->receive_buffer, nano::state_block::size, [this_l, type] (boost::system::error_code const & ec, std::size_t size_a) {
this_l->received_block (ec, size_a, type);
});
break;
}
case nano::block_type::not_a_block:
{
// Avoid re-using slow peers, or peers that sent the wrong blocks.
if (!connection->pending_stop && (expected == pull.end || (pull.count != 0 && pull.count == pull_blocks)))
{
connection->connections.pool_connection (connection);
}
break;
}
default:
{
if (connection->node->config.logging.network_packet_logging ())
{
connection->node->logger.try_log (boost::str (boost::format ("Unknown type received as block type: %1%") % static_cast<int> (type)));
}
break;
}
}
}
void nano::bulk_pull_client::received_block (boost::system::error_code const & ec, std::size_t size_a, nano::block_type type_a)
{ {
if (ec) if (ec)
{ {
network_error = true; network_error = true;
return; return;
} }
nano::bufferstream stream{ connection->receive_buffer->data (), size_a };
auto block = nano::deserialize_block (stream, type_a);
if (block == nullptr) if (block == nullptr)
{ {
// Avoid re-using slow peers, or peers that sent the wrong blocks. // Avoid re-using slow peers, or peers that sent the wrong blocks.
@ -226,7 +150,7 @@ void nano::bulk_pull_client::received_block (boost::system::error_code const & e
connection->node->stats.inc_detail_only (nano::stat::type::error, nano::stat::detail::insufficient_work); connection->node->stats.inc_detail_only (nano::stat::type::error, nano::stat::detail::insufficient_work);
return; return;
} }
auto hash (block->hash ()); auto hash = block->hash ();
if (connection->node->config.logging.bulk_pull_logging ()) if (connection->node->config.logging.bulk_pull_logging ())
{ {
std::string block_l; std::string block_l;
@ -267,7 +191,7 @@ void nano::bulk_pull_client::received_block (boost::system::error_code const & e
throttled_receive_block (); throttled_receive_block ();
} }
} }
else if (stop_pull && block_expected) else if (!stop_pull && block_expected)
{ {
connection->connections.pool_connection (connection); connection->connections.pool_connection (connection);
} }

8
nano/node/bootstrap/bootstrap_bulk_pull.hpp
View file

@ -8,6 +8,10 @@
namespace nano namespace nano
{ {
class bootstrap_attempt; class bootstrap_attempt;
namespace bootstrap
{
class block_deserializer;
};
class pull_info class pull_info
{ {
public: public:
@ -37,8 +41,7 @@ public:
void request (); void request ();
void receive_block (); void receive_block ();
void throttled_receive_block (); void throttled_receive_block ();
void received_type (); void received_block (boost::system::error_code ec, std::shared_ptr<nano::block> block);
void received_block (boost::system::error_code const &, std::size_t, nano::block_type);
nano::block_hash first (); nano::block_hash first ();
std::shared_ptr<nano::bootstrap_client> connection; std::shared_ptr<nano::bootstrap_client> connection;
std::shared_ptr<nano::bootstrap_attempt> attempt; std::shared_ptr<nano::bootstrap_attempt> attempt;
@ -67,6 +70,7 @@ private:
* Tracks the number of times an unexpected block was received * Tracks the number of times an unexpected block was received
*/ */
uint64_t unexpected_count{ 0 }; uint64_t unexpected_count{ 0 };
std::shared_ptr<nano::bootstrap::block_deserializer> block_deserializer;
}; };
class bootstrap_attempt_wallet; class bootstrap_attempt_wallet;
class bulk_pull_account_client final : public std::enable_shared_from_this<nano::bulk_pull_account_client> class bulk_pull_account_client final : public std::enable_shared_from_this<nano::bulk_pull_account_client>