如何在libevent回调外操作bufferevent？求解

问题 #

用libevent编写客户端-服务器应用程序，客户端需要持续等待控制台输入并发送数据到服务端。尝试在子线程运行event_base_dispatch，主线程读取输入后调用bufferevent_write发送，但该方式无效；但将bufferevent_write放在回调函数内却能正常工作。相关代码如下：

主线程输入发送代码片段 #

std::thread libevthr(libev_start, base);    

std::string s;
do
{
    cin >> s;
    bufferevent_write(bev, "Hello, world!", 13); 

} while(s != "xxx");

libevthr.join();      

回调函数代码片段 #

void event_cb(struct bufferevent *bev, short events, void *ptr)
{
    if (events & BEV_EVENT_CONNECTED) {
         /* We're connected to 127.0.0.1   Ordinarily we'd do
            something here, like start reading or writing. */
        bufferevent_write(bev, "Hello, world!", 13);
        std::cout << "Connected" << std::endl;
    }
    
    if (events & BEV_EVENT_ERROR) {  
        if (EVUTIL_SOCKET_ERROR() == 10054) 
            cout << "Server stopped working!" << endl;
        else 
            cout  << "Error has happened" << endl;
    }

    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR))
    {
        bufferevent_free(bev);   
    }     
}

完整代码 #

#include "UClient.h"
#include <iostream>
#include "event2/event.h"
#include "event2/listener.h"
#include "event2/bufferevent.h"
#include "event2/buffer.h"

#include <thread>

using std::cout, std::cin, std::endl;

void event_cb(struct bufferevent *bev, short events, void *ptr)
{
    if (events & BEV_EVENT_CONNECTED) {
         /* We're connected to 127.0.0.1   Ordinarily we'd do
            something here, like start reading or writing. */
        bufferevent_write(bev, "Hello, world!", 13);
        std::cout << "Connected" << std::endl;
    }
    
    if (events & BEV_EVENT_ERROR) {  
        if (EVUTIL_SOCKET_ERROR() == 10054) 
            cout << "Server stopped working!" << endl;
        else 
            cout  << "Error has happened" << endl;
    }

    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR))
    {
        bufferevent_free(bev);     
    } 

}

void write_cb(struct bufferevent *bev, void *ctx)
{
    cout << 'Data was written' << endl;
}

void libev_start(event_base *base)
{
    event_base_dispatch(base);    
}

int main()
{
    int port = 9554;
    
    struct event_base *base;         
    struct bufferevent *bev;        
    struct sockaddr_in cl_inf;         

    if (!initWinsock()) {
        perror("Failed to initialize Winsock");
        return 1;        
    }

    base = event_base_new();

    ZeroMemory(&cl_inf, sizeof(cl_inf));
    in_addr serv_ip;

    inet_pton(AF_INET, "127.0.0.1", &serv_ip);

    cl_inf.sin_family = AF_INET;
    cl_inf.sin_addr = serv_ip;
    cl_inf.sin_port = htons(port);

    bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);

    bufferevent_setcb(bev, NULL, write_cb, event_cb, NULL);  

    if (bufferevent_socket_connect(bev,
        (struct sockaddr *)&cl_inf, sizeof(cl_inf)) < 0) {
        /* Error starting connection */
        std::cout << "Can't connect to server!" << std::endl;
        bufferevent_free(bev);
        return -1;
    }

    bufferevent_enable(bev, EV_READ | EV_WRITE);

    
    std::thread libevthr(libev_start, base);    

    std::string s;
    do
    {
        cin >> s;
    } while(s != "xxx");
    
    libevthr.join();        
    std::cout << "client finished working";
    return 0;

}

解决方案 #

问题原因 #

libevent的核心事件循环（event_base_dispatch）并非线程安全，直接在主线程调用bufferevent_write会绕过libevent的线程同步机制，导致操作无法被事件循环正确处理，数据无法发送。而回调函数是在事件循环线程内执行的，所有操作都符合libevent的线程模型，因此能正常工作。

正确实现方式 #

要在非事件循环线程中触发libevent操作，必须通过自定义事件+线程安全队列的方式，将发送操作投递到事件循环线程中执行。具体步骤：

1. 用线程安全队列存储主线程读取的输入内容
2. 主线程读取输入后，通知事件循环线程有新消息待发送
3. 在事件循环线程内触发自定义事件，执行bufferevent_write操作

修改后的完整代码 #

#include "UClient.h"
#include <iostream>
#include "event2/event.h"
#include "event2/listener.h"
#include "event2/bufferevent.h"
#include "event2/buffer.h"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>

using std::cout, std::cin, std::endl;

// 线程安全的消息队列及同步工具
std::queue<std::string> msg_queue;
std::mutex queue_mutex;
std::condition_variable queue_cv;
bool stop_flag = false;

void event_cb(struct bufferevent *bev, short events, void *ptr)
{
    if (events & BEV_EVENT_CONNECTED) {
        bufferevent_write(bev, "Hello, world!", 13);
        std::cout << "Connected" << std::endl;
    }
    
    if (events & BEV_EVENT_ERROR) {  
        if (EVUTIL_SOCKET_ERROR() == 10054) 
            cout << "Server stopped working!" << endl;
        else 
            cout  << "Error has happened" << endl;
    }

    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR))
    {
        bufferevent_free(bev);
        stop_flag = true;
        queue_cv.notify_one();
    } 
}

void write_cb(struct bufferevent *bev, void *ctx)
{
    cout << "Data was written" << endl;
}

// 自定义事件回调：在事件循环线程内执行发送操作
void input_event_cb(evutil_socket_t fd, short events, void *arg)
{
    struct bufferevent *bev = (struct bufferevent*)arg;
    std::unique_lock<std::mutex> lock(queue_mutex);
    
    while (!msg_queue.empty()) {
        std::string msg = msg_queue.front();
        msg_queue.pop();
        lock.unlock();
        
        bufferevent_write(bev, msg.c_str(), msg.size());
        
        lock.lock();
    }
    
    // 重置事件，等待下一次激活
    event_del((struct event*)fd);
}

void libev_start(event_base *base, struct bufferevent *bev)
{
    // 创建自定义事件，用于触发发送逻辑
    struct event *input_event = event_new(base, -1, EV_PERSIST, input_event_cb, bev);
    
    while (!stop_flag) {
        // 等待新消息或退出信号
        std::unique_lock<std::mutex> lock(queue_mutex);
        queue_cv.wait(lock, []{ return !msg_queue.empty() || stop_flag; });
        
        if (!msg_queue.empty()) {
            event_active(input_event, EV_WRITE, 0);
            event_base_dispatch(base);
        }
    }
    
    event_free(input_event);
    event_base_loopbreak(base);
}

int main()
{
    int port = 9554;
    
    struct event_base *base;         
    struct bufferevent *bev;        
    struct sockaddr_in cl_inf;         

    if (!initWinsock()) {
        perror("Failed to initialize Winsock");
        return 1;        
    }

    base = event_base_new();

    ZeroMemory(&cl_inf, sizeof(cl_inf));
    in_addr serv_ip;

    inet_pton(AF_INET, "127.0.0.1", &serv_ip);

    cl_inf.sin_family = AF_INET;
    cl_inf.sin_addr = serv_ip;
    cl_inf.sin_port = htons(port);

    bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);

    bufferevent_setcb(bev, NULL, write_cb, event_cb, NULL);  

    if (bufferevent_socket_connect(bev,
        (struct sockaddr *)&cl_inf, sizeof(cl_inf)) < 0) {
        std::cout << "Can't connect to server!" << std::endl;
        bufferevent_free(bev);
        return -1;
    }

    bufferevent_enable(bev, EV_READ | EV_WRITE);

    std::thread libevthr(libev_start, base, bev);    

    std::string s;
    do
    {
        cin >> s;
        if (s == "xxx") {
            stop_flag = true;
            queue_cv.notify_one();
            break;
        }
        
        // 将输入存入线程安全队列
        std::lock_guard<std::mutex> lock(queue_mutex);
        msg_queue.push(s);
        queue_cv.notify_one();
        
    } while(!stop_flag);
    
    libevthr.join();        
    std::cout << "client finished working";
    return 0;
}

关键说明 #

• 用线程安全队列存储输入内容，通过互斥锁和条件变量实现线程间同步
• 自定义事件确保bufferevent_write在事件循环线程内执行，符合libevent线程安全要求
• 添加stop_flag实现主线程和事件循环线程的优雅退出

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