Guest User

Untitled

a guest
Jul 18th, 2012
567
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
text 3.92 KB | None | 0 0
  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