Use start/stop pattern in block_processor (#4462)
This commit is contained in:
parent
7561a796c4
commit
5c7b115c15
3 changed files with 28 additions and 11 deletions
|
|
@ -46,9 +46,21 @@ nano::block_processor::block_processor (nano::node & node_a, nano::write_databas
|
||||||
block_processed.notify (result, context);
|
block_processed.notify (result, context);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
processing_thread = std::thread ([this] () {
|
}
|
||||||
|
|
||||||
|
nano::block_processor::~block_processor ()
|
||||||
|
{
|
||||||
|
// Thread must be stopped before destruction
|
||||||
|
debug_assert (!thread.joinable ());
|
||||||
|
}
|
||||||
|
|
||||||
|
void nano::block_processor::start ()
|
||||||
|
{
|
||||||
|
debug_assert (!thread.joinable ());
|
||||||
|
|
||||||
|
thread = std::thread ([this] () {
|
||||||
nano::thread_role::set (nano::thread_role::name::block_processing);
|
nano::thread_role::set (nano::thread_role::name::block_processing);
|
||||||
this->process_blocks ();
|
run ();
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -59,7 +71,10 @@ void nano::block_processor::stop ()
|
||||||
stopped = true;
|
stopped = true;
|
||||||
}
|
}
|
||||||
condition.notify_all ();
|
condition.notify_all ();
|
||||||
nano::join_or_pass (processing_thread);
|
if (thread.joinable ())
|
||||||
|
{
|
||||||
|
thread.join ();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
std::size_t nano::block_processor::size ()
|
std::size_t nano::block_processor::size ()
|
||||||
|
|
@ -172,14 +187,13 @@ void nano::block_processor::rollback_competitor (store::write_transaction const
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void nano::block_processor::process_blocks ()
|
void nano::block_processor::run ()
|
||||||
{
|
{
|
||||||
nano::unique_lock<nano::mutex> lock{ mutex };
|
nano::unique_lock<nano::mutex> lock{ mutex };
|
||||||
while (!stopped)
|
while (!stopped)
|
||||||
{
|
{
|
||||||
if (have_blocks_ready ())
|
if (have_blocks_ready ())
|
||||||
{
|
{
|
||||||
active = true;
|
|
||||||
lock.unlock ();
|
lock.unlock ();
|
||||||
|
|
||||||
auto processed = process_batch (lock);
|
auto processed = process_batch (lock);
|
||||||
|
|
@ -194,7 +208,6 @@ void nano::block_processor::process_blocks ()
|
||||||
batch_processed.notify (processed);
|
batch_processed.notify (processed);
|
||||||
|
|
||||||
lock.lock ();
|
lock.lock ();
|
||||||
active = false;
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
|
|
||||||
|
|
@ -67,8 +67,11 @@ public: // Context
|
||||||
|
|
||||||
public:
|
public:
|
||||||
block_processor (nano::node &, nano::write_database_queue &);
|
block_processor (nano::node &, nano::write_database_queue &);
|
||||||
|
~block_processor ();
|
||||||
|
|
||||||
|
void start ();
|
||||||
void stop ();
|
void stop ();
|
||||||
|
|
||||||
std::size_t size ();
|
std::size_t size ();
|
||||||
bool full ();
|
bool full ();
|
||||||
bool half_full ();
|
bool half_full ();
|
||||||
|
|
@ -78,7 +81,7 @@ public:
|
||||||
bool should_log ();
|
bool should_log ();
|
||||||
bool have_blocks_ready ();
|
bool have_blocks_ready ();
|
||||||
bool have_blocks ();
|
bool have_blocks ();
|
||||||
void process_blocks ();
|
|
||||||
std::unique_ptr<container_info_component> collect_container_info (std::string const & name);
|
std::unique_ptr<container_info_component> collect_container_info (std::string const & name);
|
||||||
|
|
||||||
std::atomic<bool> flushing{ false };
|
std::atomic<bool> flushing{ false };
|
||||||
|
|
@ -93,6 +96,7 @@ public: // Events
|
||||||
nano::observer_set<std::shared_ptr<nano::block> const &> rolled_back;
|
nano::observer_set<std::shared_ptr<nano::block> const &> rolled_back;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
|
void run ();
|
||||||
// Roll back block in the ledger that conflicts with 'block'
|
// Roll back block in the ledger that conflicts with 'block'
|
||||||
void rollback_competitor (store::write_transaction const &, nano::block const & block);
|
void rollback_competitor (store::write_transaction const &, nano::block const & block);
|
||||||
nano::block_status process_one (store::write_transaction const &, context const &, bool forced = false);
|
nano::block_status process_one (store::write_transaction const &, context const &, bool forced = false);
|
||||||
|
|
@ -106,15 +110,14 @@ private: // Dependencies
|
||||||
nano::write_database_queue & write_database_queue;
|
nano::write_database_queue & write_database_queue;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
bool stopped{ false };
|
|
||||||
bool active{ false };
|
|
||||||
|
|
||||||
std::deque<context> blocks;
|
std::deque<context> blocks;
|
||||||
std::deque<context> forced;
|
std::deque<context> forced;
|
||||||
|
|
||||||
std::chrono::steady_clock::time_point next_log;
|
std::chrono::steady_clock::time_point next_log;
|
||||||
|
|
||||||
|
bool stopped{ false };
|
||||||
nano::condition_variable condition;
|
nano::condition_variable condition;
|
||||||
nano::mutex mutex{ mutex_identifier (mutexes::block_processor) };
|
nano::mutex mutex{ mutex_identifier (mutexes::block_processor) };
|
||||||
std::thread processing_thread;
|
std::thread thread;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -635,6 +635,7 @@ void nano::node::start ()
|
||||||
}
|
}
|
||||||
wallets.start ();
|
wallets.start ();
|
||||||
vote_processor.start ();
|
vote_processor.start ();
|
||||||
|
block_processor.start ();
|
||||||
active.start ();
|
active.start ();
|
||||||
generator.start ();
|
generator.start ();
|
||||||
final_generator.start ();
|
final_generator.start ();
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue