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