Used named_thread instead of named_thread_pool

libraries/chain/controller.cpp

@@ -301,7 +301,7 @@ struct controller_impl {
     conf( cfg ),
     chain_id( chain_id ),
     read_mode( cfg.read_mode ),
-    thread_pool( "chain" )
+    thread_pool( "chain"_n )
    {
       fork_db.open( [this]( block_timestamp_type timestamp,
                             const flat_set<digest_type>& cur_features,

libraries/chain/include/eosio/chain/thread_utils.hpp

@@ -1,7 +1,9 @@
 #pragma once
 
+#include <eosio/chain/name.hpp>
 #include <fc/exception/exception.hpp>
 #include <boost/asio/io_context.hpp>
+#include <boost/asio/strand.hpp>
 #include <boost/asio/post.hpp>
 #include <future>
 #include <memory>
@@ -21,12 +23,12 @@ namespace eosio { namespace chain {
       /// @param name_prefix is name appended with -## of thread.
       ///                    A short name_prefix (6 chars or under) is recommended as console_appender uses 9 chars
       ///                    for the thread name.
-      explicit named_thread_pool( std::string name_prefix );
+      explicit named_thread_pool( eosio::chain::name name_prefix );
 
       /// calls stop()
       ~named_thread_pool();
 
-      boost::asio::io_context& get_executor() { return _ioc; }
+      auto& get_executor() { return _ioc; }
 
       /// Spawn threads, can be re-started after stop().
       /// Assumes start()/stop() called from the same thread or externally protected.
@@ -42,12 +44,40 @@ namespace eosio { namespace chain {
    private:
       using ioc_work_t = boost::asio::executor_work_guard<boost::asio::io_context::executor_type>;
 
-      std::string                    _name_prefix;
+      eosio::chain::name             _name_prefix;
       boost::asio::io_context        _ioc;
       std::vector<std::thread>       _thread_pool;
       std::optional<ioc_work_t>      _ioc_work;
    };
 
+   class named_thread {
+   public:
+      /// @param name_prefix is name appended with -1 for thread name.
+      ///                    A short name_prefix (6 chars or under) is recommended as console_appender uses 9 chars
+      ///                    for the thread name.
+      explicit named_thread( eosio::chain::name name_prefix )
+      : _pool( name_prefix ) {}
+
+      /// stop before destroying strand
+      ~named_thread() { stop(); }
+
+      auto& get_executor() { return _strand; }
+
+      /// Spawn thread, can be re-started after stop().
+      /// Assumes start()/stop() called from the same thread or externally protected.
+      /// @param on_except is the function to call if io_context throws an exception, is called from thread pool thread.
+      ///                  if an empty function then logs and rethrows exception on thread which will terminate.
+      /// @throw assert_exception if already started and not stopped.
+      void start( named_thread_pool::on_except_t on_except ) { _pool.start( 1, std::move(on_except) ); }
+
+      /// destroy work guard, stop io_context, join thread_pool
+      void stop() { _pool.stop(); }
+
+   private:
+      named_thread_pool _pool;
+      boost::asio::io_context::strand _strand{_pool.get_executor()};
+   };
+
 
    // async on io_context and return future
    template<typename F>

libraries/chain/thread_utils.cpp

@@ -1,11 +1,12 @@
 #include <eosio/chain/thread_utils.hpp>
+#include <eosio/chain/name.hpp>
 #include <fc/log/logger_config.hpp>
 #include <fc/exception/exception.hpp>
 
 namespace eosio { namespace chain {
 
-named_thread_pool::named_thread_pool( std::string name_prefix )
-: _name_prefix( std::move(name_prefix) )
+named_thread_pool::named_thread_pool( eosio::chain::name name_prefix )
+: _name_prefix( name_prefix )
 , _ioc()
 {
 }
@@ -19,9 +20,11 @@ void named_thread_pool::start( size_t num_threads, on_except_t on_except ) {
    _ioc_work.emplace( boost::asio::make_work_guard( _ioc ) );
    _ioc.restart();
    _thread_pool.reserve( num_threads );
+
    for( size_t i = 0; i < num_threads; ++i ) {
-      _thread_pool.emplace_back( [&ioc = _ioc, &name_prefix = _name_prefix, on_except, i]() {
-         std::string tn = name_prefix + "-" + std::to_string( i );
+      // capture eosio::name so it is available in the stack for debugger, use abieos/tools/num2name to get string name
+      _thread_pool.emplace_back( [name_prefix = _name_prefix, &ioc = _ioc, on_except, i]() {
+         std::string tn = name_prefix.to_string() + "-" + std::to_string( i );
          try {
             fc::set_os_thread_name( tn );
             ioc.run();

plugins/chain_plugin/test/test_account_query_db.cpp

@@ -115,7 +115,7 @@ BOOST_FIXTURE_TEST_CASE(updateauth_test_multi_threaded, TESTER) { try {
    produce_block();
    create_account(tester_account);
 
-   named_thread_pool thread_pool( "test" );
+   named_thread_pool thread_pool( "test"_n );
    thread_pool.start( 5, {} );
 
    for( size_t i = 0; i < 100; ++i ) {

plugins/http_plugin/include/eosio/http_plugin/common.hpp

@@ -129,7 +129,7 @@ struct http_plugin_state {
    bool keep_alive = false;
 
    uint16_t thread_pool_size = 2;
-   eosio::chain::named_thread_pool thread_pool{ "http" };
+   eosio::chain::named_thread_pool thread_pool{ eosio::chain::name("http") };
 
    fc::logger& logger;
 

plugins/net_plugin/net_plugin.cpp

@@ -281,7 +281,7 @@ namespace eosio {
       compat::channels::transaction_ack::channel_type::handle  incoming_transaction_ack_subscription;
 
       uint16_t                              thread_pool_size = 4;
-      eosio::chain::named_thread_pool       thread_pool{ "net" };
+      eosio::chain::named_thread_pool       thread_pool{ "net"_n };
 
       boost::asio::deadline_timer           accept_error_timer{thread_pool.get_executor()};
 

plugins/producer_plugin/producer_plugin.cpp

@@ -307,7 +307,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       pending_block_mode                                        _pending_block_mode = pending_block_mode::speculating;
       unapplied_transaction_queue                               _unapplied_transactions;
       size_t                                                    _thread_pool_size = config::default_controller_thread_pool_size;
-      named_thread_pool                                         _thread_pool{ "prod" };
+      named_thread_pool                                         _thread_pool{ "prod"_n };
 
       std::atomic<int32_t>                                      _max_transaction_time_ms; // modified by app thread, read by net_plugin thread pool
       std::atomic<bool>                                         _received_block{false}; // modified by net_plugin thread pool and app thread

plugins/state_history_plugin/state_history_plugin.cpp

@@ -87,7 +87,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
    template <class ACCEPTOR>
    struct generic_acceptor  {
       using socket_type = typename ACCEPTOR::protocol_type::socket;
-      generic_acceptor(boost::asio::io_context& ioc) : acceptor_(ioc), socket_(ioc), error_timer_(ioc) {}
+      generic_acceptor(boost::asio::io_context::strand& s) : acceptor_(s), socket_(s), error_timer_(s) {}
       ACCEPTOR                    acceptor_;
       socket_type                 socket_;
       boost::asio::deadline_timer error_timer_;
@@ -99,7 +99,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
    using acceptor_type = std::variant<std::unique_ptr<tcp_acceptor>, std::unique_ptr<unix_acceptor>>;
    std::set<acceptor_type>          acceptors;
 
-   named_thread_pool                thread_pool{"SHiP"};
+   named_thread                     ship_thread{"ship"_n};
 
    static void get_log_entry(state_history_log& log, uint32_t block_num, std::optional<bytes>& result) {
       if (block_num < log.begin_block() || block_num >= log.end_block())
@@ -224,7 +224,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
 
       template <typename T>
       void send(T obj) {
-         boost::asio::post(this->plugin->thread_pool.get_executor(), [self = this->shared_from_this(), obj = std::move(obj) ]() mutable {
+         boost::asio::post(this->plugin->ship_thread.get_executor(), [self = this->shared_from_this(), obj = std::move(obj) ]() mutable {
             self->send_queue.emplace_back(fc::raw::pack(state_result{std::move(obj)}));
             self->send();
          });
@@ -405,7 +405,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
       }
 
       void close() override {
-         boost::asio::post(plugin->thread_pool.get_executor(), [self = this->shared_from_this()]() {
+         boost::asio::post(plugin->ship_thread.get_executor(), [self = this->shared_from_this()]() {
             self->close_i();
          });
       }
@@ -434,7 +434,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
          FC_THROW_EXCEPTION(plugin_exception, "unable to open listen socket");
       };
 
-      auto init_tcp_acceptor  = [&]() { acceptors.insert(std::make_unique<tcp_acceptor>(thread_pool.get_executor())); };
+      auto init_tcp_acceptor  = [&]() { acceptors.insert(std::make_unique<tcp_acceptor>(ship_thread.get_executor())); };
       auto init_unix_acceptor = [&]() {
          // take a sniff and see if anything is already listening at the given socket path, or if the socket path exists
          //  but nothing is listening
@@ -453,7 +453,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
                ec = test_ec;
          }
          check_ec("open");
-         acceptors.insert(std::make_unique<unix_acceptor>(thread_pool.get_executor()));
+         acceptors.insert(std::make_unique<unix_acceptor>(ship_thread.get_executor()));
       };
 
       // create and configure acceptors, can be both
@@ -727,8 +727,7 @@ void state_history_plugin::plugin_startup() {
          fc_ilog( _log, "Done storing initial state on startup" );
       }
       my->listen();
-      // use of executor assumes only one thread
-      my->thread_pool.start( 1, [](const fc::exception& e) {
+      my->ship_thread.start( [](const fc::exception& e) {
          fc_elog( _log, "Exception in SHiP thread pool, exiting: ${e}", ("e", e.to_detail_string()) );
          app().quit();
       } );
@@ -745,7 +744,7 @@ void state_history_plugin::plugin_shutdown() {
    my->stopping = true;
    my->trace_log->stop();
    my->chain_state_log->stop();
-   my->thread_pool.stop();
+   my->ship_thread.stop();
 }
 
 void state_history_plugin::handle_sighup() { fc::logger::update(logger_name, _log); }

unittests/misc_tests.cpp

@@ -910,7 +910,7 @@ BOOST_AUTO_TEST_CASE(transaction_metadata_test) { try {
       BOOST_CHECK_EQUAL(trx.id(), ptrx->id());
       BOOST_CHECK_EQUAL(trx.id(), ptrx2->id());
 
-      named_thread_pool thread_pool( "misc" );
+      named_thread_pool thread_pool( "misc"_n );
       thread_pool.start( 5, {} );
 
       auto fut = transaction_metadata::start_recover_keys( ptrx, thread_pool.get_executor(), test.control->get_chain_id(), fc::microseconds::maximum(), transaction_metadata::trx_type::input );
@@ -1202,7 +1202,7 @@ BOOST_AUTO_TEST_CASE(bad_alloc_test) {
 
 BOOST_AUTO_TEST_CASE(named_thread_pool_test) {
    {
-      named_thread_pool thread_pool( "misc" );
+      named_thread_pool thread_pool( "misc"_n );
       thread_pool.start( 5, {} );
 
       std::promise<void> p;
@@ -1213,7 +1213,7 @@ BOOST_AUTO_TEST_CASE(named_thread_pool_test) {
       BOOST_TEST( (f.wait_for( 100ms ) == std::future_status::ready) );
    }
    { // delayed start
-      named_thread_pool thread_pool( "misc" );
+      named_thread_pool thread_pool( "misc"_n );
 
       std::promise<void> p;
       auto f = p.get_future();
@@ -1227,7 +1227,7 @@ BOOST_AUTO_TEST_CASE(named_thread_pool_test) {
    { // exception
       std::promise<fc::exception> ep;
       auto ef = ep.get_future();
-      named_thread_pool thread_pool( "misc" );
+      named_thread_pool thread_pool( "misc"_n );
       thread_pool.start( 5, [&ep](const fc::exception& e) { ep.set_value(e); } );
 
       boost::asio::post( thread_pool.get_executor(), [](){
@@ -1250,6 +1250,56 @@ BOOST_AUTO_TEST_CASE(named_thread_pool_test) {
    }
 }
 
+BOOST_AUTO_TEST_CASE(named_thread_test) {
+   {
+      named_thread thread( "misc"_n );
+      thread.start( {} );
+
+      std::promise<void> p;
+      auto f = p.get_future();
+      boost::asio::post( thread.get_executor(), [&p](){
+         p.set_value();
+      });
+      BOOST_TEST( (f.wait_for( 100ms ) == std::future_status::ready) );
+   }
+   { // delayed start
+      named_thread thread( "misc"_n );
+
+      std::promise<void> p;
+      auto f = p.get_future();
+      boost::asio::post( thread.get_executor(), [&p](){
+         p.set_value();
+      });
+      BOOST_TEST( (f.wait_for( 10ms ) == std::future_status::timeout) );
+      thread.start( {} );
+      BOOST_TEST( (f.wait_for( 100ms ) == std::future_status::ready) );
+   }
+   { // exception
+      std::promise<fc::exception> ep;
+      auto ef = ep.get_future();
+      named_thread thread( "misc"_n );
+      thread.start( [&ep](const fc::exception& e) { ep.set_value(e); } );
+
+      boost::asio::post( thread.get_executor(), [](){
+         FC_ASSERT( false, "oops throw on thread" );
+      });
+      BOOST_TEST( (ef.wait_for( 100ms ) == std::future_status::ready) );
+      BOOST_TEST( ef.get().to_detail_string().find("oops throw on thread") != std::string::npos );
+
+      // we can restart, after a stop
+      BOOST_REQUIRE_THROW( thread.start( [&ep](const fc::exception& e) { ep.set_value(e); } ), fc::assert_exception );
+      thread.stop();
+
+      std::promise<void> p;
+      auto f = p.get_future();
+      boost::asio::post( thread.get_executor(), [&p](){
+         p.set_value();
+      });
+      thread.start( [&ep](const fc::exception& e) { ep.set_value(e); } );
+      BOOST_TEST( (f.wait_for( 100ms ) == std::future_status::ready) );
+   }
+}
+
 BOOST_AUTO_TEST_CASE(public_key_from_hash) {
    auto private_key_string = std::string("5KQwrPbwdL6PhXujxW37FSSQZ1JiwsST4cqQzDeyXtP79zkvFD3");
    auto expected_public_key = std::string("EOS6MRyAjQq8ud7hVNYcfnVPJqcVpscN5So8BhtHuGYqET5GDW5CV");