redis 响应处理流程

相关资料

networking.c

请求处理完成后，通过调用 addReply 函数族来完成响应。

void addReply(redisClient *c, robj *obj);
void addReplySds(redisClient *c, sds s);
void addReplyString(redisClient *c, char *s, size_t len);

void addReplyString(redisClient *c, char *s, size_t len) {
    if (prepareClientToWrite(c) != REDIS_OK) return;
    if (_addReplyToBuffer(c,s,len) != REDIS_OK)
        _addReplyStringToList(c,s,len);
}

以 addReplyString 为例，在 addReply 函数中：
首先调用 prepareClientToWrite 函数，完成准备工作。
然后向客户端缓存写入响应内容，写入响应内容时，总是先尝试写入到固定 buf，如果写入失败，再写入到动态分配的链表中。

客户端缓存由两部分组成：固定大小的 buf 和动态分配的 reply（链表）。

#define REDIS_REPLY_CHUNK_BYTES (16*1024) /* 16k output buffer */
char buf[REDIS_REPLY_CHUNK_BYTES];

c->reply = listCreate();

1. 写入响应内容之前，完成准备工作

/* This function is called every time we are going to transmit new data
 * to the client. The behavior is the following:
 *
 * If the client should receive new data (normal clients will) the function
 * returns REDIS_OK, and make sure to install the write handler in our event
 * loop so that when the socket is writable new data gets written.
 *
 * If the client should not receive new data, because it is a fake client,
 * a master, a slave not yet online, or because the setup of the write handler
 * failed, the function returns REDIS_ERR.
 *
 * Typically gets called every time a reply is built, before adding more
 * data to the clients output buffers. If the function returns REDIS_ERR no
 * data should be appended to the output buffers. */
int prepareClientToWrite(redisClient *c) {
    if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
    if ((c->flags & REDIS_MASTER) &&
        !(c->flags & REDIS_MASTER_FORCE_REPLY)) return REDIS_ERR;
    if (c->fd <= 0) return REDIS_ERR; /* Fake client */
    if (c->bufpos == 0 && listLength(c->reply) == 0 &&
        (c->replstate == REDIS_REPL_NONE ||
         c->replstate == REDIS_REPL_ONLINE) &&
        aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
        sendReplyToClient, c) == AE_ERR) return REDIS_ERR;
    return REDIS_OK;
}

在向客户端缓存写入响应内容之前，先向事件处理程序注册写事件，回调函数是：sendReplyToClient()。
此时，客户端缓存为空：c->bufpos == 0 && listLength(c->reply) == 0

2. 写入响应内容

写入响应内容，通过调用以下函数来完成：

int _addReplyToBuffer(redisClient *c, char *s, size_t len);
void _addReplyObjectToList(redisClient *c, robj *o);
void _addReplySdsToList(redisClient *c, sds s);
void _addReplyStringToList(redisClient *c, char *s, size_t len);

注：写入响应内容时，总是先尝试写入到固定 buf，如果写入失败，再写入到动态分配的链表中。

int _addReplyToBuffer(redisClient *c, char *s, size_t len) {
    size_t available = sizeof(c->buf)-c->bufpos;

    if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK;

    /* If there already are entries in the reply list, we cannot
     * add anything more to the static buffer. */
    if (listLength(c->reply) > 0) return REDIS_ERR;

    /* Check that the buffer has enough space available for this string. */
    if (len > available) return REDIS_ERR;

    memcpy(c->buf+c->bufpos,s,len);
    c->bufpos+=len;
    return REDIS_OK;
}

void _addReplyStringToList(redisClient *c, char *s, size_t len) {
    robj *tail;

    if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;

    if (listLength(c->reply) == 0) {
        robj *o = createStringObject(s,len);

        listAddNodeTail(c->reply,o);
        c->reply_bytes += zmalloc_size_sds(o->ptr);
    } else {
        tail = listNodeValue(listLast(c->reply));

        /* Append to this object when possible. */
        if (tail->ptr != NULL &&
            sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
        {
            c->reply_bytes -= zmalloc_size_sds(tail->ptr);
            tail = dupLastObjectIfNeeded(c->reply);
            tail->ptr = sdscatlen(tail->ptr,s,len);
            c->reply_bytes += zmalloc_size_sds(tail->ptr);
        } else {
            robj *o = createStringObject(s,len);

            listAddNodeTail(c->reply,o);
            c->reply_bytes += zmalloc_size_sds(o->ptr);
        }
    }
    asyncCloseClientOnOutputBufferLimitReached(c);
}

3. 发送响应内容

发送响应内容，通过 sendReplyToClient 函数来实现。

while(c->bufpos > 0 || listLength(c->reply)) {
    if (c->bufpos > 0) {
        nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
        if (nwritten <= 0) break;
        c->sentlen += nwritten;
        totwritten += nwritten;

        /* If the buffer was sent, set bufpos to zero to continue with
         * the remainder of the reply. */
        if (c->sentlen == c->bufpos) {
            c->bufpos = 0;
            c->sentlen = 0;
        }
    } else {
        o = listNodeValue(listFirst(c->reply));
        objlen = sdslen(o->ptr);
        objmem = zmalloc_size_sds(o->ptr);

        if (objlen == 0) {
            listDelNode(c->reply,listFirst(c->reply));
            continue;
        }

        nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
        if (nwritten <= 0) break;
        c->sentlen += nwritten;
        totwritten += nwritten;

        /* If we fully sent the object on head go to the next one */
        if (c->sentlen == objlen) {
            listDelNode(c->reply,listFirst(c->reply));
            c->sentlen = 0;
            c->reply_bytes -= objmem;
        }
    }
    /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT
     * bytes, in a single threaded server it's a good idea to serve
     * other clients as well, even if a very large request comes from
     * super fast link that is always able to accept data (in real world
     * scenario think about 'KEYS *' against the loopback interface).
     *
     * However if we are over the maxmemory limit we ignore that and
     * just deliver as much data as it is possible to deliver. */
    server.stat_net_output_bytes += totwritten;
    if (totwritten > REDIS_MAX_WRITE_PER_EVENT &&
        (server.maxmemory == 0 ||
         zmalloc_used_memory() < server.maxmemory)) break;
}