/*

* Copyright (c) 2009-Present, Redis Ltd.

* All rights reserved.

*

* Copyright (c) 2024-present, Valkey contributors.

* All rights reserved.

*

* Licensed under your choice of the Redis Source Available License 2.0

* (RSALv2) or the Server Side Public License v1 (SSPLv1).

*

* Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information.

*/

#include "server.h"

#include "atomicvar.h"

#include "cluster.h"

#include "script.h"

#include "fpconv_dtoa.h"

#include "fmtargs.h"

#include <sys/socket.h>

#include <sys/uio.h>

#include <math.h>

#include <ctype.h>

static void setProtocolError(const char *errstr, client *c);

static void pauseClientsByClient(mstime_t end, int isPauseClientAll);

int postponeClientRead(client *c);

char *getClientSockname(client *c);

int ProcessingEventsWhileBlocked = 0; /* See processEventsWhileBlocked(). */

__thread sds thread_reusable_qb = NULL;

__thread int thread_reusable_qb_used = 0; /* Avoid multiple clients using reusable query

* buffer due to nested command execution. */

/* 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_t sdsZmallocSize(sds s) {

void *sh = sdsAllocPtr(s);

return zmalloc_size(sh);

}

/* Return the size consumed from the allocator, for the specified hfield with

* metadata (mstr), including internal fragmentation. This function is used in

* order to compute the client output buffer size. */

size_t hfieldZmallocSize(hfield s) {

void *sh = hfieldGetAllocPtr(s);

return zmalloc_size(sh);

}

/* Return the amount of memory used by the sds string at object->ptr

* for a string object. This includes internal fragmentation. */

size_t getStringObjectSdsUsedMemory(robj *o) {

serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);

switch(o->encoding) {

case OBJ_ENCODING_RAW: return sdsZmallocSize(o->ptr);

case OBJ_ENCODING_EMBSTR: return zmalloc_size(o)-sizeof(robj);

default: return 0; /* Just integer encoding for now. */

}

}

/* Return the length of a string object.

* This does NOT includes internal fragmentation or sds unused space. */

size_t getStringObjectLen(robj *o) {

serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);

switch(o->encoding) {

case OBJ_ENCODING_RAW: return sdslen(o->ptr);

case OBJ_ENCODING_EMBSTR: return sdslen(o->ptr);

default: return 0; /* 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);

return buf;

}

void freeClientReplyValue(void *o) {

zfree(o);

}

/* This function links the client to the global linked list of clients.

* unlinkClient() does the opposite, among other things. */

void linkClient(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_t id = htonu64(c->id);

raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);

}

/* Initialize client authentication state.

*/

static void clientSetDefaultAuth(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);

}

int authRequired(client *c) {

/* Check if the user is authenticated. This check is skipped in case

* the default user is flagged as "nopass" and is active. */

int auth_required = (!(DefaultUser->flags & USER_FLAG_NOPASS) ||

(DefaultUser->flags & USER_FLAG_DISABLED)) &&

!c->authenticated;

return auth_required;

}

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) {

connEnableTcpNoDelay(conn);

if (server.tcpkeepalive)

connKeepAlive(conn,server.tcpkeepalive);

connSetReadHandler(conn, readQueryFromClient);

connSetPrivateData(conn, c);

}

c->buf = zmalloc_usable(PROTO_REPLY_CHUNK_BYTES, &c->buf_usable_size);

selectDb(c,0);

uint64_t client_id;

atomicGetIncr(server.next_client_id, client_id, 1);

c->id = client_id;

#ifdef LOG_REQ_RES

reqresReset(c, 0);

c->resp = server.client_default_resp;

#else

c->resp = 2;

#endif

c->conn = conn;

c->name = NULL;

c->lib_name = NULL;

c->lib_ver = NULL;

c->bufpos = 0;

c->buf_peak = c->buf_usable_size;

c->buf_peak_last_reset_time = server.unixtime;

c->ref_repl_buf_node = NULL;

c->ref_block_pos = 0;

c->qb_pos = 0;

c->querybuf = NULL;

c->querybuf_peak = 0;

c->reqtype = 0;

c->argc = 0;

c->argv = NULL;

c->argv_len = 0;

c->argv_len_sum = 0;

c->original_argc = 0;

c->original_argv = NULL;

c->cmd = c->lastcmd = c->realcmd = NULL;

c->cur_script = NULL;

c->multibulklen = 0;

c->bulklen = -1;

c->sentlen = 0;

c->flags = 0;

c->slot = -1;

c->ctime = c->lastinteraction = server.unixtime;

c->duration = 0;

clientSetDefaultAuth(c);

c->replstate = REPL_STATE_NONE;

c->repl_start_cmd_stream_on_ack = 0;

c->reploff = 0;

c->read_reploff = 0;

c->repl_applied = 0;

c->repl_ack_off = 0;

c->repl_ack_time = 0;

c->repl_aof_off = 0;

c->repl_last_partial_write = 0;

c->slave_listening_port = 0;

c->slave_addr = NULL;

c->slave_capa = SLAVE_CAPA_NONE;

c->slave_req = SLAVE_REQ_NONE;

c->reply = listCreate();

c->deferred_reply_errors = NULL;

c->reply_bytes = 0;

c->obuf_soft_limit_reached_time = 0;

listSetFreeMethod(c->reply,freeClientReplyValue);

listSetDupMethod(c->reply,dupClientReplyValue);

initClientBlockingState(c);

c->woff = 0;

c->watched_keys = listCreate();

c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType);

c->pubsub_patterns = dictCreate(&objectKeyPointerValueDictType);

c->pubsubshard_channels = dictCreate(&objectKeyPointerValueDictType);

c->peerid = NULL;

c->sockname = NULL;

c->client_list_node = NULL;

c->postponed_list_node = NULL;

c->pending_read_list_node = NULL;

c->client_tracking_redirection = 0;

c->client_tracking_prefixes = NULL;

c->last_memory_usage = 0;

c->last_memory_type = CLIENT_TYPE_NORMAL;

c->module_blocked_client = NULL;

c->module_auth_ctx = NULL;

c->auth_callback = NULL;

c->auth_callback_privdata = NULL;

c->auth_module = NULL;

listInitNode(&c->clients_pending_write_node, c);

c->mem_usage_bucket = NULL;

c->mem_usage_bucket_node = NULL;

if (conn) linkClient(c);

initClientMultiState(c);

return c;

}

void installClientWriteHandler(client *c) {

int ae_barrier = 0;

/* For the fsync=always policy, we want that a given FD is never

* served for reading and writing in the same event loop iteration,

* so that in the middle of receiving the query, and serving it

* to the client, we'll call beforeSleep() that will do the

* actual fsync of AOF to disk. the write barrier ensures that. */

if (server.aof_state == AOF_ON &&

server.aof_fsync == AOF_FSYNC_ALWAYS)

{

ae_barrier = 1;

}

if (connSetWriteHandlerWithBarrier(c->conn, sendReplyToClient, ae_barrier) == C_ERR) {

freeClientAsync(c);

}

}

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

void putClientInPendingWriteQueue(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_start_cmd_stream_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;

listLinkNodeHead(server.clients_pending_write, &c->clients_pending_write_node);

}

}

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

int prepareClientToWrite(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_SCRIPT|CLIENT_MODULE)) return C_OK;

/* If CLIENT_CLOSE_ASAP flag is set, we need not write anything. */

if (c->flags & CLIENT_CLOSE_ASAP) return C_ERR;

/* CLIENT REPLY OFF / SKIP handling: don't send replies.

* CLIENT_PUSHING handling: disables the reply silencing flags. */

if ((c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) &&

!(c->flags & CLIENT_PUSHING)) return C_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)) return C_ERR;

if (!c->conn) return C_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 put the client in pending write queue. Instead, it will be

* done by handleClientsWithPendingReadsUsingThreads() upon return.

*/

if (!clientHasPendingReplies(c) && io_threads_op == IO_THREADS_OP_IDLE)

putClientInPendingWriteQueue(c);

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

return C_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.

*

* Sanitizer suppression: client->buf_usable_size determined by

* zmalloc_usable_size() call. Writing beyond client->buf boundaries confuses

* sanitizer and generates a false positive out-of-bounds error */

REDIS_NO_SANITIZE("bounds")

size_t _addReplyToBuffer(client *c, const char *s, size_t len) {

size_t available = 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) return 0;

size_t reply_len = len > available ? available : len;

memcpy(c->buf+c->bufpos,s,reply_len);

c->bufpos+=reply_len;

/* We update the buffer peak after appending the reply to the buffer */

if(c->buf_peak < (size_t)c->bufpos)

c->buf_peak = (size_t)c->bufpos;

return reply_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, list *reply_list, const char *s, size_t len) {

listNode *ln = listLast(reply_list);

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

* to 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_t avail = tail->size - tail->used;

size_t copy = 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_t usable_size;

size_t size = 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(reply_list, tail);

c->reply_bytes += tail->size;

closeClientOnOutputBufferLimitReached(c, 1);

}

}

/* The subscribe / unsubscribe command family has a push as a reply,

* or in other words, it responds with a push (or several of them

* depending on how many arguments it got), and has no reply. */

int cmdHasPushAsReply(struct redisCommand *cmd) {

if (!cmd) return 0;

return cmd->proc == subscribeCommand || cmd->proc == unsubscribeCommand ||

cmd->proc == psubscribeCommand || cmd->proc == punsubscribeCommand ||

cmd->proc == ssubscribeCommand || cmd->proc == sunsubscribeCommand;

}

void _addReplyToBufferOrList(client *c, const char *s, size_t len) {

if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return;

/* Replicas should normally not cause any writes to the reply buffer. In case a rogue replica sent a command on the

* replication link that caused a reply to be generated we'll simply disconnect it.

* Note this is the simplest way to check a command added a response. Replication links are used to write data but

* not for responses, so we should normally never get here on a replica client. */

if (getClientType(c) == CLIENT_TYPE_SLAVE) {

sds cmdname = c->lastcmd ? c->lastcmd->fullname : NULL;

logInvalidUseAndFreeClientAsync(c, "Replica generated a reply to command '%s'",

cmdname ? cmdname : "<unknown>");

return;

}

/* We call it here because this function may affect the reply

* buffer offset (see function comment) */

reqresSaveClientReplyOffset(c);

/* If we're processing a push message into the current client (i.e. executing PUBLISH

* to a channel which we are subscribed to, then we wanna postpone that message to be added

* after the command's reply (specifically important during multi-exec). the exception is

* the SUBSCRIBE command family, which (currently) have a push message instead of a proper reply.

* The check for executing_client also avoids affecting push messages that are part of eviction.

* Check CLIENT_PUSHING first to avoid race conditions, as it's absent in module's fake client. */

if ((c->flags & CLIENT_PUSHING) && c == server.current_client &&

server.executing_client && !cmdHasPushAsReply(server.executing_client->cmd))

{

_addReplyProtoToList(c,server.pending_push_messages,s,len);

return;

}

size_t reply_len = _addReplyToBuffer(c,s,len);

if (len > reply_len) _addReplyProtoToList(c,c->reply,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. */

void addReply(client *c, robj *obj) {

if (prepareClientToWrite(c) != C_OK) return;

if (sdsEncodedObject(obj)) {

_addReplyToBufferOrList(c,obj->ptr,sdslen(obj->ptr));

} else if (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. */

char buf[32];

size_t len = 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. */

void addReplySds(client *c, sds s) {

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

void addReplyProto(client *c, const char *s, size_t len) {

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

void addReplyErrorLength(client *c, const char *s, size_t len) {

/* 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.)

* Possible flags:

* * ERR_REPLY_FLAG_NO_STATS_UPDATE - indicate not to update any error stats. */

void afterErrorReply(client *c, const char *s, size_t len, int flags) {

/* Module clients fall into two categories:

* Calls to RM_Call, in which case the error isn't being returned to a client, so should not be counted.

* Module thread safe context calls to RM_ReplyWithError, which will be added to a real client by the main thread later. */

if (c->flags & CLIENT_MODULE) {

if (!c->deferred_reply_errors) {

c->deferred_reply_errors = listCreate();

listSetFreeMethod(c->deferred_reply_errors, (void (*)(void*))sdsfree);

}

listAddNodeTail(c->deferred_reply_errors, sdsnewlen(s, len));

return;

}

if (!(flags & ERR_REPLY_FLAG_NO_STATS_UPDATE)) {

/* 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) {

const size_t errEndPos = (size_t)(spaceloc - s);

incrementErrorCount(s+1, errEndPos-1);

} else {

/* Fallback to ERR if we can't retrieve the error prefix */

incrementErrorCount("ERR", 3);

}

}

} else {

/* stat_total_error_replies will not be updated, which means that

* the cmd stats will not be updated as well, we still want this command

* to be counted as failed so we update it here. We update c->realcmd in

* case c->cmd was changed (like in GEOADD). */

c->realcmd->failed_calls++;

}

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

int ctype = 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";

} else if (ctype == CLIENT_TYPE_MASTER) {

to = "master";

from = "replica";

} else {

to = "replica";

from = "master";

}

if (len > 4096) len = 4096;

sds cmdname = c->lastcmd ? c->lastcmd->fullname : NULL;

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 ? cmdname : "<unknown>");

if (ctype == CLIENT_TYPE_MASTER && server.repl_backlog &&

server.repl_backlog->histlen > 0)

{

showLatestBacklog();

}

server.stat_unexpected_error_replies++;

/* Based off the propagation error behavior, check if we need to panic here. There

* are currently two checked cases:

* * If this command was from our master and we are not a writable replica.

* * We are reading from an AOF file. */

int panic_in_replicas = (ctype == CLIENT_TYPE_MASTER && server.repl_slave_ro)

&& (server.propagation_error_behavior == PROPAGATION_ERR_BEHAVIOR_PANIC ||

server.propagation_error_behavior == PROPAGATION_ERR_BEHAVIOR_PANIC_ON_REPLICAS);

int panic_in_aof = c->id == CLIENT_ID_AOF

&& server.propagation_error_behavior == PROPAGATION_ERR_BEHAVIOR_PANIC;

if (panic_in_replicas || panic_in_aof) {

serverPanic("This %s panicked sending an error to its %s"

" after processing the command '%s'",

from, to, cmdname ? cmdname : "<unknown>");

}

}

}

/* The 'err' object is expected to start with -ERRORCODE and end with \r\n.

* Unlike addReplyErrorSds and others alike which rely on addReplyErrorLength. */

void addReplyErrorObject(client *c, robj *err) {

addReply(c, err);

afterErrorReply(c, err->ptr, sdslen(err->ptr)-2, 0); /* Ignore trailing \r\n */

}

/* Sends either a reply or an error reply by checking the first char.

* If the first char is '-' the reply is considered an error.

* In any case the given reply is sent, if the reply is also recognize

* as an error we also perform some post reply operations such as

* logging and stats update. */

void addReplyOrErrorObject(client *c, robj *reply) {

serverAssert(sdsEncodedObject(reply));

sds rep = reply->ptr;

if (sdslen(rep) > 1 && rep[0] == '-') {

addReplyErrorObject(c, reply);

} else {

addReply(c, reply);

}

}

/* See addReplyErrorLength for expectations from the input string. */

void addReplyError(client *c, const char *err) {

addReplyErrorLength(c,err,strlen(err));

afterErrorReply(c,err,strlen(err),0);

}

/* Add error reply to the given client.

* Supported flags:

* * ERR_REPLY_FLAG_NO_STATS_UPDATE - indicate not to perform any error stats updates */

void addReplyErrorSdsEx(client *c, sds err, int flags) {

addReplyErrorLength(c,err,sdslen(err));

afterErrorReply(c,err,sdslen(err),flags);

sdsfree(err);

}

/* See addReplyErrorLength for expectations from the input string. */

/* As a side effect the SDS string is freed. */

void addReplyErrorSds(client *c, sds err) {

addReplyErrorSdsEx(c, err, 0);

}

/* See addReplyErrorLength for expectations from the input string. */

/* As a side effect the SDS string is freed. */

void addReplyErrorSdsSafe(client *c, sds err) {

err = sdsmapchars(err, "\r\n", " ", 2);

addReplyErrorSdsEx(c, err, 0);

}

/* Internal function used by addReplyErrorFormat, addReplyErrorFormatEx and RM_ReplyWithErrorFormat.

* Refer to afterErrorReply for more information about the flags. */

void addReplyErrorFormatInternal(client *c, int flags, const char *fmt, va_list ap) {

va_list cpy;

va_copy(cpy,ap);

sds s = sdscatvprintf(sdsempty(),fmt,cpy);

va_end(cpy);

/* 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),flags);

sdsfree(s);

}

void addReplyErrorFormatEx(client *c, int flags, const char *fmt, ...) {

va_list ap;

va_start(ap,fmt);

addReplyErrorFormatInternal(c, flags, fmt, ap);

va_end(ap);

}

/* See addReplyErrorLength for expectations from the formatted string.

* The formatted string is safe to contain \r and \n anywhere. */

void addReplyErrorFormat(client *c, const char *fmt, ...) {

va_list ap;

va_start(ap,fmt);

addReplyErrorFormatInternal(c, 0, fmt, ap);

va_end(ap);

}

void addReplyErrorArity(client *c) {

addReplyErrorFormat(c, "wrong number of arguments for '%s' command",

c->cmd->fullname);

}

void addReplyErrorExpireTime(client *c) {

addReplyErrorFormat(c, "invalid expire time in '%s' command",

c->cmd->fullname);

}

void addReplyStatusLength(client *c, const char *s, size_t len) {

addReplyProto(c,"+",1);

addReplyProto(c,s,len);

addReplyProto(c,"\r\n",2);

}

void addReplyStatus(client *c, const char *status) {

addReplyStatusLength(c,status,strlen(status));

}

void addReplyStatusFormat(client *c, const char *fmt, ...) {

va_list ap;

va_start(ap,fmt);

sds s = 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. */

void trimReplyUnusedTailSpace(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_t usable_size;

size_t old_size = tail->size;

tail = zrealloc_usable(tail, tail->used + sizeof(clientReplyBlock), &usable_size);

/* take over the allocation's internal fragmentation (at least for

* memory usage tracking) */

tail->size = usable_size - 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) return NULL;

/* Replicas should normally not cause any writes to the reply buffer. In case a rogue replica sent a command on the

* replication link that caused a reply to be generated we'll simply disconnect it.

* Note this is the simplest way to check a command added a response. Replication links are used to write data but

* not for responses, so we should normally never get here on a replica client. */

if (getClientType(c) == CLIENT_TYPE_SLAVE) {

sds cmdname = c->lastcmd ? c->lastcmd->fullname : NULL;

logInvalidUseAndFreeClientAsync(c, "Replica generated a reply to command '%s'",

cmdname ? cmdname : "<unknown>");

return NULL;

}

/* We call it here because this function conceptually affects the reply

* buffer offset (see function comment) */

reqresSaveClientReplyOffset(c);

trimReplyUnusedTailSpace(c);

listAddNodeTail(c->reply,NULL); /* NULL is our placeholder. */

return listLast(c->reply);

}

void setDeferredReply(client *c, void *node, const char *s, size_t length) {

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_t len_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 */

size_t usable_size;

clientReplyBlock *buf = zmalloc_usable(length + sizeof(clientReplyBlock), &usable_size);

/* Take over the allocation's internal fragmentation */

buf->size = usable_size - 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. */

void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {

serverAssert(length >= 0);

/* Abort when *node is NULL: when the client should not accept writes

* we return NULL in addReplyDeferredLen() */

if (node == NULL) return;

/* Things like *2\r\n, %3\r\n or ~4\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. */

const size_t hdr_len = OBJ_SHARED_HDR_STRLEN(length);

const int opt_hdr = length < OBJ_SHARED_BULKHDR_LEN;

if (prefix == '*' && opt_hdr) {

setDeferredReply(c, node, shared.mbulkhdr[length]->ptr, hdr_len);

return;

}

if (prefix == '%' && opt_hdr) {

setDeferredReply(c, node, shared.maphdr[length]->ptr, hdr_len);

return;

}

if (prefix == '~' && opt_hdr) {

setDeferredReply(c, node, shared.sethdr[length]->ptr, hdr_len);

return;

}

char lenstr[128];

size_t lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);

setDeferredReply(c, node, lenstr, lenstr_len);

}

void setDeferredArrayLen(client *c, void *node, long length) {

setDeferredAggregateLen(c,node,length,'*');

}

void setDeferredMapLen(client *c, void *node, long length) {

int prefix = c->resp == 2 ? '*' : '%';

if (c->resp == 2) length *= 2;

setDeferredAggregateLen(c,node,length,prefix);

}

void setDeferredSetLen(client *c, void *node, long length) {

int prefix = c->resp == 2 ? '*' : '~';

setDeferredAggregateLen(c,node,length,prefix);

}

void setDeferredAttributeLen(client *c, void *node, long length) {

serverAssert(c->resp >= 3);

setDeferredAggregateLen(c,node,length,'|');

}

void setDeferredPushLen(client *c, void *node, long length) {

serverAssert(c->resp >= 3);

setDeferredAggregateLen(c,node,length,'>');

}

/* Add a double as a bulk reply */

void addReplyDouble(client *c, double d) {

if (c->resp == 3) {

char dbuf[MAX_D2STRING_CHARS+3];

dbuf[0] = ',';

const int dlen = d2string(dbuf+1,sizeof(dbuf)-1,d);

dbuf[dlen+1] = '\r';

dbuf[dlen+2] = '\n';

dbuf[dlen+3] = '\0';

addReplyProto(c,dbuf,dlen+3);

} else {

char dbuf[MAX_LONG_DOUBLE_CHARS+32];

/* In order to prepend the string length before the formatted number,

* but still avoid an extra memcpy of the whole number, we reserve space

* for maximum header `$0000\r\n`, print double, add the resp header in

* front of it, and then send the buffer with the right `start` offset. */

const int dlen = d2string(dbuf+7,sizeof(dbuf)-7,d);

int digits = digits10(dlen);

int start = 4 - digits;

serverAssert(start >= 0);

dbuf[start] = '$';

/* Convert `dlen` to string, putting it's digits after '$' and before the

* formatted double string. */

for(int i = digits, val = dlen; val && i > 0 ; --i, val /= 10) {

dbuf[start + i] = "0123456789"[val % 10];

}

dbuf[5] = '\r';

dbuf[6] = '\n';

dbuf[dlen+7] = '\r';

dbuf[dlen+8] = '\n';

dbuf[dlen+9] = '\0';

addReplyProto(c,dbuf+start,dlen+9-start);

}

}

void addReplyBigNum(client *c, const char* num, size_t len) {

if (c->resp == 2) {

addReplyBulkCBuffer(c, num, len);

} else {

addReplyProto(c,"(",1);

addReplyProto(c,num,len);

addReplyProto(c,"\r\n",2);

}

}

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

void addReplyHumanLongDouble(client *c, long double d) {

if (c->resp == 2) {

robj *o = createStringObjectFromLongDouble(d,1);

addReplyBulk(c,o);

decrRefCount(o);

} else {

char buf[MAX_LONG_DOUBLE_CHARS];

int len = 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>. */

static void _addReplyLongLongWithPrefix(client *c, long long ll, char prefix) {

char buf[128];

int len;

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

const int opt_hdr = ll < OBJ_SHARED_BULKHDR_LEN && ll >= 0;

const size_t hdr_len = OBJ_SHARED_HDR_STRLEN(ll);

if (prefix == '*' && opt_hdr) {

_addReplyToBufferOrList(c, shared.mbulkhdr[ll]->ptr, hdr_len);

return;

} else if (prefix == '$' && opt_hdr) {

_addReplyToBufferOrList(c, shared.bulkhdr[ll]->ptr, hdr_len);

return;

} else if (prefix == '%' && opt_hdr) {

_addReplyToBufferOrList(c, shared.maphdr[ll]->ptr, hdr_len);

return;

} else if (prefix == '~' && opt_hdr) {

_addReplyToBufferOrList(c, shared.sethdr[ll]->ptr, hdr_len);

return;

}

buf[0] = prefix;

len = ll2string(buf + 1, sizeof(buf) - 1, ll);

buf[len + 1] = '\r';

buf[len + 2] = '\n';

_addReplyToBufferOrList(c, buf, len + 3);

}

void addReplyLongLong(client *c, long long ll) {

if (ll == 0)

addReply(c,shared.czero);

else if (ll == 1)

addReply(c, shared.cone);

else {

if (prepareClientToWrite(c) != C_OK) return;

_addReplyLongLongWithPrefix(c, ll, ':');

}

}

void addReplyLongLongFromStr(client *c, robj *str) {

addReplyProto(c,":",1);

addReply(c,str);

addReplyProto(c,"\r\n",2);

}

void addReplyAggregateLen(client *c, long length, int prefix) {

serverAssert(length >= 0);

if (prepareClientToWrite(c) != C_OK) return;

_addReplyLongLongWithPrefix(c, length, prefix);

}

void addReplyArrayLen(client *c, long length) {

addReplyAggregateLen(c,length,'*');

}

void addReplyMapLen(client *c, long length) {

int prefix = c->resp == 2 ? '*' : '%';

if (c->resp == 2) length *= 2;

addReplyAggregateLen(c,length,prefix);

}

void addReplySetLen(client *c, long length) {

int prefix = c->resp == 2 ? '*' : '~';

addReplyAggregateLen(c,length,prefix);