networking.c

/*
 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 * * Neither the name of Redis nor the names of its contributors may be used
 * to endorse or promote products derived from this software without
 * specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include"server.h"
#include"atomicvar.h"
#include"cluster.h"
#include<sys/socket.h>
#include<sys/uio.h>
#include<math.h>
#include<ctype.h>

staticvoidsetProtocolError(constchar*errstr, client*c);
intpostponeClientRead(client*c);
intProcessingEventsWhileBlocked=0; /* See processEventsWhileBlocked(). */

/* Return the size consumed from the allocator, for the specified SDS string,
 * including internal fragmentation. This function is used in order to compute
 * the client output buffer size. */
size_tsdsZmallocSize(sdss) {
void*sh=sdsAllocPtr(s);
returnzmalloc_size(sh);
}

/* Return the amount of memory used by the sds string at object->ptr
 * for a string object. This includes internal fragmentation. */
size_tgetStringObjectSdsUsedMemory(robj*o) {
serverAssertWithInfo(NULL,o,o->type==OBJ_STRING);
switch(o->encoding) {
caseOBJ_ENCODING_RAW: returnsdsZmallocSize(o->ptr);
caseOBJ_ENCODING_EMBSTR: returnzmalloc_size(o)-sizeof(robj);
default: return0; /* Just integer encoding for now. */
 }
}

/* Return the length of a string object.
 * This does NOT includes internal fragmentation or sds unused space. */
size_tgetStringObjectLen(robj*o) {
serverAssertWithInfo(NULL,o,o->type==OBJ_STRING);
switch(o->encoding) {
caseOBJ_ENCODING_RAW: returnsdslen(o->ptr);
caseOBJ_ENCODING_EMBSTR: returnsdslen(o->ptr);
default: return0; /* Just integer encoding for now. */
 }
}

/* Client.reply list dup and free methods. */
void*dupClientReplyValue(void*o) {
clientReplyBlock*old=o;
clientReplyBlock*buf=zmalloc(sizeof(clientReplyBlock) +old->size);
memcpy(buf, o, sizeof(clientReplyBlock) +old->size);
returnbuf;
}

voidfreeClientReplyValue(void*o) {
zfree(o);
}

intlistMatchObjects(void*a, void*b) {
returnequalStringObjects(a,b);
}

/* This function links the client to the global linked list of clients.
 * unlinkClient() does the opposite, among other things. */
voidlinkClient(client*c) {
listAddNodeTail(server.clients,c);
/* Note that we remember the linked list node where the client is stored,
 * this way removing the client in unlinkClient() will not require
 * a linear scan, but just a constant time operation. */
c->client_list_node=listLast(server.clients);
uint64_tid=htonu64(c->id);
raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);
}

/* Initialize client authentication state.
 */
staticvoidclientSetDefaultAuth(client*c) {
/* If the default user does not require authentication, the user is
 * directly authenticated. */
c->user=DefaultUser;
c->authenticated= (c->user->flags&USER_FLAG_NOPASS) &&
 !(c->user->flags&USER_FLAG_DISABLED);
}

client*createClient(connection*conn) {
client*c=zmalloc(sizeof(client));

/* passing NULL as conn it is possible to create a non connected client.
 * This is useful since all the commands needs to be executed
 * in the context of a client. When commands are executed in other
 * contexts (for instance a Lua script) we need a non connected client. */
if (conn) {
connNonBlock(conn);
connEnableTcpNoDelay(conn);
if (server.tcpkeepalive)
connKeepAlive(conn,server.tcpkeepalive);
connSetReadHandler(conn, readQueryFromClient);
connSetPrivateData(conn, c);
 }

selectDb(c,0);
uint64_tclient_id;
atomicGetIncr(server.next_client_id, client_id, 1);
c->id=client_id;
c->resp=2;
c->conn=conn;
c->name=NULL;
c->bufpos=0;
c->buf_usable_size=zmalloc_usable_size(c)-offsetof(client,buf);
c->qb_pos=0;
c->querybuf=sdsempty();
c->pending_querybuf=sdsempty();
c->querybuf_peak=0;
c->reqtype=0;
c->argc=0;
c->argv=NULL;
c->argv_len_sum=0;
c->original_argc=0;
c->original_argv=NULL;
c->cmd=c->lastcmd=NULL;
c->multibulklen=0;
c->bulklen=-1;
c->sentlen=0;
c->flags=0;
c->ctime=c->lastinteraction=server.unixtime;
clientSetDefaultAuth(c);
c->replstate=REPL_STATE_NONE;
c->repl_put_online_on_ack=0;
c->reploff=0;
c->read_reploff=0;
c->repl_ack_off=0;
c->repl_ack_time=0;
c->repl_last_partial_write=0;
c->slave_listening_port=0;
c->slave_addr=NULL;
c->slave_capa=SLAVE_CAPA_NONE;
c->reply=listCreate();
c->reply_bytes=0;
c->obuf_soft_limit_reached_time=0;
listSetFreeMethod(c->reply,freeClientReplyValue);
listSetDupMethod(c->reply,dupClientReplyValue);
c->btype=BLOCKED_NONE;
c->bpop.timeout=0;
c->bpop.keys=dictCreate(&objectKeyHeapPointerValueDictType,NULL);
c->bpop.target=NULL;
c->bpop.xread_group=NULL;
c->bpop.xread_consumer=NULL;
c->bpop.xread_group_noack=0;
c->bpop.numreplicas=0;
c->bpop.reploffset=0;
c->woff=0;
c->watched_keys=listCreate();
c->pubsub_channels=dictCreate(&objectKeyPointerValueDictType,NULL);
c->pubsub_patterns=listCreate();
c->peerid=NULL;
c->sockname=NULL;
c->client_list_node=NULL;
c->paused_list_node=NULL;
c->client_tracking_redirection=0;
c->client_tracking_prefixes=NULL;
c->client_cron_last_memory_usage=0;
c->client_cron_last_memory_type=CLIENT_TYPE_NORMAL;
c->auth_callback=NULL;
c->auth_callback_privdata=NULL;
c->auth_module=NULL;
listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
if (conn) linkClient(c);
initClientMultiState(c);
returnc;
}

/* This function puts the client in the queue of clients that should write
 * their output buffers to the socket. Note that it does not *yet* install
 * the write handler, to start clients are put in a queue of clients that need
 * to write, so we try to do that before returning in the event loop (see the
 * handleClientsWithPendingWrites() function).
 * If we fail and there is more data to write, compared to what the socket
 * buffers can hold, then we'll really install the handler. */
voidclientInstallWriteHandler(client*c) {
/* Schedule the client to write the output buffers to the socket only
 * if not already done and, for slaves, if the slave can actually receive
 * writes at this stage. */
if (!(c->flags&CLIENT_PENDING_WRITE) &&
 (c->replstate==REPL_STATE_NONE||
 (c->replstate==SLAVE_STATE_ONLINE&& !c->repl_put_online_on_ack)))
 {
/* Here instead of installing the write handler, we just flag the
 * client and put it into a list of clients that have something
 * to write to the socket. This way before re-entering the event
 * loop, we can try to directly write to the client sockets avoiding
 * a system call. We'll only really install the write handler if
 * we'll not be able to write the whole reply at once. */
c->flags |= CLIENT_PENDING_WRITE;
listAddNodeHead(server.clients_pending_write,c);
 }
}

/* 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 C_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
 * (used to load AOF in memory), a master or because the setup of the write
 * handler failed, the function returns C_ERR.
 *
 * The function may return C_OK without actually installing the write
 * event handler in the following cases:
 *
 * 1) The event handler should already be installed since the output buffer
 * already contains something.
 * 2) The client is a slave but not yet online, so we want to just accumulate
 * writes in the buffer but not actually sending them yet.
 *
 * Typically gets called every time a reply is built, before adding more
 * data to the clients output buffers. If the function returns C_ERR no
 * data should be appended to the output buffers. */
intprepareClientToWrite(client*c) {
/* If it's the Lua client we always return ok without installing any
 * handler since there is no socket at all. */
if (c->flags& (CLIENT_LUA|CLIENT_MODULE)) returnC_OK;

/* If CLIENT_CLOSE_ASAP flag is set, we need not write anything. */
if (c->flags&CLIENT_CLOSE_ASAP) returnC_ERR;

/* CLIENT REPLY OFF / SKIP handling: don't send replies. */
if (c->flags& (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) returnC_ERR;

/* Masters don't receive replies, unless CLIENT_MASTER_FORCE_REPLY flag
 * is set. */
if ((c->flags&CLIENT_MASTER) &&
 !(c->flags&CLIENT_MASTER_FORCE_REPLY)) returnC_ERR;

if (!c->conn) returnC_ERR; /* Fake client for AOF loading. */

/* Schedule the client to write the output buffers to the socket, unless
 * it should already be setup to do so (it has already pending data).
 *
 * If CLIENT_PENDING_READ is set, we're in an IO thread and should
 * not install a write handler. Instead, it will be done by
 * handleClientsWithPendingReadsUsingThreads() upon return.
 */
if (!clientHasPendingReplies(c) && !(c->flags&CLIENT_PENDING_READ))
clientInstallWriteHandler(c);

/* Authorize the caller to queue in the output buffer of this client. */
returnC_OK;
}

/* -----------------------------------------------------------------------------
 * Low level functions to add more data to output buffers.
 * -------------------------------------------------------------------------- */

/* Attempts to add the reply to the static buffer in the client struct.
 * Returns the length of data that is added to the reply buffer. */
size_t_addReplyToBuffer(client*c, constchar*s, size_tlen) {
size_tavailable=c->buf_usable_size-c->bufpos;

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

size_treply_len=len>available ? available : len;
memcpy(c->buf+c->bufpos,s,reply_len);
c->bufpos+=reply_len;
returnreply_len;
}

/* Adds the reply to the reply linked list.
 * Note: some edits to this function need to be relayed to AddReplyFromClient. */
void_addReplyProtoToList(client*c, constchar*s, size_tlen) {
listNode*ln=listLast(c->reply);
clientReplyBlock*tail=ln? listNodeValue(ln): NULL;

/* Note that 'tail' may be NULL even if we have a tail node, because when
 * addReplyDeferredLen() is used, it sets a dummy node to NULL just
 * fo fill it later, when the size of the bulk length is set. */

/* Append to tail string when possible. */
if (tail) {
/* Copy the part we can fit into the tail, and leave the rest for a
 * new node */
size_tavail=tail->size-tail->used;
size_tcopy=avail >= len? len: avail;
memcpy(tail->buf+tail->used, s, copy);
tail->used+=copy;
s+=copy;
len-=copy;
 }
if (len) {
/* Create a new node, make sure it is allocated to at
 * least PROTO_REPLY_CHUNK_BYTES */
size_tusable_size;
size_tsize=len<PROTO_REPLY_CHUNK_BYTES? PROTO_REPLY_CHUNK_BYTES: len;
tail=zmalloc_usable(size+sizeof(clientReplyBlock), &usable_size);
/* take over the allocation's internal fragmentation */
tail->size=usable_size-sizeof(clientReplyBlock);
tail->used=len;
memcpy(tail->buf, s, len);
listAddNodeTail(c->reply, tail);
c->reply_bytes+=tail->size;

closeClientOnOutputBufferLimitReached(c, 1);
 }
}

void_addReplyToBufferOrList(client*c, constchar*s, size_tlen) {
if (c->flags&CLIENT_CLOSE_AFTER_REPLY) return;

size_treply_len=_addReplyToBuffer(c,s,len);
if (len>reply_len) _addReplyProtoToList(c,s+reply_len,len-reply_len);
}

/* -----------------------------------------------------------------------------
 * Higher level functions to queue data on the client output buffer.
 * The following functions are the ones that commands implementations will call.
 * -------------------------------------------------------------------------- */

/* Add the object 'obj' string representation to the client output buffer. */
voidaddReply(client*c, robj*obj) {
if (prepareClientToWrite(c) !=C_OK) return;

if (sdsEncodedObject(obj)) {
_addReplyToBufferOrList(c,obj->ptr,sdslen(obj->ptr));
 } elseif (obj->encoding==OBJ_ENCODING_INT) {
/* For integer encoded strings we just convert it into a string
 * using our optimized function, and attach the resulting string
 * to the output buffer. */
charbuf[32];
size_tlen=ll2string(buf,sizeof(buf),(long)obj->ptr);
_addReplyToBufferOrList(c,buf,len);
 } else {
serverPanic("Wrong obj->encoding in addReply()");
 }
}

/* Add the SDS 's' string to the client output buffer, as a side effect
 * the SDS string is freed. */
voidaddReplySds(client*c, sdss) {
if (prepareClientToWrite(c) !=C_OK) {
/* The caller expects the sds to be free'd. */
sdsfree(s);
return;
 }
_addReplyToBufferOrList(c,s,sdslen(s));
sdsfree(s);
}

/* This low level function just adds whatever protocol you send it to the
 * client buffer, trying the static buffer initially, and using the string
 * of objects if not possible.
 *
 * It is efficient because does not create an SDS object nor an Redis object
 * if not needed. The object will only be created by calling
 * _addReplyProtoToList() if we fail to extend the existing tail object
 * in the list of objects. */
voidaddReplyProto(client*c, constchar*s, size_tlen) {
if (prepareClientToWrite(c) !=C_OK) return;
_addReplyToBufferOrList(c,s,len);
}

/* Low level function called by the addReplyError...() functions.
 * It emits the protocol for a Redis error, in the form:
 *
 * -ERRORCODE Error Message<CR><LF>
 *
 * If the error code is already passed in the string 's', the error
 * code provided is used, otherwise the string "-ERR " for the generic
 * error code is automatically added.
 * Note that 's' must NOT end with \r\n. */
voidaddReplyErrorLength(client*c, constchar*s, size_tlen) {
/* If the string already starts with "-..." then the error code
 * is provided by the caller. Otherwise we use "-ERR". */
if (!len||s[0] !='-') addReplyProto(c,"-ERR ",5);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
}

/* Do some actions after an error reply was sent (Log if needed, updates stats, etc.) */
voidafterErrorReply(client*c, constchar*s, size_tlen) {
/* Increment the global error counter */
server.stat_total_error_replies++;
/* Increment the error stats
 * If the string already starts with "-..." then the error prefix
 * is provided by the caller ( we limit the search to 32 chars). Otherwise we use "-ERR". */
if (s[0] !='-') {
incrementErrorCount("ERR", 3);
 } else {
char*spaceloc=memchr(s, ' ', len<32 ? len : 32);
if (spaceloc) {
constsize_terrEndPos= (size_t)(spaceloc-s);
incrementErrorCount(s+1, errEndPos-1);
 } else {
/* Fallback to ERR if we can't retrieve the error prefix */
incrementErrorCount("ERR", 3);
 }
 }

/* Sometimes it could be normal that a slave replies to a master with
 * an error and this function gets called. Actually the error will never
 * be sent because addReply*() against master clients has no effect...
 * A notable example is:
 *
 * EVAL 'redis.call("incr",KEYS[1]); redis.call("nonexisting")' 1 x
 *
 * Where the master must propagate the first change even if the second
 * will produce an error. However it is useful to log such events since
 * they are rare and may hint at errors in a script or a bug in Redis. */
intctype=getClientType(c);
if (ctype==CLIENT_TYPE_MASTER||ctype==CLIENT_TYPE_SLAVE||c->id==CLIENT_ID_AOF) {
char*to, *from;

if (c->id==CLIENT_ID_AOF) {
to="AOF-loading-client";
from="server";
 } elseif (ctype==CLIENT_TYPE_MASTER) {
to="master";
from="replica";
 } else {
to="replica";
from="master";
 }

if (len>4096) len=4096;
char*cmdname=c->lastcmd ? c->lastcmd->name : "<unknown>";
serverLog(LL_WARNING,"== CRITICAL == This %s is sending an error "
"to its %s: '%.*s' after processing the command "
"'%s'", from, to, (int)len, s, cmdname);
if (ctype==CLIENT_TYPE_MASTER&&server.repl_backlog&&
server.repl_backlog_histlen>0)
 {
showLatestBacklog();
 }
server.stat_unexpected_error_replies++;
 }
}

/* The 'err' object is expected to start with -ERRORCODE and end with \r\n.
 * Unlike addReplyErrorSds and others alike which rely on addReplyErrorLength. */
voidaddReplyErrorObject(client*c, robj*err) {
addReply(c, err);
afterErrorReply(c, err->ptr, sdslen(err->ptr)-2); /* Ignore trailing \r\n */
}

/* See addReplyErrorLength for expectations from the input string. */
voidaddReplyError(client*c, constchar*err) {
addReplyErrorLength(c,err,strlen(err));
afterErrorReply(c,err,strlen(err));
}

/* See addReplyErrorLength for expectations from the input string. */
/* As a side effect the SDS string is freed. */
voidaddReplyErrorSds(client*c, sdserr) {
addReplyErrorLength(c,err,sdslen(err));
afterErrorReply(c,err,sdslen(err));
sdsfree(err);
}

/* See addReplyErrorLength for expectations from the formatted string.
 * The formatted string is safe to contain \r and \n anywhere. */
voidaddReplyErrorFormat(client*c, constchar*fmt, ...) {
va_listap;
va_start(ap,fmt);
sdss=sdscatvprintf(sdsempty(),fmt,ap);
va_end(ap);
/* Trim any newlines at the end (ones will be added by addReplyErrorLength) */
s=sdstrim(s, "\r\n");
/* Make sure there are no newlines in the middle of the string, otherwise
 * invalid protocol is emitted. */
s=sdsmapchars(s, "\r\n", " ", 2);
addReplyErrorLength(c,s,sdslen(s));
afterErrorReply(c,s,sdslen(s));
sdsfree(s);
}

voidaddReplyStatusLength(client*c, constchar*s, size_tlen) {
addReplyProto(c,"+",1);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
}

voidaddReplyStatus(client*c, constchar*status) {
addReplyStatusLength(c,status,strlen(status));
}

voidaddReplyStatusFormat(client*c, constchar*fmt, ...) {
va_listap;
va_start(ap,fmt);
sdss=sdscatvprintf(sdsempty(),fmt,ap);
va_end(ap);
addReplyStatusLength(c,s,sdslen(s));
sdsfree(s);
}

/* Sometimes we are forced to create a new reply node, and we can't append to
 * the previous one, when that happens, we wanna try to trim the unused space
 * at the end of the last reply node which we won't use anymore. */
voidtrimReplyUnusedTailSpace(client*c) {
listNode*ln=listLast(c->reply);
clientReplyBlock*tail=ln? listNodeValue(ln): NULL;

/* Note that 'tail' may be NULL even if we have a tail node, because when
 * addReplyDeferredLen() is used */
if (!tail) return;

/* We only try to trim the space is relatively high (more than a 1/4 of the
 * allocation), otherwise there's a high chance realloc will NOP.
 * Also, to avoid large memmove which happens as part of realloc, we only do
 * that if the used part is small. */
if (tail->size-tail->used>tail->size / 4&&
tail->used<PROTO_REPLY_CHUNK_BYTES)
 {
size_told_size=tail->size;
tail=zrealloc(tail, tail->used+sizeof(clientReplyBlock));
/* take over the allocation's internal fragmentation (at least for
 * memory usage tracking) */
tail->size=zmalloc_usable_size(tail) -sizeof(clientReplyBlock);
c->reply_bytes=c->reply_bytes+tail->size-old_size;
listNodeValue(ln) =tail;
 }
}

/* Adds an empty object to the reply list that will contain the multi bulk
 * length, which is not known when this function is called. */
void*addReplyDeferredLen(client*c) {
/* Note that we install the write event here even if the object is not
 * ready to be sent, since we are sure that before returning to the
 * event loop setDeferredAggregateLen() will be called. */
if (prepareClientToWrite(c) !=C_OK) returnNULL;
trimReplyUnusedTailSpace(c);
listAddNodeTail(c->reply,NULL); /* NULL is our placeholder. */
returnlistLast(c->reply);
}

voidsetDeferredReply(client*c, void*node, constchar*s, size_tlength) {
listNode*ln= (listNode*)node;
clientReplyBlock*next, *prev;

/* Abort when *node is NULL: when the client should not accept writes
 * we return NULL in addReplyDeferredLen() */
if (node==NULL) return;
serverAssert(!listNodeValue(ln));

/* Normally we fill this dummy NULL node, added by addReplyDeferredLen(),
 * with a new buffer structure containing the protocol needed to specify
 * the length of the array following. However sometimes there might be room
 * in the previous/next node so we can instead remove this NULL node, and
 * suffix/prefix our data in the node immediately before/after it, in order
 * to save a write(2) syscall later. Conditions needed to do it:
 *
 * - The prev node is non-NULL and has space in it or
 * - The next node is non-NULL,
 * - It has enough room already allocated
 * - And not too large (avoid large memmove) */
if (ln->prev!=NULL&& (prev=listNodeValue(ln->prev)) &&
prev->size-prev->used>0)
 {
size_tlen_to_copy=prev->size-prev->used;
if (len_to_copy>length)
len_to_copy=length;
memcpy(prev->buf+prev->used, s, len_to_copy);
prev->used+=len_to_copy;
length-=len_to_copy;
if (length==0) {
listDelNode(c->reply, ln);
return;
 }
s+=len_to_copy;
 }

if (ln->next!=NULL&& (next=listNodeValue(ln->next)) &&
next->size-next->used >= length&&
next->used<PROTO_REPLY_CHUNK_BYTES*4)
 {
memmove(next->buf+length, next->buf, next->used);
memcpy(next->buf, s, length);
next->used+=length;
listDelNode(c->reply,ln);
 } else {
/* Create a new node */
clientReplyBlock*buf=zmalloc(length+sizeof(clientReplyBlock));
/* Take over the allocation's internal fragmentation */
buf->size=zmalloc_usable_size(buf) -sizeof(clientReplyBlock);
buf->used=length;
memcpy(buf->buf, s, length);
listNodeValue(ln) =buf;
c->reply_bytes+=buf->size;

closeClientOnOutputBufferLimitReached(c, 1);
 }
}

/* Populate the length object and try gluing it to the next chunk. */
voidsetDeferredAggregateLen(client*c, void*node, longlength, charprefix) {
serverAssert(length >= 0);

/* Abort when *node is NULL: when the client should not accept writes
 * we return NULL in addReplyDeferredLen() */
if (node==NULL) return;

charlenstr[128];
size_tlenstr_len=sprintf(lenstr, "%c%ld\r\n", prefix, length);
setDeferredReply(c, node, lenstr, lenstr_len);
}

voidsetDeferredArrayLen(client*c, void*node, longlength) {
setDeferredAggregateLen(c,node,length,'*');
}

voidsetDeferredMapLen(client*c, void*node, longlength) {
intprefix=c->resp==2 ? '*' : '%';
if (c->resp==2) length *= 2;
setDeferredAggregateLen(c,node,length,prefix);
}

voidsetDeferredSetLen(client*c, void*node, longlength) {
intprefix=c->resp==2 ? '*' : '~';
setDeferredAggregateLen(c,node,length,prefix);
}

voidsetDeferredAttributeLen(client*c, void*node, longlength) {
intprefix=c->resp==2 ? '*' : '|';
if (c->resp==2) length *= 2;
setDeferredAggregateLen(c,node,length,prefix);
}

voidsetDeferredPushLen(client*c, void*node, longlength) {
intprefix=c->resp==2 ? '*' : '>';
setDeferredAggregateLen(c,node,length,prefix);
}

/* Add a double as a bulk reply */
voidaddReplyDouble(client*c, doubled) {
if (isinf(d)) {
/* Libc in odd systems (Hi Solaris!) will format infinite in a
 * different way, so better to handle it in an explicit way. */
if (c->resp==2) {
addReplyBulkCString(c, d>0 ? "inf" : "-inf");
 } else {
addReplyProto(c, d>0 ? ",inf\r\n" : ",-inf\r\n",
d>0 ? 6 : 7);
 }
 } else {
chardbuf[MAX_LONG_DOUBLE_CHARS+3],
sbuf[MAX_LONG_DOUBLE_CHARS+32];
intdlen, slen;
if (c->resp==2) {
dlen=snprintf(dbuf,sizeof(dbuf),"%.17g",d);
slen=snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
addReplyProto(c,sbuf,slen);
 } else {
dlen=snprintf(dbuf,sizeof(dbuf),",%.17g\r\n",d);
addReplyProto(c,dbuf,dlen);
 }
 }
}

/* Add a long double as a bulk reply, but uses a human readable formatting
 * of the double instead of exposing the crude behavior of doubles to the
 * dear user. */
voidaddReplyHumanLongDouble(client*c, long doubled) {
if (c->resp==2) {
robj*o=createStringObjectFromLongDouble(d,1);
addReplyBulk(c,o);
decrRefCount(o);
 } else {
charbuf[MAX_LONG_DOUBLE_CHARS];
intlen=ld2string(buf,sizeof(buf),d,LD_STR_HUMAN);
addReplyProto(c,",",1);
addReplyProto(c,buf,len);
addReplyProto(c,"\r\n",2);
 }
}

/* Add a long long as integer reply or bulk len / multi bulk count.
 * Basically this is used to output <prefix><long long><crlf>. */
voidaddReplyLongLongWithPrefix(client*c, long longll, charprefix) {
charbuf[128];
intlen;

/* Things like $3\r\n or *2\r\n are emitted very often by the protocol
 * so we have a few shared objects to use if the integer is small
 * like it is most of the times. */
if (prefix=='*'&&ll<OBJ_SHARED_BULKHDR_LEN&&ll >= 0) {
addReply(c,shared.mbulkhdr[ll]);
return;
 } elseif (prefix=='$'&&ll<OBJ_SHARED_BULKHDR_LEN&&ll >= 0) {
addReply(c,shared.bulkhdr[ll]);
return;
 }

buf[0] =prefix;
len=ll2string(buf+1,sizeof(buf)-1,ll);
buf[len+1] ='\r';
buf[len+2] ='\n';
addReplyProto(c,buf,len+3);
}

voidaddReplyLongLong(client*c, long longll) {
if (ll==0)
addReply(c,shared.czero);
elseif (ll==1)
addReply(c,shared.cone);
else
addReplyLongLongWithPrefix(c,ll,':');
}

voidaddReplyAggregateLen(client*c, longlength, intprefix) {
serverAssert(length >= 0);
addReplyLongLongWithPrefix(c,length,prefix);
}

voidaddReplyArrayLen(client*c, longlength) {
addReplyAggregateLen(c,length,'*');
}

voidaddReplyMapLen(client*c, longlength) {
intprefix=c->resp==2 ? '*' : '%';
if (c->resp==2) length *= 2;
addReplyAggregateLen(c,length,prefix);
}

voidaddReplySetLen(client*c, longlength) {
intprefix=c->resp==2 ? '*' : '~';
addReplyAggregateLen(c,length,prefix);
}

voidaddReplyAttributeLen(client*c, longlength) {
intprefix=c->resp==2 ? '*' : '|';
if (c->resp==2) length *= 2;
addReplyAggregateLen(c,length,prefix);
}

voidaddReplyPushLen(client*c, longlength) {
intprefix=c->resp==2 ? '*' : '>';
addReplyAggregateLen(c,length,prefix);
}

voidaddReplyNull(client*c) {
if (c->resp==2) {
addReplyProto(c,"$-1\r\n",5);
 } else {
addReplyProto(c,"_\r\n",3);
 }
}

voidaddReplyBool(client*c, intb) {
if (c->resp==2) {
addReply(c, b ? shared.cone : shared.czero);
 } else {
addReplyProto(c, b ? "#t\r\n" : "#f\r\n",4);
 }
}

/* A null array is a concept that no longer exists in RESP3. However
 * RESP2 had it, so API-wise we have this call, that will emit the correct
 * RESP2 protocol, however for RESP3 the reply will always be just the
 * Null type "_\r\n". */
voidaddReplyNullArray(client*c) {
if (c->resp==2) {
addReplyProto(c,"*-1\r\n",5);
 } else {
addReplyProto(c,"_\r\n",3);
 }
}

/* Create the length prefix of a bulk reply, example: $2234 */
voidaddReplyBulkLen(client*c, robj*obj) {
size_tlen=stringObjectLen(obj);

addReplyLongLongWithPrefix(c,len,'$');
}

/* Add a Redis Object as a bulk reply */
voidaddReplyBulk(client*c, robj*obj) {
addReplyBulkLen(c,obj);
addReply(c,obj);
addReply(c,shared.crlf);
}

/* Add a C buffer as bulk reply */
voidaddReplyBulkCBuffer(client*c, constvoid*p, size_tlen) {
addReplyLongLongWithPrefix(c,len,'$');
addReplyProto(c,p,len);
addReply(c,shared.crlf);
}

/* Add sds to reply (takes ownership of sds and frees it) */
voidaddReplyBulkSds(client*c, sdss) {
addReplyLongLongWithPrefix(c,sdslen(s),'$');
addReplySds(c,s);
addReply(c,shared.crlf);
}

/* Set sds to a deferred reply (for symmetry with addReplyBulkSds it also frees the sds) */
voidsetDeferredReplyBulkSds(client*c, void*node, sdss) {
sdsreply=sdscatprintf(sdsempty(), "$%d\r\n%s\r\n", (unsigned)sdslen(s), s);
setDeferredReply(c, node, reply, sdslen(reply));
sdsfree(reply);
sdsfree(s);
}

/* Add a C null term string as bulk reply */
voidaddReplyBulkCString(client*c, constchar*s) {
if (s==NULL) {
addReplyNull(c);
 } else {
addReplyBulkCBuffer(c,s,strlen(s));
 }
}

/* Add a long long as a bulk reply */
voidaddReplyBulkLongLong(client*c, long longll) {
charbuf[64];
intlen;

len=ll2string(buf,64,ll);
addReplyBulkCBuffer(c,buf,len);
}

/* Reply with a verbatim type having the specified extension.
 *
 * The 'ext' is the "extension" of the file, actually just a three
 * character type that describes the format of the verbatim string.
 * For instance "txt" means it should be interpreted as a text only
 * file by the receiver, "md " as markdown, and so forth. Only the
 * three first characters of the extension are used, and if the
 * provided one is shorter than that, the remaining is filled with
 * spaces. */
voidaddReplyVerbatim(client*c, constchar*s, size_tlen, constchar*ext) {
if (c->resp==2) {
addReplyBulkCBuffer(c,s,len);
 } else {
charbuf[32];
size_tpreflen=snprintf(buf,sizeof(buf),"=%zu\r\nxxx:",len+4);
char*p=buf+preflen-4;
for (inti=0; i<3; i++) {
if (*ext=='\0') {
p[i] =' ';
 } else {
p[i] =*ext++;
 }
 }
addReplyProto(c,buf,preflen);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
 }
}

/* Add an array of C strings as status replies with a heading.
 * This function is typically invoked by from commands that support
 * subcommands in response to the 'help' subcommand. The help array
 * is terminated by NULL sentinel. */
voidaddReplyHelp(client*c, constchar**help) {
sdscmd=sdsnew((char*) c->argv[0]->ptr);
void*blenp=addReplyDeferredLen(c);
intblen=0;

sdstoupper(cmd);
addReplyStatusFormat(c,
"%s <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",cmd);
sdsfree(cmd);

while (help[blen]) addReplyStatus(c,help[blen++]);

addReplyStatus(c,"HELP");
addReplyStatus(c," Prints this help.");

blen+=1; /* Account for the header. */
blen+=2; /* Account for the footer. */
setDeferredArrayLen(c,blenp,blen);
}

/* Add a suggestive error reply.
 * This function is typically invoked by from commands that support
 * subcommands in response to an unknown subcommand or argument error. */
voidaddReplySubcommandSyntaxError(client*c) {
sdscmd=sdsnew((char*) c->argv[0]->ptr);
sdstoupper(cmd);
addReplyErrorFormat(c,
"Unknown subcommand or wrong number of arguments for '%s'. Try %s HELP.",
 (char*)c->argv[1]->ptr,cmd);
sdsfree(cmd);
}

/* Append 'src' client output buffers into 'dst' client output buffers.
 * This function clears the output buffers of 'src' */
voidAddReplyFromClient(client*dst, client*src) {
/* If the source client contains a partial response due to client output
 * buffer limits, propagate that to the dest rather than copy a partial
 * reply. We don't wanna run the risk of copying partial response in case
 * for some reason the output limits don't reach the same decision (maybe
 * they changed) */
if (src->flags&CLIENT_CLOSE_ASAP) {
sdsclient=catClientInfoString(sdsempty(),dst);
freeClientAsync(dst);
serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
sdsfree(client);
return;
 }

/* First add the static buffer (either into the static buffer or reply list) */
addReplyProto(dst,src->buf, src->bufpos);

/* We need to check with prepareClientToWrite again (after addReplyProto)
 * since addReplyProto may have changed something (like CLIENT_CLOSE_ASAP) */
if (prepareClientToWrite(dst) !=C_OK)
return;

/* We're bypassing _addReplyProtoToList, so we need to add the pre/post
 * checks in it. */
if (dst->flags&CLIENT_CLOSE_AFTER_REPLY) return;

/* Concatenate the reply list into the dest */
if (listLength(src->reply))
listJoin(dst->reply,src->reply);
dst->reply_bytes+=src->reply_bytes;
src->reply_bytes=0;
src->bufpos=0;

/* Check output buffer limits */
closeClientOnOutputBufferLimitReached(dst, 1);
}

/* Copy 'src' client output buffers into 'dst' client output buffers.
 * The function takes care of freeing the old output buffers of the
 * destination client. */
voidcopyClientOutputBuffer(client*dst, client*src) {
listEmpty(dst->reply);
dst->sentlen=0;
dst->bufpos=0;
dst->reply_bytes=0;

/* First copy src static buffer into dst (either static buffer or reply
 * list, maybe clients have different 'usable_buffer_size'). */
_addReplyToBufferOrList(dst,src->buf,src->bufpos);

/* Copy src reply list into the dest. */
list*reply=listDup(src->reply);
listJoin(dst->reply,reply);
dst->reply_bytes+=src->reply_bytes;
listRelease(reply);
}

/* Return true if the specified client has pending reply buffers to write to
 * the socket. */
intclientHasPendingReplies(client*c) {
returnc->bufpos||listLength(c->reply);
}

voidclientAcceptHandler(connection*conn) {
client*c=connGetPrivateData(conn);

if (connGetState(conn) !=CONN_STATE_CONNECTED) {
serverLog(LL_WARNING,
"Error accepting a client connection: %s",
connGetLastError(conn));
freeClientAsync(c);
return;
 }

/* If the server is running in protected mode (the default) and there
 * is no password set, nor a specific interface is bound, we don't accept
 * requests from non loopback interfaces. Instead we try to explain the
 * user what to do to fix it if needed. */
if (server.protected_mode&&
server.bindaddr_count==0&&
DefaultUser->flags&USER_FLAG_NOPASS&&
 !(c->flags&CLIENT_UNIX_SOCKET))
 {
charcip[NET_IP_STR_LEN+1] = { 0 };
connPeerToString(conn, cip, sizeof(cip)-1, NULL);