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

Comments for confirmation_height_processor (#3900)

Browse source
This commit is contained in:
Piotr Wójcik 2022-08-24 19:07:10 +02:00 • committed by GitHub
commit de16a87d71
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 30 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

2
nano/node/confirmation_height_bounded.hpp
View file

@ -19,7 +19,7 @@ class write_guard;
class confirmation_height_bounded final class confirmation_height_bounded final
{ {
public: public:
confirmation_height_bounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds, nano::logging const &, nano::logger_mt &, std::atomic<bool> &, uint64_t &, std::function<void (std::vector<std::shared_ptr<nano::block>> const &)> const &, std::function<void (nano::block_hash const &)> const &, std::function<uint64_t ()> const &); confirmation_height_bounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds batch_separate_pending_min_time, nano::logging const &, nano::logger_mt &, std::atomic<bool> & stopped, uint64_t & batch_write_size, std::function<void (std::vector<std::shared_ptr<nano::block>> const &)> const & cemented_callback, std::function<void (nano::block_hash const &)> const & already_cemented_callback, std::function<uint64_t ()> const & awaiting_processing_size_query);
bool pending_empty () const; bool pending_empty () const;
void clear_process_vars (); void clear_process_vars ();
void process (std::shared_ptr<nano::block> original_block); void process (std::shared_ptr<nano::block> original_block);

18
nano/node/confirmation_height_processor.cpp
View file

@ -14,10 +14,16 @@
nano::confirmation_height_processor::confirmation_height_processor (nano::ledger & ledger_a, nano::write_database_queue & write_database_queue_a, std::chrono::milliseconds batch_separate_pending_min_time_a, nano::logging const & logging_a, nano::logger_mt & logger_a, boost::latch & latch, confirmation_height_mode mode_a) : nano::confirmation_height_processor::confirmation_height_processor (nano::ledger & ledger_a, nano::write_database_queue & write_database_queue_a, std::chrono::milliseconds batch_separate_pending_min_time_a, nano::logging const & logging_a, nano::logger_mt & logger_a, boost::latch & latch, confirmation_height_mode mode_a) :
ledger (ledger_a), ledger (ledger_a),
write_database_queue (write_database_queue_a), write_database_queue (write_database_queue_a),
// clang-format off unbounded_processor (
unbounded_processor (ledger_a, write_database_queue_a, batch_separate_pending_min_time_a, logging_a, logger_a, stopped, batch_write_size, [this](auto & cemented_blocks) { this->notify_observers (cemented_blocks); }, [this](auto const & block_hash_a) { this->notify_observers (block_hash_a); }, [this]() { return this->awaiting_processing_size (); }), ledger_a, write_database_queue_a, batch_separate_pending_min_time_a, logging_a, logger_a, stopped, batch_write_size,
bounded_processor (ledger_a, write_database_queue_a, batch_separate_pending_min_time_a, logging_a, logger_a, stopped, batch_write_size, [this](auto & cemented_blocks) { this->notify_observers (cemented_blocks); }, [this](auto const & block_hash_a) { this->notify_observers (block_hash_a); }, [this]() { return this->awaiting_processing_size (); }), /* cemented_callback */ [this] (auto & cemented_blocks) { this->notify_cemented (cemented_blocks); },
// clang-format on /* already cemented_callback */ [this] (auto const & block_hash_a) { this->notify_already_cemented (block_hash_a); },
/* awaiting_processing_size_query */ [this] () { return this->awaiting_processing_size (); }),
bounded_processor (
ledger_a, write_database_queue_a, batch_separate_pending_min_time_a, logging_a, logger_a, stopped, batch_write_size,
/* cemented_callback */ [this] (auto & cemented_blocks) { this->notify_cemented (cemented_blocks); },
/* already cemented_callback */ [this] (auto const & block_hash_a) { this->notify_already_cemented (block_hash_a); },
/* awaiting_processing_size_query */ [this] () { return this->awaiting_processing_size (); }),
thread ([this, &latch, mode_a] () { thread ([this, &latch, mode_a] () {
nano::thread_role::set (nano::thread_role::name::confirmation_height_processing); nano::thread_role::set (nano::thread_role::name::confirmation_height_processing);
// Do not start running the processing thread until other threads have finished their operations // Do not start running the processing thread until other threads have finished their operations
@ -181,7 +187,7 @@ void nano::confirmation_height_processor::add_block_already_cemented_observer (s
block_already_cemented_observers.push_back (callback_a); block_already_cemented_observers.push_back (callback_a);
} }
void nano::confirmation_height_processor::notify_observers (std::vector<std::shared_ptr<nano::block>> const & cemented_blocks) void nano::confirmation_height_processor::notify_cemented (std::vector<std::shared_ptr<nano::block>> const & cemented_blocks)
{ {
for (auto const & block_callback_data : cemented_blocks) for (auto const & block_callback_data : cemented_blocks)
{ {
@ -192,7 +198,7 @@ void nano::confirmation_height_processor::notify_observers (std::vector<std::sha
} }
} }
void nano::confirmation_height_processor::notify_observers (nano::block_hash const & hash_already_cemented_a) void nano::confirmation_height_processor::notify_already_cemented (nano::block_hash const & hash_already_cemented_a)
{ {
for (auto const & observer : block_already_cemented_observers) for (auto const & observer : block_already_cemented_observers)
{ {

19
nano/node/confirmation_height_processor.hpp
View file

@ -44,16 +44,24 @@ public:
bool is_processing_block (nano::block_hash const &) const; bool is_processing_block (nano::block_hash const &) const;
nano::block_hash current () const; nano::block_hash current () const;
/*
* Called for each newly cemented block
* Called from confirmation height processor thread
*/
void add_cemented_observer (std::function<void (std::shared_ptr<nano::block> const &)> const &); void add_cemented_observer (std::function<void (std::shared_ptr<nano::block> const &)> const &);
/*
* Called when the block was added to the confirmation height processor but is already confirmed
* Called from confirmation height processor thread
*/
void add_block_already_cemented_observer (std::function<void (nano::block_hash const &)> const &); void add_block_already_cemented_observer (std::function<void (nano::block_hash const &)> const &);
private: private:
mutable nano::mutex mutex{ mutex_identifier (mutexes::confirmation_height_processor) }; mutable nano::mutex mutex{ mutex_identifier (mutexes::confirmation_height_processor) };
// Hashes which have been added to the confirmation height processor, but not yet processed // Hashes which have been added to the confirmation height processor, but not yet processed
// clang-format off
struct block_wrapper struct block_wrapper
{ {
block_wrapper (std::shared_ptr<nano::block> const & block_a) : explicit block_wrapper (std::shared_ptr<nano::block> const & block_a) :
block (block_a) block (block_a)
{ {
} }
@ -65,6 +73,7 @@ private:
std::shared_ptr<nano::block> block; std::shared_ptr<nano::block> block;
}; };
// clang-format off
class tag_sequence {}; class tag_sequence {};
class tag_hash {}; class tag_hash {};
boost::multi_index_container<block_wrapper, boost::multi_index_container<block_wrapper,
@ -97,10 +106,12 @@ private:
std::thread thread; std::thread thread;
void set_next_hash (); void set_next_hash ();
void notify_observers (std::vector<std::shared_ptr<nano::block>> const &); void notify_cemented (std::vector<std::shared_ptr<nano::block>> const &);
void notify_observers (nano::block_hash const &); void notify_already_cemented (nano::block_hash const &);
friend std::unique_ptr<container_info_component> collect_container_info (confirmation_height_processor &, std::string const &); friend std::unique_ptr<container_info_component> collect_container_info (confirmation_height_processor &, std::string const &);
private: // Tests
friend class confirmation_height_pending_observer_callbacks_Test; friend class confirmation_height_pending_observer_callbacks_Test;
friend class confirmation_height_dependent_election_Test; friend class confirmation_height_dependent_election_Test;
friend class confirmation_height_dependent_election_after_already_cemented_Test; friend class confirmation_height_dependent_election_after_already_cemented_Test;

3
nano/node/confirmation_height_unbounded.cpp
View file

@ -361,9 +361,6 @@ void nano::confirmation_height_unbounded::prepare_iterated_blocks_for_cementing
} }
} }
/*
* Returns true if there was an error in finding one of the blocks to write a confirmation height for, false otherwise
*/
void nano::confirmation_height_unbounded::cement_blocks (nano::write_guard & scoped_write_guard_a) void nano::confirmation_height_unbounded::cement_blocks (nano::write_guard & scoped_write_guard_a)
{ {
nano::timer<std::chrono::milliseconds> cemented_batch_timer; nano::timer<std::chrono::milliseconds> cemented_batch_timer;

2
nano/node/confirmation_height_unbounded.hpp
View file

@ -20,7 +20,7 @@ class write_guard;
class confirmation_height_unbounded final class confirmation_height_unbounded final
{ {
public: public:
confirmation_height_unbounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds, nano::logging const &, nano::logger_mt &, std::atomic<bool> &, uint64_t &, std::function<void (std::vector<std::shared_ptr<nano::block>> const &)> const &, std::function<void (nano::block_hash const &)> const &, std::function<uint64_t ()> const &); confirmation_height_unbounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds batch_separate_pending_min_time, nano::logging const &, nano::logger_mt &, std::atomic<bool> & stopped, uint64_t & batch_write_size, std::function<void (std::vector<std::shared_ptr<nano::block>> const &)> const & cemented_callback, std::function<void (nano::block_hash const &)> const & already_cemented_callback, std::function<uint64_t ()> const & awaiting_processing_size_query);
bool pending_empty () const; bool pending_empty () const;
void clear_process_vars (); void clear_process_vars ();
void process (std::shared_ptr<nano::block> original_block); void process (std::shared_ptr<nano::block> original_block);