Name application threads

libraries/chain/CMakeLists.txt

@@ -44,9 +44,9 @@ add_library( eosio_chain
 #             global_property_object.cpp
 #
 #             contracts/chain_initializer.cpp
-
-
+
              transaction_metadata.cpp
+             thread_utils.cpp
              ${HEADERS}
              )
 

libraries/chain/controller.cpp

@@ -21,6 +21,7 @@
 
 #include <chainbase/chainbase.hpp>
 #include <fc/io/json.hpp>
+#include <fc/log/logger_config.hpp>
 #include <fc/scoped_exit.hpp>
 #include <fc/variant_object.hpp>
 
@@ -132,7 +133,7 @@ struct controller_impl {
    optional<fc::microseconds>     subjective_cpu_leeway;
    bool                           trusted_producer_light_validation = false;
    uint32_t                       snapshot_head_block = 0;
-   boost::asio::thread_pool       thread_pool;
+   named_thread_pool              thread_pool;
 
    typedef pair<scope_name,action_name>                   handler_key;
    map< account_name, map<handler_key, apply_handler> >   apply_handlers;
@@ -184,7 +185,7 @@ struct controller_impl {
     conf( cfg ),
     chain_id( cfg.genesis.compute_chain_id() ),
     read_mode( cfg.read_mode ),
-    thread_pool( cfg.thread_pool_size )
+    thread_pool( "chain", cfg.thread_pool_size )
    {
 
 #define SET_APP_HANDLER( receiver, contract, action) \
@@ -403,6 +404,7 @@ struct controller_impl {
    }
 
    ~controller_impl() {
+      thread_pool.stop();
       pending.reset();
    }
 
@@ -1195,7 +1197,7 @@ struct controller_impl {
                auto& pt = receipt.trx.get<packed_transaction>();
                auto mtrx = std::make_shared<transaction_metadata>( std::make_shared<packed_transaction>( pt ) );
                if( !self.skip_auth_check() ) {
-                  transaction_metadata::start_recover_keys( mtrx, thread_pool, chain_id, microseconds::maximum() );
+                  transaction_metadata::start_recover_keys( mtrx, thread_pool.get_executor(), chain_id, microseconds::maximum() );
                }
                packed_transactions.emplace_back( std::move( mtrx ) );
             }
@@ -1273,7 +1275,7 @@ struct controller_impl {
       auto prev = fork_db.get_block( b->previous );
       EOS_ASSERT( prev, unlinkable_block_exception, "unlinkable block ${id}", ("id", id)("previous", b->previous) );
 
-      return async_thread_pool( thread_pool, [b, prev]() {
+      return async_thread_pool( thread_pool.get_executor(), [b, prev]() {
          const bool skip_validate_signee = false;
          return std::make_shared<block_state>( *prev, move( b ), skip_validate_signee );
       } );
@@ -1780,8 +1782,8 @@ void controller::abort_block() {
    my->abort_block();
 }
 
-boost::asio::thread_pool& controller::get_thread_pool() {
-   return my->thread_pool;
+boost::asio::io_context& controller::get_thread_pool() {
+   return my->thread_pool.get_executor();
 }
 
 std::future<block_state_ptr> controller::create_block_state_future( const signed_block_ptr& b ) {

libraries/chain/include/eosio/chain/controller.hpp

@@ -140,7 +140,7 @@ namespace eosio { namespace chain {
          std::future<block_state_ptr> create_block_state_future( const signed_block_ptr& b );
          void push_block( std::future<block_state_ptr>& block_state_future );
 
-         boost::asio::thread_pool& get_thread_pool();
+         boost::asio::io_context& get_thread_pool();
 
          const chainbase::database& db()const;
 

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

@@ -4,16 +4,45 @@
  */
 #pragma once
 
+#include <fc/optional.hpp>
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/thread_pool.hpp>
 #include <boost/asio/post.hpp>
 #include <future>
 #include <memory>
 
 namespace eosio { namespace chain {
 
+   /**
+    * Wrapper class for boost asio thread pool and io_context run.
+    * Also names threads so that tools like htop can see thread name.
+    */
+   class named_thread_pool {
+   public:
+      // name_prefix is name appended with -## of thread.
+      // short name_prefix (6 chars or under) is recommended as console_appender uses 9 chars for thread name
+      named_thread_pool( std::string name_prefix, size_t num_threads );
+
+      // calls stop()
+      ~named_thread_pool();
+
+      boost::asio::io_context& get_executor() { return _ioc; }
+
+      // destroy work guard, stop io_context, join thread_pool, and stop thread_pool
+      void stop();
+
+   private:
+      using ioc_work_t = boost::asio::executor_work_guard<boost::asio::io_context::executor_type>;
+
+      boost::asio::thread_pool       _thread_pool;
+      boost::asio::io_context        _ioc;
+      fc::optional<ioc_work_t>       _ioc_work;
+   };
+
+
    // async on thread_pool and return future
    template<typename F>
-   auto async_thread_pool( boost::asio::thread_pool& thread_pool, F&& f ) {
+   auto async_thread_pool( boost::asio::io_context& thread_pool, F&& f ) {
       auto task = std::make_shared<std::packaged_task<decltype( f() )()>>( std::forward<F>( f ) );
       boost::asio::post( thread_pool, [task]() { (*task)(); } );
       return task->get_future();

libraries/chain/include/eosio/chain/transaction_metadata.hpp

@@ -53,13 +53,11 @@ class transaction_metadata {
 
       // must be called from main application thread
       static signing_keys_future_type
-      start_recover_keys( const transaction_metadata_ptr& mtrx, boost::asio::thread_pool& thread_pool,
+      start_recover_keys( const transaction_metadata_ptr& mtrx, boost::asio::io_context& thread_pool,
                           const chain_id_type& chain_id, fc::microseconds time_limit );
 
       // start_recover_keys must be called first
       recovery_keys_type recover_keys( const chain_id_type& chain_id );
-
-
 };
 
 } } // eosio::chain

libraries/chain/thread_utils.cpp

@@ -0,0 +1,40 @@
+/**
+ *  @file
+ *  @copyright defined in eos/LICENSE.txt
+ */
+
+#include <eosio/chain/thread_utils.hpp>
+#include <fc/log/logger_config.hpp>
+
+namespace eosio { namespace chain {
+
+
+//
+// named_thread_pool
+//
+named_thread_pool::named_thread_pool( std::string name_prefix, size_t num_threads )
+: _thread_pool( num_threads )
+{
+   _ioc_work.emplace( boost::asio::make_work_guard( _ioc ) );
+   for( size_t i = 0; i < num_threads; ++i ) {
+      boost::asio::post( _thread_pool, [&ioc = _ioc, name_prefix, i]() {
+         std::string tn = name_prefix + "-" + std::to_string( i );
+         fc::set_os_thread_name( tn );
+         ioc.run();
+      } );
+   }
+}
+
+named_thread_pool::~named_thread_pool() {
+   stop();
+}
+
+void named_thread_pool::stop() {
+   _ioc_work.reset();
+   _ioc.stop();
+   _thread_pool.join();
+   _thread_pool.stop();
+}
+
+
+} } // eosio::chain

libraries/chain/transaction_metadata.cpp

@@ -26,7 +26,7 @@ recovery_keys_type transaction_metadata::recover_keys( const chain_id_type& chai
 }
 
 signing_keys_future_type transaction_metadata::start_recover_keys( const transaction_metadata_ptr& mtrx,
-                                                                   boost::asio::thread_pool& thread_pool,
+                                                                   boost::asio::io_context& thread_pool,
                                                                    const chain_id_type& chain_id,
                                                                    fc::microseconds time_limit )
 {

plugins/bnet_plugin/bnet_plugin.cpp

@@ -51,6 +51,7 @@
 #include <eosio/chain_plugin/chain_plugin.hpp>
 
 #include <fc/io/json.hpp>
+#include <fc/log/logger_config.hpp>
 
 #include <boost/beast/core.hpp>
 #include <boost/beast/websocket.hpp>
@@ -1398,7 +1399,13 @@ namespace eosio {
 
       my->_socket_threads.reserve( my->_num_threads );
       for( auto i = 0; i < my->_num_threads; ++i ) {
-         my->_socket_threads.emplace_back( [&ioc]{ wlog( "start thread" ); ioc.run(); wlog( "end thread" ); } );
+         my->_socket_threads.emplace_back( [&ioc, i]{
+            std::string tn = "bnet-" + std::to_string( i );
+            fc::set_os_thread_name( tn );
+            wlog( "start thread" );
+            ioc.run();
+            wlog( "end thread" );
+         } );
       }
 
       for( const auto& peer : my->_connect_to_peers ) {

plugins/chain_plugin/chain_plugin.cpp

@@ -770,8 +770,6 @@ void chain_plugin::plugin_shutdown() {
    my->irreversible_block_connection.reset();
    my->accepted_transaction_connection.reset();
    my->applied_transaction_connection.reset();
-   my->chain->get_thread_pool().stop();
-   my->chain->get_thread_pool().join();
    my->chain.reset();
 }
 

plugins/http_plugin/http_plugin.cpp

@@ -5,6 +5,7 @@
 #include <eosio/http_plugin/http_plugin.hpp>
 #include <eosio/http_plugin/local_endpoint.hpp>
 #include <eosio/chain/exceptions.hpp>
+#include <eosio/chain/thread_utils.hpp>
 
 #include <fc/network/ip.hpp>
 #include <fc/log/logger_config.hpp>
@@ -123,7 +124,6 @@ namespace eosio {
    using websocket_local_server_type = websocketpp::server<detail::asio_local_with_stub_log>;
    using websocket_server_tls_type =  websocketpp::server<detail::asio_with_stub_log<websocketpp::transport::asio::tls_socket::endpoint>>;
    using ssl_context_ptr =  websocketpp::lib::shared_ptr<websocketpp::lib::asio::ssl::context>;
-   using io_work_t = boost::asio::executor_work_guard<boost::asio::io_context::executor_type>;
 
    static bool verbose_http_errors = false;
 
@@ -140,9 +140,7 @@ namespace eosio {
          websocket_server_type    server;
 
          uint16_t                                    thread_pool_size = 2;
-         optional<boost::asio::thread_pool>          thread_pool;
-         std::shared_ptr<boost::asio::io_context>    server_ioc;
-         optional<io_work_t>                         server_ioc_work;
+         optional<eosio::chain::named_thread_pool>   thread_pool;
          std::atomic<int64_t>                        bytes_in_flight{0};
          size_t                                      max_bytes_in_flight = 0;
 
@@ -301,12 +299,12 @@ namespace eosio {
                   con->defer_http_response();
                   bytes_in_flight += body.size();
                   app().post( appbase::priority::low,
-                              [ioc = this->server_ioc, &bytes_in_flight = this->bytes_in_flight, handler_itr,
+                              [&ioc = thread_pool->get_executor(), &bytes_in_flight = this->bytes_in_flight, handler_itr,
                                resource{std::move( resource )}, body{std::move( body )}, con]() {
                      try {
                         handler_itr->second( resource, body,
-                              [ioc{std::move(ioc)}, &bytes_in_flight, con]( int code, fc::variant response_body ) {
-                           boost::asio::post( *ioc, [ioc, response_body{std::move( response_body )}, &bytes_in_flight, con, code]() mutable {
+                              [&ioc, &bytes_in_flight, con]( int code, fc::variant response_body ) {
+                           boost::asio::post( ioc, [response_body{std::move( response_body )}, &bytes_in_flight, con, code]() mutable {
                               std::string json = fc::json::to_string( response_body );
                               response_body.clear();
                               const size_t json_size = json.size();
@@ -340,11 +338,11 @@ namespace eosio {
          void create_server_for_endpoint(const tcp::endpoint& ep, websocketpp::server<detail::asio_with_stub_log<T>>& ws) {
             try {
                ws.clear_access_channels(websocketpp::log::alevel::all);
-               ws.init_asio(&(*server_ioc));
+               ws.init_asio( &thread_pool->get_executor() );
                ws.set_reuse_addr(true);
                ws.set_max_http_body_size(max_body_size);
                // capture server_ioc shared_ptr in http handler to keep it alive while in use
-               ws.set_http_handler([&, ioc = this->server_ioc](connection_hdl hdl) {
+               ws.set_http_handler([&](connection_hdl hdl) {
                   handle_http_request<detail::asio_with_stub_log<T>>(ws.get_con_from_hdl(hdl));
                });
             } catch ( const fc::exception& e ){
@@ -518,12 +516,7 @@ namespace eosio {
 
    void http_plugin::plugin_startup() {
 
-      my->thread_pool.emplace( my->thread_pool_size );
-      my->server_ioc = std::make_shared<boost::asio::io_context>();
-      my->server_ioc_work.emplace( boost::asio::make_work_guard(*my->server_ioc) );
-      for( uint16_t i = 0; i < my->thread_pool_size; ++i ) {
-         boost::asio::post( *my->thread_pool, [ioc = my->server_ioc]() { ioc->run(); } );
-      }
+      my->thread_pool.emplace( "http", my->thread_pool_size );
 
       if(my->listen_endpoint) {
          try {
@@ -547,10 +540,10 @@ namespace eosio {
       if(my->unix_endpoint) {
          try {
             my->unix_server.clear_access_channels(websocketpp::log::alevel::all);
-            my->unix_server.init_asio(&(*my->server_ioc));
+            my->unix_server.init_asio( &my->thread_pool->get_executor() );
             my->unix_server.set_max_http_body_size(my->max_body_size);
             my->unix_server.listen(*my->unix_endpoint);
-            my->unix_server.set_http_handler([&, ioc = my->server_ioc](connection_hdl hdl) {
+            my->unix_server.set_http_handler([&, &ioc = my->thread_pool->get_executor()](connection_hdl hdl) {
                my->handle_http_request<detail::asio_local_with_stub_log>( my->unix_server.get_con_from_hdl(hdl));
             });
             my->unix_server.start_accept();
@@ -610,12 +603,7 @@ namespace eosio {
       if(my->unix_server.is_listening())
          my->unix_server.stop_listening();
 
-      if( my->server_ioc_work )
-         my->server_ioc_work->reset();
-      if( my->server_ioc )
-         my->server_ioc->stop();
       if( my->thread_pool ) {
-         my->thread_pool->join();
          my->thread_pool->stop();
       }
    }

plugins/mongo_db_plugin/mongo_db_plugin.cpp

@@ -10,6 +10,7 @@
 #include <eosio/chain/types.hpp>
 
 #include <fc/io/json.hpp>
+#include <fc/log/logger_config.hpp>
 #include <fc/utf8.hpp>
 #include <fc/variant.hpp>
 
@@ -1527,7 +1528,10 @@ void mongo_db_plugin_impl::init() {
 
    ilog("starting db plugin thread");
 
-   consume_thread = std::thread([this] { consume_blocks(); });
+   consume_thread = std::thread( [this] {
+      fc::set_os_thread_name( "mongodb" );
+      consume_blocks();
+   } );
 
    startup = false;
 }