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