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

switch(o->encoding) {

case OBJ_ENCODING_RAW: return sdsZmallocSize((sds)ptrFromObj(o));

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

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

}

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

switch(o->encoding) {

case OBJ_ENCODING_RAW: return sdslen(szFromObj(o));

case OBJ_ENCODING_EMBSTR: return sdslen(szFromObj(o));

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

}

serverAssert(GlobalLocksAcquired());

listAddNodeTail(g_pserver->clients,c);

/* Note that we remember the linked list node where the client is stored,

c->client_list_node = listLast(g_pserver->clients);

if (c->conn != nullptr) atomicIncr(g_pserver->rgthreadvar[c->iel].cclients, 1);

uint64_t id = htonu64(c->id);

raxInsert(g_pserver->clients_index,(unsigned char*)&id,sizeof(id),c,NULL);

/* Schedule the client to write the output buffers to the socket only

if (!(c->flags & CLIENT_PENDING_WRITE) &&

(c->replstate == REPL_STATE_NONE ||

(c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack)))

{

AssertCorrectThread(c);

serverAssert(c->lock.fOwnLock());

/* Here instead of installing the write handler, we just flag the

c->flags |= CLIENT_PENDING_WRITE;

std::unique_lock<fastlock> lockf(g_pserver->rgthreadvar[c->iel].lockPendingWrite);

g_pserver->rgthreadvar[c->iel].clients_pending_write.push_back(c);

}

serverAssert(GlobalLocksAcquired());

if (!(c->fPendingAsyncWrite)) {

c->fPendingAsyncWrite = TRUE;

listAddNodeHead(serverTL->clients_pending_asyncwrite,c);

}

bool fAsync = !FCorrectThread(c); // Not async if we're on the right thread

if (!fAsync) {

serverAssert(c->conn == nullptr || c->lock.fOwnLock());

} else {

serverAssert(GlobalLocksAcquired());

}

auto flags = c->flags.load(std::memory_order_relaxed);

if (flags & CLIENT_FORCE_REPLY) return C_OK; // FORCE REPLY means we're doing something else with the buffer.

// do not install a write handler

/* If it's the Lua client we always return ok without installing any

if (flags & (CLIENT_LUA|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. */

if (flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) return C_ERR;

/* Masters don't receive replies, unless CLIENT_MASTER_FORCE_REPLY flag

if ((flags & CLIENT_MASTER) &&

!(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

if (!fAsync && !clientHasPendingReplies(c)) clientInstallWriteHandler(c);

if (fAsync && !(c->fPendingAsyncWrite)) clientInstallAsyncWriteHandler(c);

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

return C_OK;

if (c->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_AFTER_REPLY) return C_OK;

bool fAsync = !FCorrectThread(c);

if (fAsync)

{

serverAssert(GlobalLocksAcquired());

if (c->replyAsync == nullptr || (c->replyAsync->size - c->replyAsync->used) < len)

{

if (c->replyAsync == nullptr) {

size_t newsize = std::max(len, (size_t)PROTO_ASYNC_REPLY_CHUNK_BYTES);

clientReplyBlock *replyNew = (clientReplyBlock*)zmalloc(sizeof(clientReplyBlock) + newsize);

replyNew->size = zmalloc_usable(replyNew) - sizeof(clientReplyBlock);

replyNew->used = 0;

c->replyAsync = replyNew;

} else {

size_t newsize = std::max(c->replyAsync->used + len, c->replyAsync->size*2);

clientReplyBlock *replyNew = (clientReplyBlock*)zmalloc(sizeof(clientReplyBlock) + newsize);

replyNew->size = zmalloc_usable(replyNew) - sizeof(clientReplyBlock);

replyNew->used = c->replyAsync->used;

memcpy(replyNew->buf(), c->replyAsync->buf(), c->replyAsync->used);

zfree(c->replyAsync);

c->replyAsync = replyNew;

}

}

memcpy(c->replyAsync->buf() + c->replyAsync->used,s,len);

c->replyAsync->used += len;

}

else

{

size_t available = sizeof(c->buf)-c->bufpos;

/* If there already are entries in the reply list, we cannot

if (listLength(c->reply) > 0) return C_ERR;

/* Check that the buffer has enough space available for this string. */

if (len > available) return C_ERR;

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

c->bufpos+=len;

}

return C_OK;

if (c->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_AFTER_REPLY) return;

AssertCorrectThread(c);

listNode *ln = listLast(c->reply);

clientReplyBlock *tail = (clientReplyBlock*) (ln? listNodeValue(ln): NULL);

/* Note that 'tail' may be NULL even if we have a tail node, because when

/* Append to tail string when possible. */

if (tail) {

/* Copy the part we can fit into the tail, and leave the rest for a

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

size_t size = len < PROTO_REPLY_CHUNK_BYTES? PROTO_REPLY_CHUNK_BYTES: len;

tail = (clientReplyBlock*)zmalloc(size + sizeof(clientReplyBlock), MALLOC_LOCAL);

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

tail->size = zmalloc_usable(tail) - sizeof(clientReplyBlock);

tail->used = len;

memcpy(tail->buf(), s, len);

listAddNodeTail(c->reply, tail);

c->reply_bytes += tail->size;

}

asyncCloseClientOnOutputBufferLimitReached(c);

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

if (sdsEncodedObject(obj)) {

if (_addReplyToBuffer(c,(const char*)ptrFromObj(obj),sdslen((sds)ptrFromObj(obj))) != C_OK)

_addReplyProtoToList(c,(const char*)ptrFromObj(obj),sdslen((sds)ptrFromObj(obj)));

} else if (obj->encoding == OBJ_ENCODING_INT) {

/* For integer encoded strings we just convert it into a string

char buf[32];

size_t len = ll2string(buf,sizeof(buf),(long)ptrFromObj(obj));

if (_addReplyToBuffer(c,buf,len) != C_OK)

_addReplyProtoToList(c,buf,len);

} else {

serverPanic("Wrong obj->encoding in addReply()");

}

if (prepareClientToWrite(c) != C_OK) {

/* The caller expects the sds to be free'd. */

sdsfree(s);

return;

}

if (_addReplyToBuffer(c,s,sdslen(s)) != C_OK)

_addReplyProtoToList(c,s,sdslen(s));

sdsfree(s);

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

if (_addReplyToBuffer(c,s,len) != C_OK)

_addReplyProtoToList(c,s,len);

{

std::string newstr;

size_t len = sdslen(str);

for (size_t ich = 0; ich < len; ++ich)

{

char ch = str[ich];

switch (ch)

{

case '\n':

newstr += "\\n";

break;

case '\t':

newstr += "\\t";

break;

case '\r':

newstr += "\\r";

break;

default:

newstr += ch;

}

}

return newstr;

/* If the string already starts with "-..." then the error code

if (!len || s[0] != '-') addReplyProto(c,"-ERR ",5);

addReplyProto(c,s,len);

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

/* Sometimes it could be normal that a replica replies to a master with

int ctype = getClientType(c);

if (ctype == CLIENT_TYPE_MASTER || ctype == CLIENT_TYPE_SLAVE || c->id == CLIENT_ID_AOF) {

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

const char *cmdname = c->lastcmd ? c->lastcmd->name : "<unknown>";

switch (severity) {

case ERR_NOTICE:

serverLog(LL_NOTICE,"== NOTICE == This %s is rejecting a command "

"from its %s: '%.*s' after processing the command "

"'%s'", from, to, (int)len, s, cmdname);

break;

case ERR_WARNING:

serverLog(LL_WARNING,"== WARNING == This %s is rejecting a command "

"from its %s: '%.*s' after processing the command "

"'%s'", from, to, (int)len, s, cmdname);

break;

case ERR_ERROR:

serverLog(LL_WARNING,"== ERROR == This %s is sending an error "

"to its %s: '%.*s' after processing the command "

"'%s'", from, to, (int)len, s, cmdname);

break;

case ERR_CRITICAL:

default:

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

break;

}

if (ctype == CLIENT_TYPE_MASTER && g_pserver->repl_backlog &&

g_pserver->repl_backlog_histlen > 0)

{

showLatestBacklog();

}

g_pserver->stat_unexpected_error_replies++;

}

va_list ap;

va_start(ap,fmt);

sds s = 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

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

addReplyErrorLength(c,s,sdslen(s));

afterErrorReply(c,s,sdslen(s));

sdsfree(s);

addReplyProto(c,"+",1);

addReplyProto(c,s,len);

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

va_list ap;

va_start(ap,fmt);

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

va_end(ap);

addReplyStatusLength(c,s,sdslen(s));

sdsfree(s);

listNode *ln = listLast(c->reply);

clientReplyBlock *tail = ln? (clientReplyBlock*)listNodeValue(ln): NULL;

/* Note that 'tail' may be NULL even if we have a tail node, becuase when

if (!tail) return;

/* We only try to trim the space is relatively high (more than a 1/4 of the

if (tail->size - tail->used > tail->size / 4 &&

tail->used < PROTO_REPLY_CHUNK_BYTES)

{

size_t old_size = tail->size;

tail = (clientReplyBlock*)zrealloc(tail, tail->used + sizeof(clientReplyBlock));

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

tail->size = zmalloc_usable(tail) - sizeof(clientReplyBlock);

c->reply_bytes = c->reply_bytes + tail->size - old_size;

listNodeValue(ln) = tail;

}

serverAssert(length >= 0);

listNode *ln = (listNode*)node;

clientReplyBlock *next;

char lenstr[128];

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

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

if (node == NULL) return;

serverAssert(!listNodeValue(ln));

/* Normally we fill this dummy NULL node, added by addReplyDeferredLen(),

if (ln->next != NULL && (next = (clientReplyBlock*)listNodeValue(ln->next)) &&

next->size - next->used >= lenstr_len &&

next->used < PROTO_REPLY_CHUNK_BYTES * 4) {

memmove(next->buf() + lenstr_len, next->buf(), next->used);

memcpy(next->buf(), lenstr, lenstr_len);

next->used += lenstr_len;

listDelNode(c->reply,ln);

} else {

/* Create a new node */

clientReplyBlock *buf = (clientReplyBlock*)zmalloc(lenstr_len + sizeof(clientReplyBlock), MALLOC_LOCAL);

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

buf->size = zmalloc_usable(buf) - sizeof(clientReplyBlock);

buf->used = lenstr_len;

memcpy(buf->buf(), lenstr, lenstr_len);

listNodeValue(ln) = buf;

c->reply_bytes += buf->size;

}

asyncCloseClientOnOutputBufferLimitReached(c);

{

if (FCorrectThread(c)) {

setDeferredAggregateLen(c, node, length, prefix);

return;

}

char lenstr[128];

int lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);

size_t idxSplice = (size_t)node;

serverAssert(idxSplice <= c->replyAsync->used);

if (c->replyAsync->size < (c->replyAsync->used + lenstr_len))

{

int newsize = std::max(c->replyAsync->used + lenstr_len, c->replyAsync->size*2);

clientReplyBlock *replyNew = (clientReplyBlock*)zmalloc(sizeof(clientReplyBlock) + newsize);

replyNew->size = zmalloc_usable(replyNew) - sizeof(clientReplyBlock);

replyNew->used = c->replyAsync->used;

memcpy(replyNew->buf(), c->replyAsync->buf(), c->replyAsync->used);

zfree(c->replyAsync);

c->replyAsync = replyNew;

}

memmove(c->replyAsync->buf() + idxSplice + lenstr_len, c->replyAsync->buf() + idxSplice, c->replyAsync->used - idxSplice);

memcpy(c->replyAsync->buf() + idxSplice, lenstr, lenstr_len);

c->replyAsync->used += lenstr_len;

if (FCorrectThread(c))

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

else

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

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

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

setDeferredAggregateLen(c,node,length,prefix);

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

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

setDeferredAggregateLen(c,node,length,prefix);

if (std::isinf(d)) {

/* Libc in odd systems (Hi Solaris!) will format infinite in a

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 {

char dbuf[MAX_LONG_DOUBLE_CHARS+3],

sbuf[MAX_LONG_DOUBLE_CHARS+32];

int dlen, 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);

}

}

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

}

char buf[128];

int len;

/* Things like $3\r\n or *2\r\n are emitted very often by the protocol

if (prefix == '*' && ll < OBJ_SHARED_BULKHDR_LEN && ll >= 0) {

addReply(c,shared.mbulkhdr[ll]);

return;

} else if (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);

if (ll == 0)

addReply(c,shared.czero);

else if (ll == 1)

addReply(c,shared.cone);

else

addReplyLongLongWithPrefix(c,ll,':');

if (prefix == '*' && length < OBJ_SHARED_BULKHDR_LEN)

addReply(c,shared.mbulkhdr[length]);

else

addReplyLongLongWithPrefix(c,length,prefix);

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

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

addReplyAggregateLen(c,length,prefix);

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

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

addReplyAggregateLen(c,length,prefix);

if (c->resp == 2) {

addReplyProto(c,"$-1\r\n",5);

} else {

addReplyProto(c,"_\r\n",3);

}

if (c->resp == 2) {

addReply(c, b ? shared.cone : shared.czero);

} else {

addReplyProto(c, b ? "#t\r\n" : "#f\r\n",4);

}

{

if (c->resp == 2) {

addReplyProto(c,"*-1\r\n",5);

} else {

addReplyProto(c,"_\r\n",3);

}

size_t len = stringObjectLen(obj);

if (len < OBJ_SHARED_BULKHDR_LEN)

addReply(c,shared.bulkhdr[len]);

else

addReplyLongLongWithPrefix(c,len,'$');

addReplyBulkLen(c,obj);

addReply(c,obj);

addReply(c,shared.crlf);

addReplyLongLongWithPrefix(c,len,'$');

addReplyProto(c,(const char*)p,len);

addReply(c,shared.crlf);

addReplyLongLongWithPrefix(c,sdslen(s),'$');

addReplySds(c,s);

addReply(c,shared.crlf);

if (s == NULL) {

if (c->resp < 3)

addReply(c,shared.nullbulk);

else

addReplyNull(c);

} else {

addReplyBulkCBuffer(c,s,strlen(s));

}

char buf[64];

int len;

len = ll2string(buf,64,ll);

addReplyBulkCBuffer(c,buf,len);

if (c->resp == 2) {

addReplyBulkCBuffer(c,s,len);

} else {

char buf[32];

size_t preflen = snprintf(buf,sizeof(buf),"=%zu\r\nxxx:",len+4);

char *p = buf+preflen-4;

for (int i = 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);

}

sds cmd = sdsnew((char*) ptrFromObj(c->argv[0]));

void *blenp = addReplyDeferredLen(c);

int blen = 0;

sdstoupper(cmd);

addReplyStatusFormat(c,

"%s <subcommand> arg arg ... arg. Subcommands are:",cmd);

sdsfree(cmd);

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

blen++; /* Account for the header line(s). */

setDeferredArrayLen(c,blenp,blen);