boost::asio::async_write - 确保只有一个未完成的调用

Question

根据文档：

“在此操作完成之前，程序必须确保流不执行其他写入操作（例如 async_write、流的 async_write_some 函数或执行写入的任何其他组合操作）。”

这是否意味着，在调用第一次的处理程序之前，我不能第二次调用 boost::asio::async_write？如何实现这一点并且仍然是异步的？

如果我有方法发送：

//--------------------------------------------------------------------
void Connection::Send(const std::vector<char> & data)
{
    auto callback = boost::bind(&Connection::OnSend, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred);
    boost::asio::async_write(m_socket, boost::asio::buffer(data), callback);
}

我是否必须将其更改为：

//--------------------------------------------------------------------
void Connection::Send(const std::vector<char> & data)
{
    // Issue a send
    std::lock_guard<std::mutex> lock(m_numPostedSocketIOMutex);
    ++m_numPostedSocketIO;

    m_numPostedSocketIOConditionVariable.wait(lock, [this]() {return m_numPostedSocketIO == 0; });

    auto callback = boost::bind(&Connection::OnSend, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred);
    boost::asio::async_write(m_socket, boost::asio::buffer(data), callback);
}

如果是这样，那我不是在第一次通话后再次阻塞吗？

Answer 1

是的，您需要等待完成处理程序才能再次调用async_write。你确定你会被屏蔽？当然，这取决于您生成数据的速度，但即使是，也无法以比您的网络处理速度更快的速度发送数据。如果这确实是一个问题，请考虑发送更大的块。

Answer 2

async_write() 中的 async 指的是函数在后台写入时立即返回的事实。在任何给定时间，仍然应该只有一个未完成的写入。

如果您有一个异步生产者来保留新的数据块直到当前活动的写入完成，您需要使用 buffer，然后在完成处理程序中发出一个新的 async_write。

也就是说，Connection::Send 必须只调用一次async_write 来启动进程，在随后的调用中它应该缓冲其数据，这些数据将在当前正在执行的async_write 的完成处理程序中被拾取。

出于性能原因，您希望避免将数据复制到缓冲区中，而是将新块附加到缓冲区列表中，并使用接受 ConstBufferSequence 的 async_write 的 scatter-gather overload。也可以使用一个大的streambuf 作为缓冲区并直接追加到其中。

当然缓冲区需要同步，除非Connection::Send 和io_service 在同一个线程中运行。空缓冲区可以重复使用，表示没有async_write 正在进行中。

这里有一些代码来说明我的意思：

struct Connection
{
    void Connection::Send(std::vector<char>&& data)
    {
        std::lock_guard<std::mutex> lock(buffer_mtx);
        buffers[active_buffer ^ 1].push_back(std::move(data)); // move input data to the inactive buffer
        doWrite();
    }

private:

    void Connection::doWrite()
    {
        if (buffer_seq.empty()) { // empty buffer sequence == no writing in progress
            active_buffer ^= 1; // switch buffers
            for (const auto& data : buffers[active_buffer]) {
                buffer_seq.push_back(boost::asio::buffer(data));
            }
            boost::asio::async_write(m_socket, buffer_seq, [this] (const boost::system::error_code& ec, size_t bytes_transferred) {
                std::lock_guard<std::mutex> lock(buffer_mtx);
                buffers[active_buffer].clear();
                buffer_seq.clear();
                if (!ec) {
                    if (!buffers[active_buffer ^ 1].empty()) { // have more work
                        doWrite();
                    }
                }
            });
        }
    }

    std::mutex buffer_mtx;
    std::vector<std::vector<char>> buffers[2]; // a double buffer
    std::vector<boost::asio::const_buffer> buffer_seq;
    int active_buffer = 0;
    . . .
};

完整的工作源码可以在this answer找到。

Answer 3

这是一个完整的、可编译的和经过测试的示例，我在阅读了 RustyX 的答案和随后的编辑后，研究并通过反复试验来工作。

Connection.h

#pragma once

#include <boost/asio.hpp>

#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>

//--------------------------------------------------------------------
class ConnectionManager;

//--------------------------------------------------------------------
class Connection : public std::enable_shared_from_this<Connection>
{
public:

    typedef std::shared_ptr<Connection> SharedPtr;

    // Ensure all instances are created as shared_ptr in order to fulfill requirements for shared_from_this
    static Connection::SharedPtr Create(ConnectionManager * connectionManager, boost::asio::ip::tcp::socket & socket);

    //
    static std::string ErrorCodeToString(const boost::system::error_code & errorCode);

    Connection(const Connection &) = delete;
    Connection(Connection &&) = delete;
    Connection & operator = (const Connection &) = delete;
    Connection & operator = (Connection &&) = delete;
    ~Connection();

    // We have to defer the start until we are fully constructed because we share_from_this()
    void Start();
    void Stop();

    void Send(const std::vector<char> & data);

private:

    static size_t                                           m_nextClientId;

    size_t                                                  m_clientId;
    ConnectionManager *                                     m_owner;
    boost::asio::ip::tcp::socket                            m_socket;
    std::atomic<bool>                                       m_stopped;
    boost::asio::streambuf                                  m_receiveBuffer;
    mutable std::mutex                                      m_sendMutex;
    std::vector<char>                                       m_sendBuffers[2];         // Double buffer
    int                                                     m_activeSendBufferIndex;
    bool                                                    m_sending;

    std::vector<char>                                       m_allReadData;            // Strictly for test purposes

    Connection(ConnectionManager * connectionManager, boost::asio::ip::tcp::socket socket);

    void DoReceive();
    void DoSend();
};

//--------------------------------------------------------------------

Connection.cpp

#include "Connection.h"
#include "ConnectionManager.h"

#include <boost/bind.hpp>

#include <algorithm>
#include <cstdlib>

//--------------------------------------------------------------------
size_t Connection::m_nextClientId(0);

//--------------------------------------------------------------------
Connection::SharedPtr Connection::Create(ConnectionManager * connectionManager, boost::asio::ip::tcp::socket & socket)
{
    return Connection::SharedPtr(new Connection(connectionManager, std::move(socket)));
}

//--------------------------------------------------------------------------------------------------
std::string Connection::ErrorCodeToString(const boost::system::error_code & errorCode)
{
    std::ostringstream debugMsg;
    debugMsg << " Error Category: " << errorCode.category().name() << ". "
             << " Error Message: "  << errorCode.message() << ". ";

    // IMPORTANT - These comparisons only work if you dynamically link boost libraries
    //             Because boost chose to implement boost::system::error_category::operator == by comparing addresses
    //             The addresses are different in one library and the other when statically linking.
    //
    // We use make_error_code macro to make the correct category as well as error code value.
    // Error code value is not unique and can be duplicated in more than one category.
    if (errorCode == boost::asio::error::make_error_code(boost::asio::error::connection_refused))
    {
        debugMsg << " (Connection Refused)";
    }
    else if (errorCode == boost::asio::error::make_error_code(boost::asio::error::eof))
    {
        debugMsg << " (Remote host has disconnected)";
    }
    else
    {
        debugMsg << " (boost::system::error_code has not been mapped to a meaningful message)";
    }

    return debugMsg.str();
}

//--------------------------------------------------------------------
Connection::Connection(ConnectionManager * connectionManager, boost::asio::ip::tcp::socket socket)
    :
    m_clientId                          (m_nextClientId++)
  , m_owner                             (connectionManager)
  , m_socket                            (std::move(socket))
  , m_stopped                           (false)
  , m_receiveBuffer                     ()
  , m_sendMutex                         ()
  , m_sendBuffers                       ()
  , m_activeSendBufferIndex             (0)
  , m_sending                           (false)
  , m_allReadData                       ()
{
    printf("Client connection with id %zd has been created.", m_clientId);
}

//--------------------------------------------------------------------
Connection::~Connection()
{
    // Boost uses RAII, so we don't have anything to do. Let thier destructors take care of business
    printf("Client connection with id %zd has been destroyed.", m_clientId);
}

//--------------------------------------------------------------------
void Connection::Start()
{
    DoReceive();
}

//--------------------------------------------------------------------
void Connection::Stop()
{
    // The entire connection class is only kept alive, because it is a shared pointer and always has a ref count
    // as a consequence of the outstanding async receive call that gets posted every time we receive.
    // Once we stop posting another receive in the receive handler and once our owner release any references to
    // us, we will get destroyed.
    m_stopped = true;
    m_owner->OnConnectionClosed(shared_from_this());
}

//--------------------------------------------------------------------
void Connection::Send(const std::vector<char> & data)
{
    std::lock_guard<std::mutex> lock(m_sendMutex);

    // Append to the inactive buffer
    std::vector<char> & inactiveBuffer = m_sendBuffers[m_activeSendBufferIndex ^ 1];
    inactiveBuffer.insert(inactiveBuffer.end(), data.begin(), data.end());

    //
    DoSend();
}

//--------------------------------------------------------------------
void Connection::DoSend()
{
    // Check if there is an async send in progress
    // An empty active buffer indicates there is no outstanding send
    if (m_sendBuffers[m_activeSendBufferIndex].empty())
    {
        m_activeSendBufferIndex ^= 1;

        std::vector<char> & activeBuffer = m_sendBuffers[m_activeSendBufferIndex];
        auto self(shared_from_this());

        boost::asio::async_write(m_socket, boost::asio::buffer(activeBuffer),
            [self](const boost::system::error_code & errorCode, size_t bytesTransferred)
            {
                std::lock_guard<std::mutex> lock(self->m_sendMutex);

                self->m_sendBuffers[self->m_activeSendBufferIndex].clear();

                if (errorCode)
                {
                    printf("An error occured while attemping to send data to client id %zd. %s", self->m_clientId, ErrorCodeToString(errorCode).c_str());

                    // An error occurred
                    // We do not stop or close on sends, but instead let the receive error out and then close
                    return;
                }

                // Check if there is more to send that has been queued up on the inactive buffer,
                // while we were sending what was on the active buffer
                if (!self->m_sendBuffers[self->m_activeSendBufferIndex ^ 1].empty())
                {
                    self->DoSend();
                }
            });
    }
}

//--------------------------------------------------------------------
void Connection::DoReceive()
{
    auto self(shared_from_this());

    boost::asio::async_read_until(m_socket, m_receiveBuffer, '#',
        [self](const boost::system::error_code & errorCode, size_t bytesRead)
        {
            if (errorCode)
            {
                // Check if the other side hung up
                if (errorCode == boost::asio::error::make_error_code(boost::asio::error::eof))
                {
                    // This is not really an error. The client is free to hang up whenever they like
                    printf("Client %zd has disconnected.", self->m_clientId);
                }
                else
                {
                    printf("An error occured while attemping to receive data from client id %zd. Error Code: %s", self->m_clientId, ErrorCodeToString(errorCode).c_str());
                }

                // Notify our masters that we are ready to be destroyed
                self->m_owner->OnConnectionClosed(self);

                // An error occured
                return;
            }

            // Grab the read data
            std::istream stream(&self->m_receiveBuffer);
            std::string data;
            std::getline(stream, data, '#');
            data += "#";

            printf("Received data from client %zd: %s", self->m_clientId, data.c_str());

            // Issue the next receive
            if (!self->m_stopped)
            {
                self->DoReceive();
            }
        });
}

//--------------------------------------------------------------------

ConnectionManager.h

#pragma once

#include "Connection.h"

// Boost Includes
#include <boost/asio.hpp>

// Standard Includes
#include <thread>
#include <vector>

//--------------------------------------------------------------------
class ConnectionManager
{
public:

    ConnectionManager(unsigned port, size_t numThreads);
    ConnectionManager(const ConnectionManager &) = delete;
    ConnectionManager(ConnectionManager &&) = delete;
    ConnectionManager & operator = (const ConnectionManager &) = delete;
    ConnectionManager & operator = (ConnectionManager &&) = delete;
    ~ConnectionManager();

    void Start();
    void Stop();

    void OnConnectionClosed(Connection::SharedPtr connection);

protected:

    boost::asio::io_service            m_io_service;
    boost::asio::ip::tcp::acceptor     m_acceptor;
    boost::asio::ip::tcp::socket       m_listenSocket;
    std::vector<std::thread>           m_threads;

    mutable std::mutex                 m_connectionsMutex;
    std::vector<Connection::SharedPtr> m_connections;

    boost::asio::deadline_timer        m_timer;

    void IoServiceThreadProc();

    void DoAccept();
    void DoTimer();
};

//--------------------------------------------------------------------

ConnectionManager.cpp

#include "ConnectionManager.h"

#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <system_error>
#include <cstdio>

//------------------------------------------------------------------------------
ConnectionManager::ConnectionManager(unsigned port, size_t numThreads)
    :
    m_io_service  ()
  , m_acceptor    (m_io_service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port))
  , m_listenSocket(m_io_service)
  , m_threads     (numThreads)
  , m_timer       (m_io_service)
{
}

//------------------------------------------------------------------------------
ConnectionManager::~ConnectionManager()
{
    Stop();
}

//------------------------------------------------------------------------------
void ConnectionManager::Start()
{
    if (m_io_service.stopped())
    {
        m_io_service.reset();
    }

    DoAccept();

    for (auto & thread : m_threads)
    {
        if (!thread.joinable())
        {
            thread.swap(std::thread(&ConnectionManager::IoServiceThreadProc, this));
        }
    }

    DoTimer();
}

//------------------------------------------------------------------------------
void ConnectionManager::Stop()
{
    {
        std::lock_guard<std::mutex> lock(m_connectionsMutex);
        m_connections.clear();
    }

    // TODO - Will the stopping of the io_service be enough to kill all the connections and ultimately have them get destroyed?
    //        Because remember they have outstanding ref count to thier shared_ptr in the async handlers
    m_io_service.stop();

    for (auto & thread : m_threads)
    {
        if (thread.joinable())
        {
            thread.join();
        }
    }
}

//------------------------------------------------------------------------------
void ConnectionManager::IoServiceThreadProc()
{
    try
    {
        // Log that we are starting the io_service thread
        {
            printf("io_service socket thread starting.");
        }

        // Run the asynchronous callbacks from the socket on this thread
        // Until the io_service is stopped from another thread
        m_io_service.run();
    }
    catch (std::system_error & e)
    {
        printf("System error caught in io_service socket thread. Error Code: %d", e.code().value());
    }
    catch (std::exception & e)
    {
        printf("Standard exception caught in io_service socket thread. Exception: %s", e.what());
    }
    catch (...)
    {
        printf("Unhandled exception caught in io_service socket thread.");
    }

    {
        printf("io_service socket thread exiting.");
    }
}

//------------------------------------------------------------------------------
void ConnectionManager::DoAccept()
{
    m_acceptor.async_accept(m_listenSocket,
        [this](const boost::system::error_code errorCode)
        {
            if (errorCode)
            {
                printf("An error occured while attemping to accept connections. Error Code: %s", Connection::ErrorCodeToString(errorCode).c_str());
                return;
            }

            // Create the connection from the connected socket
            std::lock_guard<std::mutex> lock(m_connectionsMutex);
            Connection::SharedPtr connection = Connection::Create(this, m_listenSocket);
            m_connections.push_back(connection);
            connection->Start();

            DoAccept();
        });
}

//------------------------------------------------------------------------------
void ConnectionManager::OnConnectionClosed(Connection::SharedPtr connection)
{
    std::lock_guard<std::mutex> lock(m_connectionsMutex);

    auto itConnection = std::find(m_connections.begin(), m_connections.end(), connection);
    if (itConnection != m_connections.end())
    {
        m_connections.erase(itConnection);
    }
}

//------------------------------------------------------------------------------
void ConnectionManager::DoTimer()
{
    if (!m_io_service.stopped())
    {
        // Send messages every second
        m_timer.expires_from_now(boost::posix_time::seconds(30));
        m_timer.async_wait(
            [this](const boost::system::error_code & errorCode)
            {
                std::lock_guard<std::mutex> lock(m_connectionsMutex);
                for (auto connection : m_connections)
                {
                    connection->Send(std::vector<char>{'b', 'e', 'e', 'p', '#'});
                }

                DoTimer();
            });
    }
}

ma​​in.cpp

#include "ConnectionManager.h"

#include <cstring>
#include <iostream>
#include <string>

int main()
{
    // Start up the server
    ConnectionManager connectionManager(5000, 2);
    connectionManager.Start();

    // Pretend we are doing other things or just waiting for shutdown
    std::this_thread::sleep_for(std::chrono::minutes(5));

    // Stop the server
    connectionManager.Stop();

    return 0;
}

Answer 4

我们是否可以通过将 write(...) 作为异步操作发布到 strand1 并将 handler(...) 发布到 strand2 来解决这个问题？ 非常感谢您对代码的建议。

boost::asio::strand<boost::asio::io_context::executor_type> strand1, strand2;
std::vector<char> empty_vector(0);

void Connection::Send(const std::vector<char> & data)
{
    boost::asio::post(boost::asio::bind_executor(strand1, std::bind(&Connection::write, this, true, data)));
}

void Connection::write(bool has_data, const std::vector<char> & data)
{
     // Append to the inactive buffer
    std::vector<char> & inactiveBuffer = m_sendBuffers[m_activeSendBufferIndex ^ 1];

    if (has_data)
    {            
        inactiveBuffer.insert(inactiveBuffer.end(), data.begin(), data.end());
    }

    //
    if (!inactiveBuffer.empty() && m_sendBuffers[m_activeSendBufferIndex].empty())
    {
        m_activeSendBufferIndex ^= 1;
        std::vector<char> & activeBuffer = m_sendBuffers[m_activeSendBufferIndex];
        boost::asio::async_write(m_socket, boost::asio::buffer(activeBuffer), boost::asio::bind_executor(strand2, std::bind(&Connection::handler, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)));
    }
} 

void Connection::handler(const boost::system::error_code & errorCode, size_t bytesTransferred)
{
    self->m_sendBuffers[self->m_activeSendBufferIndex].clear();

    if (errorCode)
    {
        printf("An error occured while attemping to send data to client id %zd. %s", self->m_clientId, ErrorCodeToString(errorCode).c_str());

        // An error occurred
        // We do not stop or close on sends, but instead let the receive error out and then close
        return;
    }

    boost::asio::post(boost::asio::bind_executor(strand1, std::bind(&Connection::write, this, false, empty_vector)));
    }
}