3

我正在尝试运行一个 websocket++ 示例，该示例包括从 websocket 客户端接收消息并向所有连接的客户端广播，但是我遇到了线程同步问题。

在代码示例中，方法 process_messages 等待 std:queue 上的消息

        boost::unique_lock<boost::mutex> lock(m_action_lock);

        while(m_actions.empty()) {
            m_action_cond.wait(lock);
        }

并且 on_message 处理程序在推送从客户端接收到的新消息之前锁定队列，但是当它尝试 notify_one() 时，程序失败并出现分段错误 11。

    void on_message(connection_hdl hdl, server::message_ptr msg) {
    // queue message up for sending by processing thread
    {
        boost::unique_lock<boost::mutex> lock(m_action_lock);
        m_actions.push(action(MESSAGE,msg));
        lock.unlock();
    }
    m_action_cond.notify_one();
}

该程序工作的唯一方法是评论等待（锁定），但我不确定这是否安全。

有些人可以帮助我找到去分割错误的原因吗？

完整的代码是：

#include <websocketpp/config/asio_no_tls.hpp>

#include <websocketpp/server.hpp>

#include <iostream>

#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>

typedef websocketpp::server<websocketpp::config::asio> server;

 using websocketpp::connection_hdl;
 using websocketpp::lib::placeholders::_1;
 using websocketpp::lib::placeholders::_2;
 using websocketpp::lib::bind;

/* on_open insert connection_hdl into channel
 * on_close remove connection_hdl from channel
 * on_message queue send to all channels
 */

enum action_type {
  SUBSCRIBE,
  UNSUBSCRIBE,
  MESSAGE
};

struct action {
  action(action_type t, connection_hdl h) : type(t), hdl(h) {}
  action(action_type t, server::message_ptr m) : type(t), msg(m) {}

  action_type type;
  websocketpp::connection_hdl hdl;
  server::message_ptr msg;
};

class broadcast_server {
public:
  broadcast_server() {
    // Initialize Asio Transport
    m_server.init_asio();

    // Register handler callbacks
    m_server.set_open_handler(bind(&broadcast_server::on_open,this,::_1));
    m_server.set_close_handler(bind(&broadcast_server::on_close,this,::_1));
    m_server.set_message_handler(bind(&broadcast_server::on_message,this,::_1,::_2));
}

void run(uint16_t port) {
    // listen on specified port
    m_server.listen(port);

    // Start the server accept loop
    m_server.start_accept();

    // Start the ASIO io_service run loop
    try {
        m_server.run();
    } catch (const std::exception & e) {
        std::cout << e.what() << std::endl;
    } catch (websocketpp::lib::error_code e) {
        std::cout << e.message() << std::endl;
    } catch (...) {
        std::cout << "other exception" << std::endl;
    }
}

void on_open(connection_hdl hdl) {
    boost::unique_lock<boost::mutex> lock(m_action_lock);
    //std::cout << "on_open" << std::endl;
    m_actions.push(action(SUBSCRIBE,hdl));
    lock.unlock();
    m_action_cond.notify_one();
}

void on_close(connection_hdl hdl) {
    boost::unique_lock<boost::mutex> lock(m_action_lock);
    //std::cout << "on_close" << std::endl;
    m_actions.push(action(UNSUBSCRIBE,hdl));
    lock.unlock();
    m_action_cond.notify_one();
}

void on_message(connection_hdl hdl, server::message_ptr msg) {
    // queue message up for sending by processing thread
    boost::unique_lock<boost::mutex> lock(m_action_lock);
    //std::cout << "on_message" << std::endl;
    m_actions.push(action(MESSAGE,msg));
    lock.unlock();
    m_action_cond.notify_one();
}

void process_messages() {
    while(1) {
        boost::unique_lock<boost::mutex> lock(m_action_lock);

        while(m_actions.empty()) {
            m_action_cond.wait(lock);
        }

        action a = m_actions.front();
        m_actions.pop();

        lock.unlock();

        if (a.type == SUBSCRIBE) {
            boost::unique_lock<boost::mutex> lock(m_connection_lock);
            m_connections.insert(a.hdl);
        } else if (a.type == UNSUBSCRIBE) {
            boost::unique_lock<boost::mutex> lock(m_connection_lock);
            m_connections.erase(a.hdl);
        } else if (a.type == MESSAGE) {
            boost::unique_lock<boost::mutex> lock(m_connection_lock);

            con_list::iterator it;
            for (it = m_connections.begin(); it != m_connections.end(); ++it) {
                m_server.send(*it,a.msg);
            }
        } else {
            // undefined.
        }
      }
  }
 private:
    typedef std::set<connection_hdl,std::owner_less<connection_hdl>> con_list;

    server m_server;
    con_list m_connections;
    std::queue<action> m_actions;

    boost::mutex m_action_lock;
    boost::mutex m_connection_lock;
    boost::condition_variable m_action_cond;
  };

 int main() {
broadcast_server server;

// Start a thread to run the processing loop
boost::thread(bind(&broadcast_server::process_messages,&server));

// Run the asio loop with the main thread
server.run(9002);
 }
4

1 回答 1

3

当使用 g++ 和 libstdc++ 编译 Boost 但链接到它的程序是使用 clang 和 libc++ 编译时，我可以重现此行为。libstdc++ 和 libc++ 标准库与 ABI 不兼容，因此您需要使用其中一个构建所有内容或使用另一个构建所有内容。

关于如何使用 clang/libc++ 在 C++11 模式下编译 Boost 的详细信息：

如何使用 clang++/libc++ 编译/链接 Boost？

于 2013-05-16T15:11:19.923 回答