使用Google Test测试boost::asio async_write时数据交织问题求助

算力阿峰

2026-7-1

TCP服务器数据交织问题排查与修复

问题描述

用boost::asio实现支持多客户端的TCP服务器，通过Google Test编写测试用例检测数据交织：将5个缓冲区分别填充单一数值，发送后读取并统计每个值的出现次数，预期每个读取缓冲区中只有一种数值。已知async_write底层依赖async_write_some，因此使用strand保证操作顺序，但测试未达到预期，出现数据交织。

代码实现

#include <boost/asio.hpp>
#include <gtest/gtest.h>
using tcp = boost::asio::ip::tcp;

// tcp_session.h
using boost_tcp = boost::asio::ip::tcp;

class tcp_session : public std::enable_shared_from_this<tcp_session> {
  public:
    explicit tcp_session(boost_tcp::socket socket);
    ~tcp_session();

    void   async_write(char const* data, size_t size);
    void   async_write(boost::asio::const_buffer buff);
    size_t write(char const* data, size_t size);
    size_t write(boost::asio::const_buffer buff);

    void close();

  private:
    boost_tcp::socket                                     m_socket;
    std::mutex                                            m_socket_mutex{};
    boost::asio::strand<boost_tcp::socket::executor_type> m_strand;

    void handle_async_write(boost::system::error_code const& err, size_t bytes_transferred);
};

// tcp_svr.h
class tcp_svr {
  public:
    explicit tcp_svr(int16_t port);
    ~tcp_svr();

    void async_write(char const* data, size_t size);
    void async_write(boost::asio::const_buffer buff);
    void run();
    void stop();

    size_t              get_session_count();
    boost_tcp::endpoint get_local_endpoint() { return m_acceptor.local_endpoint(); }

  private:
    using session_ptr = std::shared_ptr<tcp_session>;
    void                     do_accept();
    std::atomic_bool         m_stopped;
    std::vector<session_ptr> m_sessions{};
    std::mutex               m_sessions_mutex{};
    boost::asio::io_service  m_io_service{};
    boost_tcp::acceptor      m_acceptor;
};
// tcp_svr.cpp

tcp_svr::tcp_svr(int16_t port) : m_acceptor(m_io_service, tcp::endpoint(tcp::v4(), port)) { do_accept(); }

tcp_svr::~tcp_svr() { stop(); }

void tcp_svr::async_write(boost::asio::const_buffer buff) {
    std::lock_guard lock(m_sessions_mutex);
    for (auto const& client : m_sessions) {
        client->async_write(buff);
    }
}

void tcp_svr::run() { m_io_service.run(); }

void tcp_svr::stop() {
    if (m_stopped)
        return;

    m_stopped   = true;
    auto thread = std::thread{[this]() { m_acceptor.cancel(); }};
    thread.join();
    {
        std::lock_guard lock(m_sessions_mutex);
        for (auto const& client : m_sessions) {
            client->close();
        }

        m_sessions.clear();
    }

    m_io_service.stop();
}

size_t tcp_svr::get_session_count() {
    std::lock_guard lock(m_sessions_mutex);
    return m_sessions.size();
}

void tcp_svr::do_accept() {
    m_acceptor.async_accept(boost::asio::make_strand(m_io_service),
                            [this](boost::system::error_code err, boost_tcp::socket socket) {
                                if (m_stopped) {
                                    return;
                                }

                                if (!err) {
                                    std::lock_guard lock(m_sessions_mutex);
                                    m_sessions.push_back(std::make_shared<tcp_session>(std::move(socket)));
                                }

                                do_accept();
                            });
}

// tcp_session.cpp
tcp_session::tcp_session(boost_tcp::socket socket)
    : m_socket(std::move(socket))
    , m_strand(socket.get_executor()) {}

tcp_session::~tcp_session() { close(); }

void tcp_session::async_write(boost::asio::const_buffer buff) {
    boost::asio::post(m_strand, [this, buff = buff]() {
        std::lock_guard lock(m_socket_mutex);
        boost::asio::async_write(m_socket, buff,
                                 [this, self = shared_from_this()](const boost::system::error_code& ec,
                                                                   std::size_t bytes_transferred) {
                                     handle_async_write(ec, bytes_transferred);
                                 });
    });
}

void tcp_session::async_write(char const* data, size_t size) { async_write(boost::asio::buffer(data, size)); }

void tcp_session::close() {
    std::lock_guard lock(m_socket_mutex);
    if (m_socket.is_open()) {
        m_socket.close();
    }
}

void tcp_session::handle_async_write(boost::system::error_code const& err,
                                     [[maybe_unused]] size_t          bytes_transferred) {
    if (err) {
        // log
    }
}

class TcpFixture : public testing::Test {
  protected:
    boost::asio::io_service  m_client_io_service{};
    std::unique_ptr<tcp_svr> m_server;
    std::thread              m_client_io_service_thread;
    std::thread              m_server_thread;

    void SetUp() override {
        int16_t port               = 1234;
        m_server                   = std::make_unique<tcp_svr>(port);
        m_client_io_service_thread = std::thread([&]() { m_client_io_service.run(); });
        m_server_thread            = std::thread([&]() { m_server->run(); });
    }

    void TearDown() override {
        m_client_io_service.stop();
        m_client_io_service_thread.join();
        m_server->stop();
        m_server_thread.join();
    }
};

TEST_F(TcpFixture, NoInterleavingAsyncWrite) {
    tcp::socket client_socket(m_client_io_service);
    client_socket.connect(m_server->get_local_endpoint());

    static constexpr int    kMessageCount = 5;
    static constexpr size_t kDataSize     = 65536;

    std::array<std::array<int, kDataSize>, kMessageCount> data{};
    for (int i = 0; i < kMessageCount; i++) {
        std::fill(data.at(i).begin(), data.at(i).end(), i + 1);
    }

    while (m_server->get_session_count() != 1) {
    }

    for (auto const& arr : data) {
        m_server->async_write(boost::asio::buffer(arr));
    }

    for (int i = 0; i < kMessageCount; i++) {
        std::array<int, kDataSize> read_buffer{};
        boost::asio::read(client_socket, boost::asio::buffer(read_buffer));

        std::map<int, int> value_count{};
        for (int val : read_buffer) {
            value_count[val]++;
        }

        EXPECT_EQ(value_count.size(), 1);
    }
}

测试输出

Running main() from ./googletest/src/gtest_main.cc
[==========] Running 1 test from 1 test suite.
[----------] Global test environment set-up.
[----------] 1 test from TcpFixture
[ RUN      ] TcpFixture.NoInterleavingAsyncWrite
/home/sehe/Projects/stackoverflow/test.cpp:190: Failure
Expected equality of these values:
  value_count.size()
    Which is: 4
  1
/home/sehe/Projects/stackoverflow/test.cpp:190: Failure
Expected equality of these values:
  value_count.size()
    Which is: 4
  1
/home/sehe/Projects/stackoverflow/test.cpp:190: Failure
Expected equality of these values:
  value_count.size()
    Which is: 4
  1
/home/sehe/Projects/stackoverflow/test.cpp:190: Failure
Expected equality of these values:
  value_count.size()
    Which is: 4
  1
/home/sehe/Projects/stackoverflow/test.cpp:190: Failure
Expected equality of these values:
  value_count.size()
    Which is: 4
  1
[  FAILED  ] TcpFixture.NoInterleavingAsyncWrite (102 ms)
[----------] 1 test from TcpFixture (102 ms total)

[----------] Global test environment tear-down
[==========] 1 test from 1 test suite ran. (102 ms total)
[  PASSED  ] 0 tests.
[  FAILED  ] 1 test, listed below:
[  FAILED  ] TcpFixture.NoInterleavingAsyncWrite

 1 FAILED TEST

问题根源与修复方案

核心问题：strand未正确约束所有异步操作

当前代码中，虽然将async_write的发起逻辑post到了strand，但async_write内部会自动发起多次async_write_some调用，这些底层操作的handler并没有绑定到strand，导致多个async_write请求的底层操作可能交织执行。此外，代码中冗余的m_socket_mutex不仅没必要，还可能掩盖strand的使用问题。

修复步骤

移除冗余的socket互斥锁：strand已保证同一session的操作串行执行，无需额外加锁保护socket。
将async_write的handler绑定到strand：确保async_write内部所有async_write_some的回调都在strand上执行，彻底序列化整个写操作序列。

修改后的tcp_session::async_write方法：

void tcp_session::async_write(boost::asio::const_buffer buff) {
    boost::asio::post(m_strand, [this, self = shared_from_this(), buff]() {
        boost::asio::async_write(m_socket, buff,
            boost::asio::bind_executor(m_strand,
                [self](const boost::system::error_code& ec,
                       std::size_t bytes_transferred) {
                    if (ec) {
                        // 处理错误，比如关闭session
                        self->close();
                    }
                }));
    });
}

或者更简洁的方式，直接在发起async_write时绑定strand：

void tcp_session::async_write(boost::asio::const_buffer buff) {
    boost::asio::async_write(m_socket, buff,
        boost::asio::bind_executor(m_strand,
            [self = shared_from_this()](const boost::system::error_code& ec,
                                       std::size_t bytes_transferred) {
                if (ec) {
                    self->close();
                }
            }));
}

优化测试时序（可选）：原测试中通过忙等判断session是否建立，可改为用条件变量实现同步，避免不必要的CPU占用：
在tcp_svr中添加条件变量：

// tcp_svr.h
private:
    // ... 原有成员 ...
    std::condition_variable m_session_cv;

在do_accept中添加notify：

if (!err) {
    std::lock_guard lock(m_sessions_mutex);
    m_sessions.push_back(std::make_shared<tcp_session>(std::move(socket)));
    m_session_cv.notify_all();
}

修改测试中的等待逻辑：

// 测试中替换忙等循环
std::unique_lock<std::mutex> lock(m_server->m_sessions_mutex); // 需要将m_sessions_mutex改为public或提供接口
m_server->m_session_cv.wait(lock, [&](){ return m_server->get_session_count() == 1; });

修复原理

strand的核心作用是保证同一strand上的异步操作按提交顺序串行执行。通过将async_write的handler绑定到strand，async_write内部的所有async_write_some回调都会在strand上排队执行，确保多个async_write请求不会交织，每个缓冲区的完整数据会连续发送，客户端读取到的每个缓冲区只会包含单一数值。

内容的提问来源于stack exchange，提问作者Adwo

使用Google Test测试boost::asio async_write时数据交织问题求助

问题描述 #

代码实现 #

测试输出 #

问题根源与修复方案 #

核心问题：strand未正确约束所有异步操作 #

修复步骤 #

修复原理 #