Create a named_thread with strand for single thread use of thread_pool

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

@@ -5,6 +5,7 @@
 #include <fc/log/logger_config.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/post.hpp>
+#include <boost/asio/strand.hpp>
 #include <future>
 #include <memory>
 #include <optional>
@@ -29,7 +30,7 @@ namespace eosio { namespace chain {
    public:
       using on_except_t = std::function<void(const fc::exception& e)>;
 
-      explicit named_thread_pool()
+      named_thread_pool()
             : _name_prefix(NamePrefix)
             , _ioc()
       {}
@@ -38,7 +39,7 @@ namespace eosio { namespace chain {
          stop();
       }
 
-      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.
@@ -109,6 +110,33 @@ namespace eosio { namespace chain {
       std::optional<ioc_work_t>      _ioc_work;
    };
 
+   template<uint64_t NamePrefix>
+   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.
+      named_thread() = default;
+
+      /// 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( typename named_thread_pool<NamePrefix>::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<NamePrefix> _pool;
+      decltype(boost::asio::make_strand(_pool.get_executor())) _strand{ boost::asio::make_strand( _pool.get_executor() ) };
+   };
 
    // async on io_context and return future
    template<typename F>

plugins/state_history_plugin/state_history_plugin.cpp

@@ -87,7 +87,8 @@ 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) {}
+      template <typename IOC>
+      generic_acceptor(IOC& ioc) : acceptor_(ioc), socket_(ioc), error_timer_(ioc) {}
       ACCEPTOR                    acceptor_;
       socket_type                 socket_;
       boost::asio::deadline_timer error_timer_;
@@ -99,7 +100,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<eosio::chain::make_name_v("ship")>                thread_pool;
+   named_thread<eosio::chain::make_name_v("ship")> ship_thread;
 
    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 +225,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 +406,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 +435,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 +454,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 +728,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 +745,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

@@ -1250,6 +1250,57 @@ BOOST_AUTO_TEST_CASE(named_thread_pool_test) {
    }
 }
 
+BOOST_AUTO_TEST_CASE(named_thread_test) {
+   {
+      named_thread<eosio::chain::make_name_v("misc")> thread;
+      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<eosio::chain::make_name_v("misc")> thread;
+
+      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<eosio::chain::make_name_v("misc")> thread;
+      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");