c++ - 将 asio read_some 转换为异步版本

Question

我有以下使用 boost asio read_some 函数从 TCP 套接字读取的代码。目前代码是同步的，我需要将其转换为异步版本。问题最初是读取了一些字节以识别数据包类型并获取数据包的长度。然后我们有一个读取数据的循环。我是否需要使用两个回调来异步执行此操作，还是可以使用一个回调来完成(这会更可取)。

   void Transport::OnReadFromTcp()
    {
        int read = 0;
        // read 7 bytes from TCP into mTcpBuffer
        m_sslsock->read_some(asio::buffer(mTcpBuffer, 7));

        bool tag = true;
        for (unsigned char i = 0; i < 5; i++)
        {
            tag = tag && (mTcpBuffer[i] == g_TcpPacketTag[i]);

        }

        // get the length from the last two bytes
        unsigned short dataLen = (mTcpBuffer[5] ) | (mTcpBuffer[6] << 8);
        mBuff = new char[dataLen];

        int readTotal = 0;
        while (readTotal < dataLen)
        {
            // read lengths worth of data from tcp pipe into buffer
            int readlen = dataLen;

           size_t read = m_sslsock->read_some(asio::buffer(&mBuff[readTotal], readlen));
           readlen = dataLen - read;
            readTotal += read;

        }
        // Process data .....
    }

Answer 1

第一步是意识到您可以使用自由函数 read 删除整个 read_some 循环:

void Transport::OnReadFromTcp() {
    int read = 0;

    // read 7 bytes from TCP into mTcpBuffer
    size_t bytes = asio::read(*m_sslsock, asio::buffer(mTcpBuffer, 7), asio::transfer_all());
    assert(bytes == 7);

    bool tag = g_TcpPacketTag.end() == std::mismatch(
                    g_TcpPacketTag.begin(), g_TcpPacketTag.end(),
                    mTcpBuffer.begin(), mTcpBuffer.end())
                .first;

    // get the length from the last two bytes
    uint16_t const dataLen = mTcpBuffer[5] | (mTcpBuffer[6] << 8);
    mBuff.resize(dataLen);

    size_t readTotal = asio::read(*m_sslsock, asio::buffer(mBuff), asio::transfer_exactly(dataLen));

    assert(mBuff.size() == readTotal);
    assert(dataLen == readTotal);
}

这甚至与执行是否异步无关。

让它异步有点复杂，因为它需要假设缓冲区/Transport 实例的生命周期以及潜在的多线程。我会在早上喝完咖啡后提供一个 sample :)

---

没有线程/生命周期并发症的演示:

Live On Coliru

#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/bind.hpp>
#include <iostream>
#include <array>
#include <cassert>

namespace asio = boost::asio;
namespace ssl  = asio::ssl;


namespace {
    static std::array<char, 5> g_TcpPacketTag {{'A','B','C','D','E'}};
}

struct Transport {
    using tcp = asio::ip::tcp;
    using SslSocket = std::shared_ptr<asio::ssl::stream<tcp::socket> >;

    Transport(SslSocket s) : m_sslsock(s) { }

    void OnReadFromTcp();
    void OnHeaderReceived(boost::system::error_code ec, size_t transferred);
    void OnContentReceived(boost::system::error_code ec, size_t transferred);
  private:
    uint16_t datalen() const {
        return mTcpBuffer[5] | (mTcpBuffer[6] << 8);
    }

    SslSocket m_sslsock;

    std::array<char, 7> mTcpBuffer;
    std::vector<char> mBuff;
};

void Transport::OnReadFromTcp() {
    // read 7 bytes from TCP into mTcpBuffer
    asio::async_read(*m_sslsock, asio::buffer(mTcpBuffer, 7), asio::transfer_all(),
                boost::bind(&Transport::OnHeaderReceived, this, asio::placeholders::error, asio::placeholders::bytes_transferred)
            );
}

#include <boost/range/algorithm/mismatch.hpp> // I love sugar

void Transport::OnHeaderReceived(boost::system::error_code ec, size_t bytes) {
    if (ec) {
        std::cout << "Error: " << ec.message() << "\n";
    }

    assert(bytes == 7);

    bool tag = (g_TcpPacketTag.end() == boost::mismatch(g_TcpPacketTag, mTcpBuffer).first);

    if (tag) {
        // get the length from the last two bytes
        mBuff.resize(datalen());

        asio::async_read(*m_sslsock, asio::buffer(mBuff), asio::transfer_exactly(datalen()),
                boost::bind(&Transport::OnContentReceived, this, asio::placeholders::error, asio::placeholders::bytes_transferred)
            );

    } else {
        std::cout << "TAG MISMATCH\n"; // TODO handle error
    }
}

void Transport::OnContentReceived(boost::system::error_code ec, size_t readTotal) {
    assert(mBuff.size() == readTotal);
    assert(datalen() == readTotal);

    std::cout << "Successfully completed receive of " << datalen() << " bytes\n";
}

int main() {
    asio::io_service svc;

    using Socket = Transport::SslSocket::element_type;

    // connect to localhost:6767 with SSL
    ssl::context ctx(ssl::context::sslv23);
    auto s = std::make_shared<Socket>(svc, ctx);
    s->lowest_layer().connect({ {}, 6767 });
    s->handshake(Socket::handshake_type::client);

    // do transport
    Transport tx(s);
    tx.OnReadFromTcp();

    svc.run();

    // all done
    std::cout << "All done\n";
}

针对在端口 6767 上接受 SSL 连接的示例服务器使用时:

(printf "ABCDE\x01\x01F"; cat main.cpp) |
     openssl s_server -accept 6767 -cert so.crt -pass pass:test

打印:

Successfully completed receive of 257 bytes
All done