libevent

Posted 2020-06-14

http://blog.csdn.net/wind19/article/details/42678875

 

libevent的基本使用  

http://www.cnblogs.com/cnspace/archive/2011/07/19/2110891.html

http://yifangyou.blog.51cto.com/900206/611414

======================================================================================================================
libevent的多线程模型  
转自http://blog.csdn.net/bokee/article/details/6670550
最近在阅读memcached的源代码，打算将自己学习所得成文留念，更因为是第一次正式接触memcached，水平有限，希望大家多多交流。此系列文章按自己的理解将memcached分成几个模块分别分析。这里以memcached-1.4.6为例。


一，libevent简介
        memcached中的网络数据传输与处理完全依赖libevent。我会在另一篇文章介绍libevent。这里简单介绍libevent的用法。首先介绍相关定义。
        1）文件描述符（file descriptor）状态为可读或可写（readable/writable），是指用户线程在对此状态的文件描述符进行IO操作时，read/write系统调用会马上从内核buff读取或向内核buff写入数据并返回，而不会因为无可读数据，或无可写入空间而阻塞，直到描述符满足IO条件（IO conditions are ready），即变为可读或可写。
        2）IO事件（event）是指文件描述符状态从不可读到可读，或从不可写到可写的一次状态变化。由此可知，一个IO事件一定与一个文件描述符关联，而且分为可读事件或可写事件等不同事件类型。
        用户线程使用libevent则通常按以下步骤：
        1）用户线程通过event_init()函数创建一个event_base对象。event_base对象管理所有注册到自己内部的IO事件。多线程环境下，event_base对象不能被多个线程共享，即一个event_base对象只能对应一个线程。
        2）然后该线程通过event_add函数，将与自己感兴趣的文件描述符相关的IO事件，注册到event_base对象，同时指定事件发生时所要调用的事件处理函数（event handler）。服务器程序通常监听套接字（socket）的可读事件。比如，服务器线程注册套接字sock1的EV_READ事件，并指定event_handler1()为该事件的回调函数。libevent将IO事件封装成struct event类型对象，事件类型用EV_READ/EV_WRITE等常量标志。
        3） 注册完事件之后，线程调用event_base_loop进入循环监听（monitor）状态。该循环内部会调用epoll等IO复用函数进入阻塞状态，直到描述符上发生自己感兴趣的事件。此时，线程会调用事先指定的回调函数处理该事件。例如，当套接字sock1发生可读事件，即sock1的内核buff中已有可读数据时，被阻塞的线程立即返回（wake up）并调用event_handler1()函数来处理该次事件。
        4）处理完这次监听获得的事件后，线程再次进入阻塞状态并监听，直到下次事件发生。


二，memcached线程模型
        1，多线程的初始化与启动。
        memcached是一个典型的单进程多线程服务器。memcached启动后，main thread线程会初始化各个模块，如调用slabs_init()函数初始化内存管理模块，当然也包括创建多个worker thread线程以及初始化相关数据，最后调用event_base_loop()进入监听循环。
        本节介绍各个线程以及相关数据的创建以及初始化工作。描述具体代码前，先介绍主要数据结构。memcached将原始线程id（pthread_t）封装成LIBEVENT_THREAD对象，该对象与线程一一对应，此对象定义如下：
[cpp] view plaincopy
/*  
 * File: memcached.h  
 */  
typedef struct {  
    pthread_t thread_id; /* 线程id */  
    struct event_base *base; /* 该event_base对象管理该线程所有的IO事件 */  
    struct event notify_event; /* 此事件对象与下面的notify_receive_fd描述符关联 */  
    int notify_receive_fd; /* 与main thread通信的管道(pipe)的接收端描述符 */  
    int notify_send_fd; /* 与main thread通信的管道的发送端描述符 */  
    struct thread_stats stats; /* Stats generated by this thread */  
    struct conn_queue *new_conn_queue; /* 此队列是被锁保护的同步对象，主要用来在main thread线程与该worker thread线程之间传递初始化conn对象所需数据 */  
    cache_t *suffix_cache; /* suffix cache */} LIBEVENT_THREAD;  
  
/* 
 * File: thread.c  
 * 与所有worker thread线程对应的线程对象数组  
 */  
static LIBEVENT_THREAD *threads;  
        我们重点关注LIBEVENT_THREAD定义中，添加了中文注释的字段。具体功能见对应注释。
        main thread线程创建以及初始化worker thread的操作主要通过thread_init()和setup_thread()函数来完成。thread_init()主要代码如下：
[cpp] view plaincopy
/* 
 * File: thread.c 
 * thread_init() 
 */  
// 1) 此for循环初始化worker thread线程对象数组。  
for (i = 0; i < nthreads; i++) {  
// 1.1) 创建与main thread线程通信的管道，并初始化notify_*_fd描述符。  
    int fds[2];  
    if (pipe(fds)) {  
        perror("Can‘t create notify pipe");  
        exit(1);  
    }  
    threads[i].notify_receive_fd = fds[0];  
    threads[i].notify_send_fd = fds[1];  
  
// 1.2) 主要用来注册与threads[i]线程的notify_event_fd描述符相关的IO事件。  
    setup_thread(&threads[i]);  
}  
  
// 2) 此for循环启动worker thread线程。worker_libevent()函数内部主要调用event_base_loop()函数，即循环监听该线程注册的IO事件。  
/* Create threads after we‘ve done all the libevent setup. */  
for (i = 0; i < nthreads; i++) {  
    create_worker(worker_libevent, &threads[i]);  
}  
  
// 3) 等待所有子线程，即worker thread线程启动后，此函数才返回。  
/* Wait for all the threads to set themselves up before returning. */  
pthread_mutex_lock(&init_lock);  
while (init_count < nthreads) {  
    pthread_cond_wait(&init_cond, &init_lock);  
}  
pthread_mutex_unlock(&init_lock);  
       thread_init()函数的重点是通过setup_thread()函数为每个worker thread线程注册与notify_event_fd描述符有关的IO事件，这里的notify_event_fd描述符是该worker thread线程与main thread线程通信的管道的接收端描述符。通过注册与该描述符有关的IO事件，worker thread线程就能监听main thread线程发给自己的数据（事件）。setup_thread()函数主要代码如下：
[cpp] view plaincopy
/* 
 * File: thread.c 
 * setup_thread() 
 */  
// 1.2.1) 初始化线程对象中notify_event事件对象，并将其注册到event_base对象。  
/* Listen for notifications from other threads */  
event_set(&me->notify_event, me->notify_receive_fd,  
          EV_READ | EV_PERSIST, thread_libevent_process, me);  
event_base_set(me->base, &me->notify_event);  
  
if (event_add(&me->notify_event, 0) == -1) {  
    fprintf(stderr, "Can‘t monitor libevent notify pipe\\n");  
    exit(1);  
}  
  
// 1.2.2） 创建与初始化new_conn_queue队列。  
me->new_conn_queue = malloc(sizeof(struct conn_queue));  
if (me->new_conn_queue == NULL) {  
    perror("Failed to allocate memory for connection queue");  
    exit(EXIT_FAILURE);  
}  
cq_init(me->new_conn_queue);  
        由1.2.1)处代码段知，该worker thread线程将监听notify_event_fd描述符上的可读事件，即监听与main thread线程t通信的管道上的可读事件，并指定用thread_libevent_process()函数处理该事件。
        在3)处的代码段执行完毕后，各个worker thread线程就已经完成初始化并启动，而且各个worker thread线程开始监听并等待处理与notify_receive_fd描述符有关的IO事件。


        在worker thread线程启动后，main thread线程就要创建监听套接字（listening socket）来等待客户端连接请求。这里的监听（listen）客户端连接请求与libevent中的监听（monitor）IO事件有一定区别。在memcached中，套接字跟线程id一样，都被进一步封装。套接字被封装成conn对象，表示与客户端的连接（connection），该结构体定义很大，现选择与主题相关的几个字段，定义如下：
[cpp] view plaincopy
/*  
 * File: memcache.h  
 */    
typedef struct conn conn;    
struct conn {    
    int    sfd;    // 原始套接字    
    sasl_conn_t *sasl_conn;    
    enum conn_states  state;    // 此连接的态变变量，用于标记此连接在运行过程中的各个状态。此字段很重要。取值范围由conn_states枚举定义。        
    enum bin_substates substate;  // 与state字段类似  
    struct event event;    // 此事件对象与该套接字，即sfd字段关联。    
    short  ev_flags; // 与上一字段有关，指定监听的事件类型，如EV_READ。    
    short  which;   /** which events were just triggered */    
// 以下字段略    
}  
        下面是main thread线程创建listening socket的地方：
[cpp] view plaincopy
/* 
 * File: memcached.c 
 * server_socket() 
 */  
// 4) main thread线程在这里创建并初始化listening socket，包括注册与该conn对象相关的IO事件。注意conn_listening参数，它指定了该conn对象的初始化状态。  
if (!(listen_conn_add = conn_new(sfd, conn_listening,  
                                             EV_READ | EV_PERSIST, 1,  
                                             transport, main_base))) {  
    fprintf(stderr, "failed to create listening connection\\n");  
    exit(EXIT_FAILURE);  
}  
listen_conn_add->next = listen_conn;  
listen_conn = listen_conn_add;  
        conn_new()是memcached中一个重要的函数，此函数负责将原始套接字封装成为一个conn对象，同时会注册与该conn对象相关的IO事件，并指定该连接（conn）的初始状态。这里要注意的是listening socket的conn对象被初始化为conn_listening状态，这个细节会在后面用到。conn_new()函数的部分代码如下：
[cpp] view plaincopy
/* 
 * File: memcached.c 
 * conn_new() 
 */  
// 4.1) 初始化conn对象的相关字段。注意state字段。  
c->sfd = sfd;  
c->state = init_state;  
  
// 中间初始化步骤略  
  
// 4.2) 注册与该连接有关的IO事件  
event_set(&c->event, sfd, event_flags, event_handler, (void *)c);  
event_base_set(base, &c->event);  
c->ev_flags = event_flags;  
  
if (event_add(&c->event, 0) == -1) {  
    if (conn_add_to_freelist(c)) {  
       conn_free(c);  
    }  
    perror("event_add");  
    return NULL;  
}  
        再次提醒，连接对象的state字段是一个很重要的变量，它标志了该conn对象在运行过程中的各个状态，该字段的取值范围由conn_states枚举定义。由4处代码段，传递给conn_new()函数的conn_listening常量知，main thread线程创建了一个初始状态为conn_listening的连接。这里可以提前透露下，worker thread线程在接受main thread线程的分派后（下一节会介绍），会创建初始状态为conn_new_cmd的conn对象。        大家应该熟悉了如何注册IO事件，就不赘述了。这里要提醒的是，你会发现memcached中所有conn对象相关的处理函数都是event_handler()函数，它在内部将主要的事件处理部分交给drive_machine()函数。这个函数就全权负责处理与客户连接相关的事件。        主线程在完成初始化后，会通过event_base_loop()进入监听循环，此时主线程开始等待listening socket上的连接请求。
         2，客户端连接的建立与分派        
        上一节介绍的启动步骤完成之后，memcached的主线程开始监听listening socket上的可读事件，即等待客户端连接请求，而worker thread监听各自notify_receive_fd描述符上的可读事件，即等待来自main thread线程的数据。现在，我们来看当客户端向memcached服务器发来连接请求，memcached会如何处理。        参考上一节关于创建listening socket的部分内容，我们知道，当客户端发来连接请求，main thread线程会因listening socket发生可读事件而返回（wake up），并调用event_handler()函数来处理该请求，此函数会调用drive_machie()函数，其中处理客户端连接请求的部分如下：
[cpp] view plaincopy
/* 
 * File: memcached.c 
 * drive_machine() 
 */  
switch(c->state) {  
        case conn_listening:  
// 5) 以下数行建立与客户端的连接，得到sfd套接字。  
            addrlen = sizeof(addr);  
            if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) {  
                if (errno == EAGAIN || errno == EWOULDBLOCK) {  
                    /* these are transient, so don‘t log anything */  
                    stop = true;  
                } else if (errno == EMFILE) {  
                    if (settings.verbose > 0)  
                        fprintf(stderr, "Too many open connections\\n");  
                    accept_new_conns(false);  
                    stop = true;  
                } else {  
                    perror("accept()");  
                    stop = true;  
                }  
                break;  
            }  
            if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||  
                fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {  
                perror("setting O_NONBLOCK");  
                close(sfd);  
                break;  
            }  
// 6) 此函数将main thread线程创建的原始套接字以及一些初始化数据，传递给某个指定的worker thread线程。  
            dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,  
                                     DATA_BUFFER_SIZE, tcp_transport);  
            stop = true;  
            break;  
        这里就是conn对象的state字段发挥作用的地方了：），drive_machine()函数是一个巨大的switch语句，它根据conn对象的当前状态，即state字段的值选择执行不同的分支，因为listening socket的conn对象被初始化为conn_listening状态，所以drive_machine()函数会执行switch语句中case conn_listenning的分支，即创建并分派客户端连接部分。见5)处代码段。
        在这里，main thread线程利用dispatch_conn_new()函数，来将客户端连接套接字（这里还只是原始套接字）以及其它相关初始化数据，传递给某个worker thread线程。这里就要用到上一节提到的，main thread线程与worker thread线程之间的管道（pipe），还有线程对象中的new_conn_queue队列。代码如下：
[cpp] view plaincopy
void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,  
                       int read_buffer_size, enum network_transport transport) {  
// 6.1) 新建一个CQ_ITEM对象，并通过一个简单的取余机制选择将该CQ_ITEM对象传递给哪个worker thread。  
    CQ_ITEM *item = cqi_new();  
    int tid = (last_thread + 1) % settings.num_threads;  
    LIBEVENT_THREAD *thread = threads + tid;  
    last_thread = tid;  
  
// 6.2) 初始化新建的CQ_ITEM对象  
    item->sfd = sfd;  
    item->init_state = init_state;  
    item->event_flags = event_flags;  
    item->read_buffer_size = read_buffer_size;  
    item->transport = transport;  
  
// 6.3) 将CQ_ITEM对象推入new_conn_queue队列。  
    cq_push(thread->new_conn_queue, item);  
  
// 6.4) 向与worker thread线程连接的管道写入一字节的数据。  
    MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);  
    if (write(thread->notify_send_fd, "", 1) != 1) {  
        perror("Writing to thread notify pipe");  
    }  
}  
        此函数主要新建并初始化了一个CQ_ITEM对象，该对象包含许多创建conn对象所需用的初始化数据，如原始套接字（sfd），初始化状态（init_state）等，然后该函数将该CQ_ITEM对象传递给某个被选定的worker thread线程。在上一节介绍LIBEVENT_THREAD线程对象时说过，new_conn_queue队列用来在两个线程之间传递数据，这里就被用来向worker thread线程传递一个CQ_ITEM对象。除此之外，还要注意main thread线程向与worker thread线程连接的管道写入了一个字节的数据。此举意在触发管道另一端，即notify_receive_fd描述符的可读事件。现在我们看管道另一端的worker thread线程会发生什么。
        我们知道memcached启动后，worker thread线程会监听notify_receive_fd描述符上的可读事件。因为main thread线程向管道写入了一个字节的数据，worker thread线程会因notify_receive_fd描述符上发生可读事件而返回，并调用事先注册时指定的thread_libevent_process()函数来处理该事件，该函数主要代码如下：
[cpp] view plaincopy
/* 
 * File: thread.c 
 * thread_libevent_process() 
 */  
// 7) 从管道中读出一个字节数据，此字节即main thread线程先前向notify_send_fd描述符写入的字节。  
if (read(fd, buf, 1) != 1)  
        if (settings.verbose > 0)  
            fprintf(stderr, "Can‘t read from libevent pipe\\n");  
  
// 8) 从new_conn_queue队列中弹出一个CQ_ITEM对象，此对象即先前main thread线程推入new_conn_queue队列的对象。  
    item = cq_pop(me->new_conn_queue);  
  
// 9) 根据这个CQ_ITEM对象，创建并初始化conn对象，该对象负责客户端与该worker thread线程之间的通信。  
    if (NULL != item) {  
        conn *c = conn_new(item->sfd, item->init_state, item->event_flags,  
                           item->read_buffer_size, item->transport, me->base);  
  
// 以下略  
        注意，在7)处代码段，从管道读出的一个字节数据就是main thread线程在2.4处写入的数据。显然，该数据本身没有意义，它的目的只是触发worker thread线程这边notify_receive_fd描述符的可读事件。然后根据取得的CQ_ITEM对象创建并初始化conn对象。这里要注意的是，在6)处代码段，main thread线程将该CQ_ITEM对象的init_state字段初始化为conn_new_cmd，那么worker thread线程创建的conn对象的state字段将被初始化为conn_new_cmd。
        到这里，就完成了从客户端发送连接请求，到main thread线程创建原始套接字，再到将原始套接字等初始化数据分派到各个worker thread线程，到最后worker thread线程创建conn对象，开始负责与客户端之间通信的整个流程。worker thread就从这里开始监听该客户端连接的可读事件，并准备用event_handler()函数处理从客户端发来的数据。


参考：1）http://bachmozart.iteye.com/blog/344172