include/fc/asio.hpp

@@ -5,6 +5,8 @@
 #pragma once
 #include <boost/asio.hpp>
 #include <boost/bind.hpp>
+#include <boost/thread.hpp>
+#include <vector>
 #include <fc/thread/future.hpp>
 #include <fc/io/iostream.hpp>
 
@@ -65,7 +67,26 @@ namespace asio {
             bool operator()( C&, bool ) { return false; } 
         };
         #endif 
-    }
+    } // end of namespace detail
+
+    /***
+     * A structure for holding the boost io service and associated
+     * threads
+     */
+    class default_io_service_scope
+    {
+       public:
+          default_io_service_scope();
+          ~default_io_service_scope();
+          static void set_num_threads(uint16_t num_threads);
+          boost::asio::io_service*          io;
+       private:
+          std::vector<boost::thread*>       asio_threads;
+          boost::asio::io_service::work*    the_work;
+       protected:
+          static uint16_t num_io_threads; // marked protected to help with testing
+    };
+
     /**
      * @return the default boost::asio::io_service for use with fc::asio
      * 

src/asio.cpp

@@ -4,6 +4,8 @@
 #include <fc/log/logger.hpp>
 #include <fc/exception/exception.hpp>
 #include <boost/scope_exit.hpp>
+#include <algorithm>
+#include <thread>
 
 namespace fc {
   namespace asio {
@@ -91,63 +93,88 @@ namespace fc {
         }
     }
 
-    struct default_io_service_scope
-    {
-       boost::asio::io_service*          io;
-       std::vector<boost::thread*>       asio_threads;
-       boost::asio::io_service::work*    the_work;
+    uint16_t fc::asio::default_io_service_scope::num_io_threads = 0;
 
-       default_io_service_scope()
+    /***
+     * @brief set the default number of threads for the io service
+     *
+     * Sets the number of threads for the io service. This will throw
+     * an exception if called more than once.
+     *
+     * @param num_threads the number of threads
+     */
+    void default_io_service_scope::set_num_threads(uint16_t num_threads) {
+       FC_ASSERT(fc::asio::default_io_service_scope::num_io_threads == 0);
+       fc::asio::default_io_service_scope::num_io_threads = num_threads;
+    }
+
+    /***
+     * Default constructor
+     */
+    default_io_service_scope::default_io_service_scope()
+    {
+       io           = new boost::asio::io_service();
+       the_work     = new boost::asio::io_service::work(*io);
+
+       if (this->num_io_threads == 0)
        {
-            io           = new boost::asio::io_service();
-            the_work     = new boost::asio::io_service::work(*io);
-            for( int i = 0; i < 8; ++i ) {
-               asio_threads.push_back( new boost::thread( [=]()
-               {
+          // the default was not set by the configuration. Determine a good
+          // number of threads. Minimum of 8, maximum of hardware_concurrency
+          this->num_io_threads = std::max( boost::thread::hardware_concurrency(), 8u );
+       }
+
+       for( uint16_t i = 0; i < this->num_io_threads; ++i )
+       {
+          asio_threads.push_back( new boost::thread( [=]()
+                {
                  fc::thread::current().set_name("asio");
                  
                  BOOST_SCOPE_EXIT(void)
                  {
-                   fc::thread::cleanup();
+                    fc::thread::cleanup();
                  } 
                  BOOST_SCOPE_EXIT_END
 
                  while (!io->stopped())
                  {
-                   try
-                   {
-                     io->run();
-                   }
-                   catch (const fc::exception& e)
-                   {
-                     elog("Caught unhandled exception in asio service loop: ${e}", ("e", e));
-                   }
-                   catch (const std::exception& e)
-                   {
-                     elog("Caught unhandled exception in asio service loop: ${e}", ("e", e.what()));
-                   }
-                   catch (...)
-                   {
-                     elog("Caught unhandled exception in asio service loop");
-                   }
+                    try
+                    {
+                       io->run();
+                    }
+                    catch (const fc::exception& e)
+                    {
+                       elog("Caught unhandled exception in asio service loop: ${e}", ("e", e));
+                    }
+                    catch (const std::exception& e)
+                    {
+                       elog("Caught unhandled exception in asio service loop: ${e}", ("e", e.what()));
+                    }
+                    catch (...)
+                    {
+                       elog("Caught unhandled exception in asio service loop");
+                    }
                  }
-               }) );
-            }
-       }
+                }) );
+       } // build thread loop
+    } // end of constructor
 
-       ~default_io_service_scope()
+    /***
+     * destructor
+     */
+    default_io_service_scope::~default_io_service_scope()
+    {
+       delete the_work;
+       io->stop();
+       for( auto asio_thread : asio_threads )
        {
-          delete the_work;
-          io->stop();
-          for( auto asio_thread : asio_threads ) {
-             asio_thread->join();
-          }
-          delete io;
-          for( auto asio_thread : asio_threads ) {
-             delete asio_thread;
-          }
+          asio_thread->join();
        }
-    };
+       delete io;
+       for( auto asio_thread : asio_threads )
+       {
+          delete asio_thread;
+       }
+    } // end of destructor
 
     /***
      * @brief create an io_service

tests/io/tcp_test.cpp

@@ -1,6 +1,7 @@
 #include <boost/test/unit_test.hpp>
 
 #include <fc/network/tcp_socket.hpp>
+#include <fc/asio.hpp>
 
 BOOST_AUTO_TEST_SUITE(tcp_tests)
 
@@ -14,4 +15,44 @@ BOOST_AUTO_TEST_CASE(tcpconstructor_test)
    fc::tcp_socket socket;
 }
 
+class my_io_class : public fc::asio::default_io_service_scope
+{
+public:
+   uint16_t get_num_threads()
+   {
+      return fc::asio::default_io_service_scope::num_io_threads;
+   }
+   static void reset_num_threads() { fc::asio::default_io_service_scope::num_io_threads = 0; }
+};
+
+/***
+ * Test the control of number of threads from outside
+ */
+BOOST_AUTO_TEST_CASE( number_threads_test )
+{
+   // to erase leftovers from previous tests
+   my_io_class::reset_num_threads();
+
+   fc::asio::default_io_service_scope::set_num_threads(12);
+
+   my_io_class my_class;
+
+   BOOST_CHECK_EQUAL( 12, my_class.get_num_threads() );
+}
+
+/***
+ * Test the control of number of threads from outside
+ */
+BOOST_AUTO_TEST_CASE( default_number_threads_test )
+{
+   // to erase leftovers from previous tests
+   my_io_class::reset_num_threads();
+
+   my_io_class my_class;
+
+   fc::asio::default_io_service();
+
+   BOOST_CHECK( my_class.get_num_threads() > 1 );
+}
+
 BOOST_AUTO_TEST_SUITE_END()