设为首页收藏本站
切换到窄版

 找回密码
 用户注册

QQ登录

只需一步，快速开始

ACE Developer»论坛 › 开发基础 › 开源类库和工具 › Redis源码分析：Event-driven programming library |yfk ...
返回列表 发新帖
查看: 25045|回复: 0

Redis源码分析：Event-driven programming library |yfkiss

[复制链接]
发表于 2011-12-14 14:07:46 | 显示全部楼层 |阅读模式
Event-driven programming  libray，提到这个，很容易想到如雷贯耳的libevent库（libeven封装了以下三种事件的响应:IO事件,定时器事件,信号事件）。
redis的没有采用庞大的libevent库，而是自己写了一个，牺牲了一些平台通用性，但是性能非常强劲。memcache采用了libevent，有人认为这是redis的优于性能比memcache性能。没有测试过，保留意见。

x相关源码：ae.h ae.c
networking.c   anet.c  net.h  ae_epoll.c ae_select.c ae_kqueue.c
ae.h、ae.c ：event library具体实现
networking.c ：　与客户端的交互
anet.h anet.c ：　网络通信
ae_epoll.c ae_select.c ae_kqueue.c  : 不同系统多路IO封装
             ae_epoll.c ： linux平台
             ae_select.c ：unix平台
             ae_kqueue.c : BSD、APPLE

ae.c多路IO选择：
  2. #ifdef __linux__
  3. #define HAVE_EPOLL 1
  4. #endif
  6. #if (defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
  7. #define HAVE_KQUEUE 1
  8. #endif
  10. #ifdef HAVE_EPOLL
  11. #include "ae_epoll.c"
  12. #else
  13.     #ifdef HAVE_KQUEUE
  14.     #include "ae_kqueue.c"
  15.     #else
  16.     #include "ae_select.c"
  17.     #endif
  18. #endif
复制代码
多路IO封装（以ae_epoll.c为例）：
aeApiCreate：创建句柄（epoll_create）
aeApiFree：关闭句柄（close）
aeApiAddEvent：事件添加（epoll_ctl）
aeApiDelEvent：事件删除（epoll_ctl）
aeApiPoll：等待事件发生（epoll_wait）

网络简要说明：   
1. 初始化server, 等待客户端连接，并注册事件，回调函数acceptTcpHandler/acceptUnixHandler
  2. if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,
  3.         acceptTcpHandler,NULL) == AE_ERR) oom("creating file event");
  4.     if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
  5.         acceptUnixHandler,NULL) == AE_ERR) oom("creating file event");
复制代码
2. 回调函数acceptTcpHandler/acceptUnixHandler
在监听到新连接请求时，接收连接，创建redisClient对象，并注册事件（回调函数readQueryFromClient）
  2. void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
  3.     redisClient *c = (redisClient*) privdata;
  4.     char buf[REDIS_IOBUF_LEN];
  5.     int nread;
  6.     REDIS_NOTUSED(el);
  7.     REDIS_NOTUSED(mask);
  9.     nread = read(fd, buf, REDIS_IOBUF_LEN);
  10.     if (nread == -1) {
  11.         if (errno == EAGAIN) {
  12.             nread = 0;
  13.         } else {
  14.             redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
  15.             freeClient(c);
  16.             return;
  17.         }
  18.     } else if (nread == 0) {
  19.         redisLog(REDIS_VERBOSE, "Client closed connection");
  20.         freeClient(c);
  21.         return;
  22.     }
  23.     if (nread) {
  24.         c->querybuf = sdscatlen(c->querybuf,buf,nread);
  25.         c->lastinteraction = time(NULL);
  26.     } else {
  27.         return;
  28.     }
  29.     if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
  30.         sds ci = getClientInfoString(c), bytes = sdsempty();
  32.         bytes = sdscatrepr(bytes,c->querybuf,64);
  33.         redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
  34.         sdsfree(ci);
  35.         sdsfree(bytes);
  36.         freeClient(c);
  37.         return;
  38.     }
  39.     processInputBuffer(c);
  40. }
复制代码
3. 客户端请求处理：
在接收到客户端请求数据后，首先对请求进行解析，解析完成后反馈请求
  2. void processInputBuffer(redisClient *c) {
  3.     /* Keep processing while there is something in the input buffer */
  4.     while(sdslen(c->querybuf)) {
  5.         /* Immediately abort if the client is in the middle of something. */
  6.         if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
  8.         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
  9.          * written to the client. Make sure to not let the reply grow after
  10.          * this flag has been set (i.e. don't process more commands). */
  11.         if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
  13.         /* Determine request type when unknown. */
  14.         if (!c->reqtype) {
  15.             if (c->querybuf[0] == '*') {
  16.                 c->reqtype = REDIS_REQ_MULTIBULK;
  17.             } else {
  18.                 c->reqtype = REDIS_REQ_INLINE;
  19.             }
  20.         }
  22.         if (c->reqtype == REDIS_REQ_INLINE) {
  23.             if (processInlineBuffer(c) != REDIS_OK) break;
  24.         } else if (c->reqtype == REDIS_REQ_MULTIBULK) {
  25.             if (processMultibulkBuffer(c) != REDIS_OK) break;
  26.         } else {
  27.             redisPanic("Unknown request type");
  28.         }
  30.         /* Multibulk processing could see a <= 0 length. */
  31.         if (c->argc == 0) {
  32.             resetClient(c);
  33.         } else {
  34.             /* Only reset the client when the command was executed. */
  35.             if (processCommand(c) == REDIS_OK)
  36.                 resetClient(c);
  37.         }
  38.     }
  39. }
复制代码
在请求处理完成后，反馈结果.
  2. int processCommand(redisClient *c) {
  3.     .......
  4.     /* Exec the command */
  5.     if (c->flags & REDIS_MULTI &&
  6.         c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
  7.         c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
  8.     {
  9.         queueMultiCommand(c);
  10.         addReply(c,shared.queued);
  11.     } else {
  12.         if (server.vm_enabled && server.vm_max_threads > 0 &&
  13.             blockClientOnSwappedKeys(c)) return REDIS_ERR;
  14.         call(c);
  15.     }
  16.     return REDIS_OK;
  17. }void addReply(redisClient *c, robj *obj) {
  18.     if (_installWriteEvent(c) != REDIS_OK) return;
  19.     redisAssert(!server.vm_enabled || obj->storage == REDIS_VM_MEMORY);
  21.     /* This is an important place where we can avoid copy-on-write
  22.      * when there is a saving child running, avoiding touching the
  23.      * refcount field of the object if it's not needed.
  24.      *
  25.      * If the encoding is RAW and there is room in the static buffer
  26.      * we'll be able to send the object to the client without
  27.      * messing with its page. */
  28.     if (obj->encoding == REDIS_ENCODING_RAW) {
  29.         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
  30.             _addReplyObjectToList(c,obj);
  31.     } else {
  32.         /* FIXME: convert the long into string and use _addReplyToBuffer()
  33.          * instead of calling getDecodedObject. As this place in the
  34.          * code is too performance critical. */
  35.         obj = getDecodedObject(obj);
  36.         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
  37.             _addReplyObjectToList(c,obj);
  38.         decrRefCount(obj);
  39.     }
  40. }int _installWriteEvent(redisClient *c) {
  41.     if (c->fd <= 0) return REDIS_ERR;
  42.     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
  43.         (c->replstate == REDIS_REPL_NONE ||
  44.          c->replstate == REDIS_REPL_ONLINE) &&
  45.         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
  46.         sendReplyToClient, c) == AE_ERR) return REDIS_ERR;
  47.     return REDIS_OK;
  48. }
复制代码
Redis执行完客户端请求后，会调用addReply，在addReply中调用installWriteEvent来注册一个事件，并绑定事件处理函数sendReplyToClient，用来把数据发送到client。
4.主循环
处理定时事件和注册事件
  2. void aeMain(aeEventLoop *eventLoop) {
  3.     eventLoop->stop = 0;
  4.     while (!eventLoop->stop) {
  5.         if (eventLoop->beforesleep != NULL)
  6.             eventLoop->beforesleep(eventLoop);
  7.         aeProcessEvents(eventLoop, AE_ALL_EVENTS);
  8.     }
  9. }
复制代码
beforesleep通过aeSetBeforeSleepProc定义，主要是特殊处理vm和aof相关的请求
  2. int aeProcessEvents(aeEventLoop *eventLoop, int flags)
  3. {
  4.     int processed = 0, numevents;
  6.     /* Nothing to do? return ASAP */
  7.     if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
  9.     /* Note that we want call select() even if there are no
  10.      * file events to process as long as we want to process time
  11.      * events, in order to sleep until the next time event is ready
  12.      * to fire. */
  13.     if (eventLoop->maxfd != -1 ||
  14.         ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
  15.         int j;
  16.         aeTimeEvent *shortest = NULL;
  17.         struct timeval tv, *tvp;
  19.         if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
  20.             shortest = aeSearchNearestTimer(eventLoop);
  21.         if (shortest) {
  22.             long now_sec, now_ms;
  24.             /* Calculate the time missing for the nearest
  25.              * timer to fire. */
  26.             aeGetTime(&now_sec, &now_ms);
  27.             tvp = &tv;
  28.             tvp->tv_sec = shortest->when_sec - now_sec;
  29.             if (shortest->when_ms < now_ms) {
  30.                 tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
  31.                 tvp->tv_sec --;
  32.             } else {
  33.                 tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
  34.             }
  35.             if (tvp->tv_sec < 0) tvp->tv_sec = 0;
  36.             if (tvp->tv_usec < 0) tvp->tv_usec = 0;
  37.         } else {
  38.             /* If we have to check for events but need to return
  39.              * ASAP because of AE_DONT_WAIT we need to se the timeout
  40.              * to zero */
  41.             if (flags & AE_DONT_WAIT) {
  42.                 tv.tv_sec = tv.tv_usec = 0;
  43.                 tvp = &tv;
  44.             } else {
  45.                 /* Otherwise we can block */
  46.                 tvp = NULL; /* wait forever */
  47.             }
  48.         }
  50.         numevents = aeApiPoll(eventLoop, tvp);
  51.         for (j = 0; j < numevents; j++) {
  52.             aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
  53.             int mask = eventLoop->fired[j].mask;
  54.             int fd = eventLoop->fired[j].fd;
  55.             int rfired = 0;
  57.             /* note the fe->mask & mask & ... code: maybe an already processed
  58.              * event removed an element that fired and we still didn't
  59.              * processed, so we check if the event is still valid. */
  60.             if (fe->mask & mask & AE_READABLE) {
  61.                 rfired = 1;
  62.                 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
  63.             }
  64.             if (fe->mask & mask & AE_WRITABLE) {
  65.                 if (!rfired || fe->wfileProc != fe->rfileProc)
  66.                     fe->wfileProc(eventLoop,fd,fe->clientData,mask);
  67.             }
  68.             processed++;
  69.         }
  70.     }
  71.     /* Check time events */
  72.     if (flags & AE_TIME_EVENTS)
  73.         processed += processTimeEvents(eventLoop);
  75.     return processed; /* return the number of processed file/time events */
  76. }
复制代码
rfileProc和wfileProc即注册事件时定义的回调函数

作者：yfkiss 发表于2011-12-13 23:33:27 原文链接
傲游高品质VPS主机，特价促销
回复

使用道具 举报

返回列表 发新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 用户注册

本版积分规则

Archiver|手机版|小黑屋|ACE Developer ( 京ICP备06055248号 )

GMT+8, 2024-4-28 02:49 , Processed in 0.124782 second(s), 5 queries , Redis On.

Powered by Discuz! X3.5

© 2001-2023 Discuz! Team.

快速回复 返回顶部 返回列表