Redis is an event-driven server. Its event loop handles two broad categories: file events, which represent sockets becoming readable or writable, and time events, which represent scheduled work. Both are represented inside an aeEventLoop structure:
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;
The stop field tells the loop whether to exit. events is the array of registered file events, while fired holds file events that have been triggered. Redis can have multiple scheduled tasks, so time events form a linked list; timeEventHead points to its first node. Before the process blocks waiting for file events, it calls beforesleep; after it is woken for any reason, it can call aftersleep. Redis supports several I/O multiplexing backends, such as kqueue and epoll, and apidata points to the backend-specific state.
The loop itself is a simple while loop:
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
}
}
aeProcessEvents is the main processing function. Its second argument is a flag set: AE_ALL_EVENTS means both file events and time events should be handled, while the sleep flags control whether the before-sleep and after-sleep callbacks run around the blocking wait. The function follows a compact sequence: find the nearest time event and derive a timeout, block until a file event arrives or the timeout expires, process file events, then run any due time events. Time events are executed by processTimeEvents.
File events and socket readiness
Redis clients communicate with the server over TCP sockets, so a file event is really a socket readiness event: readable or writable. Socket I/O can be blocking or non-blocking. In blocking mode, a process can handle only one network connection's read/write events at a time; handling many connections would typically require multiple threads or processes and would be inefficient. In non-blocking mode, Redis can use a mature I/O multiplexing model such as select, epoll, or kqueue, depending on the operating system.
The backend is selected at compile time. The conditional chain in the code chooses evport when available, then epoll, kqueue, and finally select:
#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
Taking epoll as an example, aeApiCreate wraps epoll_create; aeApiAddEvent adds an event and wraps epoll_ctl; aeApiDelEvent removes an event and also wraps epoll_ctl; and aeApiPoll wraps epoll_wait.
static int aeApiCreate(aeEventLoop *eventLoop);
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask);
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask)
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp);
These functions take an eventLoop whose most relevant file-event fields are threefold. apidata points to the I/O multiplexing object; because the four backend object types differ, this field is void *. events stores the events that need to be monitored, indexed by socket file descriptor. fired stores events that have already been triggered.
Time events and scheduled work
Redis has many periodic jobs inside the server: clearing timed-out client connections, deleting expired keys, and so on. Each scheduled task is represented by an aeTimeEvent object. Multiple time events form a linked list, stored through the timeEventHead field of aeEventLoop, which points to the first node. The structure is defined as follows:
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;
int refcount; /* refcount to prevent timer events from being
* freed in recursive time event calls. */
} aeTimeEvent;
The fields have the following roles. id is the unique time event identifier, driven by eventLoop->timeEventNextId. when_sec and when_ms give the second and millisecond at which the event should fire. timeProc is a function pointer to the time event handler. finalizerProc is called before a time event node is removed. clientData points to the corresponding client object. prev and next link neighboring time events, and refcount prevents timer events from being freed during recursive time event calls.
processTimeEvents is straightforward: it walks the time event list, checks whether each event is due, and calls the handler timeProc when it is.
static int processTimeEvents(aeEventLoop *eventLoop) {
int processed = 0;
aeTimeEvent *te;
long long maxId;
time_t now = time(NULL);
/* Check system clock */
if (now < eventLoop->lastTime) {
te = eventLoop->timeEventHead;
while(te) {
te->when_sec = 0;
te = te->next;
}
}
eventLoop->lastTime = now;
te = eventLoop->timeEventHead;
maxId = eventLoop->timeEventNextId-1;
while(te) {
long now_sec, now_ms;
long long id;
/* Remove events marked for deletion */
if (te->id == AE_DELETED_EVENT_ID) {
aeTimeEvent *next = te->next;
if (te->refcount) {
te = next;
continue;
}
if (te->prev)
te->prev->next = te->next;
else
eventLoop->timeEventHead = te->next;
if (te->next)
te->next->prev = te->prev;
if (te->finalizerProc)
te->finalizerProc(eventLoop, te->clientData);
zfree(te);
te = next;
continue;
}
/* Make sure only time events created by Redis itself run */
if (te->id > maxId) {
te = te->next;
continue;
}
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;
te->refcount++;
// Handle time event
retval = te->timeProc(eventLoop, id, te->clientData);
te->refcount--;
processed++;
// Reset time event expiration time
if (retval != AE_NOMORE) {
aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
} else {
te->id = AE_DELETED_EVENT_ID;
}
}
te = te->next;
}
return processed;
}
The return value retval from timeProc indicates when this time event should fire next. It is a relative value in milliseconds: from the current time, the event will fire again after retval milliseconds. If retval is AE_NOMORE, the event is marked for deletion.
Redis creates time event nodes with aeCreateTimeEvent. Internally it is simple: create the event and add it to the time event list. Its signature is:
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
aeTimeProc *proc, void *clientData,
aeEventFinalizerProc *finalizerProc);
eventLoop points to the event loop structure. milliseconds is the relative trigger time in milliseconds from now. proc is the handler. clientData points to the associated structure object. finalizerProc is another function pointer, called before the time event node is removed.
A central example is serverCron, which bundles many of Redis's periodic checks and returns the delay until its next invocation:
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
run_with_period(100) {
// Run every 100 milliseconds
}
run_with_period(5000) {
// Run every 5000 milliseconds
}
/* Show information about connected clients */
if (!server.sentinel_mode) {
run_with_period(5000) {
serverLog(LL_DEBUG,
"%lu clients connected (%lu replicas), %zu bytes in use",
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
zmalloc_used_memory());
}
}
/* Clean up unused client connections */
clientsCron();
/* Run database periodic tasks */
databasesCron();
/* Check whether an AOF write command was triggered manually */
if (!hasActiveChildProcess() &&
server.aof_rewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
/* Check whether AOF write is complete, etc. */
if (hasActiveChildProcess() || ldbPendingChildren())
{
checkChildrenDone();
} else {
}
/* Flush AOF to disk periodically */
if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);
/* Periodically check AOF write result */
run_with_period(1000) {
if (server.aof_last_write_status == C_ERR)
flushAppendOnlyFile(0);
}
/* Clean up unused client connections */
clientsArePaused(); /* Don't check return value, just use the side effect.*/
/* Periodic cluster check */
run_with_period(1000) replicationCron();
/* Run cluster periodic tasks */
run_with_period(100) {
if (server.cluster_enabled) clusterCron();
}
/* Run Sentinel mode periodic tasks */
if (server.sentinel_mode) sentinelTimer();
/* Cleanup expired MIGRATE cached sockets. */
run_with_period(1000) {
migrateCloseTimedoutSockets();
}
/* Stop unused I/O threads */
stopThreadedIOIfNeeded();
/* Periodically check slot information */
if (server.tracking_clients) trackingLimitUsedSlots();
/* Run module periodic tasks */
RedisModuleCronLoopV1 ei = {REDISMODULE_CRON_LOOP_VERSION,server.hz};
moduleFireServerEvent(REDISMODULE_EVENT_CRON_LOOP,
0,
&ei);
server.cronloops++;
return 1000/server.hz;
}