<轉>淺談 Boost.Asio 的多線程模型

本文轉自：http://senlinzhan.github.io/2017/09/17/boost-asio/

Boost.Asio 有兩種支持多線程的方式，第一種方式比較簡單：在多線程的場景下，每個線程都持有一個io_service，並且每個線程都調用各自的io_service的run()方法。
　　另一種支持多線程的方式：全局只分配一個io_service，並且讓這個io_service在多個線程之間共享，每個線程都調用全局的io_service的run()方法。

每個線程一個 I/O Service

　　讓我們先分析第一種方案：在多線程的場景下，每個線程都持有一個io_service (通常的做法是，讓線程數和 CPU 核心數保持一致)。那么這種方案有什么特點呢？

• 在多核的機器上，這種方案可以充分利用多個 CPU 核心。
• 某個 socket 描述符並不會在多個線程之間共享，所以不需要引入同步機制。
• 在 event handler 中不能執行阻塞的操作，否則將會阻塞掉io_service所在的線程。

　　下面我們實現了一個AsioIOServicePool，封裝了線程池的創建操作 [完整代碼]：

class AsioIOServicePool
{
public:
    using IOService = boost::asio::io_service;
    using Work = boost::asio::io_service::work;
    using WorkPtr = std::unique_ptr<Work>;
    AsioIOServicePool(std::size_t size = std::thread::hardware_concurrency())
        : ioServices_(size),
          works_(size),
          nextIOService_(0)
    {
        for (std::size_t i = 0; i < size; ++i)
        {
            works_[i] = std::unique_ptr<Work>(new Work(ioServices_[i]));
        }
        for (std::size_t i = 0; i < ioServices_.size(); ++i)
        {
            threads_.emplace_back([this, i] ()
                                  {
                                      ioServices_[i].run();
                                  });
        }
    }
    AsioIOServicePool(const AsioIOServicePool &) = delete;
    AsioIOServicePool &operator=(const AsioIOServicePool &) = delete;
    // 使用 round-robin 的方式返回一個 io_service
    boost::asio::io_service &getIOService()
    {
        auto &service = ioServices_[nextIOService_++];
        if (nextIOService_ == ioServices_.size())
        {
            nextIOService_ = 0;
        }
        return service;
    }
    void stop()
    {
        for (auto &work: works_)
        {
            work.reset();
        }
        for (auto &t: threads_)
        {
            t.join();
        }
    }
private:
    std::vector<IOService>       ioServices_;
    std::vector<WorkPtr>         works_;
    std::vector<std::thread>     threads_;
    std::size_t                  nextIOService_;
};

AsioIOServicePool使用起來也很簡單：

std::mutex mtx;             // protect std::cout
AsioIOServicePool pool;
 
boost::asio::steady_timer timer{pool.getIOService(), std::chrono::seconds{2}};
timer.async_wait([&mtx] (const boost::system::error_code &ec)
                  {
                      std::lock_guard<std::mutex> lock(mtx);
                      std::cout << "Hello, World! " << std::endl;
                  });
pool.stop();

一個 I/O Service 與多個線程

　　另一種方案則是先分配一個全局io_service，然后開啟多個線程，每個線程都調用這個io_service的run()方法。這樣，當某個異步事件完成時，io_service就會將相應的 event handler 交給任意一個線程去執行。
　　然而這種方案在實際使用中，需要注意一些問題：

• 在 event handler 中允許執行阻塞的操作 (例如數據庫查詢操作)。
• 線程數可以大於 CPU 核心數，譬如說，如果需要在 event handler 中執行阻塞的操作，為了提高程序的響應速度，這時就需要提高線程的數目。
• 由於多個線程同時運行事件循環(event loop)，所以會導致一個問題：即一個 socket 描述符可能會在多個線程之間共享，容易出現競態條件 (race condition)。譬如說，如果某個 socket 的可讀事件很快發生了兩次，那么就會出現兩個線程同時讀同一個 socket 的問題 (可以使用strand解決這個問題)。

　　下面實現了一個線程池，在每個 worker 線程中執行io_service的run()方法 [完整代碼]：

class AsioThreadPool
{
public:
    AsioThreadPool(int threadNum = std::thread::hardware_concurrency())
        : work_(new boost::asio::io_service::work(service_))
    {
        for (int i = 0; i < threadNum; ++i)
        {
            threads_.emplace_back([this] () { service_.run(); });
        }
    }
    AsioThreadPool(const AsioThreadPool &) = delete;
    AsioThreadPool &operator=(const AsioThreadPool &) = delete;
    boost::asio::io_service &getIOService()
    {
        return service_;
    }
    void stop()
    {
        work_.reset();
        for (auto &t: threads_)
        {
            t.join();            
        }        
    }
private:
    boost::asio::io_service service_;
    std::unique_ptr<boost::asio::io_service::work> work_;
    std::vector<std::thread> threads_;
};

無鎖的同步方式

　　要怎樣解決前面提到的競態條件呢？Boost.Asio 提供了io_service::strand：如果多個 event handler 通過同一個 strand 對象分發 (dispatch)，那么這些 event handler 就會保證順序地執行。
　　例如，下面的例子使用 strand，所以不需要使用互斥鎖保證同步了 [完整代碼]：

AsioThreadPool pool(4);    // 開啟 4 個線程
boost::asio::steady_timer timer1{pool.getIOService(), std::chrono::seconds{1}};
boost::asio::steady_timer timer2{pool.getIOService(), std::chrono::seconds{1}};
int value = 0;
boost::asio::io_service::strand strand{pool.getIOService()};    
    
timer1.async_wait(strand.wrap([&value] (const boost::system::error_code &ec)
                              {
                                  std::cout << "Hello, World! " << value++ << std::endl;
                              }));
timer2.async_wait(strand.wrap([&value] (const boost::system::error_code &ec)
                              {
                                  std::cout << "Hello, World! " << value++ << std::endl;
                              }));
pool.stop();

多線程 Echo Server

　　下面的EchoServer可以在多線程中使用，它使用asio::strand來解決前面提到的競態問題 [完整代碼]：

class TCPConnection : public std::enable_shared_from_this<TCPConnection> 
{
public:
    TCPConnection(boost::asio::io_service &io_service)
        : socket_(io_service),
          strand_(io_service)
    { }
    
    tcp::socket &socket() {  return socket_;  }
    void start() {  doRead();  }
    
private:
    void doRead()
    {
        auto self = shared_from_this();
        socket_.async_read_some(
            boost::asio::buffer(buffer_, buffer_.size()),
            strand_.wrap([this, self](boost::system::error_code ec, 
                                      std::size_t bytes_transferred)
                         {
                             if (!ec) {  doWrite(bytes_transferred);  }
                         }));
    }
    void doWrite(std::size_t length)
    {
        auto self = shared_from_this();
        boost::asio::async_write(
            socket_, boost::asio::buffer(buffer_, length),
            strand_.wrap([this, self](boost::system::error_code ec,
                                      std::size_t /* bytes_transferred */)
                         {
                             if (!ec) {  doRead();  }
                         }));
    }
private:
    tcp::socket socket_;
    boost::asio::io_service::strand strand_;
    std::array<char, 8192> buffer_;
};
class EchoServer
{
public:
    EchoServer(boost::asio::io_service &io_service, unsigned short port)
        : io_service_(io_service),
          acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
    {
        doAccept();
    }
    void doAccept()
    {
        auto conn = std::make_shared<TCPConnection>(io_service_);
        acceptor_.async_accept(conn->socket(),
                               [this, conn](boost::system::error_code ec)
                               {
                                   if (!ec) {  conn->start();  }
                                   this->doAccept();
                               });
    }
    
private: 
    boost::asio::io_service &io_service_;
    tcp::acceptor acceptor_;
};

參考資料

• The Boost C++ Libraries Chapter 32. Boost.Asio
• A guide to getting started with boost::asio
• Strands: Use Threads Without Explicit Locking
• Post on ASIO strand
• How strands guarantee correct execution of pending events in boost.asio
• asio C++ library
• Using Asio with C++11