redis源码阅读2----------ae事件

书接上回，redis进入ae事件监听了。

相关的c文件都以ae开头，比如ae.c   ae_epoll.c   ae_ecport.c   ae_select.c   ae_kqueue.c   

ae相关的结构体在server.h的redisServer下的aeEventLoop结构体（eventloop）。

/* 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;     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;   然后在开始源码阅读前，我想先记录下ae是干什么的。 1.处理file event。最简单的事，客户端来了个set key value。你得处理了。 2.time event。在给定时间点去执行一些事情。比如持久化。     最后，声明，本次的ae流程，没有任何的外部操作，仅仅只看一个空循环的ae流程，到底干了什么事。   好的，现在我们从上次的结尾，ae.c/aeMain开始（484行）   指定stop的值。eventloop结构体在上面有提到。然后开启事件循环，其中AE_ALL_EVENTS是3（其实是(AE_FILE_EVENTS|AE_TIME_EVENTS)，前者是1，后者是2），AE_CALL_BEFORE_SLEEP是8， AE_CALL_AFTER_SLEEP是16.为什么这么设定暂时还不知道。看看后面会不会有解答。   void aeMain(aeEventLoop *eventLoop) {     eventLoop->stop = 0;     while (!eventLoop->stop) {         aeProcessEvents(eventLoop, AE_ALL_EVENTS|                                    AE_CALL_BEFORE_SLEEP|                                    AE_CALL_AFTER_SLEEP);     } }     进入aeProcessEvents 函数注释如下： /* Process every pending time event, then every pending file event  * (that may be registered by time event callbacks just processed).  * Without special flags the function sleeps until some file event  * fires, or when the next time event occurs (if any).  *  * If flags is 0, the function does nothing and returns.  * if flags has AE_ALL_EVENTS set, all the kind of events are processed.  * if flags has AE_FILE_EVENTS set, file events are processed.  * if flags has AE_TIME_EVENTS set, time events are processed.  * if flags has AE_DONT_WAIT set the function returns ASAP until all  * the events that's possible to process without to wait are processed.  * if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.  * if flags has AE_CALL_BEFORE_SLEEP set, the beforesleep callback is called.  *  * The function returns the number of events processed. */   首先判定是否有AE_TIME_EVENT或者AE_FILEEVENT. 就像上面说的，时间事件或者文件事件。       if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0; 此处，初始的flags是27，AE_TIME_EVENTS和AE_FILE_EVENTS都是初始值。      然后会进行一个if判断 if (eventLoop->maxfd != -1 ||         ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT)))   判断的作用是，即使没有文件事件需要去处理，但是只要我们想处理时间时间，就能等待到下一次事件触发。   接下来进入ae.c/ static int64_t usUntilEarliestTimer(aeEventLoop *eventLoop) 作用是返回一个时间，离第一次计时器启动还有多少微妙。这个计时器还不知道是啥玩意。只是看的注释。 没有计时器的话，就返回-1，有的话做一下如下判断： monotime now = getMonotonicUs();     return (now >= earliest->when) ? 0 : earliest->when - now; 代码非常简单，我一步一步调试得很慢，随意return的判断是1，整体函数的返回值是0. 注释还特别注明了时间复杂度是O(n)，因为时间事件是未排序的(unsorted)。 作者还注释了一下未来可能的改进措施，希望将此处的时间复杂度降到O(1)或者O（log(n)），不过看注释，作者暂时还不想动这段代码。   接下来判断刚刚的返回值： if (usUntilTimer >= 0) {             tv.tv_sec = usUntilTimer / 1000000;             tv.tv_usec = usUntilTimer % 1000000;             tvp = &tv;         } else {             /* If we have to check for events but need to return              * ASAP because of AE_DONT_WAIT we need to set the timeout              * to zero */             if (flags & AE_DONT_WAIT) {                 tv.tv_sec = tv.tv_usec = 0;                 tvp = &tv;             } else {                 /* Otherwise we can block */                 tvp = NULL; /* wait forever */             }         }   很好奇usUntilTimer<0会是什么情况，看看接下来能不能遇见。     然后下一个逻辑是： if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)             eventLoop->beforesleep(eventLoop); 官方对这段函数的注释为：redis每次进入事件驱动库的主循环时（main loop of the event driven library），在休眠以获取文件描述符之前（before to sleep for ready file descriptors），都会调用这个函数。 并且注明了此循环由两个函数调用， 第一个就是aeMain. 第二个是processEventsWhileBlocked.在加载aof/rdb期间处理客户端。（我猜测是bgaofrewrite，bgsave那几个不阻塞用户进程的命令调用的，注意是猜测） 接下来这段注释就挺让人费解了。 如果是第二种方式调用的，说明我们不想执行所有操作（比如密钥过期），但是有部分操作又必须执行。 其中最重要的是freeClientsInAsyncFreeQueue，但同时也调用其他低风险函数。 注释完毕。接下来我们来看看这个冗长的函数。   size_t zmalloc_used = zmalloc_used_memory(); 这个函数是把um赋值给used_memory。这步是个原子操作，有锁。然后return um。       if (zmalloc_used > server.stat_peak_memory)         server.stat_peak_memory = zmalloc_used; stat_peak_memory初始为0.判断成功。    以下判断第一次没有进入，ProcessingEventsWhileBlocked初始为0.注释在这里提到这里只要调用重要函数的子集，以防止从processEventsWhileBlocked()重新进入事件循环。同时还会跟踪事件个数，因为想要停止ASAP，在没有事件处理时。 暂时不明白为什么这么做。等后面看到processEventsWhileBlocked()时再说。     if (ProcessingEventsWhileBlocked) {         uint64_t processed = 0;         processed += handleClientsWithPendingReadsUsingThreads();         processed += tlsProcessPendingData();         processed += handleClientsWithPendingWrites();         processed += freeClientsInAsyncFreeQueue();         server.events_processed_while_blocked += processed;         return;     }     下一个函数：timeout.c下 handleBlockedClientsTimeout(); 注释：处理超时的被阻塞的客户端 接下来的注释有点拗口，我先上原文： this function is called in beforeSleep() in order to unblock clients that are waiting in blocking operations with a timeout set。 没理解错的话，主要是释放客户端连接，而且还是超时的被阻塞的客户端。   涉及到的结构体是 server.clients_timeout_table。 在rax.h下98-137行 因为我这里没有被阻塞或者超时的客户端，所以在第一个判断 if (raxSize(server.clients_timeout_table) == 0) return; 就直接跳出此函数了。   下一个函数：networking.c下 handleClientsWithPendingReadsUsingThreads(); 处理挂起的读客户端。 When threaded I/O is also enabled for the reading + parsing side, the  readable handler will just put normal clients into a queue of clients to  process (instead of serving them synchronously). This function runs  the queue using the I/O threads, and process them in order to accumulate  the reads in the buffers, and also parse the first command available  rendering it in the client structures 将客户端放入队列，用IO线程运行队列，并进行处理，还解析在客户端结构中呈现的第一个命令。 这个函数看上去非常复杂，应该是一个很关键的函数。但是如前文所说，我们这里只研究在空事件下的事件循环，这里不展开研究。因为在第一个循环已经return了。不过我相信在接下来的2，3章内，这个函数会被重点研究。     下一个函数：tls.c下 tlsProcessPendingData(); 处理TLS挂起的数据。（必须在flushAppendOnlyFile之前完成） 进入函数后竟然直接return 0;没有任何操作，把我吓一跳。然后仔细一看，在tls.c下有两个宏定义，如果打开了openssl，那么会进入到对应的宏下。但是我们没有打开，所以直接return了。对这部分有兴趣的同学可以把ssl打开，研究一下。     下一个函数：ae.c下 aeSetDontWait(server.el, tlsHasPendingData()); 如果tls还有没处理的数据，就不要sleep。并通知事件处理的下一次迭代将超时设置为0   下一个函数： if (server.cluster_enabled) clusterBeforeSleep(); 在sleep前调用Redis群集功能。请注意，此函数可能会更改Redis Cluster的状态（从ok更改为fail，反之亦然），因此最好在稍后使用此函数为未阻塞的客户端提供服务之前调用它。 很明显，我们没有开启cluster。这个函数的研究要在很久很久以后了。（估计至少半年后吧）   下一个函数：expire.c下 if (server.active_expire_enabled && server.masterhost == NULL)         activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST); 运行快速过期循环（如果不需要快速循环，被调用函数将尽快返回） 根据配置的过期工作调整运行参数。默认工作量为1，最大可配置工作量为10 然后对 config_keys_per_loop=20 config_cycle_fast_duration=1000 config_cycle_slow_time_perc=25 config_cycle_acceptable_stale=10 四个变量进行设置。第一次设置的大小如上   然后定义了几个全局变量，方便其他调用 static unsigned int current_db = 0; /* Next DB to test. */ static int timelimit_exit = 0;      /* Time limit hit in previous call? */ static long long last_fast_cycle = 0; /* When last fast cycle ran. */   if (checkClientPauseTimeoutAndReturnIfPaused()) return; 当客户端暂停时，数据集应该是静态的，不仅是因为客户端无法写入，还因为过期和未执行key收回。 如果前一个循环没有在时间限制内退出，不要启动快速循环，除非估计的过期key的百分比太高。此外，切勿在与快速循环总持续时间相同的时间内重复快速循环。 if (!timelimit_exit &&             server.stat_expired_stale_perc < config_cycle_acceptable_stale)             return; 这里竟然直接return了。而且看了下接下来的判断，即使这里没有return，下面的判断也会return。   不过还是把接下来的注释翻译一下，起码知道干了啥。 。。。。。。。     下一个函数： if (listLength(server.clients_waiting_acks))         processClientsWaitingReplicas(); 解除WAIT下的synchronous replication阻塞客户端     下一个函数： if (moduleCount()) moduleHandleBlockedClients(); 检查是否有客户端被实现阻塞状态的modules解锁   下一个函数： if (listLength(server.unblocked_clients))         processUnblockedClients(); 尝试为刚刚解除阻塞状态的客户端处理命令   下一个函数： if (server.get_ack_from_slaves && !checkClientPauseTimeoutAndReturnIfPaused()) {         robj *argv[3];
        argv[0] = shared.replconf;         argv[1] = shared.getack;         argv[2] = shared.special_asterick; /* Not used argument. */         replicationFeedSlaves(server.slaves, server.slaveseldb, argv, 3);         server.get_ack_from_slaves = 0;     } 如果在上一个事件循环迭代期间至少有一个客户端被阻塞，则向所有从属服务器发送ACK请求。请注意，我们在processUnblockedClients（）之后执行此操作，因此如果有多个流水线等待，并且刚刚解除阻止的等待再次被阻止，我们不必在没有其他事件循环事件的情况下等待服务器cron周期。见#6623 我们也不会在客户端暂停时发送ACK，因为它会增加复制积压工作，如果我们仍然是主服务器，它们将在暂停后发送。      下一个函数： updateFailoverStatus(); 我们可能已收到客户关于其当前偏移量的更新。注意：在接收到ACK的情况下无法执行此操作，因为故障切换将断开客户端的连接   下一个函数： trackingBroadcastInvalidationMessages(); 以广播（BCAST）模式向参与客户端缓存协议的客户端发送失效消息。   下一个函数： if (server.aof_state == AOF_ON)         flushAppendOnlyFile(0); 将AOF buffer写入磁盘。   下一个函数： handleClientsWithPendingWritesUsingThreads(); 处理写操作。   下一个操作： freeClientsInAsyncFreeQueue(); 关闭需要异步关闭的客户端   下一个函数： handleClientsBlockedOnKeys(); 尝试每隔一段时间处理被阻塞的客户端。示例：一个模块在计时器回调中调用RM_SignalKeyAsReady（因此我们根本不访问processCommand（）。   下一个函数： if (moduleCount()) moduleReleaseGIL(); 在sleep之前，让线程通过释放GIL来访问数据集。Redis主线程此时不会接触任何东西   redis的作者还特别强调，一定要将这个函数放在beforesleep的最后。 到这里beforesleep的逻辑就结束了。我们又回到aeProcessEvents。     numevents = aeApiPoll(eventLoop, tvp); 调用多路复用API，将仅在超时或某些事件触发时返回 这里返回的是event的个数。   if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)     eventLoop->aftersleep(eventLoop); 这个函数在返回事件循环多路复用API后立即被调用，control很快将通过调用不同的事件回调返回到Redis。   后面有一大段函数我们没有进入，因为那个用于write或者read的。会在后续的章节中用到。不过我在这里，先把它贴过来，然后把注释翻译一下，留一个印象。   for (j = 0; j < numevents; j++) {             aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];             int mask = eventLoop->fired[j].mask;             int fd = eventLoop->fired[j].fd;             int fired = 0; /* Number of events fired for current fd. */
            /* Normally we execute the readable event first, and the writable              * event later. This is useful as sometimes we may be able              * to serve the reply of a query immediately after processing the              * query.通常我们先执行可读事件，然后执行可写事件。因为有时我们可以在处理查询后立即提供查询回复。              *              * However if AE_BARRIER is set in the mask, our application is              * asking us to do the reverse: never fire the writable event              * after the readable. In such a case, we invert the calls.              * This is useful when, for instance, we want to do things              * in the beforeSleep() hook, like fsyncing a file to disk,              * before replying to a client. 然而，如果设置了AE_BARRIER，我们的应用程序会要求我们做相反的事情：永远不要在可读取事件之后触发可写事件。在这种情况下，我们反转调用。例如，当我们想在beforeSleep（）钩子中做一些事情，比如在回复客户机之前将文件同步到磁盘时*/             int invert = fe->mask & AE_BARRIER;
            /* Note the "fe->mask & mask & ..." code: maybe an already              * processed event removed an element that fired and we still              * didn't processed, so we check if the event is still valid.注意“fe->mask&mask&…”代码：可能一个已经处理的事件删除了一个触发的元素，而我们仍然没有处理，所以我们检查该事件是否仍然有效。              *              * Fire the readable event if the call sequence is not如果调用顺序没有反转，则触发可读事件              * inverted. */             if (!invert && fe->mask & mask & AE_READABLE) {                 fe->rfileProc(eventLoop,fd,fe->clientData,mask);                 fired++;                 fe = &eventLoop->events[fd]; /* Refresh in case of resize. */调整大小时刷新             }
            /* Fire the writable event. */触发可写事件             if (fe->mask & mask & AE_WRITABLE) {                 if (!fired || fe->wfileProc != fe->rfileProc) {                     fe->wfileProc(eventLoop,fd,fe->clientData,mask);                     fired++;                 }             }
            /* If we have to invert the call, fire the readable event now              * after the writable one. */如果我们必须反转调用，现在在可写事件之后触发可读事件。             if (invert) {                 fe = &eventLoop->events[fd]; /* Refresh in case of resize. */                 if ((fe->mask & mask & AE_READABLE) &&                     (!fired || fe->wfileProc != fe->rfileProc))                 {                     fe->rfileProc(eventLoop,fd,fe->clientData,mask);                     fired++;                 }             }
            processed++;         }     }     最后检查时间事件，代码很少，但是其中涉及的模块很多。我们后续再研究。 if (flags & AE_TIME_EVENTS)         processed += processTimeEvents(eventLoop);   最后返回全部事件的总数： return processed   至此，整个空的事件循环就结束了。