织梦CMS - 轻松建站从此开始!

罗索实验室

当前位置: 主页 > 基础技术 > 高性能服务器 >

boost.asio服务器使用io_service作为work pool

jackyhwei 发布于 2010-12-02 09:37 点击:次 
无论如何使用,都能感觉到使用boost.asio实现服务器,不仅是一件非常轻松的事,而且代码很漂亮,逻辑也相当清晰,这点上很不同于ACE。
TAG:

使用io_service作为处理工作的work pool,可以看到,就是通过io_service.post投递一个Handler到io_service的队列,Handler在这个io_service.run内部得到执行,有可能你会发现,io_services.dispatch的接口也和io_service.post一样,但不同的是它是直接调用而不是经过push到队列然后在io_services.run中执行,而在这个示例当中,显然我们需要把工作交到另一个线程去完成,这样才不会影响网络接收线程池的工作以达到高效率的接收数据,这种设计与前面的netsever其实相同,这就是典型的Half Sync/Half Async。二者的区别就是netsever自己实现了工作队列,而不是直接使用io_service,这种设计实际上在win下是使用了iocp作为工作队列。

不过我更倾向于前一种设计,因为那样做,代码一切都在自己的掌握中,而io_service则是经过许多封装代码,并且本身设计只是用于处理网络完成事件的。

无论如何使用,都能感觉到使用boost.asio实现服务器,不仅是一件非常轻松的事,而且代码很漂亮,逻辑也相当清晰,这点上很不同于ACE。

  1. #include <stdio.h> 
  2. #include <cstdlib> 
  3. #include <iostream> 
  4. #include <boost/thread.hpp> 
  5. #include <boost/aligned_storage.hpp> 
  6. #include <boost/array.hpp> 
  7. #include <boost/bind.hpp> 
  8. #include <boost/enable_shared_from_this.hpp> 
  9. #include <boost/noncopyable.hpp> 
  10. #include <boost/shared_ptr.hpp> 
  11. #include <boost/asio.hpp> 
  12.  
  13. using boost::asio::ip::tcp; 
  14.  
  15. class handler_allocator 
  16.     : private boost::noncopyable 
  17. public
  18.     handler_allocator() 
  19.         : in_use_(false
  20.     { 
  21.     } 
  22.  
  23.     void* allocate(std::size_t size) 
  24.     { 
  25.         if (!in_use_ && size < storage_.size) 
  26.         { 
  27.             in_use_ = true
  28.             return storage_.address(); 
  29.         } 
  30.         else 
  31.         { 
  32.             return ::operator new(size); 
  33.         } 
  34.     } 
  35.  
  36.     void deallocate(void* pointer) 
  37.     { 
  38.         if (pointer == storage_.address()) 
  39.         { 
  40.             in_use_ = false
  41.         } 
  42.         else 
  43.         { 
  44.             ::operator delete(pointer); 
  45.         } 
  46.     } 
  47.  
  48. private
  49.     // Storage space used for handler-based custom memory allocation. 
  50.     boost::aligned_storage<1024> storage_; 
  51.  
  52.     // Whether the handler-based custom allocation storage has been used. 
  53.     bool in_use_; 
  54. }; 
  55.  
  56. template <typename Handler> 
  57. class custom_alloc_handler 
  58. public
  59.     custom_alloc_handler(handler_allocator& a, Handler h) 
  60.         : allocator_(a), 
  61.         handler_(h) 
  62.     { 
  63.     } 
  64.  
  65.     template <typename Arg1> 
  66.     void operator()(Arg1 arg1) 
  67.     { 
  68.         handler_(arg1); 
  69.     } 
  70.  
  71.     template <typename Arg1, typename Arg2> 
  72.     void operator()(Arg1 arg1, Arg2 arg2) 
  73.     { 
  74.         handler_(arg1, arg2); 
  75.     } 
  76.  
  77.     friend void* asio_handler_allocate(std::size_t size, 
  78.         custom_alloc_handler<Handler>* this_handler) 
  79.     { 
  80.         return this_handler->allocator_.allocate(size); 
  81.     } 
  82.  
  83.     friend void asio_handler_deallocate(void* pointer, std::size_t /*size*/
  84.         custom_alloc_handler<Handler>* this_handler) 
  85.     { 
  86.         this_handler->allocator_.deallocate(pointer); 
  87.     } 
  88.  
  89. private
  90.     handler_allocator& allocator_; 
  91.     Handler handler_; 
  92. }; 
  93.  
  94. // Helper function to wrap a handler object to add custom allocation. 
  95. template <typename Handler> 
  96. inline custom_alloc_handler<Handler> make_custom_alloc_handler( 
  97.     handler_allocator& a, Handler h) 
  98.     return custom_alloc_handler<Handler>(a, h); 
  99.  
  100. /// A pool of io_service objects. 
  101. class io_service_pool 
  102.     : private boost::noncopyable 
  103. public
  104.     /// Construct the io_service pool. 
  105.     explicit io_service_pool(std::size_t pool_size) : next_io_service_(0) 
  106.     { 
  107.         if (pool_size == 0) 
  108.             throw std::runtime_error("io_service_pool size is 0"); 
  109.  
  110. // Give all the io_services work to do so that their run() functions will not 
  111. // exit until they are explicitly stopped. 
  112.         for (std::size_t i = 0; i < pool_size; ++i) 
  113.         { 
  114.             io_service_ptr io_service(new boost::asio::io_service); 
  115.             work_ptr work(new boost::asio::io_service::work(*io_service)); 
  116.             io_services_.push_back(io_service); 
  117.             work_.push_back(work); 
  118.         } 
  119.     } 
  120.  
  121. // Run all io_service objects in the pool. 
  122.     void run() 
  123.     { 
  124. // Create a pool of threads to run all of the io_services. 
  125.         std::vector<boost::shared_ptr<boost::thread> > threads; 
  126.         for (std::size_t i = 0; i < io_services_.size(); ++i) 
  127.         { 
  128.             boost::shared_ptr<boost::threadthread(new boost::thread
  129. boost::bind(&boost::asio::io_service::run, io_services_[i]))); 
  130.             threads.push_back(thread); 
  131.         } 
  132.  
  133. // Wait for all threads in the pool to exit. 
  134.         for (std::size_t i = 0; i < threads.size(); ++i) 
  135.             threads[i]->join(); 
  136.     } 
  137.  
  138. // Stop all io_service objects in the pool. 
  139.     void stop() 
  140.     { 
  141. // Explicitly stop all io_services. 
  142.         for (std::size_t i = 0; i < io_services_.size(); ++i) 
  143.             io_services_[i]->stop(); 
  144.     } 
  145.  
  146. // Get an io_service to use. 
  147.     boost::asio::io_service& get_io_service() 
  148.     { 
  149. // Use a round-robin scheme to choose the next io_service to use. 
  150.         boost::asio::io_service& io_service = *io_services_[next_io_service_]; 
  151.         ++next_io_service_; 
  152.         if (next_io_service_ == io_services_.size()) 
  153.             next_io_service_ = 0; 
  154.         return io_service; 
  155.     } 
  156.  
  157. private
  158.     typedef boost::shared_ptr<boost::asio::io_service> io_service_ptr; 
  159.     typedef boost::shared_ptr<boost::asio::io_service::work> work_ptr; 
  160.  
  161.     /// The pool of io_services. 
  162.     std::vector<io_service_ptr> io_services_; 
  163.  
  164.     /// The work that keeps the io_services running. 
  165.     std::vector<work_ptr> work_; 
  166.  
  167.     /// The next io_service to use for a connection. 
  168.     std::size_t next_io_service_; 
  169. }; 
  170.  
  171. class session 
  172.     : public boost::enable_shared_from_this<session> 
  173. public
  174.     session(boost::asio::io_service& work_service
  175. , boost::asio::io_service& io_service) 
  176.         : socket_(io_service) 
  177.         , io_work_service(work_service) 
  178.     { 
  179.     } 
  180.  
  181.     tcp::socket& socket() 
  182.     { 
  183.         return socket_; 
  184.     } 
  185.  
  186.     void start() 
  187.     { 
  188.         socket_.async_read_some(boost::asio::buffer(data_), 
  189.             make_custom_alloc_handler(allocator_, 
  190.             boost::bind(&session::handle_read, 
  191.             shared_from_this(), 
  192.             boost::asio::placeholders::error, 
  193.             boost::asio::placeholders::bytes_transferred))); 
  194.     } 
  195.  
  196.     void handle_read(const boost::system::error_code& error, 
  197.         size_t bytes_transferred) 
  198.     { 
  199.         if (!error) 
  200.         { 
  201.             boost::shared_ptr<std::vector<char> > buf(new std::vector<char>); 
  202.  
  203.             buf->resize(bytes_transferred); 
  204.             std::copy(data_.begin(), data_.begin() + bytes_transferred, buf->begin()); 
  205.             io_work_service.post(boost::bind(&session::on_receive
  206. , shared_from_this(), buf, bytes_transferred)); 
  207.  
  208.             socket_.async_read_some(boost::asio::buffer(data_), 
  209.                 make_custom_alloc_handler(allocator_, 
  210.                 boost::bind(&session::handle_read, 
  211.                 shared_from_this(), 
  212.                 boost::asio::placeholders::error, 
  213.                 boost::asio::placeholders::bytes_transferred))); 
  214.         } 
  215.     } 
  216.  
  217.     void handle_write(const boost::system::error_code& error) 
  218.     { 
  219.         if (!error) 
  220.         { 
  221.         } 
  222.     } 
  223.  
  224.     void on_receive(boost::shared_ptr<std::vector<char> > buffers
  225. size_t bytes_transferred) 
  226.     { 
  227.         char* data_stream = &(*buffers->begin()); 
  228. // in here finish the work. 
  229.         std::cout << "receive :" << bytes_transferred << " bytes." << 
  230. "message :" << data_stream << std::endl; 
  231.     } 
  232.  
  233. private
  234. // The io_service used to finish the work. 
  235.     boost::asio::io_service& io_work_service; 
  236.  
  237. // The socket used to communicate with the client. 
  238.     tcp::socket socket_; 
  239.  
  240. // Buffer used to store data received from the client. 
  241.     boost::array<char, 1024> data_; 
  242.  
  243. // The allocator to use for handler-based custom memory allocation. 
  244.     handler_allocator allocator_; 
  245. }; 
  246.  
  247. typedef boost::shared_ptr<session> session_ptr; 
  248.  
  249. class server 
  250. public
  251.     server(short port, std::size_t io_service_pool_size) 
  252.         : io_service_pool_(io_service_pool_size) 
  253.         , io_service_work_pool_(io_service_pool_size) 
  254.         , acceptor_(io_service_pool_.get_io_service(), tcp::endpoint(tcp::v4(), port)) 
  255.     { 
  256.         session_ptr new_session(new session(io_service_work_pool_.get_io_service()
  257. , io_service_pool_.get_io_service())); 
  258.         acceptor_.async_accept(new_session->socket(), 
  259.             boost::bind(&server::handle_accept, this, new_session, 
  260.             boost::asio::placeholders::error)); 
  261.     } 
  262.  
  263.     void handle_accept(session_ptr new_session, 
  264.         const boost::system::error_code& error) 
  265.     { 
  266.         if (!error) 
  267.         { 
  268.             new_session->start(); 
  269.             new_session.reset(new session(io_service_work_pool_.get_io_service()
  270. , io_service_pool_.get_io_service())); 
  271.             acceptor_.async_accept(new_session->socket(), 
  272.                 boost::bind(&server::handle_accept, this, new_session, 
  273.                 boost::asio::placeholders::error)); 
  274.         } 
  275.     } 
  276.  
  277.     void run() 
  278.     { 
  279.         io_thread_.reset(new boost::thread(boost::bind(&io_service_pool::run
  280. , &io_service_pool_))); 
  281.         work_thread_.reset(new boost::thread(boost::bind(&io_service_pool::run
  282. , &io_service_work_pool_))); 
  283.     } 
  284.  
  285.     void stop() 
  286.     { 
  287.         io_service_pool_.stop(); 
  288.         io_service_work_pool_.stop(); 
  289.  
  290.         io_thread_->join(); 
  291.         work_thread_->join(); 
  292.     } 
  293.  
  294. private
  295.     boost::shared_ptr<boost::thread> io_thread_; 
  296.     boost::shared_ptr<boost::thread> work_thread_; 
  297.     io_service_pool io_service_pool_; 
  298.     io_service_pool io_service_work_pool_; 
  299.     tcp::acceptor acceptor_; 
  300. }; 
  301.  
  302. int main(int argc, char* argv[]) 
  303.     try 
  304.     { 
  305.         if (argc != 2) 
  306.         { 
  307.             std::cerr << "Usage: server <port>\n"
  308.             return 1; 
  309.         } 
  310.  
  311.         using namespace std; // For atoi. 
  312.         server s(atoi(argv[1]), 10); 
  313.  
  314.         s.run(); 
  315.  
  316.         getchar(); 
  317.  
  318.         s.stop(); 
  319.     } 
  320.     catch (std::exception& e) 
  321.     { 
  322.         std::cerr << "Exception: " << e.what() << "\n"
  323.     } 
  324.  
  325.     return 0; 

 

(jackwgm)
本站文章除注明转载外,均为本站原创或编译欢迎任何形式的转载,但请务必注明出处,尊重他人劳动,同学习共成长。转载请注明:文章转载自:罗索实验室 [http://www1.rosoo.net/a/201012/10553.html]
本文出处:my.oschina.net/jackwgm 作者:jackwgm
顶一下
(9)
100%
踩一下
(0)
0%
------分隔线----------------------------
发表评论
请自觉遵守互联网相关的政策法规,严禁发布色情、暴力、反动的言论。
评价:
表情:
用户名: 验证码:点击我更换图片
栏目列表
将本文分享到微信
织梦二维码生成器
推荐内容