1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "redis.h" 31 #include <sys/uio.h> 32 #include <math.h> 33 34 static void setProtocolError(redisClient *c, int pos); 35 36 /* To evaluate the output buffer size of a client we need to get size of 37 * allocated objects, however we can't used zmalloc_size() directly on sds 38 * strings because of the trick they use to work (the header is before the 39 * returned pointer), so we use this helper function. */ 40 size_t zmalloc_size_sds(sds s) { 41 return zmalloc_size(s-sizeof(struct sdshdr)); 42 } 43 44 /* Return the amount of memory used by the sds string at object->ptr 45 * for a string object. */ 46 size_t getStringObjectSdsUsedMemory(robj *o) { 47 redisAssertWithInfo(NULL,o,o->type == REDIS_STRING); 48 switch(o->encoding) { 49 case REDIS_ENCODING_RAW: return zmalloc_size_sds(o->ptr); 50 case REDIS_ENCODING_EMBSTR: return sdslen(o->ptr); 51 default: return 0; /* Just integer encoding for now. */ 52 } 53 } 54 55 void *dupClientReplyValue(void *o) { 56 incrRefCount((robj*)o); 57 return o; 58 } 59 60 int listMatchObjects(void *a, void *b) { 61 return equalStringObjects(a,b); 62 } 63 64 redisClient *createClient(int fd) { 65 redisClient *c = zmalloc(sizeof(redisClient)); 66 67 /* passing -1 as fd it is possible to create a non connected client. 68 * This is useful since all the Redis commands needs to be executed 69 * in the context of a client. When commands are executed in other 70 * contexts (for instance a Lua script) we need a non connected client. */ 71 if (fd != -1) { 72 anetNonBlock(NULL,fd); 73 anetEnableTcpNoDelay(NULL,fd); 74 if (server.tcpkeepalive) 75 anetKeepAlive(NULL,fd,server.tcpkeepalive); 76 if (aeCreateFileEvent(server.el,fd,AE_READABLE, 77 readQueryFromClient, c) == AE_ERR) 78 { 79 close(fd); 80 zfree(c); 81 return NULL; 82 } 83 } 84 85 selectDb(c,0); 86 c->fd = fd; 87 c->name = NULL; 88 c->bufpos = 0; 89 c->querybuf = sdsempty(); 90 c->querybuf_peak = 0; 91 c->reqtype = 0; 92 c->argc = 0; 93 c->argv = NULL; 94 c->cmd = c->lastcmd = NULL; 95 c->multibulklen = 0; 96 c->bulklen = -1; 97 c->sentlen = 0; 98 c->flags = 0; 99 c->ctime = c->lastinteraction = server.unixtime; 100 c->authenticated = 0; 101 c->replstate = REDIS_REPL_NONE; 102 c->reploff = 0; 103 c->repl_ack_off = 0; 104 c->repl_ack_time = 0; 105 c->slave_listening_port = 0; 106 c->reply = listCreate(); 107 c->reply_bytes = 0; 108 c->obuf_soft_limit_reached_time = 0; 109 listSetFreeMethod(c->reply,decrRefCountVoid); 110 listSetDupMethod(c->reply,dupClientReplyValue); 111 c->btype = REDIS_BLOCKED_NONE; 112 c->bpop.timeout = 0; 113 c->bpop.keys = dictCreate(&setDictType,NULL); 114 c->bpop.target = NULL; 115 c->bpop.numreplicas = 0; 116 c->bpop.reploffset = 0; 117 c->woff = 0; 118 c->watched_keys = listCreate(); 119 c->pubsub_channels = dictCreate(&setDictType,NULL); 120 c->pubsub_patterns = listCreate(); 121 c->peerid = NULL; 122 listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid); 123 listSetMatchMethod(c->pubsub_patterns,listMatchObjects); 124 if (fd != -1) listAddNodeTail(server.clients,c); 125 initClientMultiState(c); 126 return c; 127 } 128 129 /* This function is called every time we are going to transmit new data 130 * to the client. The behavior is the following: 131 * 132 * If the client should receive new data (normal clients will) the function 133 * returns REDIS_OK, and make sure to install the write handler in our event 134 * loop so that when the socket is writable new data gets written. 135 * 136 * If the client should not receive new data, because it is a fake client, 137 * a master, a slave not yet online, or because the setup of the write handler 138 * failed, the function returns REDIS_ERR. 139 * 140 * Typically gets called every time a reply is built, before adding more 141 * data to the clients output buffers. If the function returns REDIS_ERR no 142 * data should be appended to the output buffers. */ 143 int prepareClientToWrite(redisClient *c) { 144 if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK; 145 if ((c->flags & REDIS_MASTER) && 146 !(c->flags & REDIS_MASTER_FORCE_REPLY)) return REDIS_ERR; 147 if (c->fd <= 0) return REDIS_ERR; /* Fake client */ 148 if (c->bufpos == 0 && listLength(c->reply) == 0 && 149 (c->replstate == REDIS_REPL_NONE || 150 c->replstate == REDIS_REPL_ONLINE) && 151 aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, 152 sendReplyToClient, c) == AE_ERR) return REDIS_ERR; 153 return REDIS_OK; 154 } 155 156 /* Create a duplicate of the last object in the reply list when 157 * it is not exclusively owned by the reply list. */ 158 robj *dupLastObjectIfNeeded(list *reply) { 159 robj *new, *cur; 160 listNode *ln; 161 redisAssert(listLength(reply) > 0); 162 ln = listLast(reply); 163 cur = listNodeValue(ln); 164 if (cur->refcount > 1) { 165 new = dupStringObject(cur); 166 decrRefCount(cur); 167 listNodeValue(ln) = new; 168 } 169 return listNodeValue(ln); 170 } 171 172 /* ----------------------------------------------------------------------------- 173 * Low level functions to add more data to output buffers. 174 * -------------------------------------------------------------------------- */ 175 176 int _addReplyToBuffer(redisClient *c, char *s, size_t len) { 177 size_t available = sizeof(c->buf)-c->bufpos; 178 179 if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK; 180 181 /* If there already are entries in the reply list, we cannot 182 * add anything more to the static buffer. */ 183 if (listLength(c->reply) > 0) return REDIS_ERR; 184 185 /* Check that the buffer has enough space available for this string. */ 186 if (len > available) return REDIS_ERR; 187 188 memcpy(c->buf+c->bufpos,s,len); 189 c->bufpos+=len; 190 return REDIS_OK; 191 } 192 193 void _addReplyObjectToList(redisClient *c, robj *o) { 194 robj *tail; 195 196 if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; 197 198 if (listLength(c->reply) == 0) { 199 incrRefCount(o); 200 listAddNodeTail(c->reply,o); 201 c->reply_bytes += getStringObjectSdsUsedMemory(o); 202 } else { 203 tail = listNodeValue(listLast(c->reply)); 204 205 /* Append to this object when possible. */ 206 if (tail->ptr != NULL && 207 tail->encoding == REDIS_ENCODING_RAW && 208 sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES) 209 { 210 c->reply_bytes -= zmalloc_size_sds(tail->ptr); 211 tail = dupLastObjectIfNeeded(c->reply); 212 tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr)); 213 c->reply_bytes += zmalloc_size_sds(tail->ptr); 214 } else { 215 incrRefCount(o); 216 listAddNodeTail(c->reply,o); 217 c->reply_bytes += getStringObjectSdsUsedMemory(o); 218 } 219 } 220 asyncCloseClientOnOutputBufferLimitReached(c); 221 } 222 223 /* This method takes responsibility over the sds. When it is no longer 224 * needed it will be free'd, otherwise it ends up in a robj. */ 225 void _addReplySdsToList(redisClient *c, sds s) { 226 robj *tail; 227 228 if (c->flags & REDIS_CLOSE_AFTER_REPLY) { 229 sdsfree(s); 230 return; 231 } 232 233 if (listLength(c->reply) == 0) { 234 listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); 235 c->reply_bytes += zmalloc_size_sds(s); 236 } else { 237 tail = listNodeValue(listLast(c->reply)); 238 239 /* Append to this object when possible. */ 240 if (tail->ptr != NULL && tail->encoding == REDIS_ENCODING_RAW && 241 sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES) 242 { 243 c->reply_bytes -= zmalloc_size_sds(tail->ptr); 244 tail = dupLastObjectIfNeeded(c->reply); 245 tail->ptr = sdscatlen(tail->ptr,s,sdslen(s)); 246 c->reply_bytes += zmalloc_size_sds(tail->ptr); 247 sdsfree(s); 248 } else { 249 listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); 250 c->reply_bytes += zmalloc_size_sds(s); 251 } 252 } 253 asyncCloseClientOnOutputBufferLimitReached(c); 254 } 255 256 void _addReplyStringToList(redisClient *c, char *s, size_t len) { 257 robj *tail; 258 259 if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; 260 261 if (listLength(c->reply) == 0) { 262 robj *o = createStringObject(s,len); 263 264 listAddNodeTail(c->reply,o); 265 c->reply_bytes += getStringObjectSdsUsedMemory(o); 266 } else { 267 tail = listNodeValue(listLast(c->reply)); 268 269 /* Append to this object when possible. */ 270 if (tail->ptr != NULL && tail->encoding == REDIS_ENCODING_RAW && 271 sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES) 272 { 273 c->reply_bytes -= zmalloc_size_sds(tail->ptr); 274 tail = dupLastObjectIfNeeded(c->reply); 275 tail->ptr = sdscatlen(tail->ptr,s,len); 276 c->reply_bytes += zmalloc_size_sds(tail->ptr); 277 } else { 278 robj *o = createStringObject(s,len); 279 280 listAddNodeTail(c->reply,o); 281 c->reply_bytes += getStringObjectSdsUsedMemory(o); 282 } 283 } 284 asyncCloseClientOnOutputBufferLimitReached(c); 285 } 286 287 /* ----------------------------------------------------------------------------- 288 * Higher level functions to queue data on the client output buffer. 289 * The following functions are the ones that commands implementations will call. 290 * -------------------------------------------------------------------------- */ 291 292 void addReply(redisClient *c, robj *obj) { 293 if (prepareClientToWrite(c) != REDIS_OK) return; 294 295 /* This is an important place where we can avoid copy-on-write 296 * when there is a saving child running, avoiding touching the 297 * refcount field of the object if it's not needed. 298 * 299 * If the encoding is RAW and there is room in the static buffer 300 * we'll be able to send the object to the client without 301 * messing with its page. */ 302 if (sdsEncodedObject(obj)) { 303 if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) 304 _addReplyObjectToList(c,obj); 305 } else if (obj->encoding == REDIS_ENCODING_INT) { 306 /* Optimization: if there is room in the static buffer for 32 bytes 307 * (more than the max chars a 64 bit integer can take as string) we 308 * avoid decoding the object and go for the lower level approach. */ 309 if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) { 310 char buf[32]; 311 int len; 312 313 len = ll2string(buf,sizeof(buf),(long)obj->ptr); 314 if (_addReplyToBuffer(c,buf,len) == REDIS_OK) 315 return; 316 /* else... continue with the normal code path, but should never 317 * happen actually since we verified there is room. */ 318 } 319 obj = getDecodedObject(obj); 320 if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) 321 _addReplyObjectToList(c,obj); 322 decrRefCount(obj); 323 } else { 324 redisPanic("Wrong obj->encoding in addReply()"); 325 } 326 } 327 328 void addReplySds(redisClient *c, sds s) { 329 if (prepareClientToWrite(c) != REDIS_OK) { 330 /* The caller expects the sds to be free'd. */ 331 sdsfree(s); 332 return; 333 } 334 if (_addReplyToBuffer(c,s,sdslen(s)) == REDIS_OK) { 335 sdsfree(s); 336 } else { 337 /* This method free's the sds when it is no longer needed. */ 338 _addReplySdsToList(c,s); 339 } 340 } 341 342 void addReplyString(redisClient *c, char *s, size_t len) { 343 if (prepareClientToWrite(c) != REDIS_OK) return; 344 if (_addReplyToBuffer(c,s,len) != REDIS_OK) 345 _addReplyStringToList(c,s,len); 346 } 347 348 void addReplyErrorLength(redisClient *c, char *s, size_t len) { 349 addReplyString(c,"-ERR ",5); 350 addReplyString(c,s,len); 351 addReplyString(c,"\r\n",2); 352 } 353 354 void addReplyError(redisClient *c, char *err) { 355 addReplyErrorLength(c,err,strlen(err)); 356 } 357 358 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { 359 size_t l, j; 360 va_list ap; 361 va_start(ap,fmt); 362 sds s = sdscatvprintf(sdsempty(),fmt,ap); 363 va_end(ap); 364 /* Make sure there are no newlines in the string, otherwise invalid protocol 365 * is emitted. */ 366 l = sdslen(s); 367 for (j = 0; j < l; j++) { 368 if (s[j] == '\r' || s[j] == '\n') s[j] = ' '; 369 } 370 addReplyErrorLength(c,s,sdslen(s)); 371 sdsfree(s); 372 } 373 374 void addReplyStatusLength(redisClient *c, char *s, size_t len) { 375 addReplyString(c,"+",1); 376 addReplyString(c,s,len); 377 addReplyString(c,"\r\n",2); 378 } 379 380 void addReplyStatus(redisClient *c, char *status) { 381 addReplyStatusLength(c,status,strlen(status)); 382 } 383 384 void addReplyStatusFormat(redisClient *c, const char *fmt, ...) { 385 va_list ap; 386 va_start(ap,fmt); 387 sds s = sdscatvprintf(sdsempty(),fmt,ap); 388 va_end(ap); 389 addReplyStatusLength(c,s,sdslen(s)); 390 sdsfree(s); 391 } 392 393 /* Adds an empty object to the reply list that will contain the multi bulk 394 * length, which is not known when this function is called. */ 395 void *addDeferredMultiBulkLength(redisClient *c) { 396 /* Note that we install the write event here even if the object is not 397 * ready to be sent, since we are sure that before returning to the 398 * event loop setDeferredMultiBulkLength() will be called. */ 399 if (prepareClientToWrite(c) != REDIS_OK) return NULL; 400 listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL)); 401 return listLast(c->reply); 402 } 403 404 /* Populate the length object and try gluing it to the next chunk. */ 405 void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { 406 listNode *ln = (listNode*)node; 407 robj *len, *next; 408 409 /* Abort when *node is NULL (see addDeferredMultiBulkLength). */ 410 if (node == NULL) return; 411 412 len = listNodeValue(ln); 413 len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length); 414 len->encoding = REDIS_ENCODING_RAW; /* in case it was an EMBSTR. */ 415 c->reply_bytes += zmalloc_size_sds(len->ptr); 416 if (ln->next != NULL) { 417 next = listNodeValue(ln->next); 418 419 /* Only glue when the next node is non-NULL (an sds in this case) */ 420 if (next->ptr != NULL) { 421 c->reply_bytes -= zmalloc_size_sds(len->ptr); 422 c->reply_bytes -= getStringObjectSdsUsedMemory(next); 423 len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr)); 424 c->reply_bytes += zmalloc_size_sds(len->ptr); 425 listDelNode(c->reply,ln->next); 426 } 427 } 428 asyncCloseClientOnOutputBufferLimitReached(c); 429 } 430 431 /* Add a double as a bulk reply */ 432 void addReplyDouble(redisClient *c, double d) { 433 char dbuf[128], sbuf[128]; 434 int dlen, slen; 435 if (isinf(d)) { 436 /* Libc in odd systems (Hi Solaris!) will format infinite in a 437 * different way, so better to handle it in an explicit way. */ 438 addReplyBulkCString(c, d > 0 ? "inf" : "-inf"); 439 } else { 440 dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d); 441 slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf); 442 addReplyString(c,sbuf,slen); 443 } 444 } 445 446 /* Add a long long as integer reply or bulk len / multi bulk count. 447 * Basically this is used to output <prefix><long long><crlf>. */ 448 void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) { 449 char buf[128]; 450 int len; 451 452 /* Things like $3\r\n or *2\r\n are emitted very often by the protocol 453 * so we have a few shared objects to use if the integer is small 454 * like it is most of the times. */ 455 if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) { 456 addReply(c,shared.mbulkhdr[ll]); 457 return; 458 } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) { 459 addReply(c,shared.bulkhdr[ll]); 460 return; 461 } 462 463 buf[0] = prefix; 464 len = ll2string(buf+1,sizeof(buf)-1,ll); 465 buf[len+1] = '\r'; 466 buf[len+2] = '\n'; 467 addReplyString(c,buf,len+3); 468 } 469 470 void addReplyLongLong(redisClient *c, long long ll) { 471 if (ll == 0) 472 addReply(c,shared.czero); 473 else if (ll == 1) 474 addReply(c,shared.cone); 475 else 476 addReplyLongLongWithPrefix(c,ll,':'); 477 } 478 479 void addReplyMultiBulkLen(redisClient *c, long length) { 480 if (length < REDIS_SHARED_BULKHDR_LEN) 481 addReply(c,shared.mbulkhdr[length]); 482 else 483 addReplyLongLongWithPrefix(c,length,'*'); 484 } 485 486 /* Create the length prefix of a bulk reply, example: $2234 */ 487 void addReplyBulkLen(redisClient *c, robj *obj) { 488 size_t len; 489 490 if (sdsEncodedObject(obj)) { 491 len = sdslen(obj->ptr); 492 } else { 493 long n = (long)obj->ptr; 494 495 /* Compute how many bytes will take this integer as a radix 10 string */ 496 len = 1; 497 if (n < 0) { 498 len++; 499 n = -n; 500 } 501 while((n = n/10) != 0) { 502 len++; 503 } 504 } 505 506 if (len < REDIS_SHARED_BULKHDR_LEN) 507 addReply(c,shared.bulkhdr[len]); 508 else 509 addReplyLongLongWithPrefix(c,len,'$'); 510 } 511 512 /* Add a Redis Object as a bulk reply */ 513 void addReplyBulk(redisClient *c, robj *obj) { 514 addReplyBulkLen(c,obj); 515 addReply(c,obj); 516 addReply(c,shared.crlf); 517 } 518 519 /* Add a C buffer as bulk reply */ 520 void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { 521 addReplyLongLongWithPrefix(c,len,'$'); 522 addReplyString(c,p,len); 523 addReply(c,shared.crlf); 524 } 525 526 /* Add a C nul term string as bulk reply */ 527 void addReplyBulkCString(redisClient *c, char *s) { 528 if (s == NULL) { 529 addReply(c,shared.nullbulk); 530 } else { 531 addReplyBulkCBuffer(c,s,strlen(s)); 532 } 533 } 534 535 /* Add a long long as a bulk reply */ 536 void addReplyBulkLongLong(redisClient *c, long long ll) { 537 char buf[64]; 538 int len; 539 540 len = ll2string(buf,64,ll); 541 addReplyBulkCBuffer(c,buf,len); 542 } 543 544 /* Copy 'src' client output buffers into 'dst' client output buffers. 545 * The function takes care of freeing the old output buffers of the 546 * destination client. */ 547 void copyClientOutputBuffer(redisClient *dst, redisClient *src) { 548 listRelease(dst->reply); 549 dst->reply = listDup(src->reply); 550 memcpy(dst->buf,src->buf,src->bufpos); 551 dst->bufpos = src->bufpos; 552 dst->reply_bytes = src->reply_bytes; 553 } 554 555 #define MAX_ACCEPTS_PER_CALL 1000 556 static void acceptCommonHandler(int fd, int flags) { 557 redisClient *c; 558 if ((c = createClient(fd)) == NULL) { 559 redisLog(REDIS_WARNING, 560 "Error registering fd event for the new client: %s (fd=%d)", 561 strerror(errno),fd); 562 close(fd); /* May be already closed, just ignore errors */ 563 return; 564 } 565 /* If maxclient directive is set and this is one client more... close the 566 * connection. Note that we create the client instead to check before 567 * for this condition, since now the socket is already set in non-blocking 568 * mode and we can send an error for free using the Kernel I/O */ 569 if (listLength(server.clients) > server.maxclients) { 570 char *err = "-ERR max number of clients reached\r\n"; 571 572 /* That's a best effort error message, don't check write errors */ 573 if (write(c->fd,err,strlen(err)) == -1) { 574 /* Nothing to do, Just to avoid the warning... */ 575 } 576 server.stat_rejected_conn++; 577 freeClient(c); 578 return; 579 } 580 server.stat_numconnections++; 581 c->flags |= flags; 582 } 583 584 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 585 int cport, cfd, max = MAX_ACCEPTS_PER_CALL; 586 char cip[REDIS_IP_STR_LEN]; 587 REDIS_NOTUSED(el); 588 REDIS_NOTUSED(mask); 589 REDIS_NOTUSED(privdata); 590 591 while(max--) { 592 cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); 593 if (cfd == ANET_ERR) { 594 if (errno != EWOULDBLOCK) 595 redisLog(REDIS_WARNING, 596 "Accepting client connection: %s", server.neterr); 597 return; 598 } 599 redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport); 600 acceptCommonHandler(cfd,0); 601 } 602 } 603 604 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 605 int cfd, max = MAX_ACCEPTS_PER_CALL; 606 REDIS_NOTUSED(el); 607 REDIS_NOTUSED(mask); 608 REDIS_NOTUSED(privdata); 609 610 while(max--) { 611 cfd = anetUnixAccept(server.neterr, fd); 612 if (cfd == ANET_ERR) { 613 if (errno != EWOULDBLOCK) 614 redisLog(REDIS_WARNING, 615 "Accepting client connection: %s", server.neterr); 616 return; 617 } 618 redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket); 619 acceptCommonHandler(cfd,REDIS_UNIX_SOCKET); 620 } 621 } 622 623 static void freeClientArgv(redisClient *c) { 624 int j; 625 for (j = 0; j < c->argc; j++) 626 decrRefCount(c->argv[j]); 627 c->argc = 0; 628 c->cmd = NULL; 629 } 630 631 /* Close all the slaves connections. This is useful in chained replication 632 * when we resync with our own master and want to force all our slaves to 633 * resync with us as well. */ 634 void disconnectSlaves(void) { 635 while (listLength(server.slaves)) { 636 listNode *ln = listFirst(server.slaves); 637 freeClient((redisClient*)ln->value); 638 } 639 } 640 641 /* This function is called when the slave lose the connection with the 642 * master into an unexpected way. */ 643 void replicationHandleMasterDisconnection(void) { 644 server.master = NULL; 645 server.repl_state = REDIS_REPL_CONNECT; 646 server.repl_down_since = server.unixtime; 647 /* We lost connection with our master, force our slaves to resync 648 * with us as well to load the new data set. 649 * 650 * If server.masterhost is NULL the user called SLAVEOF NO ONE so 651 * slave resync is not needed. */ 652 if (server.masterhost != NULL) disconnectSlaves(); 653 } 654 655 void freeClient(redisClient *c) { 656 listNode *ln; 657 658 /* If this is marked as current client unset it */ 659 if (server.current_client == c) server.current_client = NULL; 660 661 /* If it is our master that's beging disconnected we should make sure 662 * to cache the state to try a partial resynchronization later. 663 * 664 * Note that before doing this we make sure that the client is not in 665 * some unexpected state, by checking its flags. */ 666 if (server.master && c->flags & REDIS_MASTER) { 667 redisLog(REDIS_WARNING,"Connection with master lost."); 668 if (!(c->flags & (REDIS_CLOSE_AFTER_REPLY| 669 REDIS_CLOSE_ASAP| 670 REDIS_BLOCKED| 671 REDIS_UNBLOCKED))) 672 { 673 replicationCacheMaster(c); 674 return; 675 } 676 } 677 678 /* Log link disconnection with slave */ 679 if ((c->flags & REDIS_SLAVE) && !(c->flags & REDIS_MONITOR)) { 680 char ip[REDIS_IP_STR_LEN]; 681 682 if (anetPeerToString(c->fd,ip,sizeof(ip),NULL) != -1) { 683 redisLog(REDIS_WARNING,"Connection with slave %s:%d lost.", 684 ip, c->slave_listening_port); 685 } 686 } 687 688 /* Free the query buffer */ 689 sdsfree(c->querybuf); 690 c->querybuf = NULL; 691 692 /* Deallocate structures used to block on blocking ops. */ 693 if (c->flags & REDIS_BLOCKED) unblockClient(c); 694 dictRelease(c->bpop.keys); 695 696 /* UNWATCH all the keys */ 697 unwatchAllKeys(c); 698 listRelease(c->watched_keys); 699 700 /* Unsubscribe from all the pubsub channels */ 701 pubsubUnsubscribeAllChannels(c,0); 702 pubsubUnsubscribeAllPatterns(c,0); 703 dictRelease(c->pubsub_channels); 704 listRelease(c->pubsub_patterns); 705 706 /* Close socket, unregister events, and remove list of replies and 707 * accumulated arguments. */ 708 if (c->fd != -1) { 709 aeDeleteFileEvent(server.el,c->fd,AE_READABLE); 710 aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); 711 close(c->fd); 712 } 713 listRelease(c->reply); 714 freeClientArgv(c); 715 716 /* Remove from the list of clients */ 717 if (c->fd != -1) { 718 ln = listSearchKey(server.clients,c); 719 redisAssert(ln != NULL); 720 listDelNode(server.clients,ln); 721 } 722 723 /* When client was just unblocked because of a blocking operation, 724 * remove it from the list of unblocked clients. */ 725 if (c->flags & REDIS_UNBLOCKED) { 726 ln = listSearchKey(server.unblocked_clients,c); 727 redisAssert(ln != NULL); 728 listDelNode(server.unblocked_clients,ln); 729 } 730 731 /* Master/slave cleanup Case 1: 732 * we lost the connection with a slave. */ 733 if (c->flags & REDIS_SLAVE) { 734 if (c->replstate == REDIS_REPL_SEND_BULK) { 735 if (c->repldbfd != -1) close(c->repldbfd); 736 if (c->replpreamble) sdsfree(c->replpreamble); 737 } 738 list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves; 739 ln = listSearchKey(l,c); 740 redisAssert(ln != NULL); 741 listDelNode(l,ln); 742 /* We need to remember the time when we started to have zero 743 * attached slaves, as after some time we'll free the replication 744 * backlog. */ 745 if (c->flags & REDIS_SLAVE && listLength(server.slaves) == 0) 746 server.repl_no_slaves_since = server.unixtime; 747 refreshGoodSlavesCount(); 748 } 749 750 /* Master/slave cleanup Case 2: 751 * we lost the connection with the master. */ 752 if (c->flags & REDIS_MASTER) replicationHandleMasterDisconnection(); 753 754 /* If this client was scheduled for async freeing we need to remove it 755 * from the queue. */ 756 if (c->flags & REDIS_CLOSE_ASAP) { 757 ln = listSearchKey(server.clients_to_close,c); 758 redisAssert(ln != NULL); 759 listDelNode(server.clients_to_close,ln); 760 } 761 762 /* Release other dynamically allocated client structure fields, 763 * and finally release the client structure itself. */ 764 if (c->name) decrRefCount(c->name); 765 zfree(c->argv); 766 freeClientMultiState(c); 767 sdsfree(c->peerid); 768 zfree(c); 769 } 770 771 /* Schedule a client to free it at a safe time in the serverCron() function. 772 * This function is useful when we need to terminate a client but we are in 773 * a context where calling freeClient() is not possible, because the client 774 * should be valid for the continuation of the flow of the program. */ 775 void freeClientAsync(redisClient *c) { 776 if (c->flags & REDIS_CLOSE_ASAP) return; 777 c->flags |= REDIS_CLOSE_ASAP; 778 listAddNodeTail(server.clients_to_close,c); 779 } 780 781 void freeClientsInAsyncFreeQueue(void) { 782 while (listLength(server.clients_to_close)) { 783 listNode *ln = listFirst(server.clients_to_close); 784 redisClient *c = listNodeValue(ln); 785 786 c->flags &= ~REDIS_CLOSE_ASAP; 787 freeClient(c); 788 listDelNode(server.clients_to_close,ln); 789 } 790 } 791 792 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { 793 redisClient *c = privdata; 794 int nwritten = 0, totwritten = 0, objlen; 795 size_t objmem; 796 robj *o; 797 REDIS_NOTUSED(el); 798 REDIS_NOTUSED(mask); 799 800 while(c->bufpos > 0 || listLength(c->reply)) { 801 if (c->bufpos > 0) { 802 nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); 803 if (nwritten <= 0) break; 804 c->sentlen += nwritten; 805 totwritten += nwritten; 806 807 /* If the buffer was sent, set bufpos to zero to continue with 808 * the remainder of the reply. */ 809 if (c->sentlen == c->bufpos) { 810 c->bufpos = 0; 811 c->sentlen = 0; 812 } 813 } else { 814 o = listNodeValue(listFirst(c->reply)); 815 objlen = sdslen(o->ptr); 816 objmem = getStringObjectSdsUsedMemory(o); 817 818 if (objlen == 0) { 819 listDelNode(c->reply,listFirst(c->reply)); 820 c->reply_bytes -= objmem; 821 continue; 822 } 823 824 nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen); 825 if (nwritten <= 0) break; 826 c->sentlen += nwritten; 827 totwritten += nwritten; 828 829 /* If we fully sent the object on head go to the next one */ 830 if (c->sentlen == objlen) { 831 listDelNode(c->reply,listFirst(c->reply)); 832 c->sentlen = 0; 833 c->reply_bytes -= objmem; 834 } 835 } 836 /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT 837 * bytes, in a single threaded server it's a good idea to serve 838 * other clients as well, even if a very large request comes from 839 * super fast link that is always able to accept data (in real world 840 * scenario think about 'KEYS *' against the loopback interface). 841 * 842 * However if we are over the maxmemory limit we ignore that and 843 * just deliver as much data as it is possible to deliver. */ 844 if (totwritten > REDIS_MAX_WRITE_PER_EVENT && 845 (server.maxmemory == 0 || 846 zmalloc_used_memory() < server.maxmemory)) break; 847 } 848 if (nwritten == -1) { 849 if (errno == EAGAIN) { 850 nwritten = 0; 851 } else { 852 redisLog(REDIS_VERBOSE, 853 "Error writing to client: %s", strerror(errno)); 854 freeClient(c); 855 return; 856 } 857 } 858 if (totwritten > 0) { 859 /* For clients representing masters we don't count sending data 860 * as an interaction, since we always send REPLCONF ACK commands 861 * that take some time to just fill the socket output buffer. 862 * We just rely on data / pings received for timeout detection. */ 863 if (!(c->flags & REDIS_MASTER)) c->lastinteraction = server.unixtime; 864 } 865 if (c->bufpos == 0 && listLength(c->reply) == 0) { 866 c->sentlen = 0; 867 aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); 868 869 /* Close connection after entire reply has been sent. */ 870 if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c); 871 } 872 } 873 874 /* resetClient prepare the client to process the next command */ 875 void resetClient(redisClient *c) { 876 redisCommandProc *prevcmd = c->cmd ? c->cmd->proc : NULL; 877 878 freeClientArgv(c); 879 c->reqtype = 0; 880 c->multibulklen = 0; 881 c->bulklen = -1; 882 /* We clear the ASKING flag as well if we are not inside a MULTI, and 883 * if what we just executed is not the ASKING command itself. */ 884 if (!(c->flags & REDIS_MULTI) && prevcmd != askingCommand) 885 c->flags &= (~REDIS_ASKING); 886 } 887 888 int processInlineBuffer(redisClient *c) { 889 char *newline; 890 int argc, j; 891 sds *argv, aux; 892 size_t querylen; 893 894 /* Search for end of line */ 895 newline = strchr(c->querybuf,'\n'); 896 897 /* Nothing to do without a \r\n */ 898 if (newline == NULL) { 899 if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { 900 addReplyError(c,"Protocol error: too big inline request"); 901 setProtocolError(c,0); 902 } 903 return REDIS_ERR; 904 } 905 906 /* Handle the \r\n case. */ 907 if (newline && newline != c->querybuf && *(newline-1) == '\r') 908 newline--; 909 910 /* Split the input buffer up to the \r\n */ 911 querylen = newline-(c->querybuf); 912 aux = sdsnewlen(c->querybuf,querylen); 913 argv = sdssplitargs(aux,&argc); 914 sdsfree(aux); 915 if (argv == NULL) { 916 addReplyError(c,"Protocol error: unbalanced quotes in request"); 917 setProtocolError(c,0); 918 return REDIS_ERR; 919 } 920 921 /* Newline from slaves can be used to refresh the last ACK time. 922 * This is useful for a slave to ping back while loading a big 923 * RDB file. */ 924 if (querylen == 0 && c->flags & REDIS_SLAVE) 925 c->repl_ack_time = server.unixtime; 926 927 /* Leave data after the first line of the query in the buffer */ 928 sdsrange(c->querybuf,querylen+2,-1); 929 930 /* Setup argv array on client structure */ 931 if (c->argv) zfree(c->argv); 932 c->argv = zmalloc(sizeof(robj*)*argc); 933 934 /* Create redis objects for all arguments. */ 935 for (c->argc = 0, j = 0; j < argc; j++) { 936 if (sdslen(argv[j])) { 937 c->argv[c->argc] = createObject(REDIS_STRING,argv[j]); 938 c->argc++; 939 } else { 940 sdsfree(argv[j]); 941 } 942 } 943 zfree(argv); 944 return REDIS_OK; 945 } 946 947 /* Helper function. Trims query buffer to make the function that processes 948 * multi bulk requests idempotent. */ 949 static void setProtocolError(redisClient *c, int pos) { 950 if (server.verbosity >= REDIS_VERBOSE) { 951 sds client = catClientInfoString(sdsempty(),c); 952 redisLog(REDIS_VERBOSE, 953 "Protocol error from client: %s", client); 954 sdsfree(client); 955 } 956 c->flags |= REDIS_CLOSE_AFTER_REPLY; 957 sdsrange(c->querybuf,pos,-1); 958 } 959 960 int processMultibulkBuffer(redisClient *c) { 961 char *newline = NULL; 962 int pos = 0, ok; 963 long long ll; 964 965 if (c->multibulklen == 0) { 966 /* The client should have been reset */ 967 redisAssertWithInfo(c,NULL,c->argc == 0); 968 969 /* Multi bulk length cannot be read without a \r\n */ 970 newline = strchr(c->querybuf,'\r'); 971 if (newline == NULL) { 972 if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { 973 addReplyError(c,"Protocol error: too big mbulk count string"); 974 setProtocolError(c,0); 975 } 976 return REDIS_ERR; 977 } 978 979 /* Buffer should also contain \n */ 980 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) 981 return REDIS_ERR; 982 983 /* We know for sure there is a whole line since newline != NULL, 984 * so go ahead and find out the multi bulk length. */ 985 redisAssertWithInfo(c,NULL,c->querybuf[0] == '*'); 986 ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); 987 if (!ok || ll > 1024*1024) { 988 addReplyError(c,"Protocol error: invalid multibulk length"); 989 setProtocolError(c,pos); 990 return REDIS_ERR; 991 } 992 993 pos = (newline-c->querybuf)+2; 994 if (ll <= 0) { 995 sdsrange(c->querybuf,pos,-1); 996 return REDIS_OK; 997 } 998 999 c->multibulklen = ll; 1000 1001 /* Setup argv array on client structure */ 1002 if (c->argv) zfree(c->argv); 1003 c->argv = zmalloc(sizeof(robj*)*c->multibulklen); 1004 } 1005 1006 redisAssertWithInfo(c,NULL,c->multibulklen > 0); 1007 while(c->multibulklen) { 1008 /* Read bulk length if unknown */ 1009 if (c->bulklen == -1) { 1010 newline = strchr(c->querybuf+pos,'\r'); 1011 if (newline == NULL) { 1012 if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) { 1013 addReplyError(c, 1014 "Protocol error: too big bulk count string"); 1015 setProtocolError(c,0); 1016 return REDIS_ERR; 1017 } 1018 break; 1019 } 1020 1021 /* Buffer should also contain \n */ 1022 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) 1023 break; 1024 1025 if (c->querybuf[pos] != '$') { 1026 addReplyErrorFormat(c, 1027 "Protocol error: expected '$', got '%c'", 1028 c->querybuf[pos]); 1029 setProtocolError(c,pos); 1030 return REDIS_ERR; 1031 } 1032 1033 ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); 1034 if (!ok || ll < 0 || ll > 512*1024*1024) { 1035 addReplyError(c,"Protocol error: invalid bulk length"); 1036 setProtocolError(c,pos); 1037 return REDIS_ERR; 1038 } 1039 1040 pos += newline-(c->querybuf+pos)+2; 1041 if (ll >= REDIS_MBULK_BIG_ARG) { 1042 size_t qblen; 1043 1044 /* If we are going to read a large object from network 1045 * try to make it likely that it will start at c->querybuf 1046 * boundary so that we can optimize object creation 1047 * avoiding a large copy of data. */ 1048 sdsrange(c->querybuf,pos,-1); 1049 pos = 0; 1050 qblen = sdslen(c->querybuf); 1051 /* Hint the sds library about the amount of bytes this string is 1052 * going to contain. */ 1053 if (qblen < ll+2) 1054 c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-qblen); 1055 } 1056 c->bulklen = ll; 1057 } 1058 1059 /* Read bulk argument */ 1060 if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) { 1061 /* Not enough data (+2 == trailing \r\n) */ 1062 break; 1063 } else { 1064 /* Optimization: if the buffer contains JUST our bulk element 1065 * instead of creating a new object by *copying* the sds we 1066 * just use the current sds string. */ 1067 if (pos == 0 && 1068 c->bulklen >= REDIS_MBULK_BIG_ARG && 1069 (signed) sdslen(c->querybuf) == c->bulklen+2) 1070 { 1071 c->argv[c->argc++] = createObject(REDIS_STRING,c->querybuf); 1072 sdsIncrLen(c->querybuf,-2); /* remove CRLF */ 1073 c->querybuf = sdsempty(); 1074 /* Assume that if we saw a fat argument we'll see another one 1075 * likely... */ 1076 c->querybuf = sdsMakeRoomFor(c->querybuf,c->bulklen+2); 1077 pos = 0; 1078 } else { 1079 c->argv[c->argc++] = 1080 createStringObject(c->querybuf+pos,c->bulklen); 1081 pos += c->bulklen+2; 1082 } 1083 c->bulklen = -1; 1084 c->multibulklen--; 1085 } 1086 } 1087 1088 /* Trim to pos */ 1089 if (pos) sdsrange(c->querybuf,pos,-1); 1090 1091 /* We're done when c->multibulk == 0 */ 1092 if (c->multibulklen == 0) return REDIS_OK; 1093 1094 /* Still not read to process the command */ 1095 return REDIS_ERR; 1096 } 1097 1098 void processInputBuffer(redisClient *c) { 1099 /* Keep processing while there is something in the input buffer */ 1100 while(sdslen(c->querybuf)) { 1101 /* Return if clients are paused. */ 1102 if (!(c->flags & REDIS_SLAVE) && clientsArePaused()) return; 1103 1104 /* Immediately abort if the client is in the middle of something. */ 1105 if (c->flags & REDIS_BLOCKED) return; 1106 1107 /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is 1108 * written to the client. Make sure to not let the reply grow after 1109 * this flag has been set (i.e. don't process more commands). */ 1110 if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; 1111 1112 /* Determine request type when unknown. */ 1113 if (!c->reqtype) { 1114 if (c->querybuf[0] == '*') { 1115 c->reqtype = REDIS_REQ_MULTIBULK; 1116 } else { 1117 c->reqtype = REDIS_REQ_INLINE; 1118 } 1119 } 1120 1121 if (c->reqtype == REDIS_REQ_INLINE) { 1122 if (processInlineBuffer(c) != REDIS_OK) break; 1123 } else if (c->reqtype == REDIS_REQ_MULTIBULK) { 1124 if (processMultibulkBuffer(c) != REDIS_OK) break; 1125 } else { 1126 redisPanic("Unknown request type"); 1127 } 1128 1129 /* Multibulk processing could see a <= 0 length. */ 1130 if (c->argc == 0) { 1131 resetClient(c); 1132 } else { 1133 /* Only reset the client when the command was executed. */ 1134 if (processCommand(c) == REDIS_OK) 1135 resetClient(c); 1136 } 1137 } 1138 } 1139 1140 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { 1141 redisClient *c = (redisClient*) privdata; 1142 int nread, readlen; 1143 size_t qblen; 1144 REDIS_NOTUSED(el); 1145 REDIS_NOTUSED(mask); 1146 1147 server.current_client = c; 1148 readlen = REDIS_IOBUF_LEN; 1149 /* If this is a multi bulk request, and we are processing a bulk reply 1150 * that is large enough, try to maximize the probability that the query 1151 * buffer contains exactly the SDS string representing the object, even 1152 * at the risk of requiring more read(2) calls. This way the function 1153 * processMultiBulkBuffer() can avoid copying buffers to create the 1154 * Redis Object representing the argument. */ 1155 if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 1156 && c->bulklen >= REDIS_MBULK_BIG_ARG) 1157 { 1158 int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); 1159 1160 if (remaining < readlen) readlen = remaining; 1161 } 1162 1163 qblen = sdslen(c->querybuf); 1164 if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; 1165 c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); 1166 nread = read(fd, c->querybuf+qblen, readlen); 1167 if (nread == -1) { 1168 if (errno == EAGAIN) { 1169 nread = 0; 1170 } else { 1171 redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno)); 1172 freeClient(c); 1173 return; 1174 } 1175 } else if (nread == 0) { 1176 redisLog(REDIS_VERBOSE, "Client closed connection"); 1177 freeClient(c); 1178 return; 1179 } 1180 if (nread) { 1181 sdsIncrLen(c->querybuf,nread); 1182 c->lastinteraction = server.unixtime; 1183 if (c->flags & REDIS_MASTER) c->reploff += nread; 1184 } else { 1185 server.current_client = NULL; 1186 return; 1187 } 1188 if (sdslen(c->querybuf) > server.client_max_querybuf_len) { 1189 sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); 1190 1191 bytes = sdscatrepr(bytes,c->querybuf,64); 1192 redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); 1193 sdsfree(ci); 1194 sdsfree(bytes); 1195 freeClient(c); 1196 return; 1197 } 1198 processInputBuffer(c); 1199 server.current_client = NULL; 1200 } 1201 1202 void getClientsMaxBuffers(unsigned long *longest_output_list, 1203 unsigned long *biggest_input_buffer) { 1204 redisClient *c; 1205 listNode *ln; 1206 listIter li; 1207 unsigned long lol = 0, bib = 0; 1208 1209 listRewind(server.clients,&li); 1210 while ((ln = listNext(&li)) != NULL) { 1211 c = listNodeValue(ln); 1212 1213 if (listLength(c->reply) > lol) lol = listLength(c->reply); 1214 if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf); 1215 } 1216 *longest_output_list = lol; 1217 *biggest_input_buffer = bib; 1218 } 1219 1220 /* This is a helper function for genClientPeerId(). 1221 * It writes the specified ip/port to "peerid" as a null termiated string 1222 * in the form ip:port if ip does not contain ":" itself, otherwise 1223 * [ip]:port format is used (for IPv6 addresses basically). */ 1224 void formatPeerId(char *peerid, size_t peerid_len, char *ip, int port) { 1225 if (strchr(ip,':')) 1226 snprintf(peerid,peerid_len,"[%s]:%d",ip,port); 1227 else 1228 snprintf(peerid,peerid_len,"%s:%d",ip,port); 1229 } 1230 1231 /* A Redis "Peer ID" is a colon separated ip:port pair. 1232 * For IPv4 it's in the form x.y.z.k:pork, example: "127.0.0.1:1234". 1233 * For IPv6 addresses we use [] around the IP part, like in "[::1]:1234". 1234 * For Unix socekts we use path:0, like in "/tmp/redis:0". 1235 * 1236 * A Peer ID always fits inside a buffer of REDIS_PEER_ID_LEN bytes, including 1237 * the null term. 1238 * 1239 * The function returns REDIS_OK on succcess, and REDIS_ERR on failure. 1240 * 1241 * On failure the function still populates 'peerid' with the "?:0" string 1242 * in case you want to relax error checking or need to display something 1243 * anyway (see anetPeerToString implementation for more info). */ 1244 int genClientPeerId(redisClient *client, char *peerid, size_t peerid_len) { 1245 char ip[REDIS_IP_STR_LEN]; 1246 int port; 1247 1248 if (client->flags & REDIS_UNIX_SOCKET) { 1249 /* Unix socket client. */ 1250 snprintf(peerid,peerid_len,"%s:0",server.unixsocket); 1251 return REDIS_OK; 1252 } else { 1253 /* TCP client. */ 1254 int retval = anetPeerToString(client->fd,ip,sizeof(ip),&port); 1255 formatPeerId(peerid,peerid_len,ip,port); 1256 return (retval == -1) ? REDIS_ERR : REDIS_OK; 1257 } 1258 } 1259 1260 /* This function returns the client peer id, by creating and caching it 1261 * if client->perrid is NULL, otherwise returning the cached value. 1262 * The Peer ID never changes during the life of the client, however it 1263 * is expensive to compute. */ 1264 char *getClientPeerId(redisClient *c) { 1265 char peerid[REDIS_PEER_ID_LEN]; 1266 1267 if (c->peerid == NULL) { 1268 genClientPeerId(c,peerid,sizeof(peerid)); 1269 c->peerid = sdsnew(peerid); 1270 } 1271 return c->peerid; 1272 } 1273 1274 /* Concatenate a string representing the state of a client in an human 1275 * readable format, into the sds string 's'. */ 1276 sds catClientInfoString(sds s, redisClient *client) { 1277 char flags[16], events[3], *p; 1278 int emask; 1279 1280 p = flags; 1281 if (client->flags & REDIS_SLAVE) { 1282 if (client->flags & REDIS_MONITOR) 1283 *p++ = 'O'; 1284 else 1285 *p++ = 'S'; 1286 } 1287 if (client->flags & REDIS_MASTER) *p++ = 'M'; 1288 if (client->flags & REDIS_MULTI) *p++ = 'x'; 1289 if (client->flags & REDIS_BLOCKED) *p++ = 'b'; 1290 if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd'; 1291 if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c'; 1292 if (client->flags & REDIS_UNBLOCKED) *p++ = 'u'; 1293 if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A'; 1294 if (client->flags & REDIS_UNIX_SOCKET) *p++ = 'U'; 1295 if (client->flags & REDIS_READONLY) *p++ = 'r'; 1296 if (p == flags) *p++ = 'N'; 1297 *p++ = '\0'; 1298 1299 emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd); 1300 p = events; 1301 if (emask & AE_READABLE) *p++ = 'r'; 1302 if (emask & AE_WRITABLE) *p++ = 'w'; 1303 *p = '\0'; 1304 return sdscatfmt(s, 1305 "addr=%s fd=%i name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U obl=%U oll=%U omem=%U events=%s cmd=%s", 1306 getClientPeerId(client), 1307 client->fd, 1308 client->name ? (char*)client->name->ptr : "", 1309 (long long)(server.unixtime - client->ctime), 1310 (long long)(server.unixtime - client->lastinteraction), 1311 flags, 1312 client->db->id, 1313 (int) dictSize(client->pubsub_channels), 1314 (int) listLength(client->pubsub_patterns), 1315 (client->flags & REDIS_MULTI) ? client->mstate.count : -1, 1316 (unsigned long long) sdslen(client->querybuf), 1317 (unsigned long long) sdsavail(client->querybuf), 1318 (unsigned long long) client->bufpos, 1319 (unsigned long long) listLength(client->reply), 1320 (unsigned long long) getClientOutputBufferMemoryUsage(client), 1321 events, 1322 client->lastcmd ? client->lastcmd->name : "NULL"); 1323 } 1324 1325 sds getAllClientsInfoString(void) { 1326 listNode *ln; 1327 listIter li; 1328 redisClient *client; 1329 sds o = sdsempty(); 1330 1331 o = sdsMakeRoomFor(o,200*listLength(server.clients)); 1332 listRewind(server.clients,&li); 1333 while ((ln = listNext(&li)) != NULL) { 1334 client = listNodeValue(ln); 1335 o = catClientInfoString(o,client); 1336 o = sdscatlen(o,"\n",1); 1337 } 1338 return o; 1339 } 1340 1341 void clientCommand(redisClient *c) { 1342 listNode *ln; 1343 listIter li; 1344 redisClient *client; 1345 1346 if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { 1347 sds o = getAllClientsInfoString(); 1348 addReplyBulkCBuffer(c,o,sdslen(o)); 1349 sdsfree(o); 1350 } else if (!strcasecmp(c->argv[1]->ptr,"kill") && c->argc == 3) { 1351 listRewind(server.clients,&li); 1352 while ((ln = listNext(&li)) != NULL) { 1353 char *peerid; 1354 1355 client = listNodeValue(ln); 1356 peerid = getClientPeerId(client); 1357 if (strcmp(peerid,c->argv[2]->ptr) == 0) { 1358 addReply(c,shared.ok); 1359 if (c == client) { 1360 client->flags |= REDIS_CLOSE_AFTER_REPLY; 1361 } else { 1362 freeClient(client); 1363 } 1364 return; 1365 } 1366 } 1367 addReplyError(c,"No such client"); 1368 } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) { 1369 int j, len = sdslen(c->argv[2]->ptr); 1370 char *p = c->argv[2]->ptr; 1371 1372 /* Setting the client name to an empty string actually removes 1373 * the current name. */ 1374 if (len == 0) { 1375 if (c->name) decrRefCount(c->name); 1376 c->name = NULL; 1377 addReply(c,shared.ok); 1378 return; 1379 } 1380 1381 /* Otherwise check if the charset is ok. We need to do this otherwise 1382 * CLIENT LIST format will break. You should always be able to 1383 * split by space to get the different fields. */ 1384 for (j = 0; j < len; j++) { 1385 if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */ 1386 addReplyError(c, 1387 "Client names cannot contain spaces, " 1388 "newlines or special characters."); 1389 return; 1390 } 1391 } 1392 if (c->name) decrRefCount(c->name); 1393 c->name = c->argv[2]; 1394 incrRefCount(c->name); 1395 addReply(c,shared.ok); 1396 } else if (!strcasecmp(c->argv[1]->ptr,"getname") && c->argc == 2) { 1397 if (c->name) 1398 addReplyBulk(c,c->name); 1399 else 1400 addReply(c,shared.nullbulk); 1401 } else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) { 1402 long long duration; 1403 1404 if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration,UNIT_MILLISECONDS) 1405 != REDIS_OK) return; 1406 pauseClients(duration); 1407 addReply(c,shared.ok); 1408 } else { 1409 addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port | GETNAME | SETNAME connection-name)"); 1410 } 1411 } 1412 1413 /* Rewrite the command vector of the client. All the new objects ref count 1414 * is incremented. The old command vector is freed, and the old objects 1415 * ref count is decremented. */ 1416 void rewriteClientCommandVector(redisClient *c, int argc, ...) { 1417 va_list ap; 1418 int j; 1419 robj **argv; /* The new argument vector */ 1420 1421 argv = zmalloc(sizeof(robj*)*argc); 1422 va_start(ap,argc); 1423 for (j = 0; j < argc; j++) { 1424 robj *a; 1425 1426 a = va_arg(ap, robj*); 1427 argv[j] = a; 1428 incrRefCount(a); 1429 } 1430 /* We free the objects in the original vector at the end, so we are 1431 * sure that if the same objects are reused in the new vector the 1432 * refcount gets incremented before it gets decremented. */ 1433 for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]); 1434 zfree(c->argv); 1435 /* Replace argv and argc with our new versions. */ 1436 c->argv = argv; 1437 c->argc = argc; 1438 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); 1439 redisAssertWithInfo(c,NULL,c->cmd != NULL); 1440 va_end(ap); 1441 } 1442 1443 /* Rewrite a single item in the command vector. 1444 * The new val ref count is incremented, and the old decremented. */ 1445 void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) { 1446 robj *oldval; 1447 1448 redisAssertWithInfo(c,NULL,i < c->argc); 1449 oldval = c->argv[i]; 1450 c->argv[i] = newval; 1451 incrRefCount(newval); 1452 decrRefCount(oldval); 1453 1454 /* If this is the command name make sure to fix c->cmd. */ 1455 if (i == 0) { 1456 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); 1457 redisAssertWithInfo(c,NULL,c->cmd != NULL); 1458 } 1459 } 1460 1461 /* This function returns the number of bytes that Redis is virtually 1462 * using to store the reply still not read by the client. 1463 * It is "virtual" since the reply output list may contain objects that 1464 * are shared and are not really using additional memory. 1465 * 1466 * The function returns the total sum of the length of all the objects 1467 * stored in the output list, plus the memory used to allocate every 1468 * list node. The static reply buffer is not taken into account since it 1469 * is allocated anyway. 1470 * 1471 * Note: this function is very fast so can be called as many time as 1472 * the caller wishes. The main usage of this function currently is 1473 * enforcing the client output length limits. */ 1474 unsigned long getClientOutputBufferMemoryUsage(redisClient *c) { 1475 unsigned long list_item_size = sizeof(listNode)+sizeof(robj); 1476 1477 return c->reply_bytes + (list_item_size*listLength(c->reply)); 1478 } 1479 1480 /* Get the class of a client, used in order to enforce limits to different 1481 * classes of clients. 1482 * 1483 * The function will return one of the following: 1484 * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client 1485 * REDIS_CLIENT_LIMIT_CLASS_SLAVE -> Slave or client executing MONITOR command 1486 * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels 1487 */ 1488 int getClientLimitClass(redisClient *c) { 1489 if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; 1490 if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns)) 1491 return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; 1492 return REDIS_CLIENT_LIMIT_CLASS_NORMAL; 1493 } 1494 1495 int getClientLimitClassByName(char *name) { 1496 if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL; 1497 else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; 1498 else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; 1499 else return -1; 1500 } 1501 1502 char *getClientLimitClassName(int class) { 1503 switch(class) { 1504 case REDIS_CLIENT_LIMIT_CLASS_NORMAL: return "normal"; 1505 case REDIS_CLIENT_LIMIT_CLASS_SLAVE: return "slave"; 1506 case REDIS_CLIENT_LIMIT_CLASS_PUBSUB: return "pubsub"; 1507 default: return NULL; 1508 } 1509 } 1510 1511 /* The function checks if the client reached output buffer soft or hard 1512 * limit, and also update the state needed to check the soft limit as 1513 * a side effect. 1514 * 1515 * Return value: non-zero if the client reached the soft or the hard limit. 1516 * Otherwise zero is returned. */ 1517 int checkClientOutputBufferLimits(redisClient *c) { 1518 int soft = 0, hard = 0, class; 1519 unsigned long used_mem = getClientOutputBufferMemoryUsage(c); 1520 1521 class = getClientLimitClass(c); 1522 if (server.client_obuf_limits[class].hard_limit_bytes && 1523 used_mem >= server.client_obuf_limits[class].hard_limit_bytes) 1524 hard = 1; 1525 if (server.client_obuf_limits[class].soft_limit_bytes && 1526 used_mem >= server.client_obuf_limits[class].soft_limit_bytes) 1527 soft = 1; 1528 1529 /* We need to check if the soft limit is reached continuously for the 1530 * specified amount of seconds. */ 1531 if (soft) { 1532 if (c->obuf_soft_limit_reached_time == 0) { 1533 c->obuf_soft_limit_reached_time = server.unixtime; 1534 soft = 0; /* First time we see the soft limit reached */ 1535 } else { 1536 time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time; 1537 1538 if (elapsed <= 1539 server.client_obuf_limits[class].soft_limit_seconds) { 1540 soft = 0; /* The client still did not reached the max number of 1541 seconds for the soft limit to be considered 1542 reached. */ 1543 } 1544 } 1545 } else { 1546 c->obuf_soft_limit_reached_time = 0; 1547 } 1548 return soft || hard; 1549 } 1550 1551 /* Asynchronously close a client if soft or hard limit is reached on the 1552 * output buffer size. The caller can check if the client will be closed 1553 * checking if the client REDIS_CLOSE_ASAP flag is set. 1554 * 1555 * Note: we need to close the client asynchronously because this function is 1556 * called from contexts where the client can't be freed safely, i.e. from the 1557 * lower level functions pushing data inside the client output buffers. */ 1558 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) { 1559 redisAssert(c->reply_bytes < ULONG_MAX-(1024*64)); 1560 if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return; 1561 if (checkClientOutputBufferLimits(c)) { 1562 sds client = catClientInfoString(sdsempty(),c); 1563 1564 freeClientAsync(c); 1565 redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); 1566 sdsfree(client); 1567 } 1568 } 1569 1570 /* Helper function used by freeMemoryIfNeeded() in order to flush slaves 1571 * output buffers without returning control to the event loop. */ 1572 void flushSlavesOutputBuffers(void) { 1573 listIter li; 1574 listNode *ln; 1575 1576 listRewind(server.slaves,&li); 1577 while((ln = listNext(&li))) { 1578 redisClient *slave = listNodeValue(ln); 1579 int events; 1580 1581 events = aeGetFileEvents(server.el,slave->fd); 1582 if (events & AE_WRITABLE && 1583 slave->replstate == REDIS_REPL_ONLINE && 1584 listLength(slave->reply)) 1585 { 1586 sendReplyToClient(server.el,slave->fd,slave,0); 1587 } 1588 } 1589 } 1590 1591 /* Pause clients up to the specified unixtime (in ms). While clients 1592 * are paused no command is processed from clients, so the data set can't 1593 * change during that time. 1594 * 1595 * However while this function pauses normal and Pub/Sub clients, slaves are 1596 * still served, so this function can be used on server upgrades where it is 1597 * required that slaves process the latest bytes from the replication stream 1598 * before being turned to masters. 1599 * 1600 * This function is also internally used by Redis Cluster for the manual 1601 * failover procedure implemented by CLUSTER FAILOVER. 1602 * 1603 * The function always succeed, even if there is already a pause in progress. 1604 * In such a case, the pause is extended if the duration is more than the 1605 * time left for the previous duration. However if the duration is smaller 1606 * than the time left for the previous pause, no change is made to the 1607 * left duration. */ 1608 void pauseClients(mstime_t end) { 1609 if (!server.clients_paused || end > server.clients_pause_end_time) 1610 server.clients_pause_end_time = end; 1611 server.clients_paused = 1; 1612 } 1613 1614 /* Return non-zero if clients are currently paused. As a side effect the 1615 * function checks if the pause time was reached and clear it. */ 1616 int clientsArePaused(void) { 1617 if (server.clients_paused && server.clients_pause_end_time < server.mstime) { 1618 listNode *ln; 1619 listIter li; 1620 redisClient *c; 1621 1622 server.clients_paused = 0; 1623 1624 /* Put all the clients in the unblocked clients queue in order to 1625 * force the re-processing of the input buffer if any. */ 1626 listRewind(server.clients,&li); 1627 while ((ln = listNext(&li)) != NULL) { 1628 c = listNodeValue(ln); 1629 1630 if (c->flags & REDIS_SLAVE) continue; 1631 listAddNodeTail(server.unblocked_clients,c); 1632 } 1633 } 1634 return server.clients_paused; 1635 } 1636 1637 /* This function is called by Redis in order to process a few events from 1638 * time to time while blocked into some not interruptible operation. 1639 * This allows to reply to clients with the -LOADING error while loading the 1640 * data set at startup or after a full resynchronization with the master 1641 * and so forth. 1642 * 1643 * It calls the event loop in order to process a few events. Specifically we 1644 * try to call the event loop for times as long as we receive acknowledge that 1645 * some event was processed, in order to go forward with the accept, read, 1646 * write, close sequence needed to serve a client. 1647 * 1648 * The function returns the total number of events processed. */ 1649 int processEventsWhileBlocked(void) { 1650 int iterations = 4; /* See the function top-comment. */ 1651 int count = 0; 1652 while (iterations--) { 1653 int events = aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); 1654 if (!events) break; 1655 count += events; 1656 } 1657 return count; 1658 } 1659