C基础 带你手写 redis ae 事件驱动模型

引言 - 整体认识

　　redis ae 事件驱动模型, 网上聊得很多. 但当你仔细看完一篇又一篇之后, 可能你看的很舒服, 但对于

作者为什么要这么写, 出发点, 好处, 缺点 ... 可能还是好模糊, 不是吗?

我们这里基于阅读的人已经了解了 io 复用大致流程且抄写过 ae 的全部代码. 好, 那开始吧, 希望后面的

点拨, 给同学们醍醐灌顶一下. 

　　先看看 ae.h 设计 

/* a simple event-driven programming library. originally i wrote this code
 * for the jim's event-loop (jim is a tcl interpreter) but later translated
 * it in form of a library for easy reuse.
 *
 * copyright (c) 2006-2012, salvatore sanfilippo <antirez at gmail dot com>
 * all rights reserved.
 *
 * redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * neither the name of redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * this software is provided by the copyright holders and contributors "as is"
 * and any express or implied warranties, including, but not limited to, the
 * implied warranties of merchantability and fitness for a particular purpose
 * are disclaimed. in no event shall the copyright owner or contributors be
 * liable for any direct, indirect, incidental, special, exemplary, or
 * consequential damages (including, but not limited to, procurement of
 * substitute goods or services; loss of use, data, or profits; or business
 * interruption) however caused and on any theory of liability, whether in
 * contract, strict liability, or tort (including negligence or otherwise)
 * arising in any way out of the use of this software, even if advised of the
 * possibility of such damage.
 */

#ifndef __ae_h__
#define __ae_h__

#include <time.h>

#define ae_ok 0
#define ae_err -1

#define ae_none 0       /* no events registered. */
#define ae_readable 1   /* fire when descriptor is readable. */
#define ae_writable 2   /* fire when descriptor is writable. */
#define ae_barrier 4    /* with writable, never fire the event if the
                           readable event already fired in the same event
                           loop iteration. useful when you want to persist
                           things to disk before sending replies, and want
                           to do that in a group fashion. */

#define ae_file_events 1
#define ae_time_events 2
#define ae_all_events (ae_file_events|ae_time_events)
#define ae_dont_wait 4
#define ae_call_after_sleep 8

#define ae_nomore -1
#define ae_deleted_event_id -1

/* macros */
#define ae_notused(v) ((void) v)

struct aeeventloop;

/* types and data structures */
typedef void aefileproc(struct aeeventloop *eventloop, int fd, void *clientdata, int mask);
typedef int aetimeproc(struct aeeventloop *eventloop, long long id, void *clientdata);
typedef void aeeventfinalizerproc(struct aeeventloop *eventloop, void *clientdata);
typedef void aebeforesleepproc(struct aeeventloop *eventloop);

/* file event structure */
typedef struct aefileevent {
    int mask; /* one of ae_(readable|writable|barrier) */
    aefileproc *rfileproc;
    aefileproc *wfileproc;
    void *clientdata;
} aefileevent;

/* time event structure */
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 *prev;
    struct aetimeevent *next;
} aetimeevent;

/* a fired event */
typedef struct aefiredevent {
    int fd;
    int mask;
} aefiredevent;

/* state of an event based program */
typedef struct aeeventloop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    long long timeeventnextid;
    time_t lasttime;     /* used to detect system clock skew */
    aefileevent *events; /* registered events */
    aefiredevent *fired; /* fired events */
    aetimeevent *timeeventhead;
    int stop;
    void *apidata; /* this is used for polling api specific data */
    aebeforesleepproc *beforesleep;
    aebeforesleepproc *aftersleep;
    int flags;
} aeeventloop;

/* prototypes */
aeeventloop *aecreateeventloop(int setsize);
void aedeleteeventloop(aeeventloop *eventloop);
void aestop(aeeventloop *eventloop);
int aecreatefileevent(aeeventloop *eventloop, int fd, int mask,
        aefileproc *proc, void *clientdata);
void aedeletefileevent(aeeventloop *eventloop, int fd, int mask);
int aegetfileevents(aeeventloop *eventloop, int fd);
long long aecreatetimeevent(aeeventloop *eventloop, long long milliseconds,
        aetimeproc *proc, void *clientdata,
        aeeventfinalizerproc *finalizerproc);
int aedeletetimeevent(aeeventloop *eventloop, long long id);
int aeprocessevents(aeeventloop *eventloop, int flags);
int aewait(int fd, int mask, long long milliseconds);
void aemain(aeeventloop *eventloop);
char *aegetapiname(void);
void aesetbeforesleepproc(aeeventloop *eventloop, aebeforesleepproc *beforesleep);
void aesetaftersleepproc(aeeventloop *eventloop, aebeforesleepproc *aftersleep);
int aegetsetsize(aeeventloop *eventloop);
int aeresizesetsize(aeeventloop *eventloop, int setsize);
void aesetdontwait(aeeventloop *eventloop, int nowait);

#endif

很多朋友首次看, 或者第一次手写完毕 ae.h 结构设计文件, 印象里 60% 是模糊不可描述 ~ 也许大致知

道这宏有点感觉应该是和 io event 事件有关吧 ... 

我这里先稍微要剧透点, 带大家快速了解这个库的结构设计的意图. c 先看结构, 比先看接口设计更容

易获取到核心信息. 上面代码中最重要四个结构分别是

　　aefileevent, aetimeevent, aefiredevent, aeeventloop

aefileevent 是文件描述符 event, 注册在 aeeventloop 中, 当触发后会生成事件结构 aefiredevent,

用于后续处理.  aetimeevent 是 timer event 同样注册在  aeeventloop 中用于触发定时事件. (太懒,

懒画图, 有兴趣朋友可以自行理解画出好理解的图) 对于  aeeventloop 内部字段的设计,  先不剧透了. 

后面正文部分会讨论一些. 

前言 - 底层解密

　　ae 文件整体结构如下

很清晰的看出 epoll, evport, kqueue, select io 复用的核心包装. 但写完整个 ae.c 发现对其设计影响

最深可能就是 ae_select.c 中兼容 select 思路.

/* select()-based ae.c module.
 *
 * copyright (c) 2009-2012, salvatore sanfilippo <antirez at gmail dot com>
 * all rights reserved.
 *
 * redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * neither the name of redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * this software is provided by the copyright holders and contributors "as is"
 * and any express or implied warranties, including, but not limited to, the
 * implied warranties of merchantability and fitness for a particular purpose
 * are disclaimed. in no event shall the copyright owner or contributors be
 * liable for any direct, indirect, incidental, special, exemplary, or
 * consequential damages (including, but not limited to, procurement of
 * substitute goods or services; loss of use, data, or profits; or business
 * interruption) however caused and on any theory of liability, whether in
 * contract, strict liability, or tort (including negligence or otherwise)
 * arising in any way out of the use of this software, even if advised of the
 * possibility of such damage.
 */


#include <sys/select.h>
#include <string.h>

typedef struct aeapistate {
    fd_set rfds, wfds;
    /* we need to have a copy of the fd sets as it's not safe to reuse
     * fd sets after select(). */
    fd_set _rfds, _wfds;
} aeapistate;

static int aeapicreate(aeeventloop *eventloop) {
    aeapistate *state = zmalloc(sizeof(aeapistate));

    if (!state) return -1;
    fd_zero(&state->rfds);
    fd_zero(&state->wfds);
    eventloop->apidata = state;
    return 0;
}

static int aeapiresize(aeeventloop *eventloop, int setsize) {
    /* just ensure we have enough room in the fd_set type. */
    if (setsize >= fd_setsize) return -1;
    return 0;
}

static void aeapifree(aeeventloop *eventloop) {
    zfree(eventloop->apidata);
}

static int aeapiaddevent(aeeventloop *eventloop, int fd, int mask) {
    aeapistate *state = eventloop->apidata;

    if (mask & ae_readable) fd_set(fd,&state->rfds);
    if (mask & ae_writable) fd_set(fd,&state->wfds);
    return 0;
}

static void aeapidelevent(aeeventloop *eventloop, int fd, int mask) {
    aeapistate *state = eventloop->apidata;

    if (mask & ae_readable) fd_clr(fd,&state->rfds);
    if (mask & ae_writable) fd_clr(fd,&state->wfds);
}

static int aeapipoll(aeeventloop *eventloop, struct timeval *tvp) {
    aeapistate *state = eventloop->apidata;
    int retval, j, numevents = 0;

    memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));
    memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));

    retval = select(eventloop->maxfd+1,
                &state->_rfds,&state->_wfds,null,tvp);
    if (retval > 0) {
        for (j = 0; j <= eventloop->maxfd; j++) {
            int mask = 0;
            aefileevent *fe = &eventloop->events[j];

            if (fe->mask == ae_none) continue;
            if (fe->mask & ae_readable && fd_isset(j,&state->_rfds))
                mask |= ae_readable;
            if (fe->mask & ae_writable && fd_isset(j,&state->_wfds))
                mask |= ae_writable;
            eventloop->fired[numevents].fd = j;
            eventloop->fired[numevents].mask = mask;
            numevents++;
        }
    }
    return numevents;
}

static char *aeapiname(void) {
    return "select";
}

作者实现这个 select 思路不是很好, 他把 ae_select.c 当做局部文件去设计, 没有想拆出来独挡一面.

其次对于 select 的第四个参数 error fds 集合没有处理(ae_epoll.c 中 epollhub 和 epollerr 是

处理). 实现层面 aeapipoll 也不够好, 推荐采用下面实现 

#include "ae.h"
#include <string.h>
#include <sys/select.h>

static int aeapipoll(aeeventloop * eventloop, struct timeval * tvp) {
    aeapistate * state = eventloop->apidata;
    int retval, j, numevents = 0;

    memcpy(&state->_rfds, &state->rfds, sizeof(fd_set));
    memcpy(&state->_wfds, &state->wfds, sizeof(fd_set));

    retval = select(eventloop->maxfd+1, &state->_rfds, &state->_wfds, null, tvp);
    for (j = 0; j <= eventloop->maxfd && numevents < retval; j++) {
        int mask = ae_none;
        aefileevent * fe = &eventloop->events[j];

        if (fe->mask == ae_none) continue;
        if (fe->mask & ae_readable && fd_isset(j, &state->_rfds))
            mask |= ae_readable;
        if (fe->mask & ae_writable && fd_isset(j, &state->_wfds))
            mask |= ae_writable;
        if (mask == ae_none) continue;

        eventloop->fired[numevents].fd = j;
        eventloop->fired[numevents].mask = mask;
        numevents++;
    }

    return numevents;
}

降低不需要处理的 ae_none 空事件.  随后的 epoll kqueue 都差不多(evport 不熟, 有心的朋友也别看)

正文 - 细节点拨

　　整体看 ae 事件模型设计, 还是有些简陋的. 我猜测是 redis 重io和内存操作, 对很多文件描述符需求

较固定,  一个文件描述符多数自始至终. 应对的场景不是那种大量的创建, 交互, 关闭. 所以整体设计也能

接受.

1.  setsize maxfd event fired 到底想表达什么?

#include <time.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <poll.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>

#include "ae.h"
#include "config.h"
#include "zmalloc.h"

/* include the best multiplexing layer supported by this system.
 * the following should be ordered by performances, descending. */
#ifdef have_evport
#include "ae_evport.c"
#else
    #ifdef have_epoll
    #include "ae_epoll.c"
    #else
        #ifdef have_kqueue
        #include "ae_epoll.c"
        #else
        #include "ae_select.c"
        #endif
    #endif
#endif

aeeventloop * aecreateeventloop(int setsize) {
    aeeventloop * eventloop;
    int i;

    if (!(eventloop = zmalloc(sizeof(*eventloop)))) goto err;

    eventloop->events = zmalloc(sizeof(aefileevent)*setsize);
    eventloop->fired = zmalloc(sizeof(aefiredevent)*setsize);
    if (!eventloop->events || !eventloop->fired) goto err;

    eventloop->setsize = setsize;
    eventloop->lasttime = time(null);
    eventloop->timeeventhead = null;
    eventloop->timeeventnextid = 0;
    eventloop->stop = 0;
    eventloop->maxfd = -1;
    eventloop->beforesleep = null;
    eventloop->aftersleep = null;
    eventloop->flags = 0;
    if (aeapicreate(eventloop) == -1) goto err;
    /* events with mask == ae_none are not set. so let's initialize the
     * vector with it. */
    for (i = 0; i < setsize; i++)
        eventloop->events[i].mask = ae_none;
    return eventloop;

err:
    if (eventloop) {
        zfree(eventloop->events);
        zfree(eventloop->fired);
        zfree(eventloop);
    }
    return null;
}

有心的同学可以关注  eventloop->events 和 eventloop->fired zmalloc 这块, 这基本已经

把之前的 ae_select.c ae_epoll.c ae_kqueue.c ... 串起来了. 分别用于存要监控的事件和有变动的事件.

对于 setsize 也是个看点我们分别看 server.c server.h config.c 局部代码

[server.c] 
server.el = aecreateeventloop(server.maxclients+config_fdset_incr);

[server.h]
#define config_min_reserved_fds 32

/* when configuring the server eventloop, we setup it so that the total number
* of file descriptors we can handle are server.maxclients + reserved_fds +
* a few more to stay safe. since reserved_fds defaults to 32, we add 96
* in order to make sure of not over provisioning more than 128 fds. */
#define config_fdset_incr (config_min_reserved_fds+96)

[config.c]
/* unsigned int configs */
createuintconfig("maxclients", null, modifiable_config, 1, uint_max, server.maxclients, 10000, integer_config, null, updatemaxclients),

可以看出来 setsize 分为两部分, 一分部分是配置的, 默认是 10000; 另外一部分是预留 128个. 

(128 分为两部分 config_min_reserved_fds = 32 + 96, 前者是 redis fd 保留的最少个数)

和上面 aecreateeventloop 相似的功能有 aeresizesetsize 

/* resize the maximum set size of the event loop.
 * if the requested set size is smaller than the current set size, but
 * there is already a file descriptor in use that is >= the requested
 * set size minus one, ae_err is returned and the operation is not
 * performed at all.
 *
 * otherwise ae_ok is returned and the operation is successful. */
int aeresizesetsize(aeeventloop *eventloop, int setsize) {
    int i;

    if (setsize == eventloop->setsize) return ae_ok;
    if (eventloop->maxfd >= setsize) return ae_err;
    if (aeapiresize(eventloop,setsize) == -1) return ae_err;

    eventloop->events = zrealloc(eventloop->events,sizeof(aefileevent)*setsize);
    eventloop->fired = zrealloc(eventloop->fired,sizeof(aefiredevent)*setsize);
    eventloop->setsize = setsize;

    /* make sure that if we created new slots, they are initialized with
     * an ae_none mask. */
    for (i = eventloop->maxfd+1; i < setsize; i++)
        eventloop->events[i].mask = ae_none;
    return ae_ok;
}

透过这两个函数希望你对 aeeventloop 中 setsize maxfd event fired 这些字段能了解透彻. 

2. aetimeevent 怎么用, 怎么设计的?

redis 中 timer event 设计比较简单, 单纯的无序时间链表. 下面这段作者意图表达的很清晰. 

/* search the first timer to fire.
 * this operation is useful to know how many time the select can be
 * put in sleep without to delay any event.
 * if there are no timers null is returned.
 *
 * note that's o(n) since time events are unsorted.
 * possible optimizations (not needed by redis so far, but...):
 * 1) insert the event in order, so that the nearest is just the head.
 *    much better but still insertion or deletion of timers is o(n).
 * 2) use a skiplist to have this operation as o(1) and insertion as o(log(n)).
 */
static aetimeevent *aesearchnearesttimer(aeeventloop *eventloop)
{
    aetimeevent *te = eventloop->timeeventhead;
    aetimeevent *nearest = null;

    while(te) {
        if (!nearest || te->when_sec < nearest->when_sec ||
                (te->when_sec == nearest->when_sec &&
                 te->when_ms < nearest->when_ms))
            nearest = te;
        te = te->next;
    }
    return nearest;
}

而其中到底怎么跑起来的呢, 我截取 processtimeevents 中部分代码, 帮读者了然于心

/* process time events */
static int processtimeevents(aeeventloop *eventloop) {
    int processed = 0;
    aetimeevent *te;
    long long maxid;
    time_t now = time(null);
...
  {
        ...
        aegettime(&now_sec, &now_ms);
        if (now_sec > te->when_sec ||
            (now_sec == te->when_sec && now_ms >= te->when_ms))
        {
            int retval;

            id = te->id;
            retval = te->timeproc(eventloop, id, te->clientdata);
            processed++;
            if (retval != ae_nomore) {
                aeaddmillisecondstonow(retval,&te->when_sec,&te->when_ms);
            } else {
                te->id = ae_deleted_event_id;
            }
        }
        ...      
  }   
...
    return processed;
}

从 retval = te->timeproc -> if 那段. 对于 id 打标为  ae_deleted_event_id 标识轮循到的时候要删除. 

一旦 retval != ae_nomore 就再次修改这个timer event 相关时间, 方便下次接着跑. 同样我们抽一个例

子出来, 同样核心也在 server.c 中 

[server.c]
    /* create the timer callback, this is our way to process many background
     * operations incrementally, like clients timeout, eviction of unaccessed
     * expired keys and so forth. */
    if (aecreatetimeevent(server.el, 1, servercron, null, null) == ae_err) {
        serverpanic("can't create event loop timers.");
        exit(1);
    }


[server.c]
int servercron(struct aeeventloop *eventloop, long long id, void *clientdata) {
...
    return 1000/server.hz;
}

整体看他这个 timer event 很骚, 返回毫秒时间后, 继续注入进去, 继续当循环轮循定时器事件使用 

static void aeaddmillisecondstonow(long long milliseconds, long *sec, long *ms) {
    long cur_sec, cur_ms, when_sec, when_ms;

    aegettime(&cur_sec, &cur_ms);
    when_sec = cur_sec + milliseconds/1000;
    when_ms = cur_ms + milliseconds%1000;
    if (when_ms >= 1000) {
        when_sec ++;
        when_ms -= 1000;
    }
    *sec = when_sec;
    *ms = when_ms;
}

设计的思路挺巧妙的. 多数正常思路通过类型特殊处理, 或者特殊地方再次主动注册. 

3. eventloop 是怎么跑的?

eventloop 奔跑思路很简单, 一个地方轮循, 内部先跑 aefileevent, 再跑 aetimeevent

[ae.c]
void aemain(aeeventloop *eventloop) {
    eventloop->stop = 0;
    while (!eventloop->stop) {
        if (eventloop->beforesleep != null)
            eventloop->beforesleep(eventloop);
        aeprocessevents(eventloop, ae_all_events|ae_call_after_sleep);
    }
}

[server.c]
int main(int argc, char **argv) {
...

    aesetbeforesleepproc(server.el,beforesleep);
    aesetaftersleepproc(server.el,aftersleep);
    aemain(server.el);
    aedeleteeventloop(server.el);
    return 0;
}

/* the end */

整体而言 redis ae 模型还是非常简单, 处理的这些的事情完全是为 redis io 定制的. 够用了. 

后续有机会我再大家分析 redis 中特定的 socket io 是怎么处理的. 

后记 - 为爱展望

❤ 错误是难免的, 欢迎有心同学指正和补充图, 文字是干瘪的 ~

　here we are again - https://music.163.com/#/song?id=27876900