Boost.Beast异步WebSocket服务器广播时的互斥锁断言问题

问题背景 #

基于Boost.Beast的WebSocket异步服务器示例修改为广播服务器，单客户端连接时一切正常，但多客户端场景下，广播消息能正常送达，但on_read执行完成后会触发soft_mutex.hpp中的断言失败。

核心代码实现 #

消息发送与广播逻辑 #

void send_message(std::string message) {
    buffer_.consume(buffer_.size());
    auto buffer_data = buffer_.prepare(message.size());
    std::copy(message.begin(), message.end(), boost::asio::buffer_cast<char*>(buffer_data));
    buffer_.commit(message.size());

    ws_.async_write(
        buffer_.data(),
        beast::bind_front_handler(
            &session::on_write,
            shared_from_this()));
}

void broadcast_message(std::string message) {
    std::cout << "sessions_ size: " << sessions_->size() << std::endl;

    for (const auto& session_ptr : *sessions_) {
        session_ptr->send_message(message);
    }
}

void on_read(beast::error_code ec, std::size_t bytes_transferred) {
    boost::ignore_unused(bytes_transferred);

    if(ec == websocket::error::closed) return;

    if(ec) return fail(ec, "read");

    for (const auto& session_ptr : *sessions_) {
        session_ptr->send_message("test");
    }
}

会话管理与启动修改 #

// main函数中
std::list<std::shared_ptr<session>> sessions;
std::make_shared<listener>(ioc, tcp::endpoint{address, port}, &sessions)->run();

// listener中创建会话时
auto new_session = std::make_shared<session>(std::move(socket), sessions_);
sessions_->push_back(new_session);
new_session->run();

报错信息 #

触发soft_mutex.hpp中的断言：

template<class T>
bool
try_lock(T const*)
{
    // If this assert goes off it means you are attempting to
    // simultaneously initiate more than one of same asynchronous
    // operation, which is not allowed. For example, you must wait
    // for an async_read to complete before performing another
    // async_read.
    //
    BOOST_ASSERT(id_ != T::id);
    if(id_ != 0)
        return false;
    id_ = T::id;
    return true;
}

具体报错输出：

/usr/include/boost/beast/websocket/detail/soft_mutex.hpp:89: bool boost::beast::websocket::detail::soft_mutex::try_lock(const T*) [with T = boost::beast::websocket::stream<boost::beast::basic_stream<boost::asio::ip::tcp, boost::asio::execution::any_executor<boost::asio::execution::context_as_t<boost::asio::execution_context&>, boost::asio::execution::detail::blocking::never_t<0>, boost::asio::execution::prefer_only<boost::asio::execution::detail::blocking::possibly_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::outstanding_work::tracked_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::outstanding_work::untracked_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::relationship::fork_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::relationship::continuation_t<0> > >, boost::beast::unlimited_rate_policy> >::read_some_op<boost::beast::websocket::stream<boost::beast::basic_stream<boost::asio::ip::tcp, boost::asio::execution::any_executor<boost::asio::execution::context_as_t<boost::asio::execution_context&>, boost::asio::execution::detail::blocking::never_t<0>, boost::asio::execution::prefer_only<boost::asio::execution::detail::blocking::possibly_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::outstanding_work::tracked_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::outstanding_work::untracked_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::relationship::fork_t<0> >, boost::asio::execution::prefer_only<boost::asio::execution::detail::relationship::continuation_t<0> > >, boost::beast::unlimited_rate_policy> >::read_op<boost::beast::detail::bind_front_wrapper<void (session::*)(boost::system::error_code, long unsigned int), std::shared_ptr<session> >, boost::beast::basic_flat_buffer<std::allocator<char> > >, boost::asio::mutable_buffer>]: Assertion `id_ != T::id' failed.

问题分析与修复方案 #

断言失败的本质是同一个WebSocket流上同时发起了多个同类型异步操作，这里是重复触发了async_read；同时你的send_message逻辑存在并发风险，可能导致重复发起async_write。

1. 为会话添加发送消息队列，避免重复发起async_write #

Boost.Beast不允许同时发起多个async_write，需要用队列缓冲消息，确保前一个写操作完成后再发起下一个：

class session : public std::enable_shared_from_this<session> {
private:
    // ... 原有成员变量
    std::queue<std::string> send_queue_;
    bool is_writing_ = false;

    void do_write() {
        if (is_writing_ || send_queue_.empty()) return;
        is_writing_ = true;
        auto msg = std::move(send_queue_.front());
        send_queue_.pop();
        ws_.async_write(
            boost::asio::buffer(msg),
            beast::bind_front_handler(
                &session::on_write,
                shared_from_this()));
    }

public:
    // ... 原有方法
    void send_message(std::string message) {
        bool was_empty = send_queue_.empty();
        send_queue_.push(std::move(message));
        if (was_empty) {
            do_write();
        }
    }

    void on_write(beast::error_code ec, std::size_t bytes_transferred) {
        boost::ignore_unused(bytes_transferred);
        is_writing_ = false;
        if (ec) return fail(ec, "write");
        do_write(); // 继续发送队列中的下一条消息
    }
};

2. 为会话列表添加线程安全保护 #

即使单线程运行，异步回调的执行时机可能导致会话列表被并发修改，需要用互斥锁保护：

#include <mutex>

// 全局或listener类内维护
std::list<std::shared_ptr<session>> sessions;
std::mutex sessions_mutex_;

// 添加会话时
std::lock_guard<std::mutex> lock(sessions_mutex_);
sessions.push_back(new_session);

// 广播消息时
std::lock_guard<std::mutex> lock(sessions_mutex_);
for (const auto& session_ptr : sessions) {
    session_ptr->send_message(message);
}

// 会话关闭时从列表移除
void on_close(beast::error_code ec) {
    std::lock_guard<std::mutex> lock(sessions_mutex_);
    sessions.remove(shared_from_this());
    // ... 原有关闭逻辑
}

3. 确保async_read只被串行触发 #

检查on_read逻辑，确保每次读操作完成后只发起一次新的async_read，不要在其他地方重复调用。原示例中run()方法发起第一次读操作，on_read完成后再次调用ws_.async_read(buffer_, ...)的逻辑要保留，且不能被其他代码触发。

关键注意点 #

• Boost.Beast的WebSocket流严格禁止同时发起多个同类型异步操作（如多个async_read或async_write），必须等待前一个操作完成后再发起下一个。
• 会话列表的所有访问操作必须加锁，避免并发修改导致的迭代器失效或数据混乱。
• 发送消息必须通过队列缓冲，不能直接连续调用async_write。

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