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

Simplify dependencies

Browse source
This commit is contained in:
Piotr Wójcik 2024-10-19 15:39:28 +02:00
commit 58f7169cff
9 changed files with 137 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

57
nano/core_test/confirming_set.cpp
View file

@ -1,9 +1,11 @@
#include <nano/lib/blocks.hpp>
#include <nano/lib/logging.hpp>
#include <nano/node/active_elections.hpp>
#include <nano/node/blockprocessor.hpp>
#include <nano/node/confirming_set.hpp>
#include <nano/node/election.hpp>
#include <nano/node/make_store.hpp>
#include <nano/node/unchecked_map.hpp>
#include <nano/secure/ledger.hpp>
#include <nano/secure/ledger_set_confirmed.hpp>
#include <nano/test_common/ledger_context.hpp>
@ -16,44 +18,61 @@
using namespace std::chrono_literals;
namespace
{
struct confirming_set_context
{
nano::logger & logger;
nano::stats & stats;
nano::ledger & ledger;
nano::unchecked_map unchecked;
nano::block_processor block_processor;
nano::confirming_set confirming_set;
explicit confirming_set_context (nano::test::ledger_context & ledger_context, nano::node_config node_config = {}) :
logger{ ledger_context.logger () },
stats{ ledger_context.stats () },
ledger{ ledger_context.ledger () },
unchecked{ 0, stats, false },
block_processor{ node_config, ledger, unchecked, stats, logger },
confirming_set{ node_config.confirming_set, ledger, block_processor, stats, logger }
{
}
};
}
TEST (confirming_set, construction)
{
nano::test::system system;
auto & node = *system.add_node ();
auto ctx = nano::test::ledger_empty ();
nano::confirming_set_config config{};
nano::confirming_set confirming_set{ config, ctx.ledger (), node.block_processor, ctx.stats (), ctx.logger () };
auto ledger_ctx = nano::test::ledger_empty ();
confirming_set_context ctx{ ledger_ctx };
}
TEST (confirming_set, add_exists)
{
nano::test::system system;
auto & node = *system.add_node ();
auto ctx = nano::test::ledger_send_receive ();
nano::confirming_set_config config{};
nano::confirming_set confirming_set{ config, ctx.ledger (), node.block_processor, ctx.stats (), ctx.logger () };
auto send = ctx.blocks ()[0];
auto ledger_ctx = nano::test::ledger_send_receive ();
confirming_set_context ctx{ ledger_ctx };
nano::confirming_set & confirming_set = ctx.confirming_set;
auto send = ledger_ctx.blocks ()[0];
confirming_set.add (send->hash ());
ASSERT_TRUE (confirming_set.contains (send->hash ()));
}
TEST (confirming_set, process_one)
{
nano::test::system system;
auto & node = *system.add_node ();
auto ctx = nano::test::ledger_send_receive ();
nano::confirming_set_config config{};
nano::confirming_set confirming_set{ config, ctx.ledger (), node.block_processor, ctx.stats (), ctx.logger () };
auto ledger_ctx = nano::test::ledger_send_receive ();
confirming_set_context ctx{ ledger_ctx };
nano::confirming_set & confirming_set = ctx.confirming_set;
std::atomic<int> count = 0;
std::mutex mutex;
std::condition_variable condition;
confirming_set.cemented_observers.add ([&] (auto const &) { ++count; condition.notify_all (); });
confirming_set.add (ctx.blocks ()[0]->hash ());
confirming_set.add (ledger_ctx.blocks ()[0]->hash ());
nano::test::start_stop_guard guard{ confirming_set };
std::unique_lock lock{ mutex };
ASSERT_TRUE (condition.wait_for (lock, 5s, [&] () { return count == 1; }));
ASSERT_EQ (1, ctx.stats ().count (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed, nano::stat::dir::in));
ASSERT_EQ (2, ctx.ledger ().cemented_count ());
ASSERT_EQ (1, ctx.stats.count (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed, nano::stat::dir::in));
ASSERT_EQ (2, ctx.ledger.cemented_count ());
}
TEST (confirming_set, process_multiple)

8
nano/node/active_elections.cpp
View file

@ -45,6 +45,14 @@ nano::active_elections::active_elections (nano::node & node_a, nano::confirming_
publish (context.block);
}
});
// Stop all rolled back active transactions except initial
block_processor.rolled_back.add ([this] (auto const & block, auto const & rollback_root) {
if (block->qualified_root () != rollback_root)
{
erase (block->qualified_root ());
}
});
}
nano::active_elections::~active_elections ()

124
nano/node/blockprocessor.cpp
View file

@ -6,6 +6,7 @@
#include <nano/node/blockprocessor.hpp>
#include <nano/node/local_vote_history.hpp>
#include <nano/node/node.hpp>
#include <nano/node/unchecked_map.hpp>
#include <nano/secure/ledger.hpp>
#include <nano/secure/ledger_set_any.hpp>
#include <nano/store/component.hpp>
@ -38,10 +39,13 @@ void nano::block_processor::context::set_result (result_t const & result)
* block_processor
*/
nano::block_processor::block_processor (nano::node & node_a) :
config{ node_a.config.block_processor },
node (node_a),
next_log (std::chrono::steady_clock::now ())
nano::block_processor::block_processor (nano::node_config const & node_config, nano::ledger & ledger_a, nano::unchecked_map & unchecked_a, nano::stats & stats_a, nano::logger & logger_a) :
config{ node_config.block_processor },
network_params{ node_config.network_params },
ledger{ ledger_a },
unchecked{ unchecked_a },
stats{ stats_a },
logger{ logger_a }
{
batch_processed.add ([this] (auto const & items) {
// For every batch item: notify the 'processed' observer.
@ -78,6 +82,11 @@ nano::block_processor::block_processor (nano::node & node_a) :
return 1;
}
};
// Requeue blocks that could not be immediately processed
unchecked.satisfied.add ([this] (nano::unchecked_info const & info) {
add (info.block, nano::block_source::unchecked);
});
}
nano::block_processor::~block_processor ()
@ -124,14 +133,14 @@ std::size_t nano::block_processor::size (nano::block_source source) const
bool nano::block_processor::add (std::shared_ptr<nano::block> const & block, block_source const source, std::shared_ptr<nano::transport::channel> const & channel, std::function<void (nano::block_status)> callback)
{
if (node.network_params.work.validate_entry (*block)) // true => error
if (network_params.work.validate_entry (*block)) // true => error
{
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::insufficient_work);
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::insufficient_work);
return false; // Not added
}
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process);
node.logger.debug (nano::log::type::blockprocessor, "Processing block (async): {} (source: {} {})",
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process);
logger.debug (nano::log::type::blockprocessor, "Processing block (async): {} (source: {} {})",
block->hash ().to_string (),
to_string (source),
channel ? channel->to_string () : "<unknown>"); // TODO: Lazy eval
@ -141,8 +150,8 @@ bool nano::block_processor::add (std::shared_ptr<nano::block> const & block, blo
std::optional<nano::block_status> nano::block_processor::add_blocking (std::shared_ptr<nano::block> const & block, block_source const source)
{
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process_blocking);
node.logger.debug (nano::log::type::blockprocessor, "Processing block (blocking): {} (source: {})", block->hash ().to_string (), to_string (source));
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process_blocking);
logger.debug (nano::log::type::blockprocessor, "Processing block (blocking): {} (source: {})", block->hash ().to_string (), to_string (source));
context ctx{ block, source };
auto future = ctx.get_future ();
@ -155,8 +164,8 @@ std::optional<nano::block_status> nano::block_processor::add_blocking (std::shar
}
catch (std::future_error const &)
{
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process_blocking_timeout);
node.logger.error (nano::log::type::blockprocessor, "Block dropped when processing: {}", block->hash ().to_string ());
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::process_blocking_timeout);
logger.error (nano::log::type::blockprocessor, "Block dropped when processing: {}", block->hash ().to_string ());
}
return std::nullopt;
@ -164,8 +173,8 @@ std::optional<nano::block_status> nano::block_processor::add_blocking (std::shar
void nano::block_processor::force (std::shared_ptr<nano::block> const & block_a)
{
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::force);
node.logger.debug (nano::log::type::blockprocessor, "Forcing block: {}", block_a->hash ().to_string ());
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::force);
logger.debug (nano::log::type::blockprocessor, "Forcing block: {}", block_a->hash ().to_string ());
add_impl (context{ block_a, block_source::forced });
}
@ -184,45 +193,38 @@ bool nano::block_processor::add_impl (context ctx, std::shared_ptr<nano::transpo
}
else
{
node.stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::overfill);
node.stats.inc (nano::stat::type::blockprocessor_overfill, to_stat_detail (source));
stats.inc (nano::stat::type::blockprocessor, nano::stat::detail::overfill);
stats.inc (nano::stat::type::blockprocessor_overfill, to_stat_detail (source));
}
return added;
}
void nano::block_processor::rollback_competitor (secure::write_transaction const & transaction, nano::block const & block)
void nano::block_processor::rollback_competitor (secure::write_transaction const & transaction, nano::block const & fork_block)
{
auto hash = block.hash ();
auto successor_hash = node.ledger.any.block_successor (transaction, block.qualified_root ());
auto successor = successor_hash ? node.ledger.any.block_get (transaction, successor_hash.value ()) : nullptr;
auto const hash = fork_block.hash ();
auto const successor_hash = ledger.any.block_successor (transaction, fork_block.qualified_root ());
auto const successor = successor_hash ? ledger.any.block_get (transaction, successor_hash.value ()) : nullptr;
if (successor != nullptr && successor->hash () != hash)
{
// Replace our block with the winner and roll back any dependent blocks
node.logger.debug (nano::log::type::blockprocessor, "Rolling back: {} and replacing with: {}", successor->hash ().to_string (), hash.to_string ());
logger.debug (nano::log::type::blockprocessor, "Rolling back: {} and replacing with: {}", successor->hash ().to_string (), hash.to_string ());
std::vector<std::shared_ptr<nano::block>> rollback_list;
if (node.ledger.rollback (transaction, successor->hash (), rollback_list))
if (ledger.rollback (transaction, successor->hash (), rollback_list))
{
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::rollback_failed);
node.logger.error (nano::log::type::blockprocessor, "Failed to roll back: {} because it or a successor was confirmed", successor->hash ().to_string ());
stats.inc (nano::stat::type::ledger, nano::stat::detail::rollback_failed);
logger.error (nano::log::type::blockprocessor, "Failed to roll back: {} because it or a successor was confirmed", successor->hash ().to_string ());
}
else
{
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::rollback);
node.logger.debug (nano::log::type::blockprocessor, "Blocks rolled back: {}", rollback_list.size ());
stats.inc (nano::stat::type::ledger, nano::stat::detail::rollback);
logger.debug (nano::log::type::blockprocessor, "Blocks rolled back: {}", rollback_list.size ());
}
// Deleting from votes cache, stop active transaction
for (auto & i : rollback_list)
// Notify observers of the rolled back blocks
for (auto const & block : rollback_list)
{
rolled_back.notify (i);
node.history.erase (i->root ());
// Stop all rolled back active transactions except initial
if (i->hash () != successor->hash ())
{
node.active.erase (*i);
}
rolled_back.notify (block, fork_block.qualified_root ());
}
}
}
@ -237,7 +239,7 @@ void nano::block_processor::run ()
// TODO: Cleaner periodical logging
if (should_log ())
{
node.logger.info (nano::log::type::blockprocessor, "{} blocks (+ {} forced) in processing queue",
logger.info (nano::log::type::blockprocessor, "{} blocks (+ {} forced) in processing queue",
queue.size (),
queue.size ({ nano::block_source::forced }));
}
@ -319,7 +321,7 @@ auto nano::block_processor::process_batch (nano::unique_lock<nano::mutex> & lock
lock.unlock ();
auto transaction = node.ledger.tx_begin_write (nano::store::writer::blockprocessor);
auto transaction = ledger.tx_begin_write (nano::store::writer::blockprocessor);
nano::timer<std::chrono::milliseconds> timer;
timer.start ();
@ -350,7 +352,7 @@ auto nano::block_processor::process_batch (nano::unique_lock<nano::mutex> & lock
if (number_of_blocks_processed != 0 && timer.stop () > std::chrono::milliseconds (100))
{
node.logger.debug (nano::log::type::blockprocessor, "Processed {} blocks ({} forced) in {} {}", number_of_blocks_processed, number_of_forced_processed, timer.value ().count (), timer.unit ());
logger.debug (nano::log::type::blockprocessor, "Processed {} blocks ({} forced) in {} {}", number_of_blocks_processed, number_of_forced_processed, timer.value ().count (), timer.unit ());
}
return processed;
@ -360,12 +362,12 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
{
auto block = context.block;
auto const hash = block->hash ();
nano::block_status result = node.ledger.process (transaction_a, block);
nano::block_status result = ledger.process (transaction_a, block);
node.stats.inc (nano::stat::type::blockprocessor_result, to_stat_detail (result));
node.stats.inc (nano::stat::type::blockprocessor_source, to_stat_detail (context.source));
stats.inc (nano::stat::type::blockprocessor_result, to_stat_detail (result));
stats.inc (nano::stat::type::blockprocessor_source, to_stat_detail (context.source));
node.logger.trace (nano::log::type::blockprocessor, nano::log::detail::block_processed,
logger.trace (nano::log::type::blockprocessor, nano::log::detail::block_processed,
nano::log::arg{ "result", result },
nano::log::arg{ "source", context.source },
nano::log::arg{ "arrival", nano::log::microseconds (context.arrival) },
@ -376,40 +378,41 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
{
case nano::block_status::progress:
{
queue_unchecked (transaction_a, hash);
/* For send blocks check epoch open unchecked (gap pending).
For state blocks check only send subtype and only if block epoch is not last epoch.
If epoch is last, then pending entry shouldn't trigger same epoch open block for destination account. */
unchecked.trigger (hash);
/*
* For send blocks check epoch open unchecked (gap pending).
* For state blocks check only send subtype and only if block epoch is not last epoch.
* If epoch is last, then pending entry shouldn't trigger same epoch open block for destination account.
*/
if (block->type () == nano::block_type::send || (block->type () == nano::block_type::state && block->is_send () && std::underlying_type_t<nano::epoch> (block->sideband ().details.epoch) < std::underlying_type_t<nano::epoch> (nano::epoch::max)))
{
/* block->destination () for legacy send blocks
block->link () for state blocks (send subtype) */
queue_unchecked (transaction_a, block->destination ());
unchecked.trigger (block->destination ());
}
break;
}
case nano::block_status::gap_previous:
{
node.unchecked.put (block->previous (), block);
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_previous);
unchecked.put (block->previous (), block);
stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_previous);
break;
}
case nano::block_status::gap_source:
{
release_assert (block->source_field () || block->link_field ());
node.unchecked.put (block->source_field ().value_or (block->link_field ().value_or (0).as_block_hash ()), block);
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_source);
unchecked.put (block->source_field ().value_or (block->link_field ().value_or (0).as_block_hash ()), block);
stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_source);
break;
}
case nano::block_status::gap_epoch_open_pending:
{
node.unchecked.put (block->account_field ().value_or (0), block); // Specific unchecked key starting with epoch open block account public key
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_source);
unchecked.put (block->account_field ().value_or (0), block); // Specific unchecked key starting with epoch open block account public key
stats.inc (nano::stat::type::ledger, nano::stat::detail::gap_source);
break;
}
case nano::block_status::old:
{
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::old);
stats.inc (nano::stat::type::ledger, nano::stat::detail::old);
break;
}
case nano::block_status::bad_signature:
@ -426,7 +429,7 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
}
case nano::block_status::fork:
{
node.stats.inc (nano::stat::type::ledger, nano::stat::detail::fork);
stats.inc (nano::stat::type::ledger, nano::stat::detail::fork);
break;
}
case nano::block_status::opened_burn_account:
@ -453,11 +456,6 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
return result;
}
void nano::block_processor::queue_unchecked (secure::write_transaction const & transaction_a, nano::hash_or_account const & hash_or_account_a)
{
node.unchecked.trigger (hash_or_account_a);
}
nano::container_info nano::block_processor::container_info () const
{
nano::lock_guard<nano::mutex> guard{ mutex };

21
nano/node/blockprocessor.hpp
View file

@ -78,7 +78,7 @@ public: // Context
};
public:
explicit block_processor (nano::node &);
block_processor (nano::node_config const &, nano::ledger &, nano::unchecked_map &, nano::stats &, nano::logger &);
~block_processor ();
void start ();
@ -102,27 +102,32 @@ public: // Events
// The batch observer feeds the processed observer
nano::observer_set<nano::block_status const &, context const &> block_processed;
nano::observer_set<processed_batch_t const &> batch_processed;
nano::observer_set<std::shared_ptr<nano::block> const &> rolled_back;
// Rolled back blocks <rolled back block, root of rollback>
nano::observer_set<std::shared_ptr<nano::block> const &, nano::qualified_root const &> rolled_back;
private: // Dependencies
block_processor_config const & config;
nano::network_params const & network_params;
nano::ledger & ledger;
nano::unchecked_map & unchecked;
nano::stats & stats;
nano::logger & logger;
private:
void run ();
// Roll back block in the ledger that conflicts with 'block'
void rollback_competitor (secure::write_transaction const &, nano::block const & block);
nano::block_status process_one (secure::write_transaction const &, context const &, bool forced = false);
void queue_unchecked (secure::write_transaction const &, nano::hash_or_account const &);
processed_batch_t process_batch (nano::unique_lock<nano::mutex> &);
std::deque<context> next_batch (size_t max_count);
context next ();
bool add_impl (context, std::shared_ptr<nano::transport::channel> const & channel = nullptr);
private: // Dependencies
block_processor_config const & config;
nano::node & node;
private:
nano::fair_queue<context, nano::block_source> queue;
std::chrono::steady_clock::time_point next_log;
std::chrono::steady_clock::time_point next_log{ std::chrono::steady_clock::now () };
bool stopped{ false };
nano::condition_variable condition;

2
nano/node/fwd.hpp
View file

@ -32,6 +32,8 @@ class recently_cemented_cache;
class recently_confirmed_cache;
class rep_crawler;
class rep_tiers;
class telemetry;
class unchecked_map;
class stats;
class vote_cache;
class vote_generator;

2
nano/node/local_block_broadcaster.cpp
View file

@ -54,7 +54,7 @@ nano::local_block_broadcaster::local_block_broadcaster (local_block_broadcaster_
}
});
block_processor.rolled_back.add ([this] (auto const & block) {
block_processor.rolled_back.add ([this] (auto const & block, auto const & rollback_root) {
nano::lock_guard<nano::mutex> guard{ mutex };
auto erased = local_blocks.get<tag_hash> ().erase (block->hash ());
stats.add (nano::stat::type::local_block_broadcaster, nano::stat::detail::rollback, erased);

12
nano/node/node.cpp
View file

@ -114,7 +114,8 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
application_path (application_path_a),
port_mapping_impl{ std::make_unique<nano::port_mapping> (*this) },
port_mapping{ *port_mapping_impl },
block_processor (*this),
block_processor_impl{ std::make_unique<nano::block_processor> (config, ledger, unchecked, stats, logger) },
block_processor{ *block_processor_impl },
confirming_set_impl{ std::make_unique<nano::confirming_set> (config.confirming_set, ledger, block_processor, stats, logger) },
confirming_set{ *confirming_set_impl },
active_impl{ std::make_unique<nano::active_elections> (*this, confirming_set, block_processor) },
@ -162,10 +163,6 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
process_live_dispatcher.connect (block_processor);
unchecked.satisfied.add ([this] (nano::unchecked_info const & info) {
block_processor.add (info.block, nano::block_source::unchecked);
});
vote_cache.rep_weight_query = [this] (nano::account const & rep) {
return ledger.weight (rep);
};
@ -190,6 +187,11 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
active.recently_confirmed.erase (hash);
});
// Do some cleanup of rolled back blocks
block_processor.rolled_back.add ([this] (auto const & block, auto const & rollback_root) {
history.erase (block->root ());
});
if (!init_error ())
{
// Notify election schedulers when AEC frees election slot

4
nano/node/node.hpp
View file

@ -6,7 +6,6 @@
#include <nano/lib/stats.hpp>
#include <nano/lib/thread_pool.hpp>
#include <nano/lib/work.hpp>
#include <nano/node/blockprocessor.hpp>
#include <nano/node/bootstrap/bootstrap.hpp>
#include <nano/node/bootstrap/bootstrap_attempt.hpp>
#include <nano/node/bootstrap/bootstrap_server.hpp>
@ -184,7 +183,8 @@ public:
nano::node_observers observers;
std::unique_ptr<nano::port_mapping> port_mapping_impl;
nano::port_mapping & port_mapping;
nano::block_processor block_processor;
std::unique_ptr<nano::block_processor> block_processor_impl;
nano::block_processor & block_processor;
std::unique_ptr<nano::confirming_set> confirming_set_impl;
nano::confirming_set & confirming_set;
std::unique_ptr<nano::active_elections> active_impl;

8
nano/slow_test/node.cpp
View file

@ -1120,8 +1120,6 @@ TEST (confirmation_height, many_accounts_send_receive_self)
TEST (confirmation_height, many_accounts_send_receive_self_no_elections)
{
nano::test::system system;
auto & node = *system.nodes[0];
if (nano::rocksdb_config::using_rocksdb_in_tests ())
{
// Don't test this in rocksdb mode
@ -1140,8 +1138,12 @@ TEST (confirmation_height, many_accounts_send_receive_self_no_elections)
nano::block_hash block_hash_being_processed{ 0 };
nano::store::write_queue write_queue;
nano::node_config node_config;
nano::unchecked_map unchecked{ 0, stats, false };
nano::block_processor block_processor{ node_config, ledger, unchecked, stats, logger };
nano::confirming_set_config confirming_set_config{};
nano::confirming_set confirming_set{ confirming_set_config, ledger, node.block_processor, stats, logger };
nano::confirming_set confirming_set{ confirming_set_config, ledger, block_processor, stats, logger };
auto const num_accounts = 100000;