redis超时问题分析

• 06/04. 2014

Redis在分布式应用中占据着越来越重要的地位，短短的几万行代码，实现了一个高性能的数据存储服务。最近dump中心的cm8集群出现过 几次redis超时的情况，但是查看redis机器的相关内存都没有发现内存不够，或者内存发生交换的情况，查看redis源码之后，发现在某些情况下 redis会出现超时的状况，相关细节如下。

int rdbSaveBackground(char *filename) {

    pid_t childpid;

    long long start;

    if (server.rdb_child_pid != -1) return REDIS_ERR;

    server.dirty_before_bgsave = server.dirty;

    server.lastbgsave_try = time(NULL);

    start = ustime();

    if ((childpid = fork()) == 0) {

        int retval;

        /* Child */

        if (server.ipfd > 0) close(server.ipfd);

        if (server.sofd > 0) close(server.sofd);

        retval = rdbSave(filename);

        if (retval == REDIS_OK) {

            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {

                redisLog(REDIS_NOTICE,

                    "RDB: %zu MB of memory used by copy-on-write",

                    private_dirty/(1024*1024));

            }

        }

        exitFromChild((retval == REDIS_OK) ? 0 : 1);

    } else {

        /* Parent */

        server.stat_fork_time = ustime()-start;

        if (childpid == -1) {

            server.lastbgsave_status = REDIS_ERR;

            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",

                strerror(errno));

            return REDIS_ERR;

        }

        redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);

        server.rdb_save_time_start = time(NULL);

        server.rdb_child_pid = childpid;

        updateDictResizePolicy();

        return REDIS_OK;

    }

    return REDIS_OK; /* unreached */

}

程序1
另外还有一些特殊情况也会导致swap发生。当我们使用rdb做为redis集群持久化时可能会发生物理内存不够的情况（aof持久化只是保持支持不断的
追加redis集群变化操作，不太容易引起swap）。当使用rdb持久化时，如程序1所示主进程会fork一个子进程去dump
redis中所有的数据，主进程依然为客户端服务。此时主进程和子进程共享同一块内存区域，
linux内核采用写时复制来保证数据的安全性。在这种模式下如果客户端发来写请求，内核将该页赋值到一个新的页面上并标记为写，在将写请求写入该页面。
因此，在rdb持久化时，如果有其他请求，那么redis会使用更多的内存，更容易发生swap，因此在可以快速恢复的场景下尽量少使用rdb持久化可以
将rdb
dump的条件设的苛刻一点，当然也可以选择aof，但是aof也有他自身的缺点。另外也可以使用2.6以后的主从结构，将读写分离，这样不会出现
server进程上又读又写的情景发生 3.
Redis单进程处理命令。Redis支持udp和tcp两种连接，redis客户端向redis服务器发送包含redis命令的信息，redis服务器
收到信息后解析命令后执行相应的操作，redis处理命令是串行的具体流程如下。首先服务端建立连接如程序2所示，在创建
socket,bind,listen后返回文件描述符：

server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr);

程序2
对于redis这种服务来说，它需要处理成千上万个连接(最高达到655350)，需要使用多路复用来处理多个连接。这里redis提供了
epoll,select,
kqueue来实现，这里在默认使用epoll(ae.c)。拿到listen函数返回的文件描述符fd后，redis将fd和其处理
acceptTcpHandler函数加入到事件驱动的链表中.实际上在加入事件队列中，程序4事件驱动程序将套接字相关的fd文件描述符加入到
epoll的监听事件中。

 if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,

        acceptTcpHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.ipfd file event.");

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,

        aeFileProc *proc, void *clientData)

{

    if (fd >= eventLoop->setsize) {

        errno = ERANGE;

        return AE_ERR;

    }

    aeFileEvent *fe = &eventLoop->events[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;

    if (fd > eventLoop->maxfd)

        eventLoop->maxfd = fd;

    return AE_OK;

}

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {

    aeApiState *state = eventLoop->apidata;

    struct epoll_event ee;

    /* If the fd was already monitored for some event, we need a MOD

     * operation. Otherwise we need an ADD operation. */

    int op = eventLoop->events[fd].mask == AE_NONE ?

            EPOLL_CTL_ADD : EPOLL_CTL_MOD;

    ee.events = 0;

    mask |= eventLoop->events[fd].mask; /* Merge old events */

    if (mask & AE_READABLE) ee.events |= EPOLLIN;

    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;

    ee.data.u64 = 0; /* avoid valgrind warning */

    ee.data.fd = fd;

    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;

    return 0;

}

void aeMain(aeEventLoop *eventLoop) {

    eventLoop->stop = 0;

    while (!eventLoop->stop) {

        if (eventLoop->beforesleep != NULL)

            eventLoop->beforesleep(eventLoop);

        aeProcessEvents(eventLoop, AE_ALL_EVENTS);

    }

}

int aeProcessEvents(aeEventLoop *eventLoop, int flags)

{-------------------------------

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

        }

}