Redis 源码分析系列1-main函数相关调用分析

 1 typedef struct aeEventLoop {
 2     // 目前已注册的最大描述符
 3     int maxfd;   /* highest file descriptor currently registered */
 4     // 目前已追踪的最大描述符
 5     int setsize; /* max number of file descriptors tracked */
 6     // 用于生成时间事件 id
 7     long long timeEventNextId;
 8     // 最后一次执行时间事件的时间
 9     time_t lastTime;     /* Used to detect system clock skew */
10     // 已注册的文件事件
11     aeFileEvent *events; /* Registered events */
12     // 已就绪的文件事件
13     aeFiredEvent *fired; /* Fired events */
14     // 时间事件
15     aeTimeEvent *timeEventHead;
16     // 事件处理器的开关
17     int stop;
18     // 多路复用库的私有数据
19     void *apidata; /* This is used for polling API specific data */
20     // 在处理事件前要执行的函数
21     aeBeforeSleepProc *beforesleep;
22 } aeEventLoop;

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;

/* 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;

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;
}

numevents = aeApiPoll(eventLoop, tvp);
        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 rfired = 0;

            /* 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. */
            if (fe->mask & mask & AE_READABLE) {
                // 读事件
                rfired = 1; // 确保读/写事件只能执行其中一个
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
            }
            if (fe->mask & mask & AE_WRITABLE) {
                // 写事件
                if (!rfired || fe->wfileProc != fe->rfileProc)
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
            }

            processed++;

// 注册新号处理函数，对于SIGHUP，SIGPIPE，忽略。
// 初始化共享对象

// 关键，初始化redis的事件循环主结构
server.el = aeCreateEventLoop(server.maxclients+1024);

// 初始化作为server的网络连接
server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr);

//初始化db

// 初始化pubsub

//关键，关联时间时间
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);


// 将前面创建的server socket即server.ipfd与redis的事件循环关联。
if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,
        acceptTcpHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.ipfd file event.");

//初始化bio

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    if (fd >= eventLoop->setsize) return AE_ERR;
    aeFileEvent *fe = &eventLoop->events[fd];

    // 监听指定 fd
    if (aeApiAddEvent(eventLoop, fd, mask) == -1)
        return AE_ERR;

    // 设置文件事件类型
    fe->mask |= mask;
    if (mask & AE_READABLE) fe->rfileProc = proc;
    if (mask & AE_WRITABLE) fe->wfileProc = proc;

    fe->clientData = clientData;

    // 如果有需要，更新事件处理器的最大 fd
    if (fd > eventLoop->maxfd)
        eventLoop->maxfd = fd;

    return AE_OK;
}

// 其中调用accept，创建cfd服务客户端
cfd = anetTcpAccept(server.neterr, fd, cip, &cport)

// 为cfd创建redisClient，获取客户端查询命令
// 调用readQueryFromClient()获取客户端buffer。
acceptCommandHandler(cfd,0)

static void acceptCommonHandler(int fd, int flags) {
    redisClient *c;

    // 创建新客户端
    if ((c = createClient(fd)) == NULL) {
        redisLog(REDIS_WARNING,"Error allocating resources for the client");
        close(fd); /* May be already closed, just ignore errors */
        return;
    }

    /* If maxclient directive is set and this is one client more... close the
     * connection. Note that we create the client instead to check before
     * for this condition, since now the socket is already set in nonblocking
     * mode and we can send an error for free using the Kernel I/O */

    if (listLength(server.clients) > server.maxclients) {
        char *err = "-ERR max number of clients reached\r\n";

        /* That‘s a best effort error message, don‘t check write errors */
        // 发送错误信息到客户端
        if (write(c->fd,err,strlen(err)) == -1) {
            /* Nothing to do, Just to avoid the warning... */
        }
        server.stat_rejected_conn++;
        // 释放客户端
        freeClient(c);
        return;
    }
    server.stat_numconnections++;
    c->flags |= flags;
}