diff --git a/src/networking.c b/src/networking.c index 76b062a6..dfd8acfa 100644 --- a/src/networking.c +++ b/src/networking.c @@ -32,6 +32,8 @@ #endif #include "server.h" +#include "atomicvar.h" + #ifdef _WIN32 #include "Win32_Interop/Win32_QFork.h" #else @@ -39,7 +41,6 @@ #endif #include #include -#include "atomicvar.h" WIN32_ONLY(extern int WSIOCP_QueueAccept(int listenfd);) @@ -56,10 +57,10 @@ size_t sdsZmallocSize(sds s) { /* Return the amount of memory used by the sds string at object->ptr * for a string object. */ size_t getStringObjectSdsUsedMemory(robj *o) { - serverAssertWithInfo(NULL, o, o->type == OBJ_STRING); - switch (o->encoding) { + 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); + case OBJ_ENCODING_EMBSTR: return zmalloc_size(o)-sizeof(robj); default: return 0; /* Just integer encoding for now. */ } } @@ -74,7 +75,7 @@ void freeClientReplyValue(void *o) { } int listMatchObjects(void *a, void *b) { - return equalStringObjects(a, b); + return equalStringObjects(a,b); } /* This function links the client to the global linked list of clients. @@ -95,11 +96,11 @@ client *createClient(int fd) { * 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 (fd != -1) { - anetNonBlock(NULL, fd); - anetEnableTcpNoDelay(NULL, fd); + anetNonBlock(NULL,fd); + anetEnableTcpNoDelay(NULL,fd); if (server.tcpkeepalive) - anetKeepAlive(NULL, fd, server.tcpkeepalive); - if (aeCreateFileEvent(server.el, fd, AE_READABLE, + anetKeepAlive(NULL,fd,server.tcpkeepalive); + if (aeCreateFileEvent(server.el,fd,AE_READABLE, readQueryFromClient, c) == AE_ERR) { close(fd); @@ -108,9 +109,9 @@ client *createClient(int fd) { } } - selectDb(c, 0); + selectDb(c,0); uint64_t client_id; - atomicGetIncr(server.next_client_id, client_id, 1); + atomicGetIncr(server.next_client_id,client_id,1); c->id = client_id; c->fd = fd; c->name = NULL; @@ -140,22 +141,22 @@ client *createClient(int fd) { c->reply = listCreate(); c->reply_bytes = 0; c->obuf_soft_limit_reached_time = 0; - listSetFreeMethod(c->reply, freeClientReplyValue); - listSetDupMethod(c->reply, dupClientReplyValue); + listSetFreeMethod(c->reply,freeClientReplyValue); + listSetDupMethod(c->reply,dupClientReplyValue); c->btype = BLOCKED_NONE; c->bpop.timeout = 0; - c->bpop.keys = dictCreate(&objectKeyPointerValueDictType, NULL); + c->bpop.keys = dictCreate(&objectKeyPointerValueDictType,NULL); c->bpop.target = NULL; c->bpop.numreplicas = 0; c->bpop.reploffset = 0; c->woff = 0; c->watched_keys = listCreate(); - c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType, NULL); + c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType,NULL); c->pubsub_patterns = listCreate(); c->peerid = NULL; c->client_list_node = NULL; - listSetFreeMethod(c->pubsub_patterns, decrRefCountVoid); - listSetMatchMethod(c->pubsub_patterns, listMatchObjects); + listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid); + listSetMatchMethod(c->pubsub_patterns,listMatchObjects); if (fd != -1) linkClient(c); initClientMultiState(c); return c; @@ -186,10 +187,10 @@ client *createClient(int fd) { 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_LUA | CLIENT_MODULE)) return C_OK; + if (c->flags & (CLIENT_LUA|CLIENT_MODULE)) return C_OK; /* CLIENT REPLY OFF / SKIP handling: don't send replies. */ - if (c->flags & (CLIENT_REPLY_OFF | CLIENT_REPLY_SKIP)) return C_ERR; + if (c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) return C_ERR; /* Masters don't receive replies, unless CLIENT_MASTER_FORCE_REPLY flag * is set. */ @@ -205,7 +206,7 @@ int prepareClientToWrite(client *c) { if (!clientHasPendingReplies(c) && !(c->flags & CLIENT_PENDING_WRITE) && (c->replstate == REPL_STATE_NONE || - (c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack))) + (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 @@ -214,7 +215,7 @@ int prepareClientToWrite(client *c) { * 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); + listAddNodeHead(server.clients_pending_write,c); } /* Authorize the caller to queue in the output buffer of this client. */ @@ -226,7 +227,7 @@ int prepareClientToWrite(client *c) { * -------------------------------------------------------------------------- */ int _addReplyToBuffer(client *c, const char *s, size_t len) { - size_t available = sizeof(c->buf) - c->bufpos; + size_t available = sizeof(c->buf)-c->bufpos; if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return C_OK; @@ -237,9 +238,9 @@ int _addReplyToBuffer(client *c, const char *s, size_t len) { /* 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 += (int) len; WIN_PORT_FIX /* cast (int) */ - return C_OK; + memcpy(c->buf+c->bufpos,s,len); + c->bufpos+= (int)len; WIN_PORT_FIX /* cast (int) */ + return C_OK; } void _addReplyObjectToList(client *c, robj *o) { @@ -247,23 +248,21 @@ void _addReplyObjectToList(client *c, robj *o) { if (listLength(c->reply) == 0) { sds s = sdsdup(o->ptr); - listAddNodeTail(c->reply, s); + listAddNodeTail(c->reply,s); c->reply_bytes += sdslen(s); - } - else { + } else { listNode *ln = listLast(c->reply); sds tail = listNodeValue(ln); /* Append to this object when possible. If tail == NULL it was * set via addDeferredMultiBulkLength(). */ - if (tail && sdslen(tail) + sdslen(o->ptr) <= PROTO_REPLY_CHUNK_BYTES) { - tail = sdscatsds(tail, o->ptr); + if (tail && sdslen(tail)+sdslen(o->ptr) <= PROTO_REPLY_CHUNK_BYTES) { + tail = sdscatsds(tail,o->ptr); listNodeValue(ln) = tail; c->reply_bytes += sdslen(o->ptr); - } - else { + } else { sds s = sdsdup(o->ptr); - listAddNodeTail(c->reply, s); + listAddNodeTail(c->reply,s); c->reply_bytes += sdslen(s); } } @@ -279,23 +278,21 @@ void _addReplySdsToList(client *c, sds s) { } if (listLength(c->reply) == 0) { - listAddNodeTail(c->reply, s); + listAddNodeTail(c->reply,s); c->reply_bytes += sdslen(s); - } - else { + } else { listNode *ln = listLast(c->reply); sds tail = listNodeValue(ln); /* Append to this object when possible. If tail == NULL it was * set via addDeferredMultiBulkLength(). */ - if (tail && sdslen(tail) + sdslen(s) <= PROTO_REPLY_CHUNK_BYTES) { - tail = sdscatsds(tail, s); + if (tail && sdslen(tail)+sdslen(s) <= PROTO_REPLY_CHUNK_BYTES) { + tail = sdscatsds(tail,s); listNodeValue(ln) = tail; c->reply_bytes += sdslen(s); sdsfree(s); - } - else { - listAddNodeTail(c->reply, s); + } else { + listAddNodeTail(c->reply,s); c->reply_bytes += sdslen(s); } } @@ -306,24 +303,22 @@ void _addReplyStringToList(client *c, const char *s, size_t len) { if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return; if (listLength(c->reply) == 0) { - sds node = sdsnewlen(s, len); - listAddNodeTail(c->reply, node); + sds node = sdsnewlen(s,len); + listAddNodeTail(c->reply,node); c->reply_bytes += len; - } - else { + } else { listNode *ln = listLast(c->reply); sds tail = listNodeValue(ln); /* Append to this object when possible. If tail == NULL it was * set via addDeferredMultiBulkLength(). */ - if (tail && sdslen(tail) + len <= PROTO_REPLY_CHUNK_BYTES) { - tail = sdscatlen(tail, s, len); + if (tail && sdslen(tail)+len <= PROTO_REPLY_CHUNK_BYTES) { + tail = sdscatlen(tail,s,len); listNodeValue(ln) = tail; c->reply_bytes += len; - } - else { - sds node = sdsnewlen(s, len); - listAddNodeTail(c->reply, node); + } else { + sds node = sdsnewlen(s,len); + listAddNodeTail(c->reply,node); c->reply_bytes += len; } } @@ -346,10 +341,9 @@ void addReply(client *c, robj *obj) { * we'll be able to send the object to the client without * messing with its page. */ if (sdsEncodedObject(obj)) { - if (_addReplyToBuffer(c, obj->ptr, sdslen(obj->ptr)) != C_OK) - _addReplyObjectToList(c, obj); - } - else if (obj->encoding == OBJ_ENCODING_INT) { + if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK) + _addReplyObjectToList(c,obj); + } else if (obj->encoding == OBJ_ENCODING_INT) { /* Optimization: if there is room in the static buffer for 32 bytes * (more than the max chars a 64 bit integer can take as string) we * avoid decoding the object and go for the lower level approach. */ @@ -357,18 +351,17 @@ void addReply(client *c, robj *obj) { char buf[32]; int len; - len = ll2string(buf, sizeof(buf), (PORT_LONG) obj->ptr); - if (_addReplyToBuffer(c, buf, len) == C_OK) + len = ll2string(buf,sizeof(buf),(PORT_LONG)obj->ptr); + if (_addReplyToBuffer(c,buf,len) == C_OK) return; /* else... continue with the normal code path, but should never * happen actually since we verified there is room. */ } obj = getDecodedObject(obj); - if (_addReplyToBuffer(c, obj->ptr, sdslen(obj->ptr)) != C_OK) - _addReplyObjectToList(c, obj); + if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK) + _addReplyObjectToList(c,obj); decrRefCount(obj); - } - else { + } else { serverPanic("Wrong obj->encoding in addReply()"); } } @@ -379,12 +372,11 @@ void addReplySds(client *c, sds s) { sdsfree(s); return; } - if (_addReplyToBuffer(c, s, sdslen(s)) == C_OK) { + if (_addReplyToBuffer(c,s,sdslen(s)) == C_OK) { sdsfree(s); - } - else { + } else { /* This method free's the sds when it is no longer needed. */ - _addReplySdsToList(c, s); + _addReplySdsToList(c,s); } } @@ -398,14 +390,14 @@ void addReplySds(client *c, sds s) { * in the list of objects. */ void addReplyString(client *c, const char *s, size_t len) { if (prepareClientToWrite(c) != C_OK) return; - if (_addReplyToBuffer(c, s, len) != C_OK) - _addReplyStringToList(c, s, len); + if (_addReplyToBuffer(c,s,len) != C_OK) + _addReplyStringToList(c,s,len); } void addReplyErrorLength(client *c, const char *s, size_t len) { - addReplyString(c, "-ERR ", 5); - addReplyString(c, s, len); - addReplyString(c, "\r\n", 2); + addReplyString(c,"-ERR ",5); + addReplyString(c,s,len); + addReplyString(c,"\r\n",2); if (c->flags & (CLIENT_MASTER|CLIENT_SLAVE)) { char* to = c->flags & CLIENT_MASTER? "master": "slave"; char* from = c->flags & CLIENT_MASTER? "slave": "master"; @@ -417,14 +409,14 @@ void addReplyErrorLength(client *c, const char *s, size_t len) { } void addReplyError(client *c, const char *err) { - addReplyErrorLength(c, err, strlen(err)); + addReplyErrorLength(c,err,strlen(err)); } void addReplyErrorFormat(client *c, const char *fmt, ...) { size_t l, j; va_list ap; - va_start(ap, fmt); - sds s = sdscatvprintf(sdsempty(), fmt, ap); + va_start(ap,fmt); + sds s = sdscatvprintf(sdsempty(),fmt,ap); va_end(ap); /* Make sure there are no newlines in the string, otherwise invalid protocol * is emitted. */ @@ -432,26 +424,26 @@ void addReplyErrorFormat(client *c, const char *fmt, ...) { for (j = 0; j < l; j++) { if (s[j] == '\r' || s[j] == '\n') s[j] = ' '; } - addReplyErrorLength(c, s, sdslen(s)); + addReplyErrorLength(c,s,sdslen(s)); sdsfree(s); } void addReplyStatusLength(client *c, const char *s, size_t len) { - addReplyString(c, "+", 1); - addReplyString(c, s, len); - addReplyString(c, "\r\n", 2); + addReplyString(c,"+",1); + addReplyString(c,s,len); + addReplyString(c,"\r\n",2); } void addReplyStatus(client *c, const char *status) { - addReplyStatusLength(c, status, strlen(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_start(ap,fmt); + sds s = sdscatvprintf(sdsempty(),fmt,ap); va_end(ap); - addReplyStatusLength(c, s, sdslen(s)); + addReplyStatusLength(c,s,sdslen(s)); sdsfree(s); } @@ -462,29 +454,29 @@ void *addDeferredMultiBulkLength(client *c) { * ready to be sent, since we are sure that before returning to the * event loop setDeferredMultiBulkLength() will be called. */ if (prepareClientToWrite(c) != C_OK) return NULL; - listAddNodeTail(c->reply, NULL); /* NULL is our placeholder. */ + listAddNodeTail(c->reply,NULL); /* NULL is our placeholder. */ return listLast(c->reply); } /* Populate the length object and try gluing it to the next chunk. */ void setDeferredMultiBulkLength(client *c, void *node, PORT_LONG length) { - listNode *ln = (listNode*) node; + listNode *ln = (listNode*)node; sds len, next; /* Abort when *node is NULL: when the client should not accept writes * we return NULL in addDeferredMultiBulkLength() */ if (node == NULL) return; - len = sdscatprintf(sdsnewlen("*", 1), "%Id\r\n", length); WIN_PORT_FIX /* %ld -> %Id */ - listNodeValue(ln) = len; + len = sdscatprintf(sdsnewlen("*",1),"%Id\r\n",length); WIN_PORT_FIX /* %ld -> %Id */ + listNodeValue(ln) = len; c->reply_bytes += sdslen(len); if (ln->next != NULL) { next = listNodeValue(ln->next); /* Only glue when the next node is non-NULL (an sds in this case) */ if (next != NULL) { - len = sdscatsds(len, next); - listDelNode(c->reply, ln->next); + len = sdscatsds(len,next); + listDelNode(c->reply,ln->next); listNodeValue(ln) = len; /* No need to update c->reply_bytes: we are just moving the same * amount of bytes from one node to another. */ @@ -501,25 +493,24 @@ void addReplyDouble(client *c, double d) { /* Libc in odd systems (Hi Solaris!) will format infinite in a * different way, so better to handle it in an explicit way. */ addReplyBulkCString(c, d > 0 ? "inf" : "-inf"); - } - else { - dlen = snprintf(dbuf, sizeof(dbuf), "%.17g", d); - slen = snprintf(sbuf, sizeof(sbuf), "$%d\r\n%s\r\n", dlen, dbuf); - addReplyString(c, sbuf, slen); + } else { + dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d); + slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf); + addReplyString(c,sbuf,slen); } } -/* Add a PORT_LONGDOUBLE as a bulk reply, but uses a human readable formatting +/* 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, PORT_LONGDOUBLE d) { - robj *o = createStringObjectFromLongDouble(d, 1); - addReplyBulk(c, o); + robj *o = createStringObjectFromLongDouble(d,1); + addReplyBulk(c,o); decrRefCount(o); } -/* Add a PORT_LONGLONG as integer reply or bulk len / multi bulk count. - * Basically this is used to output . */ +/* Add a long long as integer reply or bulk len / multi bulk count. + * Basically this is used to output . */ void addReplyLongLongWithPrefix(client *c, PORT_LONGLONG ll, char prefix) { char buf[128]; int len; @@ -528,35 +519,34 @@ void addReplyLongLongWithPrefix(client *c, PORT_LONGLONG ll, char prefix) { * 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]); + addReply(c,shared.mbulkhdr[ll]); return; - } - else if (prefix == '$' && ll < OBJ_SHARED_BULKHDR_LEN && ll >= 0) { - addReply(c, shared.bulkhdr[ll]); + } 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'; - addReplyString(c, buf, len + 3); + len = ll2string(buf+1,sizeof(buf)-1,ll); + buf[len+1] = '\r'; + buf[len+2] = '\n'; + addReplyString(c,buf,len+3); } void addReplyLongLong(client *c, PORT_LONGLONG ll) { if (ll == 0) - addReply(c, shared.czero); + addReply(c,shared.czero); else if (ll == 1) - addReply(c, shared.cone); + addReply(c,shared.cone); else - addReplyLongLongWithPrefix(c, ll, ':'); + addReplyLongLongWithPrefix(c,ll,':'); } void addReplyMultiBulkLen(client *c, PORT_LONG length) { if (length < OBJ_SHARED_BULKHDR_LEN) - addReply(c, shared.mbulkhdr[length]); + addReply(c,shared.mbulkhdr[length]); else - addReplyLongLongWithPrefix(c, length, '*'); + addReplyLongLongWithPrefix(c,length,'*'); } /* Create the length prefix of a bulk reply, example: $2234 */ @@ -565,9 +555,8 @@ void addReplyBulkLen(client *c, robj *obj) { if (sdsEncodedObject(obj)) { len = sdslen(obj->ptr); - } - else { - PORT_LONG n = (PORT_LONG) obj->ptr; + } else { + PORT_LONG n = (PORT_LONG)obj->ptr; /* Compute how many bytes will take this integer as a radix 10 string */ len = 1; @@ -575,55 +564,54 @@ void addReplyBulkLen(client *c, robj *obj) { len++; n = -n; } - while ((n = n / 10) != 0) { + while((n = n/10) != 0) { len++; } } if (len < OBJ_SHARED_BULKHDR_LEN) - addReply(c, shared.bulkhdr[len]); + addReply(c,shared.bulkhdr[len]); else - addReplyLongLongWithPrefix(c, len, '$'); + addReplyLongLongWithPrefix(c,len,'$'); } /* Add a Redis Object as a bulk reply */ void addReplyBulk(client *c, robj *obj) { - addReplyBulkLen(c, obj); - addReply(c, obj); - addReply(c, shared.crlf); + addReplyBulkLen(c,obj); + addReply(c,obj); + addReply(c,shared.crlf); } /* Add a C buffer as bulk reply */ void addReplyBulkCBuffer(client *c, const void *p, size_t len) { - addReplyLongLongWithPrefix(c, len, '$'); - addReplyString(c, p, len); - addReply(c, shared.crlf); + addReplyLongLongWithPrefix(c,len,'$'); + addReplyString(c,p,len); + addReply(c,shared.crlf); } /* Add sds to reply (takes ownership of sds and frees it) */ -void addReplyBulkSds(client *c, sds s) { - addReplyLongLongWithPrefix(c, sdslen(s), '$'); - addReplySds(c, s); - addReply(c, shared.crlf); +void addReplyBulkSds(client *c, sds s) { + addReplyLongLongWithPrefix(c,sdslen(s),'$'); + addReplySds(c,s); + addReply(c,shared.crlf); } /* Add a C nul term string as bulk reply */ void addReplyBulkCString(client *c, const char *s) { if (s == NULL) { - addReply(c, shared.nullbulk); - } - else { - addReplyBulkCBuffer(c, s, strlen(s)); + addReply(c,shared.nullbulk); + } else { + addReplyBulkCBuffer(c,s,strlen(s)); } } -/* Add a PORT_LONGLONG as a bulk reply */ +/* Add a long long as a bulk reply */ void addReplyBulkLongLong(client *c, PORT_LONGLONG ll) { char buf[64]; int len; - len = ll2string(buf, 64, ll); - addReplyBulkCBuffer(c, buf, len); + len = ll2string(buf,64,ll); + addReplyBulkCBuffer(c,buf,len); } /* Copy 'src' client output buffers into 'dst' client output buffers. @@ -633,7 +621,7 @@ void copyClientOutputBuffer(client *dst, client *src) { listRelease(dst->reply); dst->sentlen = 0; dst->reply = listDup(src->reply); - memcpy(dst->buf, src->buf, src->bufpos); + memcpy(dst->buf,src->buf,src->bufpos); dst->bufpos = src->bufpos; dst->reply_bytes = src->reply_bytes; } @@ -650,7 +638,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip) { if ((c = createClient(fd)) == NULL) { serverLog(LL_WARNING, "Error registering fd event for the new client: %s (fd=%d)", - IF_WIN32(wsa_strerror(errno), strerror(errno)), fd); + IF_WIN32(wsa_strerror(errno), strerror(errno)),fd); close(fd); /* May be already closed, just ignore errors */ return; } @@ -662,7 +650,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip) { char *err = "-ERR max number of clients reached\r\n"; /* That's a best effort error message, don't check write errors */ - if (write(c->fd, err, strlen(err)) == -1) { + if (write(c->fd,err,strlen(err)) == -1) { /* Nothing to do, Just to avoid the warning... */ } server.stat_rejected_conn++; @@ -680,7 +668,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip) { !(flags & CLIENT_UNIX_SOCKET) && ip != NULL) { - if (strcmp(ip, "127.0.0.1") && strcmp(ip, "::1")) { + if (strcmp(ip,"127.0.0.1") && strcmp(ip,"::1")) { char *err = "-DENIED Redis is running in protected mode because protected " "mode is enabled, no bind address was specified, no " @@ -702,7 +690,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip) { "4) Setup a bind address or an authentication password. " "NOTE: You only need to do one of the above things in order for " "the server to start accepting connections from the outside.\r\n"; - if (write(c->fd, err, strlen(err)) == -1) { + if (write(c->fd,err,strlen(err)) == -1) { /* Nothing to do, Just to avoid the warning... */ } server.stat_rejected_conn++; @@ -722,7 +710,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { UNUSED(mask); UNUSED(privdata); - while (max--) { + while(max--) { cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); if (cfd == ANET_ERR) { if (errno != EWOULDBLOCK) { @@ -737,8 +725,8 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { } return; } - serverLog(LL_VERBOSE, "Accepted %s:%d", cip, cport); - acceptCommonHandler(cfd, 0, cip); + serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport); + acceptCommonHandler(cfd,0,cip); } } @@ -748,7 +736,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { UNUSED(mask); UNUSED(privdata); - while (max--) { + while(max--) { cfd = anetUnixAccept(server.neterr, fd); if (cfd == ANET_ERR) { if (errno != EWOULDBLOCK) @@ -756,8 +744,8 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { "Accepting client connection: %s", server.neterr); return; } - serverLog(LL_VERBOSE, "Accepted connection to %s", server.unixsocket); - acceptCommonHandler(cfd, CLIENT_UNIX_SOCKET, NULL); + serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket); + acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL); } } @@ -775,7 +763,7 @@ static void freeClientArgv(client *c) { void disconnectSlaves(void) { while (listLength(server.slaves)) { listNode *ln = listFirst(server.slaves); - freeClient((client*) ln->value); + freeClient((client*)ln->value); } } @@ -799,26 +787,26 @@ void unlinkClient(client *c) { } /* Unregister async I/O handlers and close the socket. */ - aeDeleteFileEvent(server.el, c->fd, AE_READABLE); - aeDeleteFileEvent(server.el, c->fd, AE_WRITABLE); + aeDeleteFileEvent(server.el,c->fd,AE_READABLE); + aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); close(c->fd); c->fd = -1; } /* Remove from the list of pending writes if needed. */ if (c->flags & CLIENT_PENDING_WRITE) { - ln = listSearchKey(server.clients_pending_write, c); + ln = listSearchKey(server.clients_pending_write,c); serverAssert(ln != NULL); - listDelNode(server.clients_pending_write, ln); + listDelNode(server.clients_pending_write,ln); c->flags &= ~CLIENT_PENDING_WRITE; } /* When client was just unblocked because of a blocking operation, * remove it from the list of unblocked clients. */ if (c->flags & CLIENT_UNBLOCKED) { - ln = listSearchKey(server.unblocked_clients, c); + ln = listSearchKey(server.unblocked_clients,c); serverAssert(ln != NULL); - listDelNode(server.unblocked_clients, ln); + listDelNode(server.unblocked_clients,ln); c->flags &= ~CLIENT_UNBLOCKED; } } @@ -832,11 +820,11 @@ void freeClient(client *c) { * Note that before doing this we make sure that the client is not in * some unexpected state, by checking its flags. */ if (server.master && c->flags & CLIENT_MASTER) { - serverLog(LL_WARNING, "Connection with master lost."); - if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY | - CLIENT_CLOSE_ASAP | - CLIENT_BLOCKED | - CLIENT_UNBLOCKED))) + serverLog(LL_WARNING,"Connection with master lost."); + if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY| + CLIENT_CLOSE_ASAP| + CLIENT_BLOCKED| + CLIENT_UNBLOCKED))) { replicationCacheMaster(c); return; @@ -845,7 +833,7 @@ void freeClient(client *c) { /* Log link disconnection with slave */ if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) { - serverLog(LL_WARNING, "Connection with slave %s lost.", + serverLog(LL_WARNING,"Connection with slave %s lost.", replicationGetSlaveName(c)); } @@ -863,8 +851,8 @@ void freeClient(client *c) { listRelease(c->watched_keys); /* Unsubscribe from all the pubsub channels */ - pubsubUnsubscribeAllChannels(c, 0); - pubsubUnsubscribeAllPatterns(c, 0); + pubsubUnsubscribeAllChannels(c,0); + pubsubUnsubscribeAllPatterns(c,0); dictRelease(c->pubsub_channels); listRelease(c->pubsub_patterns); @@ -891,9 +879,9 @@ void freeClient(client *c) { if (c->replpreamble) sdsfree(c->replpreamble); } list *l = (c->flags & CLIENT_MONITOR) ? server.monitors : server.slaves; - ln = listSearchKey(l, c); + ln = listSearchKey(l,c); serverAssert(ln != NULL); - listDelNode(l, ln); + listDelNode(l,ln); /* We need to remember the time when we started to have zero * attached slaves, as after some time we'll free the replication * backlog. */ @@ -909,9 +897,9 @@ void freeClient(client *c) { /* If this client was scheduled for async freeing we need to remove it * from the queue. */ if (c->flags & CLIENT_CLOSE_ASAP) { - ln = listSearchKey(server.clients_to_close, c); + ln = listSearchKey(server.clients_to_close,c); serverAssert(ln != NULL); - listDelNode(server.clients_to_close, ln); + listDelNode(server.clients_to_close,ln); } /* Release other dynamically allocated client structure fields, @@ -930,7 +918,7 @@ void freeClient(client *c) { void freeClientAsync(client *c) { if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return; c->flags |= CLIENT_CLOSE_ASAP; - listAddNodeTail(server.clients_to_close, c); + listAddNodeTail(server.clients_to_close,c); } void freeClientsInAsyncFreeQueue(void) { @@ -940,18 +928,18 @@ void freeClientsInAsyncFreeQueue(void) { c->flags &= ~CLIENT_CLOSE_ASAP; freeClient(c); - listDelNode(server.clients_to_close, ln); + listDelNode(server.clients_to_close,ln); } } /* Write data in output buffers to client. Return C_OK if the client -* is still valid after the call, C_ERR if it was freed. */ + * is still valid after the call, C_ERR if it was freed. */ int writeToClient(int fd, client *c, int handler_installed) { ssize_t nwritten = 0, totwritten = 0; size_t objlen; sds o; - while (clientHasPendingReplies(c)) { + while(clientHasPendingReplies(c)) { if (c->bufpos > 0) { #ifdef _WIN32 nwritten = c->bufpos - c->sentlen; @@ -971,25 +959,24 @@ int writeToClient(int fd, client *c, int handler_installed) { break; } #else - nwritten = write(fd, c->buf + c->sentlen, c->bufpos - c->sentlen); + nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); if (nwritten <= 0) break; #endif c->sentlen += nwritten; totwritten += nwritten; /* If the buffer was sent, set bufpos to zero to continue with - * the remainder of the reply. */ - if ((int) c->sentlen == c->bufpos) { + * the remainder of the reply. */ + if ((int)c->sentlen == c->bufpos) { c->bufpos = 0; c->sentlen = 0; } - } - else { + } else { o = listNodeValue(listFirst(c->reply)); objlen = sdslen(o); if (objlen == 0) { - listDelNode(c->reply, listFirst(c->reply)); + listDelNode(c->reply,listFirst(c->reply)); continue; } @@ -1019,7 +1006,7 @@ int writeToClient(int fd, client *c, int handler_installed) { /* If we fully sent the object on head go to the next one */ if (c->sentlen == objlen) { - listDelNode(c->reply, listFirst(c->reply)); + listDelNode(c->reply,listFirst(c->reply)); c->sentlen = 0; #ifdef _WIN32 c->reply_bytes -= (PORT_ULONG) objlen; @@ -1027,7 +1014,7 @@ int writeToClient(int fd, client *c, int handler_installed) { c->reply_bytes -= objlen; #endif /* If there are no longer objects in the list, we expect - * the count of reply bytes to be exactly zero. */ + * the count of reply bytes to be exactly zero. */ if (listLength(c->reply) == 0) serverAssert(c->reply_bytes == 0); } @@ -1053,8 +1040,7 @@ int writeToClient(int fd, client *c, int handler_installed) { if (nwritten == -1) { if (errno == EAGAIN) { nwritten = 0; - } - else { + } else { #ifdef _WIN32 serverLog(LL_VERBOSE, "Error writing to client: %s", wsa_strerror(errno)); @@ -1066,17 +1052,16 @@ int writeToClient(int fd, client *c, int handler_installed) { return C_ERR; } } - if (totwritten > 0) { /* For clients representing masters we don't count sending data - * as an interaction, since we always send REPLCONF ACK commands - * that take some time to just fill the socket output buffer. - * We just rely on data / pings received for timeout detection. */ + * as an interaction, since we always send REPLCONF ACK commands + * that take some time to just fill the socket output buffer. + * We just rely on data / pings received for timeout detection. */ if (!(c->flags & CLIENT_MASTER)) c->lastinteraction = server.unixtime; } if (!clientHasPendingReplies(c)) { c->sentlen = 0; - if (handler_installed) aeDeleteFileEvent(server.el, c->fd, AE_WRITABLE); + if (handler_installed) aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); /* Close connection after entire reply has been sent. */ if (c->flags & CLIENT_CLOSE_AFTER_REPLY) { @@ -1110,7 +1095,7 @@ int writeToClient(int fd, client *c, int handler_installed) { void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { UNUSED(el); UNUSED(mask); - writeToClient(fd, privdata, 1); + writeToClient(fd,privdata,1); } /* This function is called just before entering the event loop, in the hope @@ -1122,14 +1107,14 @@ int handleClientsWithPendingWrites(void) { listNode *ln; int processed = listLength(server.clients_pending_write); - listRewind(server.clients_pending_write, &li); - while ((ln = listNext(&li))) { + listRewind(server.clients_pending_write,&li); + while((ln = listNext(&li))) { client *c = listNodeValue(ln); c->flags &= ~CLIENT_PENDING_WRITE; - listDelNode(server.clients_pending_write, ln); + listDelNode(server.clients_pending_write,ln); /* Try to write buffers to the client socket. */ - if (writeToClient(c->fd, c, 0) == C_ERR) continue; + if (writeToClient(c->fd,c,0) == C_ERR) continue; /* If after the synchronous writes above we still have data to * output to the client, we need to install the writable handler. */ @@ -1147,11 +1132,11 @@ int handleClientsWithPendingWrites(void) { } if (aeCreateFileEvent(server.el, c->fd, ae_flags, sendReplyToClient, c) == AE_ERR) - { - freeClientAsync(c); + { + freeClientAsync(c); + } } } - } return processed; } @@ -1193,29 +1178,29 @@ int processInlineBuffer(client *c) { size_t querylen; /* Search for end of line */ - newline = strchr(c->querybuf, '\n'); + newline = strchr(c->querybuf,'\n'); /* Nothing to do without a \r\n */ if (newline == NULL) { if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { - addReplyError(c, "Protocol error: too big inline request"); - setProtocolError("too big inline request", c, (PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ + addReplyError(c,"Protocol error: too big inline request"); + setProtocolError("too big inline request",c,(PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ } return C_ERR; } /* Handle the \r\n case. */ - if (newline && newline != c->querybuf && *(newline - 1) == '\r') + if (newline && newline != c->querybuf && *(newline-1) == '\r') newline--, linefeed_chars++; /* Split the input buffer up to the \r\n */ - querylen = newline - (c->querybuf); - aux = sdsnewlen(c->querybuf, querylen); - argv = sdssplitargs(aux, &argc); + querylen = newline-(c->querybuf); + aux = sdsnewlen(c->querybuf,querylen); + argv = sdssplitargs(aux,&argc); sdsfree(aux); if (argv == NULL) { - addReplyError(c, "Protocol error: unbalanced quotes in request"); - setProtocolError("unbalanced quotes in inline request", c, (PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ + addReplyError(c,"Protocol error: unbalanced quotes in request"); + setProtocolError("unbalanced quotes in inline request",c,(PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ return C_ERR; } @@ -1229,18 +1214,17 @@ int processInlineBuffer(client *c) { sdsrange(c->querybuf,querylen+linefeed_chars,-1); /* Setup argv array on client structure */ - if (argc) { - if (c->argv) zfree(c->argv); - c->argv = zmalloc(sizeof(robj*)*argc); - } + if (argc) { + if (c->argv) zfree(c->argv); + c->argv = zmalloc(sizeof(robj*)*argc); + } /* Create redis objects for all arguments. */ for (c->argc = 0, j = 0; j < argc; j++) { if (sdslen(argv[j])) { - c->argv[c->argc] = createObject(OBJ_STRING, argv[j]); + c->argv[c->argc] = createObject(OBJ_STRING,argv[j]); c->argc++; - } - else { + } else { sdsfree(argv[j]); } } @@ -1253,15 +1237,14 @@ int processInlineBuffer(client *c) { #define PROTO_DUMP_LEN 128 static void setProtocolError(const char *errstr, client *c, PORT_LONG pos) { if (server.verbosity <= LL_VERBOSE) { - sds client = catClientInfoString(sdsempty(), c); + sds client = catClientInfoString(sdsempty(),c); /* Sample some protocol to given an idea about what was inside. */ char buf[256]; if (sdslen(c->querybuf) < PROTO_DUMP_LEN) { - snprintf(buf, sizeof(buf), "Query buffer during protocol error: '%s'", c->querybuf); - } - else { - snprintf(buf, sizeof(buf), "Query buffer during protocol error: '%.*s' (... more %Iu bytes ...) '%.*s'", PROTO_DUMP_LEN / 2, c->querybuf, sdslen(c->querybuf) - PROTO_DUMP_LEN, PROTO_DUMP_LEN / 2, c->querybuf + sdslen(c->querybuf) - PROTO_DUMP_LEN / 2); WIN_PORT_FIX /* %zu -> %Iu */ + snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%s'", c->querybuf); + } else { + snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%.*s' (... more %Iu bytes ...) '%.*s'", PROTO_DUMP_LEN/2, c->querybuf, sdslen(c->querybuf)-PROTO_DUMP_LEN, PROTO_DUMP_LEN/2, c->querybuf+sdslen(c->querybuf)-PROTO_DUMP_LEN/2); WIN_PORT_FIX /* %zu -> %Iu */ } /* Remove non printable chars. */ @@ -1277,7 +1260,7 @@ static void setProtocolError(const char *errstr, client *c, PORT_LONG pos) { sdsfree(client); } c->flags |= CLIENT_CLOSE_AFTER_REPLY; - sdsrange(c->querybuf, pos, -1); + sdsrange(c->querybuf,pos,-1); } /* Process the query buffer for client 'c', setting up the client argument @@ -1299,122 +1282,120 @@ int processMultibulkBuffer(client *c) { if (c->multibulklen == 0) { /* The client should have been reset */ - serverAssertWithInfo(c, NULL, c->argc == 0); + serverAssertWithInfo(c,NULL,c->argc == 0); /* Multi bulk length cannot be read without a \r\n */ - newline = strchr(c->querybuf, '\r'); + newline = strchr(c->querybuf,'\r'); if (newline == NULL) { if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { - addReplyError(c, "Protocol error: too big mbulk count string"); - setProtocolError("too big mbulk count string", c, (PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ + addReplyError(c,"Protocol error: too big mbulk count string"); + setProtocolError("too big mbulk count string",c,(PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ } return C_ERR; } /* Buffer should also contain \n */ - if (newline - (c->querybuf) > ((signed) sdslen(c->querybuf) - 2)) + if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) return C_ERR; /* We know for sure there is a whole line since newline != NULL, * so go ahead and find out the multi bulk length. */ - serverAssertWithInfo(c, NULL, c->querybuf[0] == '*'); - ok = string2ll(c->querybuf + 1, newline - (c->querybuf + 1), &ll); - if (!ok || ll > 1024 * 1024) { - addReplyError(c, "Protocol error: invalid multibulk length"); - setProtocolError("invalid mbulk count", c, pos); + serverAssertWithInfo(c,NULL,c->querybuf[0] == '*'); + ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); + if (!ok || ll > 1024*1024) { + addReplyError(c,"Protocol error: invalid multibulk length"); + setProtocolError("invalid mbulk count",c,pos); return C_ERR; } - pos = (PORT_LONG) ((newline - c->querybuf) + 2); WIN_PORT_FIX /* cast (int) */ - if (ll <= 0) { - sdsrange(c->querybuf, pos, -1); - return C_OK; - } + pos = (PORT_LONG)((newline-c->querybuf)+2); WIN_PORT_FIX /* cast (int) */ + if (ll <= 0) { + sdsrange(c->querybuf,pos,-1); + return C_OK; + } - c->multibulklen = (int) ll; WIN_PORT_FIX /* cast (int) */ + c->multibulklen = (int)ll; WIN_PORT_FIX /* cast (int) */ /* Setup argv array on client structure */ - if (c->argv) zfree(c->argv); + if (c->argv) zfree(c->argv); c->argv = zmalloc(sizeof(robj*)*c->multibulklen); } - serverAssertWithInfo(c, NULL, c->multibulklen > 0); - while (c->multibulklen) { + serverAssertWithInfo(c,NULL,c->multibulklen > 0); + while(c->multibulklen) { /* Read bulk length if unknown */ if (c->bulklen == -1) { - newline = strchr(c->querybuf + pos, '\r'); + newline = strchr(c->querybuf+pos,'\r'); if (newline == NULL) { if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { addReplyError(c, "Protocol error: too big bulk count string"); - setProtocolError("too big bulk count string", c, (PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ + setProtocolError("too big bulk count string",c,(PORT_LONG)0); WIN_PORT_FIX /* cast (PORT_LONG) */ return C_ERR; } break; } /* Buffer should also contain \n */ - if (newline - (c->querybuf) > ((signed) sdslen(c->querybuf) - 2)) + if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) break; if (c->querybuf[pos] != '$') { addReplyErrorFormat(c, "Protocol error: expected '$', got '%c'", c->querybuf[pos]); - setProtocolError("expected $ but got something else", c, pos); + setProtocolError("expected $ but got something else",c,pos); return C_ERR; } - ok = string2ll(c->querybuf + pos + 1, newline - (c->querybuf + pos + 1), &ll); + ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); if (!ok || ll < 0 || ll > server.proto_max_bulk_len) { - addReplyError(c, "Protocol error: invalid bulk length"); - setProtocolError("invalid bulk length", c, pos); + addReplyError(c,"Protocol error: invalid bulk length"); + setProtocolError("invalid bulk length",c,pos); return C_ERR; } - pos += (PORT_LONG) (newline - (c->querybuf + pos) + 2); WIN_PORT_FIX /* cast (int) */ - if (ll >= PROTO_MBULK_BIG_ARG) { - size_t qblen; + pos += (PORT_LONG) (newline-(c->querybuf+pos)+2); WIN_PORT_FIX /* cast (int) */ + if (ll >= PROTO_MBULK_BIG_ARG) { + size_t qblen; - /* If we are going to read a large object from network - * try to make it likely that it will start at c->querybuf - * boundary so that we can optimize object creation - * avoiding a large copy of data. */ - sdsrange(c->querybuf, pos, -1); - pos = 0; - qblen = sdslen(c->querybuf); - /* Hint the sds library about the amount of bytes this string is - * going to contain. */ - if (qblen < (size_t) ll + 2) - c->querybuf = sdsMakeRoomFor(c->querybuf, ll + 2 - qblen); - } - c->bulklen = (PORT_LONG) ll; WIN_PORT_FIX /* cast (PORT_LONG) */ + /* If we are going to read a large object from network + * try to make it likely that it will start at c->querybuf + * boundary so that we can optimize object creation + * avoiding a large copy of data. */ + sdsrange(c->querybuf,pos,-1); + pos = 0; + qblen = sdslen(c->querybuf); + /* Hint the sds library about the amount of bytes this string is + * going to contain. */ + if (qblen < (size_t)ll+2) + c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-qblen); + } + c->bulklen = (PORT_LONG)ll; WIN_PORT_FIX /* cast (PORT_LONG) */ } /* Read bulk argument */ - if (sdslen(c->querybuf) - pos < (size_t) (c->bulklen + 2)) { + if (sdslen(c->querybuf)-pos < (size_t)(c->bulklen+2)) { /* Not enough data (+2 == trailing \r\n) */ break; - } - else { + } else { /* Optimization: if the buffer contains JUST our bulk element * instead of creating a new object by *copying* the sds we * just use the current sds string. */ if (pos == 0 && c->bulklen >= PROTO_MBULK_BIG_ARG && - sdslen(c->querybuf) == (size_t)c->bulklen + 2) + sdslen(c->querybuf) == (size_t)(c->bulklen+2)) { - c->argv[c->argc++] = createObject(OBJ_STRING, c->querybuf); - sdsIncrLen(c->querybuf, -2); /* remove CRLF */ + c->argv[c->argc++] = createObject(OBJ_STRING,c->querybuf); + sdsIncrLen(c->querybuf,-2); /* remove CRLF */ /* Assume that if we saw a fat argument we'll see another one * likely... */ - c->querybuf = sdsnewlen(NULL, c->bulklen + 2); + c->querybuf = sdsnewlen(NULL,c->bulklen+2); sdsclear(c->querybuf); pos = 0; - } - else { + } else { c->argv[c->argc++] = - createStringObject(c->querybuf + pos, c->bulklen); + createStringObject(c->querybuf+pos,c->bulklen); pos += (PORT_LONG) c->bulklen + 2; WIN_PORT_FIX /* cast (int) */ } c->bulklen = -1; @@ -1423,7 +1404,7 @@ int processMultibulkBuffer(client *c) { } /* Trim to pos */ - if (pos) sdsrange(c->querybuf, pos, -1); + if (pos) sdsrange(c->querybuf,pos,-1); /* We're done when c->multibulk == 0 */ if (c->multibulklen == 0) return C_OK; @@ -1439,7 +1420,7 @@ int processMultibulkBuffer(client *c) { void processInputBuffer(client *c) { server.current_client = c; /* Keep processing while there is something in the input buffer */ - while (sdslen(c->querybuf)) { + while(sdslen(c->querybuf)) { /* Return if clients are paused. */ if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break; @@ -1451,33 +1432,29 @@ void processInputBuffer(client *c) { * this flag has been set (i.e. don't process more commands). * * The same applies for clients we want to terminate ASAP. */ - if (c->flags & (CLIENT_CLOSE_AFTER_REPLY | CLIENT_CLOSE_ASAP)) break; + if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break; /* Determine request type when unknown. */ if (!c->reqtype) { if (c->querybuf[0] == '*') { c->reqtype = PROTO_REQ_MULTIBULK; - } - else { + } else { c->reqtype = PROTO_REQ_INLINE; } } if (c->reqtype == PROTO_REQ_INLINE) { if (processInlineBuffer(c) != C_OK) break; - } - else if (c->reqtype == PROTO_REQ_MULTIBULK) { + } else if (c->reqtype == PROTO_REQ_MULTIBULK) { if (processMultibulkBuffer(c) != C_OK) break; - } - else { + } else { serverPanic("Unknown request type"); } /* Multibulk processing could see a <= 0 length. */ if (c->argc == 0) { resetClient(c); - } - else { + } else { /* Only reset the client when the command was executed. */ if (processCommand(c) == C_OK) { if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) { @@ -1520,85 +1497,81 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { { ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf); - if (remaining < readlen) readlen = remaining; + if (remaining < readlen) readlen = remaining; } qblen = sdslen(c->querybuf); if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); - nread = (int) read(fd, c->querybuf + qblen, readlen); WIN_PORT_FIX /* cast (int) */ - if (nread == -1) { - if (errno == EAGAIN) { - return; - } - else { - serverLog(LL_VERBOSE, "Reading from client: %s", IF_WIN32(wsa_strerror(errno), strerror(errno))); - freeClient(c); - return; - } - } - else if (nread == 0) { - serverLog(LL_VERBOSE, "Client closed connection"); + nread = (int)read(fd, c->querybuf+qblen, readlen); WIN_PORT_FIX /* cast (int) */ + if (nread == -1) { + if (errno == EAGAIN) { + return; + } else { + serverLog(LL_VERBOSE, "Reading from client: %s", IF_WIN32(wsa_strerror(errno), strerror(errno))); freeClient(c); return; } - else if (c->flags & CLIENT_MASTER) { - /* Append the query buffer to the pending (not applied) buffer - * of the master. We'll use this buffer later in order to have a - * copy of the string applied by the last command executed. */ - c->pending_querybuf = sdscatlen(c->pending_querybuf, - c->querybuf + qblen, nread); - } - WIN32_ONLY(WSIOCP_QueueNextRead(fd);) - sdsIncrLen(c->querybuf, nread); - c->lastinteraction = server.unixtime; - if (c->flags & CLIENT_MASTER) c->read_reploff += nread; - server.stat_net_input_bytes += nread; - if (sdslen(c->querybuf) > server.client_max_querybuf_len) { - sds ci = catClientInfoString(sdsempty(), c), bytes = sdsempty(); + } else if (nread == 0) { + serverLog(LL_VERBOSE, "Client closed connection"); + freeClient(c); + return; + } else if (c->flags & CLIENT_MASTER) { + /* Append the query buffer to the pending (not applied) buffer + * of the master. We'll use this buffer later in order to have a + * copy of the string applied by the last command executed. */ + c->pending_querybuf = sdscatlen(c->pending_querybuf, + c->querybuf+qblen,nread); + } + WIN32_ONLY(WSIOCP_QueueNextRead(fd);) + sdsIncrLen(c->querybuf,nread); + c->lastinteraction = server.unixtime; + if (c->flags & CLIENT_MASTER) c->read_reploff += nread; + server.stat_net_input_bytes += nread; + if (sdslen(c->querybuf) > server.client_max_querybuf_len) { + sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); - bytes = sdscatrepr(bytes, c->querybuf, 64); - serverLog(LL_WARNING, "Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); - sdsfree(ci); - sdsfree(bytes); - freeClient(c); - return; - } + bytes = sdscatrepr(bytes,c->querybuf,64); + serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); + sdsfree(ci); + sdsfree(bytes); + freeClient(c); + return; + } - /* Time to process the buffer. If the client is a master we need to - * compute the difference between the applied offset before and after - * processing the buffer, to understand how much of the replication stream - * was actually applied to the master state: this quantity, and its - * corresponding part of the replication stream, will be propagated to - * the sub-slaves and to the replication backlog. */ - if (!(c->flags & CLIENT_MASTER)) { - processInputBuffer(c); - } - else { - size_t prev_offset = c->reploff; - processInputBuffer(c); - size_t applied = c->reploff - prev_offset; - if (applied) { - replicationFeedSlavesFromMasterStream(server.slaves, + /* Time to process the buffer. If the client is a master we need to + * compute the difference between the applied offset before and after + * processing the buffer, to understand how much of the replication stream + * was actually applied to the master state: this quantity, and its + * corresponding part of the replication stream, will be propagated to + * the sub-slaves and to the replication backlog. */ + if (!(c->flags & CLIENT_MASTER)) { + processInputBuffer(c); + } else { + size_t prev_offset = c->reploff; + processInputBuffer(c); + size_t applied = c->reploff - prev_offset; + if (applied) { + replicationFeedSlavesFromMasterStream(server.slaves, c->pending_querybuf, applied); - sdsrange(c->pending_querybuf, applied, -1); - } + sdsrange(c->pending_querybuf,applied,-1); } + } } void getClientsMaxBuffers(PORT_ULONG *longest_output_list, - PORT_ULONG *biggest_input_buffer) { + PORT_ULONG *biggest_input_buffer) { client *c; listNode *ln; listIter li; PORT_ULONG lol = 0, bib = 0; - listRewind(server.clients, &li); + listRewind(server.clients,&li); while ((ln = listNext(&li)) != NULL) { c = listNodeValue(ln); if (listLength(c->reply) > lol) lol = listLength(c->reply); - if (sdslen(c->querybuf) > bib) bib = (PORT_ULONG) sdslen(c->querybuf); WIN_PORT_FIX /* cast (PORT_ULONG) */ + if (sdslen(c->querybuf) > bib) bib = (PORT_ULONG)sdslen(c->querybuf); WIN_PORT_FIX /* cast (PORT_ULONG) */ } *longest_output_list = lol; *biggest_input_buffer = bib; @@ -1616,14 +1589,13 @@ void getClientsMaxBuffers(PORT_ULONG *longest_output_list, * in case you want to relax error checking or need to display something * anyway (see anetPeerToString implementation for more info). */ void genClientPeerId(client *client, char *peerid, - size_t peerid_len) { + size_t peerid_len) { if (client->flags & CLIENT_UNIX_SOCKET) { /* Unix socket client. */ - snprintf(peerid, peerid_len, "%s:0", server.unixsocket); - } - else { + snprintf(peerid,peerid_len,"%s:0",server.unixsocket); + } else { /* TCP client. */ - anetFormatPeer(client->fd, peerid, peerid_len); + anetFormatPeer(client->fd,peerid,peerid_len); } } @@ -1635,7 +1607,7 @@ char *getClientPeerId(client *c) { char peerid[NET_PEER_ID_LEN]; if (c->peerid == NULL) { - genClientPeerId(c, peerid, sizeof(peerid)); + genClientPeerId(c,peerid,sizeof(peerid)); c->peerid = sdsnew(peerid); } return c->peerid; @@ -1666,7 +1638,7 @@ sds catClientInfoString(sds s, client *client) { if (p == flags) *p++ = 'N'; *p++ = '\0'; - emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el, client->fd); + emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd); p = events; if (emask & AE_READABLE) *p++ = 'r'; if (emask & AE_WRITABLE) *p++ = 'w'; @@ -1676,9 +1648,9 @@ sds catClientInfoString(sds s, client *client) { (PORT_ULONGLONG) client->id, getClientPeerId(client), client->fd, - client->name ? (char*) client->name->ptr : "", - (PORT_LONGLONG) (server.unixtime - client->ctime), - (PORT_LONGLONG) (server.unixtime - client->lastinteraction), + client->name ? (char*)client->name->ptr : "", + (PORT_LONGLONG)(server.unixtime - client->ctime), + (PORT_LONGLONG)(server.unixtime - client->lastinteraction), flags, client->db->id, (int) dictSize(client->pubsub_channels), @@ -1697,13 +1669,13 @@ sds getAllClientsInfoString(void) { listNode *ln; listIter li; client *client; - sds o = sdsnewlen(NULL, 200 * listLength(server.clients)); + sds o = sdsnewlen(NULL,200*listLength(server.clients)); sdsclear(o); - listRewind(server.clients, &li); + listRewind(server.clients,&li); while ((ln = listNext(&li)) != NULL) { client = listNodeValue(ln); - o = catClientInfoString(o, client); - o = sdscatlen(o, "\n", 1); + o = catClientInfoString(o,client); + o = sdscatlen(o,"\n",1); } return o; } @@ -1713,31 +1685,26 @@ void clientCommand(client *c) { listIter li; client *client; - if (!strcasecmp(c->argv[1]->ptr, "list") && c->argc == 2) { + if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { /* CLIENT LIST */ sds o = getAllClientsInfoString(); - addReplyBulkCBuffer(c, o, sdslen(o)); + addReplyBulkCBuffer(c,o,sdslen(o)); sdsfree(o); - } - else if (!strcasecmp(c->argv[1]->ptr, "reply") && c->argc == 3) { + } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) { /* CLIENT REPLY ON|OFF|SKIP */ - if (!strcasecmp(c->argv[2]->ptr, "on")) { - c->flags &= ~(CLIENT_REPLY_SKIP | CLIENT_REPLY_OFF); - addReply(c, shared.ok); - } - else if (!strcasecmp(c->argv[2]->ptr, "off")) { + if (!strcasecmp(c->argv[2]->ptr,"on")) { + c->flags &= ~(CLIENT_REPLY_SKIP|CLIENT_REPLY_OFF); + addReply(c,shared.ok); + } else if (!strcasecmp(c->argv[2]->ptr,"off")) { c->flags |= CLIENT_REPLY_OFF; - } - else if (!strcasecmp(c->argv[2]->ptr, "skip")) { + } else if (!strcasecmp(c->argv[2]->ptr,"skip")) { if (!(c->flags & CLIENT_REPLY_OFF)) c->flags |= CLIENT_REPLY_SKIP_NEXT; - } - else { - addReply(c, shared.syntaxerr); + } else { + addReply(c,shared.syntaxerr); return; } - } - else if (!strcasecmp(c->argv[1]->ptr, "kill")) { + } else if (!strcasecmp(c->argv[1]->ptr,"kill")) { /* CLIENT KILL * CLIENT KILL