bufferevent_c#中bufferevent

转载请注明出处：http://blog.csdn.net/luotuo44/article/details/39344743

  和之前的《Libevent工作流程探究》一样，这里也是用一个例子来探究bufferevent的工作流程。具体的例子可以参考《Libevent使用例子，从简单到复杂》，这里就不列出了。其实要做的例子也就是bufferevent_socket_new、bufferevent_setcb、bufferevent_enable这几个函数。

        因为本文会用到《 Libevent工作流程探究 》中提到的说法，比如将一个event插入到event_base中。所以读者最好先读一下那篇博文。

 

 

 

bufferevent结构体：

 

        bufferevent其实也就是在event_base的基础上再进行一层封装，其本质还是离不开event和event_base，从bufferevent的结构体就可以看到这一点。

        bufferevent结构体中有两个event，分别用来监听同一个fd的可读事件和可写事件。因为不用一个event同时监听可读和可写呢？这是因为监听可写是困难的， 下面 会说到原因。读者也可以自问一下，自己之前有没有试过用最原始的event监听一个fd的可写。由于socket 是全双工的，所以在bufferevent结构体中，也有两个evbuffer成员，分别是读缓冲区和写缓冲区。 bufferevent结构体定义如下：

 

[cpp]  view plain copy  

1. //bufferevent_struct.h文件  
2. struct bufferevent {  
3.     struct event_base *ev_base;  
4.       
5.     //操作结构体，成员有一些函数指针。类似struct eventop结构体  
6.     const struct bufferevent_ops *be_ops;  
7.   
8.     struct event ev_read;//读事件event  
9.     struct event ev_write;//写事件event  
10.   
11.     struct evbuffer *input;//读缓冲区  
12.   
13.     struct evbuffer *output; //写缓冲区  
14.   
15.     struct event_watermark wm_read;//读水位  
16.     struct event_watermark wm_write;//写水位  
17.   
18.       
19.     bufferevent_data_cb readcb;//可读时的回调函数指针  
20.     bufferevent_data_cb writecb;//可写时的回调函数指针  
21.     bufferevent_event_cb errorcb;//错误发生时的回调函数指针  
22.     void *cbarg;//回调函数的参数  
23.   
24.     struct timeval timeout_read;//读事件event的超时值  
25.     struct timeval timeout_write;//写事件event的超时值  
26.   
27.     /** Events that are currently enabled: currently EV_READ and EV_WRITE 
28.         are supported. */  
29.     short enabled;  
30. };  

 

        如果看过Libevent的参考手册的话，应该还会知道bufferevent除了用于socket外，还可以用于socketpair 和 filter。如果用面向对象的思维，应从这个三个应用中抽出相同的部分作为父类，然后派生出三个子类。

        Libevent虽然是用C语言写的，不过它还是提取出一些公共部分，然后定义一个bufferevent_private结构体，用于保存这些公共部分成员。从集合的角度来说，bufferevent_private应该是bufferevent的一个子集，即一部分。但在Libevent中，bufferevent确实bufferevent_private的一个成员。下面是bufferevent_private结构体。

[cpp]  view plain copy  

1. //bufferevent-internal.h文件  
2. struct bufferevent_private {  
3.     struct bufferevent bev;  
4.   
5.     //设置input evbuffer的高水位时，需要一个evbuffer回调函数配合工作  
6.     struct evbuffer_cb_entry *read_watermarks_cb;  
7.   
8.     /** If set, we should free the lock when we free the bufferevent. */  
9.     //锁是Libevent自动分配的，还是用户分配的  
10.     unsigned own_lock : 1;  
11.   
12.     ...  
13.   
14.     //这个socket是否处理正在连接服务器状态  
15.     unsigned connecting : 1;  
16.     //标志连接被拒绝  
17.     unsigned connection_refused : 1;  
18.   
19.     //标志是什么原因把 读 挂起来  
20.     bufferevent_suspend_flags read_suspended;  
21.         //标志是什么原因把 写 挂起来  
22.     bufferevent_suspend_flags write_suspended;  
23.   
24.     enum bufferevent_options options;  
25.         int refcnt;// bufferevent的引用计数  
26.   
27.     //锁变量  
28.     void *lock;  
29. };  

 

 

新建一个bufferevent：

 

        函数bufferevent_socket_new可以完成这个工作。

 

[cpp]  view plain copy  

1. //bufferevent-internal.h文件  
2. struct bufferevent_ops {  
3.     const char *type;//类型名称  
4.   
5.     off_t mem_offset;//成员bev的偏移量  
6.   
7.     //启动。将event加入到event_base中  
8.     int (*enable)(struct bufferevent *, short);  
9.   
10.     //关闭。将event从event_base中删除  
11.     int (*disable)(struct bufferevent *, short);  
12.     //销毁  
13.     void (*destruct)(struct bufferevent *);  
14.     //调整event的超时值  
15.     int (*adj_timeouts)(struct bufferevent *);  
16.     /** Called to flush data. */  
17.     int (*flush)(struct bufferevent *, short, enum bufferevent_flush_mode);  
18.     //获取成员的值。具体看实现  
19.     int (*ctrl)(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);  
20. };  
21.   
22.   
23. //bufferevent_sock.c文件  
24. const struct bufferevent_ops bufferevent_ops_socket = {  
25.     "socket",  
26.     evutil_offsetof(struct bufferevent_private, bev),  
27.     be_socket_enable,  
28.     be_socket_disable,  
29.     be_socket_destruct,  
30.     be_socket_adj_timeouts,  
31.     be_socket_flush,  
32.     be_socket_ctrl,  
33. };  
34.   
35. //由于有几个不同类型的bufferevent，而且它们的enable、disable等操作是不同的。所以  
36. //需要的一些函数指针指明某个类型的bufferevent应该使用哪些操作函数。结构体bufferevent_ops_socket  
37. //就应运而生。对于socket，其操作函数如上。  
38.   
39. //bufferevent_sock.c文件  
40. struct bufferevent *  
41. bufferevent_socket_new(struct event_base *base, evutil_socket_t fd,  
42.     int options)  
43. {  
44.     struct bufferevent_private *bufev_p;  
45.     struct bufferevent *bufev;  
46.   
47.     ...//win32  
48.   
49.     //结构体内存清零，所有成员都为0  
50.     if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private)))== NULL)  
51.         return NULL;  
52.   
53.     //如果options中需要线程安全，那么就会申请锁  
54.     //会新建一个输入和输出缓存区  
55.     if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket,  
56.                     options) < 0) {  
57.         mm_free(bufev_p);  
58.         return NULL;  
59.     }  
60.     bufev = &bufev_p->bev;  
61.     //设置将evbuffer的数据向fd传  
62.     evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);  
63.   
64.     //将fd与event相关联。同一个fd关联两个event  
65.     event_assign(&bufev->ev_read, bufev->ev_base, fd,  
66.         EV_READ|EV_PERSIST, bufferevent_readcb, bufev);  
67.     event_assign(&bufev->ev_write, bufev->ev_base, fd,  
68.         EV_WRITE|EV_PERSIST, bufferevent_writecb, bufev);  
69.   
70.     //设置evbuffer的回调函数，使得外界给写缓冲区添加数据时，能触发  
71.     //写操作，这个回调对于写事件的监听是很重要的  
72.     evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);  
73.   
74.     //冻结读缓冲区的尾部，未解冻之前不能往读缓冲区追加数据  
75.     //也就是说不能从socket fd中读取数据  
76.     evbuffer_freeze(bufev->input, 0);  
77.   
78.     //冻结写缓冲区的头部，未解冻之前不能把写缓冲区的头部数据删除  
79.     //也就是说不能把数据写到socket fd  
80.     evbuffer_freeze(bufev->output, 1);  
81.   
82.     return bufev;  
83. }  

        留意函数里面的evbuffer_add_cb调用，后面会说到。

 

 

        函数在最后面会冻结两个缓冲区。其实，虽然这里冻结了，但实际上Libevent在读数据或者写数据之前会解冻的读完或者写完数据后，又会马上冻结。这主要防止数据被意外修改。用户一般不会直接调用evbuffer_freeze或者evbuffer_unfreeze函数。一切的冻结和解冻操作都由Libevent内部完成。还有一点要注意，因为这里只是把写缓冲区的头部冻结了。所以还是可以往写缓冲区的尾部追加数据。同样，此时也是可以从读缓冲区读取数据。这个是必须的。因为在Libevent内部不解冻的时候，用户需要从读缓冲区中获取数据(这相当于从socket fd中读取数据)，用户也需要把数据写到写缓冲区中(这相当于把数据写入到socket fd中)。

 

        在bufferevent_socket_new函数里面会调用函数bufferevent_init_common完成公有部分的初始化。

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. int  
3. bufferevent_init_common(struct bufferevent_private *bufev_private,  
4.     struct event_base *base,  
5.     const struct bufferevent_ops *ops,  
6.     enum bufferevent_options options)  
7. {  
8.     struct bufferevent *bufev = &bufev_private->bev;  
9.   
10.     //分配输入缓冲区  
11.     if (!bufev->input) {  
12.         if ((bufev->input = evbuffer_new()) == NULL)  
13.             return -1;  
14.     }  
15.   
16.     //分配输出缓冲区  
17.     if (!bufev->output) {  
18.         if ((bufev->output = evbuffer_new()) == NULL) {  
19.             evbuffer_free(bufev->input);  
20.             return -1;  
21.         }  
22.     }  
23.   
24.     bufev_private->refcnt = 1;//引用次数为1  
25.     bufev->ev_base = base;  
26.   
27.     /* Disable timeouts. */  
28.     //默认情况下,读和写event都是不支持超时的  
29.     evutil_timerclear(&bufev->timeout_read);  
30.     evutil_timerclear(&bufev->timeout_write);  
31.   
32.     bufev->be_ops = ops;  
33.   
34.     /* 
35.      * Set to EV_WRITE so that using bufferevent_write is going to 
36.      * trigger a callback.  Reading needs to be explicitly enabled 
37.      * because otherwise no data will be available. 
38.      */  
39.      //可写是默认支持的  
40.     bufev->enabled = EV_WRITE;  
41.   
42. #ifndef _EVENT_DISABLE_THREAD_SUPPORT  
43.     if (options & BEV_OPT_THREADSAFE) {  
44.         //申请锁。  
45.         if (bufferevent_enable_locking(bufev, NULL) < 0) {  
46.             /* cleanup */  
47.             evbuffer_free(bufev->input);  
48.             evbuffer_free(bufev->output);  
49.             bufev->input = NULL;  
50.             bufev->output = NULL;  
51.             return -1;  
52.         }  
53.     }  
54. #endif  
55.     ...//延迟调用的初始化，一般不需要用到  
56.   
57.     bufev_private->options = options;  
58.   
59.     //将evbuffer和bufferevent相关联  
60.     evbuffer_set_parent(bufev->input, bufev);  
61.     evbuffer_set_parent(bufev->output, bufev);  
62.   
63.     return 0;  
64. }  

        代码中可以看到，默认是enable  EV_WRITE的。

 

设置回调函数：

        函数bufferevent_setcb完成这个工作。该函数相当简单，也就是进行一些赋值操作。

 

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. void  
3. bufferevent_setcb(struct bufferevent *bufev,  
4.     bufferevent_data_cb readcb, bufferevent_data_cb writecb,  
5.     bufferevent_event_cb eventcb, void *cbarg)  
6. {  
7.     //bufferevent结构体内部有一个锁变量  
8.     BEV_LOCK(bufev);  
9.   
10.     bufev->readcb = readcb;  
11.     bufev->writecb = writecb;  
12.     bufev->errorcb = eventcb;  
13.   
14.     bufev->cbarg = cbarg;  
15.     BEV_UNLOCK(bufev);  
16. }  

 

        如果不想设置某个操作的回调函数，直接设置为NULL即可。

 

 

令bufferevent可以工作:

        相信读者也知道，即使调用了bufferevent_socket_new和bufferevent_setcb，这个bufferevent还是不能工作，必须调用bufferevent_enable。为什么会这样的呢？

        如果看过之前的那些博文，相信读者知道，一个event能够工作，不仅仅需要new出来，还要调用event_add函数，把这个event添加到event_base中。在本文前面的代码中，并没有看到event_add函数的调用。所以还需要调用一个函数，把event添加到event_base中。函数bufferevent_enable就是完成这个工作的。

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. int  
3. bufferevent_enable(struct bufferevent *bufev, short event)  
4. {  
5.     struct bufferevent_private *bufev_private =  
6.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
7.     short impl_events = event;  
8.     int r = 0;  
9.   
10.     //增加引用并加锁  
11.     //增加引用是为了防止其他线程调用bufferevent_free，释放了bufferevent  
12.     _bufferevent_incref_and_lock(bufev);  
13.   
14.     //挂起了读，此时不能监听读事件  
15.     if (bufev_private->read_suspended)  
16.         impl_events &= ~EV_READ;  
17.   
18.     //挂起了写，此时不能监听写事情  
19.     if (bufev_private->write_suspended)  
20.         impl_events &= ~EV_WRITE;  
21.   
22.     bufev->enabled |= event;  
23.   
24.     //调用对应类型的enbale函数。因为不同类型的bufferevent有不同的enable函数  
25.     if (impl_events && bufev->be_ops->enable(bufev, impl_events) < 0)  
26.         r = -1;  
27.   
28.     //减少引用并解锁  
29.     _bufferevent_decref_and_unlock(bufev);  
30.     return r;  
31. }  

        上面代码可以看到，最终会调用对应bufferevent类型的enable函数，对于socket bufferevent，其enable函数是be_socket_enable，代码如下：

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. int  
3. _bufferevent_add_event(struct event *ev, const struct timeval *tv)  
4. {  
5.     if (tv->tv_sec == 0 && tv->tv_usec == 0)  
6.         return event_add(ev, NULL);  
7.     else  
8.         return event_add(ev, tv);  
9. }  
10.   
11.   
12. //bufferevent_sock.c文件  
13. #define be_socket_add(ev, t)            \  
14.     _bufferevent_add_event((ev), (t))  
15.   
16.   
17. static int  
18. be_socket_enable(struct bufferevent *bufev, short event)  
19. {  
20.     if (event & EV_READ) {  
21.         if (be_socket_add(&bufev->ev_read,&bufev->timeout_read) == -1)  
22.             return -1;  
23.     }  
24.     if (event & EV_WRITE) {  
25.         if (be_socket_add(&bufev->ev_write,&bufev->timeout_write) == -1)  
26.             return -1;  
27.     }  
28.     return 0;  
29. }  

 

        如果读者熟悉Libevent的超时事件，那么可以知道Libevent是在event_add函数里面确定一个event的超时的。上面代码也展示了这一点，如果读或者写event设置了超时(即其超时值不为0)，那么就会作为参数传给event_add函数。如果读者不熟悉的Libevent的超时事件的话，可以参考《超时event的处理》。

        用户可以调用函数bufferevent_set_timeouts，设置读或者写事件的超时。代码如下：

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. int  
3. bufferevent_set_timeouts(struct bufferevent *bufev,  
4.              const struct timeval *tv_read,  
5.              const struct timeval *tv_write)  
6. {  
7.     int r = 0;  
8.     BEV_LOCK(bufev);  
9.     if (tv_read) {  
10.         bufev->timeout_read = *tv_read;  
11.     } else {  
12.         evutil_timerclear(&bufev->timeout_read);  
13.     }  
14.     if (tv_write) {  
15.         bufev->timeout_write = *tv_write;  
16.     } else {  
17.         evutil_timerclear(&bufev->timeout_write);  
18.     }  
19.   
20.     if (bufev->be_ops->adj_timeouts)  
21.         r = bufev->be_ops->adj_timeouts(bufev);  
22.     BEV_UNLOCK(bufev);  
23.   
24.     return r;  
25. }  
26.   
27.   
28. //bufferevent_sock.c文件  
29. static int  
30. be_socket_adj_timeouts(struct bufferevent *bufev)  
31. {  
32.     int r = 0;  
33.     //用户监听了读事件  
34.     if (event_pending(&bufev->ev_read, EV_READ, NULL))  
35.         if (be_socket_add(&bufev->ev_read, &bufev->timeout_read) < 0)  
36.             r = -1;  
37.   
38.     //用户监听了写事件  
39.     if (event_pending(&bufev->ev_write, EV_WRITE, NULL)) {  
40.         if (be_socket_add(&bufev->ev_write, &bufev->timeout_write) < 0)  
41.             r = -1;  
42.     }  
43.     return r;  
44. }  

        从上面代码可以看到：用户不仅仅可以设置超时值，还可以修改超时值，也是通过这个函数进行修的。当然也是可以删除超时的，直接把超时参数设置成NULL即可。

 

 

        至此，已经完成了bufferevent的初始化工作，只需调用event_base_dispatch函数，启动发动机就可以工作了。




 

处理读事件：

        现在来看一下，底层的socket fd接收数据后，bufferevent是怎么工作的。

 

读事件的水位：

        在讲读事件之前，先来看一下水位问题，函数bufferevent_setwatermark可以设置读和写的水位。这里只讲解读事件的水位。

 

        水位有两个：低水位和高水位。

        低水位比较容易懂，就是当可读的数据量到达这个低水位后，才会调用用户设置的回调函数。比如用户想每次读取100字节，那么就可以把低水位设置为100。当可读数据的字节数小于100时，即使有数据都不会打扰用户(即不会调用用户设置的回调函数)。可读数据大于等于100字节后，才会调用用户的回调函数。

        高水位是什么呢？其实，这和用户的回调函数没有关系。它的意义是：把读事件的evbuffer的数据量限制在高水位之下。比如，用户认为读缓冲区不能太大(太大的话，链表会很长)。那么用户就会设置读事件的高水位。当读缓冲区的数据量达到这个高水位后，即使socket fd还有数据没有读，也不会读进这个读缓冲区里面。一句话说，就是控制evbuffer的大小。

 

        虽然控制了evbuffer的大小，但socket fd可能还有数据。有数据就会触发可读事件，但处理可读的时候，又会发现设置了高水位，不能读取数据evbuffer。socket fd的数据没有被读完，又触发……。这个貌似是一个死循环。实际上是不会出现这个死循环的，因为Libevent发现evbuffer的数据量到达高水位后，就会把可读事件给挂起来，让它不能再触发了。Libevent使用函数bufferevent_wm_suspend_read把监听读事件的event挂起来。下面看一下Libevent是怎么把一个event挂起来的。

[cpp]  view plain copy  

1. //bufferevent-internal.h文件  
2. #define bufferevent_wm_suspend_read(b) \  
3.     bufferevent_suspend_read((b), BEV_SUSPEND_WM)  
4.   
5.   
6. //bufferevent.c文件  
7. void  
8. bufferevent_suspend_read(struct bufferevent *bufev, bufferevent_suspend_flags what)  
9. {  
10.     struct bufferevent_private *bufev_private =  
11.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
12.     BEV_LOCK(bufev);  
13.     if (!bufev_private->read_suspended)//不能挂多次  
14.         bufev->be_ops->disable(bufev, EV_READ);//实际调用be_socket_disable函数  
15.     bufev_private->read_suspended |= what;//因何而被挂起  
16.     BEV_UNLOCK(bufev);  
17. }  
18.   
19.   
20. //bufferevent_sock.c文件  
21. static int  
22. be_socket_disable(struct bufferevent *bufev, short event)  
23. {  
24.     struct bufferevent_private *bufev_p =  
25.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
26.     if (event & EV_READ) {  
27.         if (event_del(&bufev->ev_read) == -1)  
28.             return -1;  
29.     }  
30.     /* Don't actually disable the write if we are trying to connect. */  
31.     if ((event & EV_WRITE) && ! bufev_p->connecting) {  
32.         if (event_del(&bufev->ev_write) == -1)//删掉这个event  
33.             return -1;  
34.     }  
35.     return 0;  
36. }  

        居然是直接删除这个监听读事件的event，真的是挂了!!!

        看来不能随便设置高水位，因为它会暂停读。如果只想设置低水位而不想设置高水位，那么在调用bufferevent_setwatermark函数时，高水位的参数设为0即可。

 

        那么什么时候取消挂起，让bufferevent可以继续读socket 数据呢？从高水位的意义来说，当然是当evbuffer里面的数据量小于高水位时，就能再次读取socket数据了。现在来看一下Libevent是怎么恢复读的。看一下设置水位的函数bufferevent_setwatermark吧，它进行了一些为高水位埋下了一个回调函数。对，就是evbuffer的回调函数。前一篇博文说到，当evbuffer里面的数据添加或者删除时，是会触发一些回调函数的。当用户移除evbuffer的一些数据量时，Libevent就会检查这个evbuffer的数据量是否小于高水位，如果小于的话，那么就恢复 读事件。

        不说这么多了，上代码。

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. void  
3. bufferevent_setwatermark(struct bufferevent *bufev, short events,  
4.     size_t lowmark, size_t highmark)  
5. {  
6.     struct bufferevent_private *bufev_private =  
7.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
8.   
9.     BEV_LOCK(bufev);  
10.   
11.     if (events & EV_READ) {  
12.         bufev->wm_read.low = lowmark;  
13.         bufev->wm_read.high = highmark;  
14.   
15.         if (highmark) {//高水位  
16.             /* There is now a new high-water mark for read. 
17.                enable the callback if needed, and see if we should 
18.                suspend/bufferevent_wm_unsuspend. */  
19.   
20.             //还没设置高水位的回调函数  
21.             if (bufev_private->read_watermarks_cb == NULL) {  
22.                 bufev_private->read_watermarks_cb =  
23.                     evbuffer_add_cb(bufev->input,  
24.                             bufferevent_inbuf_wm_cb,  
25.                             bufev);//添加回调函数  
26.             }  
27.             evbuffer_cb_set_flags(bufev->input,  
28.                       bufev_private->read_watermarks_cb,  
29.                       EVBUFFER_CB_ENABLED|EVBUFFER_CB_NODEFER);  
30.   
31.             //设置(修改)高水位时，evbuffer的数据量已经超过了水位值  
32.             //可能是把之前的高水位调高或者调低  
33.             //挂起操作和取消挂起操作都是幂等的(即多次挂起的作用等同于挂起一次)  
34.             if (evbuffer_get_length(bufev->input) > highmark)  
35.                 bufferevent_wm_suspend_read(bufev);  
36.             else if (evbuffer_get_length(bufev->input) < highmark)//调低了  
37.                 bufferevent_wm_unsuspend_read(bufev);  
38.         } else {  
39.             //高水位值等于0，那么就要取消挂起 读事件  
40.             //取消挂起操作是幂等的  
41.             /* There is now no high-water mark for read. */  
42.             if (bufev_private->read_watermarks_cb)  
43.                 evbuffer_cb_clear_flags(bufev->input,  
44.                     bufev_private->read_watermarks_cb,  
45.                     EVBUFFER_CB_ENABLED);  
46.             bufferevent_wm_unsuspend_read(bufev);  
47.         }  
48.     }  
49.     BEV_UNLOCK(bufev);  
50. }  

        这个函数，不仅仅为高水位设置回调函数，还会检查当前evbuffer的数据量是否超过了高水位。因为这个设置水位函数可能是在bufferevent工作一段时间后才添加的，所以evbuffer是有可能已经有数据的了，因此需要检查。如果超过了水位值，那么就需要挂起读。当然也存在另外一种可能：用户之前设置过了一个比较大的高水位，挂起了读。现在发现错了，就把高水位调低一点，此时就需要恢复读。

 

        现在假设用户移除了一些evbuffer的数据，进而触发了evbuffer的回调函数，当然也就调用了函数bufferevent_inbuf_wm_cb。下面看一下这个函数是怎么恢复读的。

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. static void  
3. bufferevent_inbuf_wm_cb(struct evbuffer *buf,  
4.     const struct evbuffer_cb_info *cbinfo,  
5.     void *arg)  
6. {  
7.     struct bufferevent *bufev = arg;  
8.     size_t size;  
9.   
10.     size = evbuffer_get_length(buf);  
11.   
12.     if (size >= bufev->wm_read.high)  
13.         bufferevent_wm_suspend_read(bufev);  
14.     else  
15.         bufferevent_wm_unsuspend_read(bufev);  
16. }  
17.   
18. //bufferevent-internal.h文件  
19. #define bufferevent_wm_unsuspend_read(b) \  
20.     bufferevent_unsuspend_read((b), BEV_SUSPEND_WM)  
21.   
22. //bufferevent.c文件  
23. void  
24. bufferevent_unsuspend_read(struct bufferevent *bufev, bufferevent_suspend_flags what)  
25. {  
26.     struct bufferevent_private *bufev_private =  
27.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
28.   
29.     BEV_LOCK(bufev);  
30.     bufev_private->read_suspended &= ~what;  
31.     if (!bufev_private->read_suspended && (bufev->enabled & EV_READ))  
32.         bufev->be_ops->enable(bufev, EV_READ);//重新把event插入到event_base中  
33.     BEV_UNLOCK(bufev);  
34. }  

 

        因为用户可以手动为这个evbuffer添加数据，此时也会调用bufferevent_inbuf_wm_cb函数。此时就要检查evbuffer的数据量是否已经超过高水位了，而不能仅仅检查是否低于高水位。

 

        高水位导致读的挂起和之后读的恢复，一切工作都是由Libevent内部完成的，用户不用做任何工作。

 

从socket中读取数据：

        从前面的一系列博文可以知道，如果一个socket可读了，那么监听可读事件的event的回调函数就会被调用。这个回调函数是在bufferevent_socket_new函数中被Libevent内部设置的，设置为bufferevent_readcb函数，用户并不知情。

        当socket有数据可读时，Libevent就会监听到，然后调用bufferevent_readcb函数处理。该函数会调用evbuffer_read函数，把数据从socket fd中读取到evbuffer中。然后再调用用户在bufferevent_setcb函数中设置的读事件回调函数。所以，当用户的读事件回调函数被调用时，数据已经在evbuffer中了，用户拿来就用，无需调用read这类会阻塞的函数。

        下面看一下bufferevent_readcb函数的具体实现。

 

[cpp]  view plain copy  

1. static void  
2. bufferevent_readcb(evutil_socket_t fd, short event, void *arg)  
3. {  
4.     struct bufferevent *bufev = arg;  
5.     struct bufferevent_private *bufev_p =  
6.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
7.     struct evbuffer *input;  
8.     int res = 0;  
9.     short what = BEV_EVENT_READING;  
10.     ev_ssize_t howmuch = -1, readmax=-1;  
11.   
12.     _bufferevent_incref_and_lock(bufev);  
13.   
14.     if (event == EV_TIMEOUT) {  
15.         /* Note that we only check for event==EV_TIMEOUT. If 
16.          * event==EV_TIMEOUT|EV_READ, we can safely ignore the 
17.          * timeout, since a read has occurred */  
18.         what |= BEV_EVENT_TIMEOUT;  
19.         goto error;  
20.     }  
21.   
22.     input = bufev->input;  
23.   
24.   
25.     //用户设置了高水位  
26.     if (bufev->wm_read.high != 0) {  
27.         howmuch = bufev->wm_read.high - evbuffer_get_length(input);  
28.         /* we somehow lowered the watermark, stop reading */  
29.         if (howmuch <= 0) {  
30.             bufferevent_wm_suspend_read(bufev);  
31.             goto done;  
32.         }  
33.     }  
34.   
35.     //因为用户可以限速，所以这么要检测最大的可读大小。  
36.     //如果没有限速的话，那么将返回16384字节，即16K  
37.     //默认情况下是没有限速的。  
38.     readmax = _bufferevent_get_read_max(bufev_p);  
39.     if (howmuch < 0 || howmuch > readmax) /* The use of -1 for "unlimited" 
40.                            * uglifies this code. XXXX */  
41.         howmuch = readmax;  
42.   
43.     //一些原因导致读 被挂起，比如加锁了。  
44.     if (bufev_p->read_suspended)  
45.         goto done;  
46.   
47.     //解冻，使得可以在input的后面追加数据  
48.     evbuffer_unfreeze(input, 0);  
49.     res = evbuffer_read(input, fd, (int)howmuch); //从socket fd中读取数据  
50.     evbuffer_freeze(input, 0);//冻结  
51.   
52.     if (res == -1) {  
53.         int err = evutil_socket_geterror(fd);  
54.         if (EVUTIL_ERR_RW_RETRIABLE(err))//EINTER or EAGAIN  
55.             goto reschedule;  
56.   
57.         //不是 EINTER or EAGAIN 这两个可以重试的错误，那么就应该是其他致命的错误  
58.         //此时，应该报告给用户  
59.         what |= BEV_EVENT_ERROR;/**< unrecoverable error encountered */  
60.     } else if (res == 0) {//断开了连接  
61.         what |= BEV_EVENT_EOF;  
62.     }  
63.   
64.     if (res <= 0)  
65.         goto error;  
66.   
67.     //速率相关的操作  
68.     _bufferevent_decrement_read_buckets(bufev_p, res);  
69.   
70.       
71.     //evbuffer的数据量大于低水位值。  
72.     if (evbuffer_get_length(input) >= bufev->wm_read.low)  
73.         _bufferevent_run_readcb(bufev);//调用用户设置的回调函数  
74.   
75.     goto done;  
76.   
77.  reschedule:  
78.     goto done;  
79.   
80.  error:  
81.     //把监听可读事件的event从event_base的事件队列中删除掉.event_del  
82.     bufferevent_disable(bufev, EV_READ);//会调用be_socket_disable函数  
83.     _bufferevent_run_eventcb(bufev, what);//会调用用户设置的错误处理函数  
84.   
85.  done:  
86.     _bufferevent_decref_and_unlock(bufev);  
87. }  

 

 

        细心的读者可能会发现：对用户的读事件回调函数的触发是边缘触发的。这也就要求，在回调函数中，用户应该尽可能地把evbuffer的所有数据都读出来。如果想等到下一次回调时再读，那么需要等到下一次socketfd接收到数据才会触发用户的回调函数。如果之后socket fd一直收不到任何数据，那么即使evbuffer还有数据，用户的回调函数也不会被调用了。

 

 

 

 

处理写事件：

 

 

        对一个可读事件进行监听是比较容易的，但对于一个可写事件进行监听则比较困难。为什么呢？因为可读监听是监听fd的读缓冲区是否有数据了，如果没有数据那么就一直等待。对于可写，首先要明白“什么是可写”，可写就是fd的写缓冲区(这个缓冲区在内核)还没满，可以往里面放数据。这就有一个问题，如果写缓冲区没有满，那么就一直是可写状态。如果一个event监听了可写事件，那么这个event就会一直被触发，因为一般情况下，如果不是发大量的数据这个写缓冲区是不会满的。

        也就是说，不能监听可写事件。但我们确实要往fd中写数据，那怎么办？Libevent的做法是：当我们确实要写入数据时，才监听可写事件。也就是说我们调用bufferevent_write写入数据时，Libevent才会把监听可写事件的那个event注册到event_base中。当Libevent把数据都写入到fd的缓冲区后，Libevent又会把这个event从event_base中删除。比较烦琐。        

 

        bufferevent_writecb函数不仅仅要处理上面说到的那个问题，还要处理另外一个坑爹的问题。那就是：判断socket fd是不是已经连接上服务器了。这是因为这个socket fd是非阻塞的，所以它调用connect时，可能还没连接上就返回了。对于非阻塞socket fd，一般是通过判断这个socket是否可写，从而得知这个socket是否已经连接上服务器。如果可写，那么它就已经成功连接上服务器了。这个问题，这里先提一下，后面会详细讲。

 

        同前面的监听可读一样，Libevent是在bufferevent_socket_new函数设置可写的回调函数，为bufferevent_writecb。

[cpp]  view plain copy  

1. //bufferevent_sock.c文件  
2. static void  
3. bufferevent_writecb(evutil_socket_t fd, short event, void *arg)  
4. {  
5.     struct bufferevent *bufev = arg;  
6.     struct bufferevent_private *bufev_p =  
7.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
8.     int res = 0;  
9.     short what = BEV_EVENT_WRITING;  
10.     int connected = 0;  
11.     ev_ssize_t atmost = -1;  
12.   
13.     _bufferevent_incref_and_lock(bufev);  
14.   
15.     if (event == EV_TIMEOUT) {  
16.         /* Note that we only check for event==EV_TIMEOUT. If 
17.          * event==EV_TIMEOUT|EV_WRITE, we can safely ignore the 
18.          * timeout, since a read has occurred */  
19.         what |= BEV_EVENT_TIMEOUT;  
20.         goto error;  
21.     }  
22.   
23.     ...//判断这个socket是否已经连接上服务器了  
24.   
25.   
26.     //用户可能设置了限速，如果没有限速，那么atmost将返回16384(16K)  
27.     atmost = _bufferevent_get_write_max(bufev_p);  
28.   
29.     //一些原因导致写被挂起来了  
30.     if (bufev_p->write_suspended)  
31.         goto done;  
32.   
33.     //如果evbuffer有数据可以写到sockfd中  
34.     if (evbuffer_get_length(bufev->output)) {  
35.         //解冻链表头  
36.         evbuffer_unfreeze(bufev->output, 1);  
37.         //将output这个evbuffer的数据写到socket fd 的缓冲区中  
38.         //会把已经写到socket fd缓冲区的数据，从evbuffer中删除  
39.         res = evbuffer_write_atmost(bufev->output, fd, atmost);  
40.         evbuffer_freeze(bufev->output, 1);  
41.           
42.         if (res == -1) {  
43.             int err = evutil_socket_geterror(fd);  
44.             if (EVUTIL_ERR_RW_RETRIABLE(err))//可以恢复的错误。一般是EINTR或者EAGAIN  
45.                 goto reschedule;  
46.             what |= BEV_EVENT_ERROR;  
47.         } else if (res == 0) {//该socket已经断开连接了  
48.             what |= BEV_EVENT_EOF;  
49.         }  
50.         if (res <= 0)  
51.             goto error;  
52.     }  
53.   
54.     //如果把写缓冲区的数据都写完成了。为了防止event_base不断地触发可写  
55.     //事件，此时要把这个监听可写的event删除。  
56.     //前面的atmost限制了一次最大的可写数据。如果还没写所有的数据  
57.     //那么就不能delete这个event，而是要继续监听可写事情，知道把所有的  
58.     //数据都写到socket fd中。  
59.     if (evbuffer_get_length(bufev->output) == 0) {  
60.         event_del(&bufev->ev_write);  
61.     }  
62.   
63.   
64.     //如果evbuffer里面的数据量已经写得七七八八了，小于设置的低水位值，那么  
65.     //就会调用用户设置的写事件回调函数  
66.     if ((res || !connected) &&  
67.         evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {  
68.         _bufferevent_run_writecb(bufev);  
69.     }  
70.   
71.     goto done;  
72.   
73.  reschedule:  
74.     if (evbuffer_get_length(bufev->output) == 0) {  
75.         event_del(&bufev->ev_write);  
76.     }  
77.     goto done;  
78.   
79.  error:  
80.     bufferevent_disable(bufev, EV_WRITE);//有错误。把这个写event删除  
81.     _bufferevent_run_eventcb(bufev, what);  
82.   
83.  done:  
84.     _bufferevent_decref_and_unlock(bufev);  
85. }  

        上面代码的逻辑比较清晰，调用evbuffer_write_atmost函数把数据从evbuffer中写到evbuffer缓冲区中，此时要注意函数的返回值，因为可能写的时候发生错误。如果发生了错误，就要调用用户设置的event回调函数(网上也有人称其为错误处理函数)。

        之后，还要判断evbuffer的数据是否已经全部写到socket 的缓冲区了。如果已经全部写了，那么就要把监听写事件的event从event_base的插入队列中删除。如果还没写完，那么就不能删除，因为还要继续监听可写事件，下次接着写。

      

        现在来看一下，把监听写事件的event从event_base的插入队列中删除后，如果下次用户有数据要写的时候，怎么把这个event添加到event_base的插入队列。

        用户一般是通过bufferevent_write函数把数据写入到evbuffer(写入evbuffer后，接着就会被写入socket，所以调用bufferevent_write就相当于把数据写入到socket。)。而这个bufferevent_write函数是直接调用evbuffer_add函数的。函数evbuffer_add没有调用什么可疑的函数，能够把监听可写的event添加到event_base中。唯一的可能就是那个回调函数。对就是evbuffer的回调函数。关于evbuffer的回调函数，可以参考这里。

[cpp]  view plain copy  

1. //bufferevent.c文件  
2. int  
3. bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)  
4. {  
5.     if (evbuffer_add(bufev->output, data, size) == -1)  
6.         return (-1);  
7.   
8.     return 0;  
9. }  
10.   
11.   
12. //buffer.c文件  
13. int  
14. evbuffer_add(struct evbuffer *buf, const void *data_in, size_t datlen)  
15. {  
16.     ...  
17.   
18. out:  
19.     evbuffer_invoke_callbacks(buf);//调用回调函数  
20.     result = 0;  
21. done:  
22.     return result;  
23. }  

 

        还记得本文前面的bufferevent_socket_new函数吗？该函数里面会有

 

[cpp]  view plain copy  

1. evbuffer_add_cb(bufev->output,bufferevent_socket_outbuf_cb, bufev);  

 

        当bufferevent的写缓冲区output的数据发生变化时，函数bufferevent_socket_outbuf_cb就会被调用。现在马上飞到这个函数。

[cpp]  view plain copy  

1. //bufferevent_sock.c文件  
2. static void  
3. bufferevent_socket_outbuf_cb(struct evbuffer *buf,  
4.     const struct evbuffer_cb_info *cbinfo,  
5.     void *arg)  
6. {  
7.     struct bufferevent *bufev = arg;  
8.     struct bufferevent_private *bufev_p =  
9.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
10.   
11.     if (cbinfo->n_added && //evbuffer添加了数据  
12.         (bufev->enabled & EV_WRITE) && //默认情况下是enable EV_WRITE的  
13.         !event_pending(&bufev->ev_write, EV_WRITE, NULL) &&//这个event已经被踢出event_base了  
14.         !bufev_p->write_suspended) {//这个bufferevent的写并没有被挂起  
15.   
16.         //把这个event添加到event_base中  
17.         if (be_socket_add(&bufev->ev_write, &bufev->timeout_write) == -1) {  
18.             /* Should we log this? */  
19.         }  
20.     }  
21. }  

 

        这个函数首先进行一些判断，满足条件后就会把这个监听写事件的event添加到event_base中。其中event_pending函数就是判断这个bufev->ev_write是否已经被event_base删除了。关于event_pending，可以参考这里。

 

        对于bufferevent_write，初次使用该函数的读者可能会有疑问：调用该函数后，参数data指向的内存空间能不能马上释放，还是要等到Libevent把data指向的数据都写到socket 缓存区才能删除？其实，从前一篇博文可以看到，evbuffer_add是直接复制一份用户要发送的数据到evbuffer缓存区的。所以，调用完bufferevent_write，就可以马上释放参数data指向的内存空间。

 

 

        网上的关于Libevent的一些使用例子，包括我写的《 Libevent使用例子，从简单到复杂》，都是在主线程中调用bufferevent_write函数写入数据的。从上面的分析可以得知，是可以马上把监听可写事件的event添加到event_base中。如果是在次线程调用该函数写入数据呢？此时，主线程可能还睡眠在poll、epoll这类的多路IO复用函数上。这种情况下能不能及时唤醒主线程呢？其实是可以的，只要你的Libevent在一开始使用了线程功能。具体的分析过程可以参考《evthread_notify_base通知主线程》。上面代码中的be_socket_add会调用event_add，而在次线程调用event_add就会调用evthread_notify_base通知主线程。

 

 

bufferevent_socket_connect：

        用户可以在调用bufferevent_socket_new函数时，传一个-1作为socket的文件描述符，然后调用bufferevent_socket_connect函数连接服务器，无需自己写代码调用connect函数连接服务器。

 

        bufferevent_socket_connect函数会调用socket函数申请一个套接字fd，然后把这个fd设置成非阻塞的(这就导致了一些坑爹的事情)。接着就connect服务器，因为该socket fd是非阻塞的，所以不会等待，而是马上返回，连接这工作交给内核来完成。所以，返回后这个socket还没有真正连接上服务器。那么什么时候连接上呢？内核又是怎么通知通知用户呢？

 

        一般来说，当可以往socket fd写东西了，那就说明已经连接上了。也就是说这个socket fd变成可写状态，就连接上了。

        所以，对于“非阻塞connect”比较流行的做法是：用select或者poll这类多路IO复用函数监听该socket的可写事件。当这个socket触发了可写事件，然后再对这个socket调用getsockopt函数，做进一步的判断。

        Libevent也是这样实现的，下面来看一下bufferevent_socket_connect函数。

 

[cpp]  view plain copy  

1. //bufferevent_sock.c文件  
2. int  
3. bufferevent_socket_connect(struct bufferevent *bev,  
4.     struct sockaddr *sa, int socklen)  
5. {  
6.     struct bufferevent_private *bufev_p =  
7.         EVUTIL_UPCAST(bev, struct bufferevent_private, bev);  
8.   
9.     evutil_socket_t fd;  
10.     int r = 0;  
11.     int result=-1;  
12.     int ownfd = 0;  
13.   
14.     _bufferevent_incref_and_lock(bev);  
15.   
16.     if (!bufev_p)  
17.         goto done;  
18.   
19.   
20.     fd = bufferevent_getfd(bev);  
21.     if (fd < 0) {//该bufferevent还没有设置fd  
22.         if (!sa)  
23.             goto done;  
24.         fd = socket(sa->sa_family, SOCK_STREAM, 0);  
25.         if (fd < 0)  
26.             goto done;  
27.         if (evutil_make_socket_nonblocking(fd)<0)//设置为非阻塞  
28.             goto done;  
29.         ownfd = 1;  
30.     }  
31.     if (sa) {  
32.         r = evutil_socket_connect(&fd, sa, socklen);//非阻塞connect  
33.         if (r < 0)  
34.             goto freesock;  
35.     }  
36.     ...  
37.   
38.     //为bufferevent里面的两个event设置监听的fd  
39.     //后面会调用bufferevent_enable  
40.     bufferevent_setfd(bev, fd);  
41.       
42.     if (r == 0) {//暂时还没连接上，因为fd是非阻塞的  
43.         //此时需要监听可写事件，当可写了，并且没有错误的话，就成功连接上了  
44.         if (! be_socket_enable(bev, EV_WRITE)) {  
45.             bufev_p->connecting = 1;//标志这个sockfd正在连接  
46.             result = 0;  
47.             goto done;  
48.         }  
49.     } else if (r == 1) {//已经连接上了  
50.         /* The connect succeeded already. How very BSD of it. */  
51.         result = 0;  
52.         bufev_p->connecting = 1;   
53.         event_active(&bev->ev_write, EV_WRITE, 1);//手动激活这个event  
54.     } else {// connection refused  
55.         /* The connect failed already.  How very BSD of it. */  
56.         bufev_p->connection_refused = 1;  
57.         bufev_p->connecting = 1;  
58.         result = 0;  
59.         event_active(&bev->ev_write, EV_WRITE, 1);//手动激活这个event  
60.     }  
61.   
62.     goto done;  
63.   
64. freesock:  
65.     _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);//出现错误  
66.     if (ownfd)  
67.         evutil_closesocket(fd);  
68. done:  
69.     _bufferevent_decref_and_unlock(bev);  
70.     return result;  
71. }  

 

        这个函数比较多错误处理的代码，大致看一下就行了。有几个地方要注意，即使connect的时候被拒绝，或者已经连接上了，都会手动激活这个event。一个event即使没有加入event_base，也是可以手动激活的。具体原理参考这里。

        无论是手动激活event，或者监听到这个event可写了，都是会调用bufferevent_writecb函数。现在再次看一下该函数，只看connect部分。

[cpp]  view plain copy  

1. //bufferevent_sock.c文件  
2. static void  
3. bufferevent_writecb(evutil_socket_t fd, short event, void *arg)  
4. {  
5.     struct bufferevent_private *bufev_p =  
6.         EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);  
7.     int connected = 0;  
8.   
9.     _bufferevent_incref_and_lock(bufev);  
10.   
11.     ...  
12.     //正在连接。因为这个sockfd可能是非阻塞的，所以可能之前的connect还没  
13.     //连接上。而判断该sockfd是否成功连接上了的一个方法是判断这个sockfd是否可写  
14.     if (bufev_p->connecting) {  
15.         //c等于1，说明已经连接成功  
16.         //c等于0，说明还没连接上  
17.         //c等于-1，说明发生错误  
18.         int c = evutil_socket_finished_connecting(fd);  
19.   
20.   
21.         if (bufev_p->connection_refused) {//在bufferevent_socket_connect中被设置  
22.           bufev_p->connection_refused = 0;  
23.           c = -1;  
24.         }  
25.   
26.         if (c == 0)//还没连接上，继续监听可写吧  
27.             goto done;  
28.   
29.   
30.         //错误，或者已经连接上了  
31.         bufev_p->connecting = 0;//修改标志值  
32.           
33.         if (c < 0) {//错误  
34.             event_del(&bufev->ev_write);  
35.             event_del(&bufev->ev_read);  
36.             _bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);  
37.             goto done;  
38.   
39.         } else {//连接上了。  
40.             connected = 1;  
41.             ...//win32  
42.   
43.             //居然会调用用户设置的错误处理函数。太神奇了  
44.             _bufferevent_run_eventcb(bufev,  
45.                     BEV_EVENT_CONNECTED);  
46.             if (!(bufev->enabled & EV_WRITE) || //默认都是enable EV_WRITE的  
47.                 bufev_p->write_suspended) {  
48.                 event_del(&bufev->ev_write);//不再需要监听可写。因为已经连接上了  
49.                 goto done;  
50.             }  
51.         }  
52.     }  
53.   
54.     ...  
55.   
56.   
57.  done:  
58.     _bufferevent_decref_and_unlock(bufev);  
59. }  

        可以看到无论是connect被拒绝、发生错误或者连接上了，都在这里做统一的处理。

 

        如果已经连接上了，那么会调用用户设置event回调函数(网上也称之为错误处理函数)，通知用户已经连接上了。并且，还会把监听可写事件的event从event_base中删除，其理由在前面已经说过了。

 

 

        函数evutil_socket_finished_connecting会检查这个socket，从而得知这个socket是处于什么状态。在bufferevent_socket_connect函数中，出现的一些错误，比如被拒绝，也是能通过这个函数检查出来的。所以可以在这里做统一的处理。该函数的内部是使用。贴一下这个函数的代码吧。

[cpp]  view plain copy  

1. //evutil.c文件  
2. //Return 1 for connected, 0 for not yet, -1 for error.  
3. int  
4. evutil_socket_finished_connecting(evutil_socket_t fd)  
5. {  
6.     int e;  
7.     ev_socklen_t elen = sizeof(e);  
8.   
9.     //用来检测这个fd是否已经连接上了，这个fd是非阻塞的  
10.     //如果e的值被设为0，那么就说明连接上了。  
11.     //否则e被设置为对应的错误值。  
12.     if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void*)&e, &elen) < 0)  
13.         return -1;  
14.   
15.     if (e) {  
16.         if (EVUTIL_ERR_CONNECT_RETRIABLE(e))//还没连接上  
17.             return 0;  
18.         EVUTIL_SET_SOCKET_ERROR(e);  
19.         return -1;  
20.     }  
21.   
22.     return 1;  
23. }  

 

 

        好长啊！终于写完了。