Add worker queued task queue to the stats (#3101)

nano/lib/threading.cpp

@@ -228,10 +228,14 @@ void nano::thread_pool::stop ()
 
 void nano::thread_pool::push_task (std::function<void()> task)
 {
+	++num_tasks;
 	nano::lock_guard<std::mutex> guard (mutex);
 	if (!stopped)
 	{
-		boost::asio::post (*thread_pool_m, task);
+		boost::asio::post (*thread_pool_m, [this, task]() {
+			task ();
+			--num_tasks;
+		});
 	}
 }
 
@@ -255,6 +259,11 @@ unsigned nano::thread_pool::get_num_threads () const
 	return num_threads;
 }
 
+uint64_t nano::thread_pool::num_queued_tasks () const
+{
+	return num_tasks;
+}
+
 // Set the names of all the threads in the thread pool for easier identification
 void nano::thread_pool::set_thread_names (unsigned num_threads, nano::thread_role::name thread_name)
 {
@@ -279,3 +288,10 @@ void nano::thread_pool::set_thread_names (unsigned num_threads, nano::thread_rol
 		future.wait ();
 	}
 }
+
+std::unique_ptr<nano::container_info_component> nano::collect_container_info (thread_pool & thread_pool, std::string const & name)
+{
+	auto composite = std::make_unique<container_info_composite> (name);
+	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "count", thread_pool.num_queued_tasks (), sizeof (std::function<void()>) }));
+	return composite;
+}

nano/lib/threading.hpp

@@ -180,12 +180,18 @@ public:
 	/** Number of threads in the thread pool */
 	unsigned get_num_threads () const;
 
+	/** Returns the number of tasks which are awaiting execution by the thread pool **/
+	uint64_t num_queued_tasks () const;
+
 private:
 	std::mutex mutex;
 	std::atomic<bool> stopped{ false };
 	unsigned num_threads;
 	std::unique_ptr<boost::asio::thread_pool> thread_pool_m;
+	relaxed_atomic_integral<uint64_t> num_tasks{ 0 };
 
 	void set_thread_names (unsigned num_threads, nano::thread_role::name thread_name);
 };
+
+std::unique_ptr<nano::container_info_component> collect_container_info (thread_pool & thread_pool, std::string const & name);
 }

nano/node/node.cpp

@@ -90,7 +90,7 @@ io_ctx (io_ctx_a),
 node_initialized_latch (1),
 config (config_a),
 stats (config.stat_config),
-workers (std::max (3u, config.io_threads / 2), nano::thread_role::name::worker),
+workers (std::max (3u, config.io_threads / 4), nano::thread_role::name::worker),
 flags (flags_a),
 work (work_a),
 distributed_work (*this),
@@ -588,6 +588,7 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (no
 	{
 		composite->add_component (collect_container_info (*node.telemetry, "telemetry"));
 	}
+	composite->add_component (collect_container_info (node.workers, "workers"));
 	composite->add_component (collect_container_info (node.observers, "observers"));
 	composite->add_component (collect_container_info (node.wallets, "wallets"));
 	composite->add_component (collect_container_info (node.vote_processor, "vote_processor"));