Throttled removal of rep crawler targets (#3154)

* Throttled removal of rep crawler targets

* Remove recursion

nano/node/repcrawler.cpp

@@ -155,7 +155,8 @@ void nano::rep_crawler::query (std::vector<std::shared_ptr<nano::transport::chan
 	node.workers.add_timed_task (std::chrono::steady_clock::now () + std::chrono::seconds (5), [node_w, hash = hash_root.first]() {
 		if (auto node_l = node_w.lock ())
 		{
-			node_l->rep_crawler.remove (hash);
+			auto target_finished_processed (node_l->vote_processor.total_processed + node_l->vote_processor.size ());
+			node_l->rep_crawler.throttled_remove (hash, target_finished_processed);
 		}
 	});
 }
@@ -167,6 +168,24 @@ void nano::rep_crawler::query (std::shared_ptr<nano::transport::channel> const &
 	query (peers);
 }
 
+void nano::rep_crawler::throttled_remove (nano::block_hash const & hash_a, uint64_t const target_finished_processed)
+{
+	if (node.vote_processor.total_processed >= target_finished_processed)
+	{
+		remove (hash_a);
+	}
+	else
+	{
+		std::weak_ptr<nano::node> node_w (node.shared ());
+		node.workers.add_timed_task (std::chrono::steady_clock::now () + std::chrono::seconds (5), [node_w, hash_a, target_finished_processed]() {
+			if (auto node_l = node_w.lock ())
+			{
+				node_l->rep_crawler.throttled_remove (hash_a, target_finished_processed);
+			}
+		});
+	}
+}
+
 bool nano::rep_crawler::is_pr (nano::transport::channel const & channel_a) const
 {
 	nano::lock_guard<nano::mutex> lock (probable_reps_mutex);

nano/node/repcrawler.hpp

@@ -83,6 +83,9 @@ public:
 	/** Remove block hash from list of active rep queries */
 	void remove (nano::block_hash const &);
 
+	/** Remove block hash from with delay depending on vote processor size */
+	void throttled_remove (nano::block_hash const &, uint64_t const);
+
 	/** Attempt to determine if the peer manages one or more representative accounts */
 	void query (std::vector<std::shared_ptr<nano::transport::channel>> const & channels_a);
 

nano/node/vote_processor.cpp

@@ -76,6 +76,7 @@ void nano::vote_processor::process_loop ()
 
 			lock.unlock ();
 			condition.notify_all ();
+			total_processed += votes_l.size ();
 			lock.lock ();
 
 			if (log_this_iteration && elapsed.stop () > std::chrono::milliseconds (100))

nano/node/vote_processor.hpp

@@ -48,6 +48,7 @@ public:
 	bool half_full ();
 	void calculate_weights ();
 	void stop ();
+	std::atomic<uint64_t> total_processed{ 0 };
 
 private:
 	void process_loop ();