Back to home page

Redis cross reference

 
 

    


0001 /*
0002  * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
0003  * All rights reserved.
0004  *
0005  * Redistribution and use in source and binary forms, with or without
0006  * modification, are permitted provided that the following conditions are met:
0007  *
0008  *   * Redistributions of source code must retain the above copyright notice,
0009  *     this list of conditions and the following disclaimer.
0010  *   * Redistributions in binary form must reproduce the above copyright
0011  *     notice, this list of conditions and the following disclaimer in the
0012  *     documentation and/or other materials provided with the distribution.
0013  *   * Neither the name of Redis nor the names of its contributors may be used
0014  *     to endorse or promote products derived from this software without
0015  *     specific prior written permission.
0016  *
0017  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
0018  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
0019  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
0020  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
0021  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
0022  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
0023  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
0024  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
0025  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
0026  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
0027  * POSSIBILITY OF SUCH DAMAGE.
0028  */
0029 
0030 #include "redis.h"
0031 #include <sys/uio.h>
0032 
0033 static void setProtocolError(redisClient *c, int pos);
0034 
0035 /* To evaluate the output buffer size of a client we need to get size of
0036  * allocated objects, however we can't used zmalloc_size() directly on sds
0037  * strings because of the trick they use to work (the header is before the
0038  * returned pointer), so we use this helper function. */
0039 size_t zmalloc_size_sds(sds s) {
0040     return zmalloc_size(s-sizeof(struct sdshdr));
0041 }
0042 
0043 void *dupClientReplyValue(void *o) {
0044     incrRefCount((robj*)o);
0045     return o;
0046 }
0047 
0048 int listMatchObjects(void *a, void *b) {
0049     return equalStringObjects(a,b);
0050 }
0051 
0052 redisClient *createClient(int fd) {
0053     redisClient *c = zmalloc(sizeof(redisClient));
0054 
0055     /* passing -1 as fd it is possible to create a non connected client.
0056      * This is useful since all the Redis commands needs to be executed
0057      * in the context of a client. When commands are executed in other
0058      * contexts (for instance a Lua script) we need a non connected client. */
0059     if (fd != -1) {
0060         anetNonBlock(NULL,fd);
0061         anetEnableTcpNoDelay(NULL,fd);
0062         if (server.tcpkeepalive)
0063             anetKeepAlive(NULL,fd,server.tcpkeepalive);
0064         if (aeCreateFileEvent(server.el,fd,AE_READABLE,
0065             readQueryFromClient, c) == AE_ERR)
0066         {
0067             close(fd);
0068             zfree(c);
0069             return NULL;
0070         }
0071     }
0072 
0073     selectDb(c,0);
0074     c->fd = fd;
0075     c->name = NULL;
0076     c->bufpos = 0;
0077     c->querybuf = sdsempty();
0078     c->querybuf_peak = 0;
0079     c->reqtype = 0;
0080     c->argc = 0;
0081     c->argv = NULL;
0082     c->cmd = c->lastcmd = NULL;
0083     c->multibulklen = 0;
0084     c->bulklen = -1;
0085     c->sentlen = 0;
0086     c->flags = 0;
0087     c->ctime = c->lastinteraction = server.unixtime;
0088     c->authenticated = 0;
0089     c->replstate = REDIS_REPL_NONE;
0090     c->slave_listening_port = 0;
0091     c->reply = listCreate();
0092     c->reply_bytes = 0;
0093     c->obuf_soft_limit_reached_time = 0;
0094     listSetFreeMethod(c->reply,decrRefCount);
0095     listSetDupMethod(c->reply,dupClientReplyValue);
0096     c->bpop.keys = dictCreate(&setDictType,NULL);
0097     c->bpop.timeout = 0;
0098     c->bpop.target = NULL;
0099     c->io_keys = listCreate();
0100     c->watched_keys = listCreate();
0101     listSetFreeMethod(c->io_keys,decrRefCount);
0102     c->pubsub_channels = dictCreate(&setDictType,NULL);
0103     c->pubsub_patterns = listCreate();
0104     listSetFreeMethod(c->pubsub_patterns,decrRefCount);
0105     listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
0106     if (fd != -1) listAddNodeTail(server.clients,c);
0107     initClientMultiState(c);
0108     return c;
0109 }
0110 
0111 /* This function is called every time we are going to transmit new data
0112  * to the client. The behavior is the following:
0113  *
0114  * If the client should receive new data (normal clients will) the function
0115  * returns REDIS_OK, and make sure to install the write handler in our event
0116  * loop so that when the socket is writable new data gets written.
0117  *
0118  * If the client should not receive new data, because it is a fake client
0119  * or a slave, or because the setup of the write handler failed, the function
0120  * returns REDIS_ERR.
0121  *
0122  * Typically gets called every time a reply is built, before adding more
0123  * data to the clients output buffers. If the function returns REDIS_ERR no
0124  * data should be appended to the output buffers. */
0125 int prepareClientToWrite(redisClient *c) {
0126     if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
0127     if (c->fd <= 0) return REDIS_ERR; /* Fake client */
0128     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
0129         (c->replstate == REDIS_REPL_NONE ||
0130          c->replstate == REDIS_REPL_ONLINE) &&
0131         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
0132         sendReplyToClient, c) == AE_ERR) return REDIS_ERR;
0133     return REDIS_OK;
0134 }
0135 
0136 /* Create a duplicate of the last object in the reply list when
0137  * it is not exclusively owned by the reply list. */
0138 robj *dupLastObjectIfNeeded(list *reply) {
0139     robj *new, *cur;
0140     listNode *ln;
0141     redisAssert(listLength(reply) > 0);
0142     ln = listLast(reply);
0143     cur = listNodeValue(ln);
0144     if (cur->refcount > 1) {
0145         new = dupStringObject(cur);
0146         decrRefCount(cur);
0147         listNodeValue(ln) = new;
0148     }
0149     return listNodeValue(ln);
0150 }
0151 
0152 /* -----------------------------------------------------------------------------
0153  * Low level functions to add more data to output buffers.
0154  * -------------------------------------------------------------------------- */
0155 
0156 int _addReplyToBuffer(redisClient *c, char *s, size_t len) {
0157     size_t available = sizeof(c->buf)-c->bufpos;
0158 
0159     if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK;
0160 
0161     /* If there already are entries in the reply list, we cannot
0162      * add anything more to the static buffer. */
0163     if (listLength(c->reply) > 0) return REDIS_ERR;
0164 
0165     /* Check that the buffer has enough space available for this string. */
0166     if (len > available) return REDIS_ERR;
0167 
0168     memcpy(c->buf+c->bufpos,s,len);
0169     c->bufpos+=len;
0170     return REDIS_OK;
0171 }
0172 
0173 void _addReplyObjectToList(redisClient *c, robj *o) {
0174     robj *tail;
0175 
0176     if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
0177 
0178     if (listLength(c->reply) == 0) {
0179         incrRefCount(o);
0180         listAddNodeTail(c->reply,o);
0181         c->reply_bytes += zmalloc_size_sds(o->ptr);
0182     } else {
0183         tail = listNodeValue(listLast(c->reply));
0184 
0185         /* Append to this object when possible. */
0186         if (tail->ptr != NULL &&
0187             sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES)
0188         {
0189             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
0190             tail = dupLastObjectIfNeeded(c->reply);
0191             tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr));
0192             c->reply_bytes += zmalloc_size_sds(tail->ptr);
0193         } else {
0194             incrRefCount(o);
0195             listAddNodeTail(c->reply,o);
0196             c->reply_bytes += zmalloc_size_sds(o->ptr);
0197         }
0198     }
0199     asyncCloseClientOnOutputBufferLimitReached(c);
0200 }
0201 
0202 /* This method takes responsibility over the sds. When it is no longer
0203  * needed it will be free'd, otherwise it ends up in a robj. */
0204 void _addReplySdsToList(redisClient *c, sds s) {
0205     robj *tail;
0206 
0207     if (c->flags & REDIS_CLOSE_AFTER_REPLY) {
0208         sdsfree(s);
0209         return;
0210     }
0211 
0212     if (listLength(c->reply) == 0) {
0213         listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
0214         c->reply_bytes += zmalloc_size_sds(s);
0215     } else {
0216         tail = listNodeValue(listLast(c->reply));
0217 
0218         /* Append to this object when possible. */
0219         if (tail->ptr != NULL &&
0220             sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES)
0221         {
0222             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
0223             tail = dupLastObjectIfNeeded(c->reply);
0224             tail->ptr = sdscatlen(tail->ptr,s,sdslen(s));
0225             c->reply_bytes += zmalloc_size_sds(tail->ptr);
0226             sdsfree(s);
0227         } else {
0228             listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
0229             c->reply_bytes += zmalloc_size_sds(s);
0230         }
0231     }
0232     asyncCloseClientOnOutputBufferLimitReached(c);
0233 }
0234 
0235 void _addReplyStringToList(redisClient *c, char *s, size_t len) {
0236     robj *tail;
0237 
0238     if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
0239 
0240     if (listLength(c->reply) == 0) {
0241         robj *o = createStringObject(s,len);
0242 
0243         listAddNodeTail(c->reply,o);
0244         c->reply_bytes += zmalloc_size_sds(o->ptr);
0245     } else {
0246         tail = listNodeValue(listLast(c->reply));
0247 
0248         /* Append to this object when possible. */
0249         if (tail->ptr != NULL &&
0250             sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
0251         {
0252             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
0253             tail = dupLastObjectIfNeeded(c->reply);
0254             tail->ptr = sdscatlen(tail->ptr,s,len);
0255             c->reply_bytes += zmalloc_size_sds(tail->ptr);
0256         } else {
0257             robj *o = createStringObject(s,len);
0258 
0259             listAddNodeTail(c->reply,o);
0260             c->reply_bytes += zmalloc_size_sds(o->ptr);
0261         }
0262     }
0263     asyncCloseClientOnOutputBufferLimitReached(c);
0264 }
0265 
0266 /* -----------------------------------------------------------------------------
0267  * Higher level functions to queue data on the client output buffer.
0268  * The following functions are the ones that commands implementations will call.
0269  * -------------------------------------------------------------------------- */
0270 
0271 void addReply(redisClient *c, robj *obj) {
0272     if (prepareClientToWrite(c) != REDIS_OK) return;
0273 
0274     /* This is an important place where we can avoid copy-on-write
0275      * when there is a saving child running, avoiding touching the
0276      * refcount field of the object if it's not needed.
0277      *
0278      * If the encoding is RAW and there is room in the static buffer
0279      * we'll be able to send the object to the client without
0280      * messing with its page. */
0281     if (obj->encoding == REDIS_ENCODING_RAW) {
0282         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
0283             _addReplyObjectToList(c,obj);
0284     } else if (obj->encoding == REDIS_ENCODING_INT) {
0285         /* Optimization: if there is room in the static buffer for 32 bytes
0286          * (more than the max chars a 64 bit integer can take as string) we
0287          * avoid decoding the object and go for the lower level approach. */
0288         if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) {
0289             char buf[32];
0290             int len;
0291 
0292             len = ll2string(buf,sizeof(buf),(long)obj->ptr);
0293             if (_addReplyToBuffer(c,buf,len) == REDIS_OK)
0294                 return;
0295             /* else... continue with the normal code path, but should never
0296              * happen actually since we verified there is room. */
0297         }
0298         obj = getDecodedObject(obj);
0299         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
0300             _addReplyObjectToList(c,obj);
0301         decrRefCount(obj);
0302     } else {
0303         redisPanic("Wrong obj->encoding in addReply()");
0304     }
0305 }
0306 
0307 void addReplySds(redisClient *c, sds s) {
0308     if (prepareClientToWrite(c) != REDIS_OK) {
0309         /* The caller expects the sds to be free'd. */
0310         sdsfree(s);
0311         return;
0312     }
0313     if (_addReplyToBuffer(c,s,sdslen(s)) == REDIS_OK) {
0314         sdsfree(s);
0315     } else {
0316         /* This method free's the sds when it is no longer needed. */
0317         _addReplySdsToList(c,s);
0318     }
0319 }
0320 
0321 void addReplyString(redisClient *c, char *s, size_t len) {
0322     if (prepareClientToWrite(c) != REDIS_OK) return;
0323     if (_addReplyToBuffer(c,s,len) != REDIS_OK)
0324         _addReplyStringToList(c,s,len);
0325 }
0326 
0327 void addReplyErrorLength(redisClient *c, char *s, size_t len) {
0328     addReplyString(c,"-ERR ",5);
0329     addReplyString(c,s,len);
0330     addReplyString(c,"\r\n",2);
0331 }
0332 
0333 void addReplyError(redisClient *c, char *err) {
0334     addReplyErrorLength(c,err,strlen(err));
0335 }
0336 
0337 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
0338     size_t l, j;
0339     va_list ap;
0340     va_start(ap,fmt);
0341     sds s = sdscatvprintf(sdsempty(),fmt,ap);
0342     va_end(ap);
0343     /* Make sure there are no newlines in the string, otherwise invalid protocol
0344      * is emitted. */
0345     l = sdslen(s);
0346     for (j = 0; j < l; j++) {
0347         if (s[j] == '\r' || s[j] == '\n') s[j] = ' ';
0348     }
0349     addReplyErrorLength(c,s,sdslen(s));
0350     sdsfree(s);
0351 }
0352 
0353 void addReplyStatusLength(redisClient *c, char *s, size_t len) {
0354     addReplyString(c,"+",1);
0355     addReplyString(c,s,len);
0356     addReplyString(c,"\r\n",2);
0357 }
0358 
0359 void addReplyStatus(redisClient *c, char *status) {
0360     addReplyStatusLength(c,status,strlen(status));
0361 }
0362 
0363 void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
0364     va_list ap;
0365     va_start(ap,fmt);
0366     sds s = sdscatvprintf(sdsempty(),fmt,ap);
0367     va_end(ap);
0368     addReplyStatusLength(c,s,sdslen(s));
0369     sdsfree(s);
0370 }
0371 
0372 /* Adds an empty object to the reply list that will contain the multi bulk
0373  * length, which is not known when this function is called. */
0374 void *addDeferredMultiBulkLength(redisClient *c) {
0375     /* Note that we install the write event here even if the object is not
0376      * ready to be sent, since we are sure that before returning to the
0377      * event loop setDeferredMultiBulkLength() will be called. */
0378     if (prepareClientToWrite(c) != REDIS_OK) return NULL;
0379     listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL));
0380     return listLast(c->reply);
0381 }
0382 
0383 /* Populate the length object and try gluing it to the next chunk. */
0384 void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
0385     listNode *ln = (listNode*)node;
0386     robj *len, *next;
0387 
0388     /* Abort when *node is NULL (see addDeferredMultiBulkLength). */
0389     if (node == NULL) return;
0390 
0391     len = listNodeValue(ln);
0392     len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
0393     c->reply_bytes += zmalloc_size_sds(len->ptr);
0394     if (ln->next != NULL) {
0395         next = listNodeValue(ln->next);
0396 
0397         /* Only glue when the next node is non-NULL (an sds in this case) */
0398         if (next->ptr != NULL) {
0399             c->reply_bytes -= zmalloc_size_sds(len->ptr);
0400             c->reply_bytes -= zmalloc_size_sds(next->ptr);
0401             len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));
0402             c->reply_bytes += zmalloc_size_sds(len->ptr);
0403             listDelNode(c->reply,ln->next);
0404         }
0405     }
0406     asyncCloseClientOnOutputBufferLimitReached(c);
0407 }
0408 
0409 /* Add a double as a bulk reply */
0410 void addReplyDouble(redisClient *c, double d) {
0411     char dbuf[128], sbuf[128];
0412     int dlen, slen;
0413     dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
0414     slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
0415     addReplyString(c,sbuf,slen);
0416 }
0417 
0418 /* Add a long long as integer reply or bulk len / multi bulk count.
0419  * Basically this is used to output <prefix><long long><crlf>. */
0420 void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
0421     char buf[128];
0422     int len;
0423 
0424     /* Things like $3\r\n or *2\r\n are emitted very often by the protocol
0425      * so we have a few shared objects to use if the integer is small
0426      * like it is most of the times. */
0427     if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) {
0428         addReply(c,shared.mbulkhdr[ll]);
0429         return;
0430     } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) {
0431         addReply(c,shared.bulkhdr[ll]);
0432         return;
0433     }
0434 
0435     buf[0] = prefix;
0436     len = ll2string(buf+1,sizeof(buf)-1,ll);
0437     buf[len+1] = '\r';
0438     buf[len+2] = '\n';
0439     addReplyString(c,buf,len+3);
0440 }
0441 
0442 void addReplyLongLong(redisClient *c, long long ll) {
0443     if (ll == 0)
0444         addReply(c,shared.czero);
0445     else if (ll == 1)
0446         addReply(c,shared.cone);
0447     else
0448         addReplyLongLongWithPrefix(c,ll,':');
0449 }
0450 
0451 void addReplyMultiBulkLen(redisClient *c, long length) {
0452     addReplyLongLongWithPrefix(c,length,'*');
0453 }
0454 
0455 /* Create the length prefix of a bulk reply, example: $2234 */
0456 void addReplyBulkLen(redisClient *c, robj *obj) {
0457     size_t len;
0458 
0459     if (obj->encoding == REDIS_ENCODING_RAW) {
0460         len = sdslen(obj->ptr);
0461     } else {
0462         long n = (long)obj->ptr;
0463 
0464         /* Compute how many bytes will take this integer as a radix 10 string */
0465         len = 1;
0466         if (n < 0) {
0467             len++;
0468             n = -n;
0469         }
0470         while((n = n/10) != 0) {
0471             len++;
0472         }
0473     }
0474     addReplyLongLongWithPrefix(c,len,'$');
0475 }
0476 
0477 /* Add a Redis Object as a bulk reply */
0478 void addReplyBulk(redisClient *c, robj *obj) {
0479     addReplyBulkLen(c,obj);
0480     addReply(c,obj);
0481     addReply(c,shared.crlf);
0482 }
0483 
0484 /* Add a C buffer as bulk reply */
0485 void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) {
0486     addReplyLongLongWithPrefix(c,len,'$');
0487     addReplyString(c,p,len);
0488     addReply(c,shared.crlf);
0489 }
0490 
0491 /* Add a C nul term string as bulk reply */
0492 void addReplyBulkCString(redisClient *c, char *s) {
0493     if (s == NULL) {
0494         addReply(c,shared.nullbulk);
0495     } else {
0496         addReplyBulkCBuffer(c,s,strlen(s));
0497     }
0498 }
0499 
0500 /* Add a long long as a bulk reply */
0501 void addReplyBulkLongLong(redisClient *c, long long ll) {
0502     char buf[64];
0503     int len;
0504 
0505     len = ll2string(buf,64,ll);
0506     addReplyBulkCBuffer(c,buf,len);
0507 }
0508 
0509 /* Copy 'src' client output buffers into 'dst' client output buffers.
0510  * The function takes care of freeing the old output buffers of the
0511  * destination client. */
0512 void copyClientOutputBuffer(redisClient *dst, redisClient *src) {
0513     listRelease(dst->reply);
0514     dst->reply = listDup(src->reply);
0515     memcpy(dst->buf,src->buf,src->bufpos);
0516     dst->bufpos = src->bufpos;
0517     dst->reply_bytes = src->reply_bytes;
0518 }
0519 
0520 static void acceptCommonHandler(int fd, int flags) {
0521     redisClient *c;
0522     if ((c = createClient(fd)) == NULL) {
0523         redisLog(REDIS_WARNING,
0524             "Error registering fd event for the new client: %s (fd=%d)",
0525             strerror(errno),fd);
0526         close(fd); /* May be already closed, just ignore errors */
0527         return;
0528     }
0529     /* If maxclient directive is set and this is one client more... close the
0530      * connection. Note that we create the client instead to check before
0531      * for this condition, since now the socket is already set in non-blocking
0532      * mode and we can send an error for free using the Kernel I/O */
0533     if (listLength(server.clients) > server.maxclients) {
0534         char *err = "-ERR max number of clients reached\r\n";
0535 
0536         /* That's a best effort error message, don't check write errors */
0537         if (write(c->fd,err,strlen(err)) == -1) {
0538             /* Nothing to do, Just to avoid the warning... */
0539         }
0540         server.stat_rejected_conn++;
0541         freeClient(c);
0542         return;
0543     }
0544     server.stat_numconnections++;
0545     c->flags |= flags;
0546 }
0547 
0548 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
0549     int cport, cfd;
0550     char cip[128];
0551     REDIS_NOTUSED(el);
0552     REDIS_NOTUSED(mask);
0553     REDIS_NOTUSED(privdata);
0554 
0555     cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
0556     if (cfd == AE_ERR) {
0557         redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
0558         return;
0559     }
0560     redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
0561     acceptCommonHandler(cfd,0);
0562 }
0563 
0564 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
0565     int cfd;
0566     REDIS_NOTUSED(el);
0567     REDIS_NOTUSED(mask);
0568     REDIS_NOTUSED(privdata);
0569 
0570     cfd = anetUnixAccept(server.neterr, fd);
0571     if (cfd == AE_ERR) {
0572         redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
0573         return;
0574     }
0575     redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
0576     acceptCommonHandler(cfd,REDIS_UNIX_SOCKET);
0577 }
0578 
0579 
0580 static void freeClientArgv(redisClient *c) {
0581     int j;
0582     for (j = 0; j < c->argc; j++)
0583         decrRefCount(c->argv[j]);
0584     c->argc = 0;
0585     c->cmd = NULL;
0586 }
0587 
0588 /* Close all the slaves connections. This is useful in chained replication
0589  * when we resync with our own master and want to force all our slaves to
0590  * resync with us as well. */
0591 void disconnectSlaves(void) {
0592     while (listLength(server.slaves)) {
0593         listNode *ln = listFirst(server.slaves);
0594         freeClient((redisClient*)ln->value);
0595     }
0596 }
0597 
0598 void freeClient(redisClient *c) {
0599     listNode *ln;
0600 
0601     /* If this is marked as current client unset it */
0602     if (server.current_client == c) server.current_client = NULL;
0603 
0604     /* Note that if the client we are freeing is blocked into a blocking
0605      * call, we have to set querybuf to NULL *before* to call
0606      * unblockClientWaitingData() to avoid processInputBuffer() will get
0607      * called. Also it is important to remove the file events after
0608      * this, because this call adds the READABLE event. */
0609     sdsfree(c->querybuf);
0610     c->querybuf = NULL;
0611     if (c->flags & REDIS_BLOCKED)
0612         unblockClientWaitingData(c);
0613     dictRelease(c->bpop.keys);
0614 
0615     /* UNWATCH all the keys */
0616     unwatchAllKeys(c);
0617     listRelease(c->watched_keys);
0618     /* Unsubscribe from all the pubsub channels */
0619     pubsubUnsubscribeAllChannels(c,0);
0620     pubsubUnsubscribeAllPatterns(c,0);
0621     dictRelease(c->pubsub_channels);
0622     listRelease(c->pubsub_patterns);
0623     /* Obvious cleanup */
0624     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
0625     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
0626     listRelease(c->reply);
0627     freeClientArgv(c);
0628     close(c->fd);
0629     /* Remove from the list of clients */
0630     ln = listSearchKey(server.clients,c);
0631     redisAssert(ln != NULL);
0632     listDelNode(server.clients,ln);
0633     /* When client was just unblocked because of a blocking operation,
0634      * remove it from the list with unblocked clients. */
0635     if (c->flags & REDIS_UNBLOCKED) {
0636         ln = listSearchKey(server.unblocked_clients,c);
0637         redisAssert(ln != NULL);
0638         listDelNode(server.unblocked_clients,ln);
0639     }
0640     listRelease(c->io_keys);
0641     /* Master/slave cleanup.
0642      * Case 1: we lost the connection with a slave. */
0643     if (c->flags & REDIS_SLAVE) {
0644         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
0645             close(c->repldbfd);
0646         list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
0647         ln = listSearchKey(l,c);
0648         redisAssert(ln != NULL);
0649         listDelNode(l,ln);
0650     }
0651 
0652     /* Case 2: we lost the connection with the master. */
0653     if (c->flags & REDIS_MASTER) {
0654         server.master = NULL;
0655         server.repl_state = REDIS_REPL_CONNECT;
0656         server.repl_down_since = server.unixtime;
0657         /* We lost connection with our master, force our slaves to resync
0658          * with us as well to load the new data set.
0659          *
0660          * If server.masterhost is NULL the user called SLAVEOF NO ONE so
0661          * slave resync is not needed. */
0662         if (server.masterhost != NULL) disconnectSlaves();
0663     }
0664 
0665     /* If this client was scheduled for async freeing we need to remove it
0666      * from the queue. */
0667     if (c->flags & REDIS_CLOSE_ASAP) {
0668         ln = listSearchKey(server.clients_to_close,c);
0669         redisAssert(ln != NULL);
0670         listDelNode(server.clients_to_close,ln);
0671     }
0672 
0673     /* Release memory */
0674     if (c->name) decrRefCount(c->name);
0675     zfree(c->argv);
0676     freeClientMultiState(c);
0677     zfree(c);
0678 }
0679 
0680 /* Schedule a client to free it at a safe time in the serverCron() function.
0681  * This function is useful when we need to terminate a client but we are in
0682  * a context where calling freeClient() is not possible, because the client
0683  * should be valid for the continuation of the flow of the program. */
0684 void freeClientAsync(redisClient *c) {
0685     if (c->flags & REDIS_CLOSE_ASAP) return;
0686     c->flags |= REDIS_CLOSE_ASAP;
0687     listAddNodeTail(server.clients_to_close,c);
0688 }
0689 
0690 void freeClientsInAsyncFreeQueue(void) {
0691     while (listLength(server.clients_to_close)) {
0692         listNode *ln = listFirst(server.clients_to_close);
0693         redisClient *c = listNodeValue(ln);
0694 
0695         c->flags &= ~REDIS_CLOSE_ASAP;
0696         freeClient(c);
0697         listDelNode(server.clients_to_close,ln);
0698     }
0699 }
0700 
0701 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
0702     redisClient *c = privdata;
0703     int nwritten = 0, totwritten = 0, objlen;
0704     size_t objmem;
0705     robj *o;
0706     REDIS_NOTUSED(el);
0707     REDIS_NOTUSED(mask);
0708 
0709     while(c->bufpos > 0 || listLength(c->reply)) {
0710         if (c->bufpos > 0) {
0711             if (c->flags & REDIS_MASTER) {
0712                 /* Don't reply to a master */
0713                 nwritten = c->bufpos - c->sentlen;
0714             } else {
0715                 nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
0716                 if (nwritten <= 0) break;
0717             }
0718             c->sentlen += nwritten;
0719             totwritten += nwritten;
0720 
0721             /* If the buffer was sent, set bufpos to zero to continue with
0722              * the remainder of the reply. */
0723             if (c->sentlen == c->bufpos) {
0724                 c->bufpos = 0;
0725                 c->sentlen = 0;
0726             }
0727         } else {
0728             o = listNodeValue(listFirst(c->reply));
0729             objlen = sdslen(o->ptr);
0730             objmem = zmalloc_size_sds(o->ptr);
0731 
0732             if (objlen == 0) {
0733                 listDelNode(c->reply,listFirst(c->reply));
0734                 continue;
0735             }
0736 
0737             if (c->flags & REDIS_MASTER) {
0738                 /* Don't reply to a master */
0739                 nwritten = objlen - c->sentlen;
0740             } else {
0741                 nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
0742                 if (nwritten <= 0) break;
0743             }
0744             c->sentlen += nwritten;
0745             totwritten += nwritten;
0746 
0747             /* If we fully sent the object on head go to the next one */
0748             if (c->sentlen == objlen) {
0749                 listDelNode(c->reply,listFirst(c->reply));
0750                 c->sentlen = 0;
0751                 c->reply_bytes -= objmem;
0752             }
0753         }
0754         /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT
0755          * bytes, in a single threaded server it's a good idea to serve
0756          * other clients as well, even if a very large request comes from
0757          * super fast link that is always able to accept data (in real world
0758          * scenario think about 'KEYS *' against the loopback interface).
0759          *
0760          * However if we are over the maxmemory limit we ignore that and
0761          * just deliver as much data as it is possible to deliver. */
0762         if (totwritten > REDIS_MAX_WRITE_PER_EVENT &&
0763             (server.maxmemory == 0 ||
0764              zmalloc_used_memory() < server.maxmemory)) break;
0765     }
0766     if (nwritten == -1) {
0767         if (errno == EAGAIN) {
0768             nwritten = 0;
0769         } else {
0770             redisLog(REDIS_VERBOSE,
0771                 "Error writing to client: %s", strerror(errno));
0772             freeClient(c);
0773             return;
0774         }
0775     }
0776     if (totwritten > 0) c->lastinteraction = server.unixtime;
0777     if (c->bufpos == 0 && listLength(c->reply) == 0) {
0778         c->sentlen = 0;
0779         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
0780 
0781         /* Close connection after entire reply has been sent. */
0782         if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);
0783     }
0784 }
0785 
0786 /* resetClient prepare the client to process the next command */
0787 void resetClient(redisClient *c) {
0788     freeClientArgv(c);
0789     c->reqtype = 0;
0790     c->multibulklen = 0;
0791     c->bulklen = -1;
0792     /* We clear the ASKING flag as well if we are not inside a MULTI. */
0793     if (!(c->flags & REDIS_MULTI)) c->flags &= (~REDIS_ASKING);
0794 }
0795 
0796 int processInlineBuffer(redisClient *c) {
0797     char *newline = strstr(c->querybuf,"\r\n");
0798     int argc, j;
0799     sds *argv;
0800     size_t querylen;
0801 
0802     /* Nothing to do without a \r\n */
0803     if (newline == NULL) {
0804         if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
0805             addReplyError(c,"Protocol error: too big inline request");
0806             setProtocolError(c,0);
0807         }
0808         return REDIS_ERR;
0809     }
0810 
0811     /* Split the input buffer up to the \r\n */
0812     querylen = newline-(c->querybuf);
0813     argv = sdssplitlen(c->querybuf,querylen," ",1,&argc);
0814 
0815     /* Leave data after the first line of the query in the buffer */
0816     c->querybuf = sdsrange(c->querybuf,querylen+2,-1);
0817 
0818     /* Setup argv array on client structure */
0819     if (c->argv) zfree(c->argv);
0820     c->argv = zmalloc(sizeof(robj*)*argc);
0821 
0822     /* Create redis objects for all arguments. */
0823     for (c->argc = 0, j = 0; j < argc; j++) {
0824         if (sdslen(argv[j])) {
0825             c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
0826             c->argc++;
0827         } else {
0828             sdsfree(argv[j]);
0829         }
0830     }
0831     zfree(argv);
0832     return REDIS_OK;
0833 }
0834 
0835 /* Helper function. Trims query buffer to make the function that processes
0836  * multi bulk requests idempotent. */
0837 static void setProtocolError(redisClient *c, int pos) {
0838     if (server.verbosity >= REDIS_VERBOSE) {
0839         sds client = getClientInfoString(c);
0840         redisLog(REDIS_VERBOSE,
0841             "Protocol error from client: %s", client);
0842         sdsfree(client);
0843     }
0844     c->flags |= REDIS_CLOSE_AFTER_REPLY;
0845     c->querybuf = sdsrange(c->querybuf,pos,-1);
0846 }
0847 
0848 int processMultibulkBuffer(redisClient *c) {
0849     char *newline = NULL;
0850     int pos = 0, ok;
0851     long long ll;
0852 
0853     if (c->multibulklen == 0) {
0854         /* The client should have been reset */
0855         redisAssertWithInfo(c,NULL,c->argc == 0);
0856 
0857         /* Multi bulk length cannot be read without a \r\n */
0858         newline = strchr(c->querybuf,'\r');
0859         if (newline == NULL) {
0860             if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
0861                 addReplyError(c,"Protocol error: too big mbulk count string");
0862                 setProtocolError(c,0);
0863             }
0864             return REDIS_ERR;
0865         }
0866 
0867         /* Buffer should also contain \n */
0868         if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
0869             return REDIS_ERR;
0870 
0871         /* We know for sure there is a whole line since newline != NULL,
0872          * so go ahead and find out the multi bulk length. */
0873         redisAssertWithInfo(c,NULL,c->querybuf[0] == '*');
0874         ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
0875         if (!ok || ll > 1024*1024) {
0876             addReplyError(c,"Protocol error: invalid multibulk length");
0877             setProtocolError(c,pos);
0878             return REDIS_ERR;
0879         }
0880 
0881         pos = (newline-c->querybuf)+2;
0882         if (ll <= 0) {
0883             c->querybuf = sdsrange(c->querybuf,pos,-1);
0884             return REDIS_OK;
0885         }
0886 
0887         c->multibulklen = ll;
0888 
0889         /* Setup argv array on client structure */
0890         if (c->argv) zfree(c->argv);
0891         c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
0892     }
0893 
0894     redisAssertWithInfo(c,NULL,c->multibulklen > 0);
0895     while(c->multibulklen) {
0896         /* Read bulk length if unknown */
0897         if (c->bulklen == -1) {
0898             newline = strchr(c->querybuf+pos,'\r');
0899             if (newline == NULL) {
0900                 if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
0901                     addReplyError(c,"Protocol error: too big bulk count string");
0902                     setProtocolError(c,0);
0903                 }
0904                 break;
0905             }
0906 
0907             /* Buffer should also contain \n */
0908             if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
0909                 break;
0910 
0911             if (c->querybuf[pos] != '$') {
0912                 addReplyErrorFormat(c,
0913                     "Protocol error: expected '$', got '%c'",
0914                     c->querybuf[pos]);
0915                 setProtocolError(c,pos);
0916                 return REDIS_ERR;
0917             }
0918 
0919             ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll);
0920             if (!ok || ll < 0 || ll > 512*1024*1024) {
0921                 addReplyError(c,"Protocol error: invalid bulk length");
0922                 setProtocolError(c,pos);
0923                 return REDIS_ERR;
0924             }
0925 
0926             pos += newline-(c->querybuf+pos)+2;
0927             if (ll >= REDIS_MBULK_BIG_ARG) {
0928                 /* If we are going to read a large object from network
0929                  * try to make it likely that it will start at c->querybuf
0930                  * boundary so that we can optimized object creation
0931                  * avoiding a large copy of data. */
0932                 c->querybuf = sdsrange(c->querybuf,pos,-1);
0933                 pos = 0;
0934                 /* Hint the sds library about the amount of bytes this string is
0935                  * going to contain. */
0936                 c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2);
0937             }
0938             c->bulklen = ll;
0939         }
0940 
0941         /* Read bulk argument */
0942         if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) {
0943             /* Not enough data (+2 == trailing \r\n) */
0944             break;
0945         } else {
0946             /* Optimization: if the buffer contains JUST our bulk element
0947              * instead of creating a new object by *copying* the sds we
0948              * just use the current sds string. */
0949             if (pos == 0 &&
0950                 c->bulklen >= REDIS_MBULK_BIG_ARG &&
0951                 (signed) sdslen(c->querybuf) == c->bulklen+2)
0952             {
0953                 c->argv[c->argc++] = createObject(REDIS_STRING,c->querybuf);
0954                 sdsIncrLen(c->querybuf,-2); /* remove CRLF */
0955                 c->querybuf = sdsempty();
0956                 /* Assume that if we saw a fat argument we'll see another one
0957                  * likely... */
0958                 c->querybuf = sdsMakeRoomFor(c->querybuf,c->bulklen+2);
0959                 pos = 0;
0960             } else {
0961                 c->argv[c->argc++] =
0962                     createStringObject(c->querybuf+pos,c->bulklen);
0963                 pos += c->bulklen+2;
0964             }
0965             c->bulklen = -1;
0966             c->multibulklen--;
0967         }
0968     }
0969 
0970     /* Trim to pos */
0971     if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1);
0972 
0973     /* We're done when c->multibulk == 0 */
0974     if (c->multibulklen == 0) return REDIS_OK;
0975 
0976     /* Still not read to process the command */
0977     return REDIS_ERR;
0978 }
0979 
0980 void processInputBuffer(redisClient *c) {
0981     /* Keep processing while there is something in the input buffer */
0982     while(sdslen(c->querybuf)) {
0983         /* Immediately abort if the client is in the middle of something. */
0984         if (c->flags & REDIS_BLOCKED) return;
0985 
0986         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
0987          * written to the client. Make sure to not let the reply grow after
0988          * this flag has been set (i.e. don't process more commands). */
0989         if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
0990 
0991         /* Determine request type when unknown. */
0992         if (!c->reqtype) {
0993             if (c->querybuf[0] == '*') {
0994                 c->reqtype = REDIS_REQ_MULTIBULK;
0995             } else {
0996                 c->reqtype = REDIS_REQ_INLINE;
0997             }
0998         }
0999 
1000         if (c->reqtype == REDIS_REQ_INLINE) {
1001             if (processInlineBuffer(c) != REDIS_OK) break;
1002         } else if (c->reqtype == REDIS_REQ_MULTIBULK) {
1003             if (processMultibulkBuffer(c) != REDIS_OK) break;
1004         } else {
1005             redisPanic("Unknown request type");
1006         }
1007 
1008         /* Multibulk processing could see a <= 0 length. */
1009         if (c->argc == 0) {
1010             resetClient(c);
1011         } else {
1012             /* Only reset the client when the command was executed. */
1013             if (processCommand(c) == REDIS_OK)
1014                 resetClient(c);
1015         }
1016     }
1017 }
1018 
1019 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
1020     redisClient *c = (redisClient*) privdata;
1021     int nread, readlen;
1022     size_t qblen;
1023     REDIS_NOTUSED(el);
1024     REDIS_NOTUSED(mask);
1025 
1026     server.current_client = c;
1027     readlen = REDIS_IOBUF_LEN;
1028     /* If this is a multi bulk request, and we are processing a bulk reply
1029      * that is large enough, try to maximize the probability that the query
1030      * buffer contains exactly the SDS string representing the object, even
1031      * at the risk of requiring more read(2) calls. This way the function
1032      * processMultiBulkBuffer() can avoid copying buffers to create the
1033      * Redis Object representing the argument. */
1034     if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
1035         && c->bulklen >= REDIS_MBULK_BIG_ARG)
1036     {
1037         int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf);
1038 
1039         if (remaining < readlen) readlen = remaining;
1040     }
1041 
1042     qblen = sdslen(c->querybuf);
1043     if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
1044     c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
1045     nread = read(fd, c->querybuf+qblen, readlen);
1046     if (nread == -1) {
1047         if (errno == EAGAIN) {
1048             nread = 0;
1049         } else {
1050             redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
1051             freeClient(c);
1052             return;
1053         }
1054     } else if (nread == 0) {
1055         redisLog(REDIS_VERBOSE, "Client closed connection");
1056         freeClient(c);
1057         return;
1058     }
1059     if (nread) {
1060         sdsIncrLen(c->querybuf,nread);
1061         c->lastinteraction = server.unixtime;
1062     } else {
1063         server.current_client = NULL;
1064         return;
1065     }
1066     if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
1067         sds ci = getClientInfoString(c), bytes = sdsempty();
1068 
1069         bytes = sdscatrepr(bytes,c->querybuf,64);
1070         redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
1071         sdsfree(ci);
1072         sdsfree(bytes);
1073         freeClient(c);
1074         return;
1075     }
1076     processInputBuffer(c);
1077     server.current_client = NULL;
1078 }
1079 
1080 void getClientsMaxBuffers(unsigned long *longest_output_list,
1081                           unsigned long *biggest_input_buffer) {
1082     redisClient *c;
1083     listNode *ln;
1084     listIter li;
1085     unsigned long lol = 0, bib = 0;
1086 
1087     listRewind(server.clients,&li);
1088     while ((ln = listNext(&li)) != NULL) {
1089         c = listNodeValue(ln);
1090 
1091         if (listLength(c->reply) > lol) lol = listLength(c->reply);
1092         if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf);
1093     }
1094     *longest_output_list = lol;
1095     *biggest_input_buffer = bib;
1096 }
1097 
1098 /* Turn a Redis client into an sds string representing its state. */
1099 sds getClientInfoString(redisClient *client) {
1100     char ip[32], flags[16], events[3], *p;
1101     int port = 0; /* initialized to zero for the unix socket case. */
1102     int emask;
1103 
1104     if (!(client->flags & REDIS_UNIX_SOCKET))
1105         anetPeerToString(client->fd,ip,&port);
1106     p = flags;
1107     if (client->flags & REDIS_SLAVE) {
1108         if (client->flags & REDIS_MONITOR)
1109             *p++ = 'O';
1110         else
1111             *p++ = 'S';
1112     }
1113     if (client->flags & REDIS_MASTER) *p++ = 'M';
1114     if (client->flags & REDIS_MULTI) *p++ = 'x';
1115     if (client->flags & REDIS_BLOCKED) *p++ = 'b';
1116     if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
1117     if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
1118     if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
1119     if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A';
1120     if (client->flags & REDIS_UNIX_SOCKET) *p++ = 'U';
1121     if (p == flags) *p++ = 'N';
1122     *p++ = '\0';
1123 
1124     emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd);
1125     p = events;
1126     if (emask & AE_READABLE) *p++ = 'r';
1127     if (emask & AE_WRITABLE) *p++ = 'w';
1128     *p = '\0';
1129     return sdscatprintf(sdsempty(),
1130         "addr=%s:%d fd=%d name=%s age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d multi=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
1131         (client->flags & REDIS_UNIX_SOCKET) ? server.unixsocket : ip,
1132         port,
1133         client->fd,
1134         client->name ? (char*)client->name->ptr : "",
1135         (long)(server.unixtime - client->ctime),
1136         (long)(server.unixtime - client->lastinteraction),
1137         flags,
1138         client->db->id,
1139         (int) dictSize(client->pubsub_channels),
1140         (int) listLength(client->pubsub_patterns),
1141         (client->flags & REDIS_MULTI) ? client->mstate.count : -1,
1142         (unsigned long) sdslen(client->querybuf),
1143         (unsigned long) sdsavail(client->querybuf),
1144         (unsigned long) client->bufpos,
1145         (unsigned long) listLength(client->reply),
1146         getClientOutputBufferMemoryUsage(client),
1147         events,
1148         client->lastcmd ? client->lastcmd->name : "NULL");
1149 }
1150 
1151 sds getAllClientsInfoString(void) {
1152     listNode *ln;
1153     listIter li;
1154     redisClient *client;
1155     sds o = sdsempty();
1156 
1157     listRewind(server.clients,&li);
1158     while ((ln = listNext(&li)) != NULL) {
1159         sds cs;
1160 
1161         client = listNodeValue(ln);
1162         cs = getClientInfoString(client);
1163         o = sdscatsds(o,cs);
1164         sdsfree(cs);
1165         o = sdscatlen(o,"\n",1);
1166     }
1167     return o;
1168 }
1169 
1170 void clientCommand(redisClient *c) {
1171     listNode *ln;
1172     listIter li;
1173     redisClient *client;
1174 
1175     if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) {
1176         sds o = getAllClientsInfoString();
1177         addReplyBulkCBuffer(c,o,sdslen(o));
1178         sdsfree(o);
1179     } else if (!strcasecmp(c->argv[1]->ptr,"kill") && c->argc == 3) {
1180         listRewind(server.clients,&li);
1181         while ((ln = listNext(&li)) != NULL) {
1182             char ip[32], addr[64];
1183             int port;
1184 
1185             client = listNodeValue(ln);
1186             if (anetPeerToString(client->fd,ip,&port) == -1) continue;
1187             snprintf(addr,sizeof(addr),"%s:%d",ip,port);
1188             if (strcmp(addr,c->argv[2]->ptr) == 0) {
1189                 addReply(c,shared.ok);
1190                 if (c == client) {
1191                     client->flags |= REDIS_CLOSE_AFTER_REPLY;
1192                 } else {
1193                     freeClient(client);
1194                 }
1195                 return;
1196             }
1197         }
1198         addReplyError(c,"No such client");
1199     } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) {
1200         int j, len = sdslen(c->argv[2]->ptr);
1201         char *p = c->argv[2]->ptr;
1202 
1203         /* Setting the client name to an empty string actually removes
1204          * the current name. */
1205         if (len == 0) {
1206             if (c->name) decrRefCount(c->name);
1207             c->name = NULL;
1208             addReply(c,shared.ok);
1209             return;
1210         }
1211 
1212         /* Otherwise check if the charset is ok. We need to do this otherwise
1213          * CLIENT LIST format will break. You should always be able to
1214          * split by space to get the different fields. */
1215         for (j = 0; j < len; j++) {
1216             if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */
1217                 addReplyError(c,
1218                     "Client names cannot contain spaces, "
1219                     "newlines or special characters.");
1220                 return;
1221             }
1222         }
1223         if (c->name) decrRefCount(c->name);
1224         c->name = c->argv[2];
1225         incrRefCount(c->name);
1226         addReply(c,shared.ok);
1227     } else if (!strcasecmp(c->argv[1]->ptr,"getname") && c->argc == 2) {
1228         if (c->name)
1229             addReplyBulk(c,c->name);
1230         else
1231             addReply(c,shared.nullbulk);
1232     } else {
1233         addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port | GETNAME | SETNAME connection-name)");
1234     }
1235 }
1236 
1237 /* Rewrite the command vector of the client. All the new objects ref count
1238  * is incremented. The old command vector is freed, and the old objects
1239  * ref count is decremented. */
1240 void rewriteClientCommandVector(redisClient *c, int argc, ...) {
1241     va_list ap;
1242     int j;
1243     robj **argv; /* The new argument vector */
1244 
1245     argv = zmalloc(sizeof(robj*)*argc);
1246     va_start(ap,argc);
1247     for (j = 0; j < argc; j++) {
1248         robj *a;
1249         
1250         a = va_arg(ap, robj*);
1251         argv[j] = a;
1252         incrRefCount(a);
1253     }
1254     /* We free the objects in the original vector at the end, so we are
1255      * sure that if the same objects are reused in the new vector the
1256      * refcount gets incremented before it gets decremented. */
1257     for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]);
1258     zfree(c->argv);
1259     /* Replace argv and argc with our new versions. */
1260     c->argv = argv;
1261     c->argc = argc;
1262     c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
1263     redisAssertWithInfo(c,NULL,c->cmd != NULL);
1264     va_end(ap);
1265 }
1266 
1267 /* Rewrite a single item in the command vector.
1268  * The new val ref count is incremented, and the old decremented. */
1269 void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
1270     robj *oldval;
1271    
1272     redisAssertWithInfo(c,NULL,i < c->argc);
1273     oldval = c->argv[i];
1274     c->argv[i] = newval;
1275     incrRefCount(newval);
1276     decrRefCount(oldval);
1277 
1278     /* If this is the command name make sure to fix c->cmd. */
1279     if (i == 0) {
1280         c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
1281         redisAssertWithInfo(c,NULL,c->cmd != NULL);
1282     }
1283 }
1284 
1285 /* This function returns the number of bytes that Redis is virtually
1286  * using to store the reply still not read by the client.
1287  * It is "virtual" since the reply output list may contain objects that
1288  * are shared and are not really using additional memory.
1289  *
1290  * The function returns the total sum of the length of all the objects
1291  * stored in the output list, plus the memory used to allocate every
1292  * list node. The static reply buffer is not taken into account since it
1293  * is allocated anyway.
1294  *
1295  * Note: this function is very fast so can be called as many time as
1296  * the caller wishes. The main usage of this function currently is
1297  * enforcing the client output length limits. */
1298 unsigned long getClientOutputBufferMemoryUsage(redisClient *c) {
1299     unsigned long list_item_size = sizeof(listNode)+sizeof(robj);
1300 
1301     return c->reply_bytes + (list_item_size*listLength(c->reply));
1302 }
1303 
1304 /* Get the class of a client, used in order to enforce limits to different
1305  * classes of clients.
1306  *
1307  * The function will return one of the following:
1308  * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client
1309  * REDIS_CLIENT_LIMIT_CLASS_SLAVE  -> Slave or client executing MONITOR command
1310  * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels
1311  */
1312 int getClientLimitClass(redisClient *c) {
1313     if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
1314     if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns))
1315         return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;
1316     return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
1317 }
1318 
1319 int getClientLimitClassByName(char *name) {
1320     if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
1321     else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
1322     else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;
1323     else return -1;
1324 }
1325 
1326 char *getClientLimitClassName(int class) {
1327     switch(class) {
1328     case REDIS_CLIENT_LIMIT_CLASS_NORMAL:   return "normal";
1329     case REDIS_CLIENT_LIMIT_CLASS_SLAVE:    return "slave";
1330     case REDIS_CLIENT_LIMIT_CLASS_PUBSUB:   return "pubsub";
1331     default:                                return NULL;
1332     }
1333 }
1334 
1335 /* The function checks if the client reached output buffer soft or hard
1336  * limit, and also update the state needed to check the soft limit as
1337  * a side effect.
1338  *
1339  * Return value: non-zero if the client reached the soft or the hard limit.
1340  *               Otherwise zero is returned. */
1341 int checkClientOutputBufferLimits(redisClient *c) {
1342     int soft = 0, hard = 0, class;
1343     unsigned long used_mem = getClientOutputBufferMemoryUsage(c);
1344 
1345     class = getClientLimitClass(c);
1346     if (server.client_obuf_limits[class].hard_limit_bytes &&
1347         used_mem >= server.client_obuf_limits[class].hard_limit_bytes)
1348         hard = 1;
1349     if (server.client_obuf_limits[class].soft_limit_bytes &&
1350         used_mem >= server.client_obuf_limits[class].soft_limit_bytes)
1351         soft = 1;
1352 
1353     /* We need to check if the soft limit is reached continuously for the
1354      * specified amount of seconds. */
1355     if (soft) {
1356         if (c->obuf_soft_limit_reached_time == 0) {
1357             c->obuf_soft_limit_reached_time = server.unixtime;
1358             soft = 0; /* First time we see the soft limit reached */
1359         } else {
1360             time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time;
1361 
1362             if (elapsed <=
1363                 server.client_obuf_limits[class].soft_limit_seconds) {
1364                 soft = 0; /* The client still did not reached the max number of
1365                              seconds for the soft limit to be considered
1366                              reached. */
1367             }
1368         }
1369     } else {
1370         c->obuf_soft_limit_reached_time = 0;
1371     }
1372     return soft || hard;
1373 }
1374 
1375 /* Asynchronously close a client if soft or hard limit is reached on the
1376  * output buffer size. The caller can check if the client will be closed
1377  * checking if the client REDIS_CLOSE_ASAP flag is set.
1378  *
1379  * Note: we need to close the client asynchronously because this function is
1380  * called from contexts where the client can't be freed safely, i.e. from the
1381  * lower level functions pushing data inside the client output buffers. */
1382 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {
1383     redisAssert(c->reply_bytes < ULONG_MAX-(1024*64));
1384     if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return;
1385     if (checkClientOutputBufferLimits(c)) {
1386         sds client = getClientInfoString(c);
1387 
1388         freeClientAsync(c);
1389         redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
1390         sdsfree(client);
1391     }
1392 }
1393 
1394 /* Helper function used by freeMemoryIfNeeded() in order to flush slaves
1395  * output buffers without returning control to the event loop. */
1396 void flushSlavesOutputBuffers(void) {
1397     listIter li;
1398     listNode *ln;
1399 
1400     listRewind(server.slaves,&li);
1401     while((ln = listNext(&li))) {
1402         redisClient *slave = listNodeValue(ln);
1403         int events;
1404 
1405         events = aeGetFileEvents(server.el,slave->fd);
1406         if (events & AE_WRITABLE &&
1407             slave->replstate == REDIS_REPL_ONLINE &&
1408             listLength(slave->reply))
1409         {
1410             sendReplyToClient(server.el,slave->fd,slave,0);
1411         }
1412     }
1413 }