提高 ASIO 并发性

Question

我正在学习使用 Boost ASIO。这是从chat example 中复制的一些代码以及 Boost ASIO 文档，

typedef std::deque<chat_message> chat_message_queue;

class chat_client
{
    public:
        chat_client(boost::asio::io_service& io_service,
                tcp::resolver::iterator endpoint_iterator)
            : io_service_(io_service),
            socket_(io_service)
    {
        boost::asio::async_connect(socket_, endpoint_iterator,
                boost::bind(&chat_client::handle_connect, this,
                    boost::asio::placeholders::error));
    }

        void write(const chat_message& msg)
        {
            io_service_.post(boost::bind(&chat_client::do_write, this, msg));
        }

        void close()
        {
            io_service_.post(boost::bind(&chat_client::do_close, this));
        }

    private:

        void handle_connect(const boost::system::error_code& error)
        {
            //Implementation
        }

        void handle_read_header(const boost::system::error_code& error)
        {
            //Implementation
        }

        void handle_read_body(const boost::system::error_code& error)
        {
            //Implementation
        }

        void do_write(chat_message msg)
        {
            bool write_in_progress = !write_msgs_.empty();
            write_msgs_.push_back(msg);
            if (!write_in_progress)
            {
                boost::asio::async_write(socket_,
                        boost::asio::buffer(write_msgs_.front().data(),
                            write_msgs_.front().length()),
                        boost::bind(&chat_client::handle_write, this,
                            boost::asio::placeholders::error));
            }
        }



        void handle_write(const boost::system::error_code& error)
        {
            //Implementation
        }

        void do_close()
        {
            socket_.close();
        }

    private:
        boost::asio::io_service& io_service_;
        tcp::socket socket_;
        chat_message read_msg_;
        chat_message_queue write_msgs_;
};

1. 写入是异步的，在成员变量 write_msgs_ 和 read_msgs_ 周围没有使用锁。这里不应该有并发的问题吗？

2. 从运行io_service::run? 的线程调用post 是否安全，那么dispatch 呢？如果从未运行 io_service::run 的线程进行相同的调用呢？

3. 在doSend()，为什么他们将消息推送到write_msgs_而不是直接发送？同样在同一个函数中，他们为什么要检查 write_msgs_ 是否为空，并且只有在不是时才继续发送？ write_msgs_.empty() = false 是否表示正在进行写入？怎么样？

4. 如果do_write() 仅在一个线程中被调用，那么为什么我需要一个队列来维护一系列发送？ io_service不会完成手头的任务，然后执行do_write调用的异步操作吗？使用dispatch 代替post 会在上述示例中产生影响吗？

Answer 1

1. 虽然写入是异步的，但这里没有多线程：do_write() 在一个线程中调用。当然，在调用完成处理程序之前，正在发送的缓冲区必须是活动的并且保持不变。

2. 从任何线程调用 post() 和 dispatch() 都是安全的。阅读io_service documentation 的“线程安全”部分。

3. 如果async_write 正在进行中，并且您再次在同一个套接字上调用async_write，则发送数据的顺序是未定义的。换句话说，数据将被弄乱。解决此问题的最简单方法是创建消息队列：每次async_write 完成时，发出另一个async_write。 （顺便说一句，同样适用于async_read。） 为什么write_msgs_.empty() = false 表示正在写入？因为只要write_msgs_ 不为空，handle_write（前一个async_write 的完成处理程序）就会发出另一个async_write。当write_msgs_ 为空时，此循环中断。

4. 请阅读the documentation says about async_write：

   这个操作是在零次或多次调用 流的 async_write_some 函数，被称为 composed 操作。程序必须确保流不执行其他 写操作（如 async_write，流的 async_write_some 函数或任何其他执行写入的组合操作）直到 此操作完成。

至于dispatch 与post - 据我所知，在上面的示例中它们是可以互换的。如果我们不希望被发布的函子被同步调用，那么使用 post 是必不可少的。