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