[原]【原创】Redis内核之事件驱动

Redis内核之事件驱动 作者：cf （360电商技术组） 概述 R edis实现了自己的事件驱动，与开源事件库 libevent 、 libev 一样，都是基于 I/O 多路复用技术实现的。出于性能和代码精炼两方面考虑， redis 未像 memcache 一样使用 libevent 或 libev 成熟的事件

Redis内核之事件驱动

作者：cf （360电商技术组）

概述

Redis实现了自己的事件驱动，与开源事件库libevent、libev一样，都是基于I/O多路复用技术实现的。出于性能和代码精炼两方面考虑，redis未像memcache一样使用libevent或libev成熟的事件库（libevent/libev为了其通用性增加了很多扩展功能降低了使用它的性能，且代码量相比redis来说是大很多的）。

它主要支持了epoll、select、kqueue、以及基于Solaris的event ports。主要提供了对两种类型的事件驱动：

1、I/O事件，包括读事件和写事件。

2、定时器事件，包括一次性定时器和循环定时器。

源码分析

主要文件有：ae.c  ae.h  ae_epoll.c  ae_evport.c  ae_kqueue.c  ae_select.c, 其中ae.c是事件处理模块主体，ae_epoll.c  ae_kqueue.c  ae_select.c  ae_evport.c是事件处理的四种实现方式，分别对应了epoll、select、kqueue、event ports，提供了相同的接口。

#ifdef HAVE_EVPORT

#include "ae_evport.c"

#else

    #ifdef HAVE_EPOLL

    #include "ae_epoll.c"

    #else

        #ifdef HAVE_KQUEUE

        #include "ae_kqueue.c"

        #else

        #include "ae_select.c"

        #endif

    #endif

#endif

ae.c分析

redis的ae事件驱动库主要逻辑在ae.c中，其中根据使用的系统事件接口分别选择include ae_epoll.c或其他的文件。用到的主要数据结构在ae.h中定义。

主要数据结构创建：

aeCreateEventLoop

首先创建一个aeCreateEventLoop对象。该对象需要一个最大文件描述符作为参数setSize，这个参数的意义需要了解ae的数据存放结构。在aeEventLoop结构中有两个数组（服务器程序惯用提前分配好内存然后用index映射到相应位置的做法），这两个数组的大小就是这里的参数值。

ae会创建一个 setSize*sizeof(aeFileEvent) 以及一个 setSize*siezeof(aeFiredEvent) 大小的内存，用文件描述符作为其索引，可以达到O(1)的速度找到事件数据所在位置。

准备系统提供的事件模型接口，以epoll为例。ae提供了一个统一的结构名aeApiState。在包装epoll的aeApiState中有一个epfd表示epoll占用的fd，一个epoll_event *events，其实也是一个aeApiState数组，和aeFiredEvent对应，当epoll_wait()返回时，会将pending的文件描述符的信息放在aeFiredEvent数组中，包括fd和mask事件类型，此时的aeFiredEvent不是以fd作为下标的，而是把这个数组当成一个缓冲区，存放epoll_wait()返回的所有fd，同时用epoll_event数组存放epoll_wait()返回的epoll_data数据，用其数据可以填充aeFiredEvent数组的内容供ae使用找到pending的aeFileEvent对象，并在下一次进入epoll_wait()前处理完。这样完成了对epoll数据封装。

typedef struct aeApiState {

    int epfd;

    struct epoll_event *events;

} aeApiState;

aeCreateFileEvent

创建I/O事件时需要指定要注册的文件的文件描述符fd,以及要监听的事件类型mask。ae先通过fd找到其对应的aeCreateFileEvent对象所在内存位置。

typedef struct aeFileEvent {

    int mask; /* one of AE_(READABLE|WRITABLE) */

    aeFileProc *rfileProc;

    aeFileProc *wfileProc;

    void *clientData;

} aeFileEvent;

添加要监听的事件类型mask fe->mask |= mask，接着根据要监听的类型添加读事件或者写事件的回调函数，即aeFileProc，并更新maxfd以备后用。在创建文件事件的过程中还要通过宏判断后include进来的底层事件模型接口来注册I/O事件。以epoll为例，通过aeApiAddEvent将文件描述符fd和事件类型mask传给epoll操作。首先通过fd为下标找到aeCreateFileEvent对应的位置，然后取得epoll的epfd。通过EPOLL_CTL_ADD和EPOLL_CTL_MOD来加入或者修改epoll在该fd上事件的类型。

aeCreateTimeEvent

ae的定时器是用一个单链表来管理的，将定时器依次从head插入到单链表中。插入的过程中会取得未来的墙上时间作为其超时的时刻。即将当前时间加上添加定时器时给定的延迟时间。定时器结构如下。并设置超时以及注销定时器时的回调函数还用clientData。

typedef struct aeTimeEvent {

    long long id; /* time event identifier. */

    long when_sec; /* seconds */

    long when_ms; /* milliseconds */

    aeTimeProc *timeProc;

    aeEventFinalizerProc *finalizerProc;

    void *clientData;

    struct aeTimeEvent *next;

} aeTimeEvent;

事件循环：

aeMain入口函数

ae事件循环的基本结构是一个无限循环，在循环中去检测各个事件的发生。当然这里不是完全意义上的轮询，因为循环里面封装了epoll,select等事件驱动机制。

while (!eventLoop->stop) {

        if (eventLoop->beforesleep != NULL)

            eventLoop->beforesleep(eventLoop);

        aeProcessEvents(eventLoop, AE_ALL_EVENTS);

     }

beforesleep是进入一次循环之前做的操作。

aeProcessEvents

ae中最主要的逻辑就是事件处理。aeProcessEvents是处理事件的入口。在进入事件处理函数时，若没有任何事件则立即返回。

if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;

然后判断是否有定时器事件，如果有就去取得最近的一个将超时定时器的时间减去当前时间作为epoll或者select等事件接口的超时时间。该寻找过程是通过遍历单链表得到的。这样指定超时时间，在有I/O事件pending时可以处理I/O事件，若没有则可以保证从epoll或者select中返回去处理定时器事件。也可以不注册定时器事件然后将事件的flags|AE_DONT_WAIT，那么就会在poll中一直等待I/O时间的到来。

在获得事件接口超时时间后，调用封装事件接口的函数aeApiPoll。以epoll为例，首先获得apidata，然后从中获得epoll的文件描述符epfd,并用events指针指向的数组内存以及超时时间调用epoll的epoll_wait().epoll()返回时会将结果放在epoll_event数组中同时返回新的文件描述符。通过对返回数据的事件类型做判断可以填充到aeFiredEvent中fd和事件类型信息。

返回到ae的逻辑中，通过遍历aeFiredEvent数组取得fd可以找到pending事件的aeFileEvent，然后根据事件的类型去调用用户定义的I/O回调函数。

当epoll或者select超时返回时并注册了定时器事件时，通过processTimeEvents处理超时事件

if (now lastTime) {

        te = eventLoop->timeEventHead;

        while(te) {

            te->when_sec = 0;

            te = te->next;

        }

     }

这么做的意义，其实就是如果系统事件变更了，就将所有的定时器时间设为0，让他在本次循环中超时并被执行

当一个定时器被处理的时候，可能会加入新的定时，比如在定时器处理函数中加入新的定时器。此时仅应该处理上一个时间段的状态，不应该在本次循环中去处理新的定时器。因此ae在EventLoop中加入了一个timeEventNextId的成员表示此次循环中最大的定时器id+1，这样在遍历定时器列表时，先保存最大的定时器id，然后遍历过程过滤掉定时器列表可能加入新的定时器即可

if (te->id > maxId) {

            te = te->next;

            continue;

        }

这里定时器的逻辑是若单链表中的定时器时间比当前时间晚就执行定时器注册的回调函数。如果该回调函数返回正值，那么就更新定时器时间为该值之后，从而可以循环执行定时器。如果该回调函数返回AE_NOMORE，那么在执行完回调函数后注销该定时器。

清理工作

注销I/O事件

注销I/O事件不是以aeFileEvent为单位而是该I/O事件加上其监听的事件类型为对象，因此其接口为aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)。首先通过fd找到去掉aeFileEvent对象，然后获得已有的mask,对其进行减操作后，构成fd上新的mask事件类型。通过修改epoll或者select中注册的I/O事件来完成。以epoll为例，会根据该文件描述符上是否还有待等待的事件类型分别调用epoll_ctr的EPOLL_CTL_MOD或者EPOLL_CTL_DEL命令。

注销Timer时间

注销定时器事件的操作比较暴力，直接遍历链表，找到定时器id匹配的项，使用单链表删除操作进行删除。这里再删除之前会调用定时器上的finalizerProc。

注销aeEventLooop

注销aeEventLooop就是释放相关内存。

总结

感觉ae比较直观，主要提供了一个I/O事件和定时器事件的事件驱动模型。定时器的单链表逻辑可以再改进，比如用最小堆或者时间轮(Timing-Wheel)等定时器解决方法。

-------------------------------------------------------------------------------------

黑夜路人，一个关注开源技术、乐于学习、喜欢分享的程序员

博客：http://blog.csdn.net/heiyeshuwu

微博：http://weibo.com/heiyeluren

微信：heiyeluren2012  

想获取更多IT开源技术相关信息，欢迎关注微信！

微信二维码扫描快速关注本号码：

Redis内核之事件驱动

作者：cf （360电商技术组）

概述

Redis实现了自己的事件驱动，与开源事件库libevent、libev一样，都是基于I/O多路复用技术实现的。出于性能和代码精炼两方面考虑，redis未像memcache一样使用libevent或libev成熟的事件库（libevent/libev为了其通用性增加了很多扩展功能降低了使用它的性能，且代码量相比redis来说是大很多的）。

它主要支持了epoll、select、kqueue、以及基于Solaris的event ports。主要提供了对两种类型的事件驱动：

1、I/O事件，包括读事件和写事件。

2、定时器事件，包括一次性定时器和循环定时器。

源码分析

主要文件有：ae.c  ae.h  ae_epoll.c  ae_evport.c  ae_kqueue.c  ae_select.c, 其中ae.c是事件处理模块主体，ae_epoll.c  ae_kqueue.c  ae_select.c  ae_evport.c是事件处理的四种实现方式，分别对应了epoll、select、kqueue、event ports，提供了相同的接口。

#ifdef HAVE_EVPORT

#include "ae_evport.c"

#else

    #ifdef HAVE_EPOLL

    #include "ae_epoll.c"

    #else

        #ifdef HAVE_KQUEUE

        #include "ae_kqueue.c"

        #else

        #include "ae_select.c"

        #endif

    #endif

#endif

ae.c分析

redis的ae事件驱动库主要逻辑在ae.c中，其中根据使用的系统事件接口分别选择include ae_epoll.c或其他的文件。用到的主要数据结构在ae.h中定义。

主要数据结构创建：

aeCreateEventLoop

首先创建一个aeCreateEventLoop对象。该对象需要一个最大文件描述符作为参数setSize，这个参数的意义需要了解ae的数据存放结构。在aeEventLoop结构中有两个数组（服务器程序惯用提前分配好内存然后用index映射到相应位置的做法），这两个数组的大小就是这里的参数值。

ae会创建一个 setSize*sizeof(aeFileEvent) 以及一个 setSize*siezeof(aeFiredEvent) 大小的内存，用文件描述符作为其索引，可以达到O(1)的速度找到事件数据所在位置。

准备系统提供的事件模型接口，以epoll为例。ae提供了一个统一的结构名aeApiState。在包装epoll的aeApiState中有一个epfd表示epoll占用的fd，一个epoll_event *events，其实也是一个aeApiState数组，和aeFiredEvent对应，当epoll_wait()返回时，会将pending的文件描述符的信息放在aeFiredEvent数组中，包括fd和mask事件类型，此时的aeFiredEvent不是以fd作为下标的，而是把这个数组当成一个缓冲区，存放epoll_wait()返回的所有fd，同时用epoll_event数组存放epoll_wait()返回的epoll_data数据，用其数据可以填充aeFiredEvent数组的内容供ae使用找到pending的aeFileEvent对象，并在下一次进入epoll_wait()前处理完。这样完成了对epoll数据封装。

typedef struct aeApiState {

    int epfd;

    struct epoll_event *events;

} aeApiState;

aeCreateFileEvent

创建I/O事件时需要指定要注册的文件的文件描述符fd,以及要监听的事件类型mask。ae先通过fd找到其对应的aeCreateFileEvent对象所在内存位置。

typedef struct aeFileEvent {

    int mask; /* one of AE_(READABLE|WRITABLE) */

    aeFileProc *rfileProc;

    aeFileProc *wfileProc;

    void *clientData;

} aeFileEvent;

添加要监听的事件类型mask fe->mask |= mask，接着根据要监听的类型添加读事件或者写事件的回调函数，即aeFileProc，并更新maxfd以备后用。在创建文件事件的过程中还要通过宏判断后include进来的底层事件模型接口来注册I/O事件。以epoll为例，通过aeApiAddEvent将文件描述符fd和事件类型mask传给epoll操作。首先通过fd为下标找到aeCreateFileEvent对应的位置，然后取得epoll的epfd。通过EPOLL_CTL_ADD和EPOLL_CTL_MOD来加入或者修改epoll在该fd上事件的类型。

aeCreateTimeEvent

ae的定时器是用一个单链表来管理的，将定时器依次从head插入到单链表中。插入的过程中会取得未来的墙上时间作为其超时的时刻。即将当前时间加上添加定时器时给定的延迟时间。定时器结构如下。并设置超时以及注销定时器时的回调函数还用clientData。

typedef struct aeTimeEvent {

    long long id; /* time event identifier. */

    long when_sec; /* seconds */

    long when_ms; /* milliseconds */

    aeTimeProc *timeProc;

    aeEventFinalizerProc *finalizerProc;

    void *clientData;

    struct aeTimeEvent *next;

} aeTimeEvent;

事件循环：

aeMain入口函数

ae事件循环的基本结构是一个无限循环，在循环中去检测各个事件的发生。当然这里不是完全意义上的轮询，因为循环里面封装了epoll,select等事件驱动机制。

while (!eventLoop->stop) {

        if (eventLoop->beforesleep != NULL)

            eventLoop->beforesleep(eventLoop);

        aeProcessEvents(eventLoop, AE_ALL_EVENTS);

     }

beforesleep是进入一次循环之前做的操作。

aeProcessEvents

ae中最主要的逻辑就是事件处理。aeProcessEvents是处理事件的入口。在进入事件处理函数时，若没有任何事件则立即返回。

if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;

然后判断是否有定时器事件，如果有就去取得最近的一个将超时定时器的时间减去当前时间作为epoll或者select等事件接口的超时时间。该寻找过程是通过遍历单链表得到的。这样指定超时时间，在有I/O事件pending时可以处理I/O事件，若没有则可以保证从epoll或者select中返回去处理定时器事件。也可以不注册定时器事件然后将事件的flags|AE_DONT_WAIT，那么就会在poll中一直等待I/O时间的到来。

在获得事件接口超时时间后，调用封装事件接口的函数aeApiPoll。以epoll为例，首先获得apidata，然后从中获得epoll的文件描述符epfd,并用events指针指向的数组内存以及超时时间调用epoll的epoll_wait().epoll()返回时会将结果放在epoll_event数组中同时返回新的文件描述符。通过对返回数据的事件类型做判断可以填充到aeFiredEvent中fd和事件类型信息。

返回到ae的逻辑中，通过遍历aeFiredEvent数组取得fd可以找到pending事件的aeFileEvent，然后根据事件的类型去调用用户定义的I/O回调函数。

当epoll或者select超时返回时并注册了定时器事件时，通过processTimeEvents处理超时事件

if (now lastTime) {

        te = eventLoop->timeEventHead;

        while(te) {

            te->when_sec = 0;

            te = te->next;

        }

     }

这么做的意义，其实就是如果系统事件变更了，就将所有的定时器时间设为0，让他在本次循环中超时并被执行

当一个定时器被处理的时候，可能会加入新的定时，比如在定时器处理函数中加入新的定时器。此时仅应该处理上一个时间段的状态，不应该在本次循环中去处理新的定时器。因此ae在EventLoop中加入了一个timeEventNextId的成员表示此次循环中最大的定时器id+1，这样在遍历定时器列表时，先保存最大的定时器id，然后遍历过程过滤掉定时器