相关资料
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;
}