Verify more state blocks if blocks deque is empty (#1450)

rai/node/node.cpp

@@ -1298,11 +1298,22 @@ bool rai::block_processor::have_blocks ()
 	return !blocks.empty () || !forced.empty () || !state_blocks.empty ();
 }
 
-void rai::block_processor::verify_state_blocks (std::unique_lock<std::mutex> & lock_a)
+void rai::block_processor::verify_state_blocks (std::unique_lock<std::mutex> & lock_a, size_t max_count)
 {
-	lock_a.lock ();
+	assert (!mutex.try_lock ());
 	std::deque<std::pair<std::shared_ptr<rai::block>, std::chrono::steady_clock::time_point>> items;
-	items.swap (state_blocks);
+	if (max_count == std::numeric_limits<size_t>::max () || max_count >= state_blocks.size ())
+	{
+		items.swap (state_blocks);
+	}
+	else
+	{
+		auto keep_size (state_blocks.size () - max_count);
+		items.resize (keep_size);
+		std::swap_ranges (state_blocks.end () - keep_size, state_blocks.end (), items.begin ());
+		state_blocks.resize (max_count);
+		items.swap (state_blocks);
+	}
 	lock_a.unlock ();
 	auto size (items.size ());
 	std::vector<rai::uint256_union> hashes;
@@ -1340,14 +1351,20 @@ void rai::block_processor::verify_state_blocks (std::unique_lock<std::mutex> & l
 		}
 		items.pop_front ();
 	}
-	lock_a.unlock ();
 }
 
 void rai::block_processor::process_receive_many (std::unique_lock<std::mutex> & lock_a)
 {
-	verify_state_blocks (lock_a);
+	lock_a.lock ();
-	auto transaction (node.store.tx_begin_write ());
 	auto start_time (std::chrono::steady_clock::now ());
+	// Limit state blocks verification time
+	while (!state_blocks.empty () && std::chrono::steady_clock::now () - start_time < std::chrono::seconds (2))
+	{
+		verify_state_blocks (lock_a, 2048);
+	}
+	lock_a.unlock ();
+	auto transaction (node.store.tx_begin_write ());
+	start_time = std::chrono::steady_clock::now ();
 	lock_a.lock ();
 	// Processing blocks
 	while ((!blocks.empty () || !forced.empty ()) && std::chrono::steady_clock::now () - start_time < node.config.block_processor_batch_max_time)
@@ -1388,6 +1405,12 @@ void rai::block_processor::process_receive_many (std::unique_lock<std::mutex> &
 		auto process_result (process_receive_one (transaction, block.first, block.second, validated_state_block));
 		(void)process_result;
 		lock_a.lock ();
+		/* Verify more state blocks if blocks deque is empty
+		Because verification is long process, avoid large deque verification inside of write transaction */
+		if (blocks.empty () && !state_blocks.empty ())
+		{
+			verify_state_blocks (lock_a, 256);
+		}
 	}
 	lock_a.unlock ();
 }

rai/node/node.hpp

@@ -429,7 +429,7 @@ public:
 
 private:
 	void queue_unchecked (rai::transaction const &, rai::block_hash const &);
-	void verify_state_blocks (std::unique_lock<std::mutex> &);
+	void verify_state_blocks (std::unique_lock<std::mutex> &, size_t = std::numeric_limits<size_t>::max ());
 	void process_receive_many (std::unique_lock<std::mutex> &);
 	bool stopped;
 	bool active;