【原版的】Redis事件驱动内核

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结构中有两个数组（server程序惯用提前分配好内存然后用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 < eventLoop->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开源技术相关信息，欢迎关注微信！

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

tp=webp" style="max-width: 100%; height: auto !important; word-wrap: break-word !important; box-sizing: border-box !important; width: auto !important; visibility: visible !important;" alt="" />

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结构中有两个数组（server程序惯用提前分配好内存然后用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 < eventLoop->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开源技术相关信息，欢迎关注微信！

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

tp=webp" _src="//bbsmax.ikafan.com/static/L3Byb3h5L2h0dHAvbW1iaXoucXBpYy5jbi9tbWJpei93RlRETUg2ZjAxSHJKaWJFZ1dMdldyaDNGUlU4dmlhd0JRQTdjeUpjb25INWxaR0ZpY2lhamJSd1Z4WnM0eUtEM0xnU3poSFEyNU01MUhzcXdqTng2c29NbXcvNjQwP3RwPXdlYnA=.jpg" style="max-width: 100%; height: auto !important; word-wrap: break-word !important; box-sizing: border-box !important; width: auto !important; visibility: visible !important;" alt="" />

版权声明：本文博客原创文章。博客，未经同意，不得转载。