Sim

Untitled

Sim
Mar 9th, 2013
347
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. Example of dynamic thread pool in boost::asio
  2. #include <boost/asio.hpp>
  3. #include <boost/bind.hpp>
  4. #include <boost/thread.hpp>
  5. #include <iostream>
  6.  
  7. class thread_pool_checker
  8.   : private boost::noncopyable
  9. {
  10. public:
  11.  
  12.   thread_pool_checker( boost::asio::io_service& io_service,
  13.                        boost::thread_group& threads,
  14.                        unsigned int max_threads,
  15.                        long threshold_seconds,
  16.                        long periodic_seconds )
  17.     : io_service_( io_service ),
  18.       timer_( io_service ),
  19.       threads_( threads ),
  20.       max_threads_( max_threads ),
  21.       threshold_seconds_( threshold_seconds ),
  22.       periodic_seconds_( periodic_seconds )
  23.     {
  24.       schedule_check();
  25.     }
  26.  
  27. private:
  28.  
  29.   void schedule_check();
  30.   void on_check( const boost::system::error_code& error );
  31.  
  32. private:
  33.  
  34.   boost::asio::io_service&    io_service_;
  35.   boost::asio::deadline_timer timer_;
  36.   boost::thread_group&        threads_;
  37.   unsigned int                max_threads_;
  38.   long                        threshold_seconds_;
  39.   long                        periodic_seconds_;
  40. };
  41.  
  42. void thread_pool_checker::schedule_check()
  43. {
  44.   // Thread pool is already at max size.
  45.   if ( max_threads_ <= threads_.size() )
  46.   {
  47.     std::cout << "Thread pool has reached its max.  Example will shutdown."
  48.               << std::endl;
  49.     io_service_.stop();
  50.     return;
  51.   }
  52.  
  53.   // Schedule check to see if pool needs to increase.
  54.   std::cout << "Will check if pool needs to increase in "
  55.             << periodic_seconds_ << " seconds." << std::endl;
  56.   timer_.expires_from_now( boost::posix_time::seconds( periodic_seconds_ ) );
  57.   timer_.async_wait(
  58.     boost::bind( &thread_pool_checker::on_check, this,
  59.                  boost::asio::placeholders::error ) );
  60. }
  61.  
  62. void thread_pool_checker::on_check( const boost::system::error_code& error )
  63. {
  64.   // On error, return early.
  65.   if ( error ) return;
  66.  
  67.   // Check how long this job was waiting in the service queue.  This
  68.   // returns the expiration time relative to now.  Thus, if it expired
  69.   // 7 seconds ago, then the delta time is -7 seconds.
  70.   boost::posix_time::time_duration delta = timer_.expires_from_now();
  71.   long wait_in_seconds = -delta.seconds();
  72.  
  73.   // If the time delta is greater than the threshold, then the job
  74.   // remained in the service queue for too long, so increase the
  75.   // thread pool.
  76.   std::cout << "Job job sat in queue for "
  77.             << wait_in_seconds << " seconds." << std::endl;
  78.   if ( threshold_seconds_ < wait_in_seconds )
  79.   {
  80.     std::cout << "Increasing thread pool." << std::endl;
  81.     threads_.create_thread(
  82.       boost::bind( &boost::asio::io_service::run,
  83.                    &io_service_ ) );
  84.   }
  85.  
  86.   // Otherwise, schedule another pool check.
  87.   run();
  88. }
  89.  
  90. // Busy work functions.
  91. void busy_work( boost::asio::io_service&,
  92.                 unsigned int );
  93.  
  94. void add_busy_work( boost::asio::io_service& io_service,
  95.                     unsigned int count )
  96. {
  97.   io_service.post(
  98.     boost::bind( busy_work,
  99.                  boost::ref( io_service ),
  100.                  count ) );
  101. }
  102.  
  103. void busy_work( boost::asio::io_service& io_service,
  104.                 unsigned int count )
  105. {
  106.   boost::this_thread::sleep( boost::posix_time::seconds( 5 ) );
  107.  
  108.   count += 1;
  109.  
  110.   // When the count is 3, spawn additional busy work.
  111.   if ( 3 == count )
  112.   {
  113.     add_busy_work( io_service, 0 );
  114.   }
  115.   add_busy_work( io_service, count );
  116. }
  117.  
  118. int main()
  119. {
  120.   using boost::asio::ip::tcp;
  121.  
  122.   // Create io service.
  123.   boost::asio::io_service io_service;
  124.  
  125.   // Add some busy work to the service.
  126.   add_busy_work( io_service, 0 );
  127.  
  128.   // Create thread group and thread_pool_checker.
  129.   boost::thread_group threads;
  130.   thread_pool_checker checker( io_service, threads,
  131.                                3,   // Max pool size.
  132.                                2,   // Create thread if job waits for 2 sec.
  133.                                3 ); // Check if pool needs to grow every 3 sec.
  134.  
  135.   // Start running the io service.
  136.   io_service.run();
  137.  
  138.   threads.join_all();
  139.  
  140.   return 0;
  141. }
Advertisement
Add Comment
Please, Sign In to add comment