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 double as a bulk reply, but uses a human readable formatting 483 * of the double instead of exposing the crude behavior of doubles to the 484 * dear user. */ 485 void addReplyHumanLongDouble(client *c, long double d) { 486 robj *o = createStringObjectFromLongDouble(d,1); 487 addReplyBulk(c,o); 488 decrRefCount(o); 489 } 490 491 /* Add a long long as integer reply or bulk len / multi bulk count. 492 * Basically this is used to output <prefix><long long><crlf>. */ 493 void addReplyLongLongWithPrefix(client *c, long long ll, char prefix) { 494 char buf[128]; 495 int len; 496 497 /* Things like $3\r\n or *2\r\n are emitted very often by the protocol 498 * so we have a few shared objects to use if the integer is small 499 * like it is most of the times. */ 500 if (prefix == '*' && ll < OBJ_SHARED_BULKHDR_LEN && ll >= 0) { 501 addReply(c,shared.mbulkhdr[ll]); 502 return; 503 } else if (prefix == '$' && ll < OBJ_SHARED_BULKHDR_LEN && ll >= 0) { 504 addReply(c,shared.bulkhdr[ll]); 505 return; 506 } 507 508 buf[0] = prefix; 509 len = ll2string(buf+1,sizeof(buf)-1,ll); 510 buf[len+1] = '\r'; 511 buf[len+2] = '\n'; 512 addReplyString(c,buf,len+3); 513 } 514 515 void addReplyLongLong(client *c, long long ll) { 516 if (ll == 0) 517 addReply(c,shared.czero); 518 else if (ll == 1) 519 addReply(c,shared.cone); 520 else 521 addReplyLongLongWithPrefix(c,ll,':'); 522 } 523 524 void addReplyMultiBulkLen(client *c, long length) { 525 if (length < OBJ_SHARED_BULKHDR_LEN) 526 addReply(c,shared.mbulkhdr[length]); 527 else 528 addReplyLongLongWithPrefix(c,length,'*'); 529 } 530 531 /* Create the length prefix of a bulk reply, example: $2234 */ 532 void addReplyBulkLen(client *c, robj *obj) { 533 size_t len; 534 535 if (sdsEncodedObject(obj)) { 536 len = sdslen(obj->ptr); 537 } else { 538 long n = (long)obj->ptr; 539 540 /* Compute how many bytes will take this integer as a radix 10 string */ 541 len = 1; 542 if (n < 0) { 543 len++; 544 n = -n; 545 } 546 while((n = n/10) != 0) { 547 len++; 548 } 549 } 550 551 if (len < OBJ_SHARED_BULKHDR_LEN) 552 addReply(c,shared.bulkhdr[len]); 553 else 554 addReplyLongLongWithPrefix(c,len,'$'); 555 } 556 557 /* Add a Redis Object as a bulk reply */ 558 void addReplyBulk(client *c, robj *obj) { 559 addReplyBulkLen(c,obj); 560 addReply(c,obj); 561 addReply(c,shared.crlf); 562 } 563 564 /* Add a C buffer as bulk reply */ 565 void addReplyBulkCBuffer(client *c, const void *p, size_t len) { 566 addReplyLongLongWithPrefix(c,len,'$'); 567 addReplyString(c,p,len); 568 addReply(c,shared.crlf); 569 } 570 571 /* Add sds to reply (takes ownership of sds and frees it) */ 572 void addReplyBulkSds(client *c, sds s) { 573 addReplySds(c,sdscatfmt(sdsempty(),"$%u\r\n", 574 (unsigned long)sdslen(s))); 575 addReplySds(c,s); 576 addReply(c,shared.crlf); 577 } 578 579 /* Add a C nul term string as bulk reply */ 580 void addReplyBulkCString(client *c, const char *s) { 581 if (s == NULL) { 582 addReply(c,shared.nullbulk); 583 } else { 584 addReplyBulkCBuffer(c,s,strlen(s)); 585 } 586 } 587 588 /* Add a long long as a bulk reply */ 589 void addReplyBulkLongLong(client *c, long long ll) { 590 char buf[64]; 591 int len; 592 593 len = ll2string(buf,64,ll); 594 addReplyBulkCBuffer(c,buf,len); 595 } 596 597 /* Copy 'src' client output buffers into 'dst' client output buffers. 598 * The function takes care of freeing the old output buffers of the 599 * destination client. */ 600 void copyClientOutputBuffer(client *dst, client *src) { 601 listRelease(dst->reply); 602 dst->reply = listDup(src->reply); 603 memcpy(dst->buf,src->buf,src->bufpos); 604 dst->bufpos = src->bufpos; 605 dst->reply_bytes = src->reply_bytes; 606 } 607 608 /* Return true if the specified client has pending reply buffers to write to 609 * the socket. */ 610 int clientHasPendingReplies(client *c) { 611 return c->bufpos || listLength(c->reply); 612 } 613 614 #define MAX_ACCEPTS_PER_CALL 1000 615 static void acceptCommonHandler(int fd, int flags, char *ip) { 616 client *c; 617 if ((c = createClient(fd)) == NULL) { 618 serverLog(LL_WARNING, 619 "Error registering fd event for the new client: %s (fd=%d)", 620 strerror(errno),fd); 621 close(fd); /* May be already closed, just ignore errors */ 622 return; 623 } 624 /* If maxclient directive is set and this is one client more... close the 625 * connection. Note that we create the client instead to check before 626 * for this condition, since now the socket is already set in non-blocking 627 * mode and we can send an error for free using the Kernel I/O */ 628 if (listLength(server.clients) > server.maxclients) { 629 char *err = "-ERR max number of clients reached\r\n"; 630 631 /* That's a best effort error message, don't check write errors */ 632 if (write(c->fd,err,strlen(err)) == -1) { 633 /* Nothing to do, Just to avoid the warning... */ 634 } 635 server.stat_rejected_conn++; 636 freeClient(c); 637 return; 638 } 639 640 /* If the server is running in protected mode (the default) and there 641 * is no password set, nor a specific interface is bound, we don't accept 642 * requests from non loopback interfaces. Instead we try to explain the 643 * user what to do to fix it if needed. */ 644 if (server.protected_mode && 645 server.bindaddr_count == 0 && 646 server.requirepass == NULL && 647 !(flags & CLIENT_UNIX_SOCKET) && 648 ip != NULL) 649 { 650 if (strcmp(ip,"127.0.0.1") && strcmp(ip,"::1")) { 651 char *err = 652 "-DENIED Redis is running in protected mode because protected " 653 "mode is enabled, no bind address was specified, no " 654 "authentication password is requested to clients. In this mode " 655 "connections are only accepted from the loopback interface. " 656 "If you want to connect from external computers to Redis you " 657 "may adopt one of the following solutions: " 658 "1) Just disable protected mode sending the command " 659 "'CONFIG SET protected-mode no' from the loopback interface " 660 "by connecting to Redis from the same host the server is " 661 "running, however MAKE SURE Redis is not publicly accessible " 662 "from internet if you do so. Use CONFIG REWRITE to make this " 663 "change permanent. " 664 "2) Alternatively you can just disable the protected mode by " 665 "editing the Redis configuration file, and setting the protected " 666 "mode option to 'no', and then restarting the server. " 667 "3) If you started the server manually just for testing, restart " 668 "it with the '--protected-mode no' option. " 669 "4) Setup a bind address or an authentication password. " 670 "NOTE: You only need to do one of the above things in order for " 671 "the server to start accepting connections from the outside.\r\n"; 672 if (write(c->fd,err,strlen(err)) == -1) { 673 /* Nothing to do, Just to avoid the warning... */ 674 } 675 server.stat_rejected_conn++; 676 freeClient(c); 677 return; 678 } 679 } 680 681 server.stat_numconnections++; 682 c->flags |= flags; 683 } 684 685 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 686 int cport, cfd, max = MAX_ACCEPTS_PER_CALL; 687 char cip[NET_IP_STR_LEN]; 688 UNUSED(el); 689 UNUSED(mask); 690 UNUSED(privdata); 691 692 while(max--) { 693 cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); 694 if (cfd == ANET_ERR) { 695 if (errno != EWOULDBLOCK) 696 serverLog(LL_WARNING, 697 "Accepting client connection: %s", server.neterr); 698 return; 699 } 700 serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport); 701 acceptCommonHandler(cfd,0,cip); 702 } 703 } 704 705 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 706 int cfd, max = MAX_ACCEPTS_PER_CALL; 707 UNUSED(el); 708 UNUSED(mask); 709 UNUSED(privdata); 710 711 while(max--) { 712 cfd = anetUnixAccept(server.neterr, fd); 713 if (cfd == ANET_ERR) { 714 if (errno != EWOULDBLOCK) 715 serverLog(LL_WARNING, 716 "Accepting client connection: %s", server.neterr); 717 return; 718 } 719 serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket); 720 acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL); 721 } 722 } 723 724 static void freeClientArgv(client *c) { 725 int j; 726 for (j = 0; j < c->argc; j++) 727 decrRefCount(c->argv[j]); 728 c->argc = 0; 729 c->cmd = NULL; 730 } 731 732 /* Close all the slaves connections. This is useful in chained replication 733 * when we resync with our own master and want to force all our slaves to 734 * resync with us as well. */ 735 void disconnectSlaves(void) { 736 while (listLength(server.slaves)) { 737 listNode *ln = listFirst(server.slaves); 738 freeClient((client*)ln->value); 739 } 740 } 741 742 /* Remove the specified client from global lists where the client could 743 * be referenced, not including the Pub/Sub channels. 744 * This is used by freeClient() and replicationCacheMaster(). */ 745 void unlinkClient(client *c) { 746 listNode *ln; 747 748 /* If this is marked as current client unset it. */ 749 if (server.current_client == c) server.current_client = NULL; 750 751 /* Certain operations must be done only if the client has an active socket. 752 * If the client was already unlinked or if it's a "fake client" the 753 * fd is already set to -1. */ 754 if (c->fd != -1) { 755 /* Remove from the list of active clients. */ 756 ln = listSearchKey(server.clients,c); 757 serverAssert(ln != NULL); 758 listDelNode(server.clients,ln); 759 760 /* Unregister async I/O handlers and close the socket. */ 761 aeDeleteFileEvent(server.el,c->fd,AE_READABLE); 762 aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); 763 close(c->fd); 764 c->fd = -1; 765 } 766 767 /* Remove from the list of pending writes if needed. */ 768 if (c->flags & CLIENT_PENDING_WRITE) { 769 ln = listSearchKey(server.clients_pending_write,c); 770 serverAssert(ln != NULL); 771 listDelNode(server.clients_pending_write,ln); 772 c->flags &= ~CLIENT_PENDING_WRITE; 773 } 774 775 /* When client was just unblocked because of a blocking operation, 776 * remove it from the list of unblocked clients. */ 777 if (c->flags & CLIENT_UNBLOCKED) { 778 ln = listSearchKey(server.unblocked_clients,c); 779 serverAssert(ln != NULL); 780 listDelNode(server.unblocked_clients,ln); 781 c->flags &= ~CLIENT_UNBLOCKED; 782 } 783 } 784 785 void freeClient(client *c) { 786 listNode *ln; 787 788 /* If it is our master that's beging disconnected we should make sure 789 * to cache the state to try a partial resynchronization later. 790 * 791 * Note that before doing this we make sure that the client is not in 792 * some unexpected state, by checking its flags. */ 793 if (server.master && c->flags & CLIENT_MASTER) { 794 serverLog(LL_WARNING,"Connection with master lost."); 795 if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY| 796 CLIENT_CLOSE_ASAP| 797 CLIENT_BLOCKED| 798 CLIENT_UNBLOCKED))) 799 { 800 replicationCacheMaster(c); 801 return; 802 } 803 } 804 805 /* Log link disconnection with slave */ 806 if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) { 807 serverLog(LL_WARNING,"Connection with slave %s lost.", 808 replicationGetSlaveName(c)); 809 } 810 811 /* Free the query buffer */ 812 sdsfree(c->querybuf); 813 c->querybuf = NULL; 814 815 /* Deallocate structures used to block on blocking ops. */ 816 if (c->flags & CLIENT_BLOCKED) unblockClient(c); 817 dictRelease(c->bpop.keys); 818 819 /* UNWATCH all the keys */ 820 unwatchAllKeys(c); 821 listRelease(c->watched_keys); 822 823 /* Unsubscribe from all the pubsub channels */ 824 pubsubUnsubscribeAllChannels(c,0); 825 pubsubUnsubscribeAllPatterns(c,0); 826 dictRelease(c->pubsub_channels); 827 listRelease(c->pubsub_patterns); 828 829 /* Free data structures. */ 830 listRelease(c->reply); 831 freeClientArgv(c); 832 833 /* Unlink the client: this will close the socket, remove the I/O 834 * handlers, and remove references of the client from different 835 * places where active clients may be referenced. */ 836 unlinkClient(c); 837 838 /* Master/slave cleanup Case 1: 839 * we lost the connection with a slave. */ 840 if (c->flags & CLIENT_SLAVE) { 841 if (c->replstate == SLAVE_STATE_SEND_BULK) { 842 if (c->repldbfd != -1) close(c->repldbfd); 843 if (c->replpreamble) sdsfree(c->replpreamble); 844 } 845 list *l = (c->flags & CLIENT_MONITOR) ? server.monitors : server.slaves; 846 ln = listSearchKey(l,c); 847 serverAssert(ln != NULL); 848 listDelNode(l,ln); 849 /* We need to remember the time when we started to have zero 850 * attached slaves, as after some time we'll free the replication 851 * backlog. */ 852 if (c->flags & CLIENT_SLAVE && listLength(server.slaves) == 0) 853 server.repl_no_slaves_since = server.unixtime; 854 refreshGoodSlavesCount(); 855 } 856 857 /* Master/slave cleanup Case 2: 858 * we lost the connection with the master. */ 859 if (c->flags & CLIENT_MASTER) replicationHandleMasterDisconnection(); 860 861 /* If this client was scheduled for async freeing we need to remove it 862 * from the queue. */ 863 if (c->flags & CLIENT_CLOSE_ASAP) { 864 ln = listSearchKey(server.clients_to_close,c); 865 serverAssert(ln != NULL); 866 listDelNode(server.clients_to_close,ln); 867 } 868 869 /* Release other dynamically allocated client structure fields, 870 * and finally release the client structure itself. */ 871 if (c->name) decrRefCount(c->name); 872 zfree(c->argv); 873 freeClientMultiState(c); 874 sdsfree(c->peerid); 875 zfree(c); 876 } 877 878 /* Schedule a client to free it at a safe time in the serverCron() function. 879 * This function is useful when we need to terminate a client but we are in 880 * a context where calling freeClient() is not possible, because the client 881 * should be valid for the continuation of the flow of the program. */ 882 void freeClientAsync(client *c) { 883 if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return; 884 c->flags |= CLIENT_CLOSE_ASAP; 885 listAddNodeTail(server.clients_to_close,c); 886 } 887 888 void freeClientsInAsyncFreeQueue(void) { 889 while (listLength(server.clients_to_close)) { 890 listNode *ln = listFirst(server.clients_to_close); 891 client *c = listNodeValue(ln); 892 893 c->flags &= ~CLIENT_CLOSE_ASAP; 894 freeClient(c); 895 listDelNode(server.clients_to_close,ln); 896 } 897 } 898 899 /* Write data in output buffers to client. Return C_OK if the client 900 * is still valid after the call, C_ERR if it was freed. */ 901 int writeToClient(int fd, client *c, int handler_installed) { 902 ssize_t nwritten = 0, totwritten = 0; 903 size_t objlen; 904 size_t objmem; 905 robj *o; 906 907 while(clientHasPendingReplies(c)) { 908 if (c->bufpos > 0) { 909 nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); 910 if (nwritten <= 0) break; 911 c->sentlen += nwritten; 912 totwritten += nwritten; 913 914 /* If the buffer was sent, set bufpos to zero to continue with 915 * the remainder of the reply. */ 916 if ((int)c->sentlen == c->bufpos) { 917 c->bufpos = 0; 918 c->sentlen = 0; 919 } 920 } else { 921 o = listNodeValue(listFirst(c->reply)); 922 objlen = sdslen(o->ptr); 923 objmem = getStringObjectSdsUsedMemory(o); 924 925 if (objlen == 0) { 926 listDelNode(c->reply,listFirst(c->reply)); 927 c->reply_bytes -= objmem; 928 continue; 929 } 930 931 nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen); 932 if (nwritten <= 0) break; 933 c->sentlen += nwritten; 934 totwritten += nwritten; 935 936 /* If we fully sent the object on head go to the next one */ 937 if (c->sentlen == objlen) { 938 listDelNode(c->reply,listFirst(c->reply)); 939 c->sentlen = 0; 940 c->reply_bytes -= objmem; 941 } 942 } 943 /* Note that we avoid to send more than NET_MAX_WRITES_PER_EVENT 944 * bytes, in a single threaded server it's a good idea to serve 945 * other clients as well, even if a very large request comes from 946 * super fast link that is always able to accept data (in real world 947 * scenario think about 'KEYS *' against the loopback interface). 948 * 949 * However if we are over the maxmemory limit we ignore that and 950 * just deliver as much data as it is possible to deliver. */ 951 server.stat_net_output_bytes += totwritten; 952 if (totwritten > NET_MAX_WRITES_PER_EVENT && 953 (server.maxmemory == 0 || 954 zmalloc_used_memory() < server.maxmemory)) break; 955 } 956 if (nwritten == -1) { 957 if (errno == EAGAIN) { 958 nwritten = 0; 959 } else { 960 serverLog(LL_VERBOSE, 961 "Error writing to client: %s", strerror(errno)); 962 freeClient(c); 963 return C_ERR; 964 } 965 } 966 if (totwritten > 0) { 967 /* For clients representing masters we don't count sending data 968 * as an interaction, since we always send REPLCONF ACK commands 969 * that take some time to just fill the socket output buffer. 970 * We just rely on data / pings received for timeout detection. */ 971 if (!(c->flags & CLIENT_MASTER)) c->lastinteraction = server.unixtime; 972 } 973 if (!clientHasPendingReplies(c)) { 974 c->sentlen = 0; 975 if (handler_installed) aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); 976 977 /* Close connection after entire reply has been sent. */ 978 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) { 979 freeClient(c); 980 return C_ERR; 981 } 982 } 983 return C_OK; 984 } 985 986 /* Write event handler. Just send data to the client. */ 987 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { 988 UNUSED(el); 989 UNUSED(mask); 990 writeToClient(fd,privdata,1); 991 } 992 993 /* This function is called just before entering the event loop, in the hope 994 * we can just write the replies to the client output buffer without any 995 * need to use a syscall in order to install the writable event handler, 996 * get it called, and so forth. */ 997 int handleClientsWithPendingWrites(void) { 998 listIter li; 999 listNode *ln; 1000 int processed = listLength(server.clients_pending_write); 1001 1002 listRewind(server.clients_pending_write,&li); 1003 while((ln = listNext(&li))) { 1004 client *c = listNodeValue(ln); 1005 c->flags &= ~CLIENT_PENDING_WRITE; 1006 listDelNode(server.clients_pending_write,ln); 1007 1008 /* Try to write buffers to the client socket. */ 1009 if (writeToClient(c->fd,c,0) == C_ERR) continue; 1010 1011 /* If there is nothing left, do nothing. Otherwise install 1012 * the write handler. */ 1013 if (clientHasPendingReplies(c) && 1014 aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, 1015 sendReplyToClient, c) == AE_ERR) 1016 { 1017 freeClientAsync(c); 1018 } 1019 } 1020 return processed; 1021 } 1022 1023 /* resetClient prepare the client to process the next command */ 1024 void resetClient(client *c) { 1025 redisCommandProc *prevcmd = c->cmd ? c->cmd->proc : NULL; 1026 1027 freeClientArgv(c); 1028 c->reqtype = 0; 1029 c->multibulklen = 0; 1030 c->bulklen = -1; 1031 1032 /* We clear the ASKING flag as well if we are not inside a MULTI, and 1033 * if what we just executed is not the ASKING command itself. */ 1034 if (!(c->flags & CLIENT_MULTI) && prevcmd != askingCommand) 1035 c->flags &= ~CLIENT_ASKING; 1036 1037 /* Remove the CLIENT_REPLY_SKIP flag if any so that the reply 1038 * to the next command will be sent, but set the flag if the command 1039 * we just processed was "CLIENT REPLY SKIP". */ 1040 c->flags &= ~CLIENT_REPLY_SKIP; 1041 if (c->flags & CLIENT_REPLY_SKIP_NEXT) { 1042 c->flags |= CLIENT_REPLY_SKIP; 1043 c->flags &= ~CLIENT_REPLY_SKIP_NEXT; 1044 } 1045 } 1046 1047 int processInlineBuffer(client *c) { 1048 char *newline; 1049 int argc, j; 1050 sds *argv, aux; 1051 size_t querylen; 1052 1053 /* Search for end of line */ 1054 newline = strchr(c->querybuf,'\n'); 1055 1056 /* Nothing to do without a \r\n */ 1057 if (newline == NULL) { 1058 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { 1059 addReplyError(c,"Protocol error: too big inline request"); 1060 setProtocolError(c,0); 1061 } 1062 return C_ERR; 1063 } 1064 1065 /* Handle the \r\n case. */ 1066 if (newline && newline != c->querybuf && *(newline-1) == '\r') 1067 newline--; 1068 1069 /* Split the input buffer up to the \r\n */ 1070 querylen = newline-(c->querybuf); 1071 aux = sdsnewlen(c->querybuf,querylen); 1072 argv = sdssplitargs(aux,&argc); 1073 sdsfree(aux); 1074 if (argv == NULL) { 1075 addReplyError(c,"Protocol error: unbalanced quotes in request"); 1076 setProtocolError(c,0); 1077 return C_ERR; 1078 } 1079 1080 /* Newline from slaves can be used to refresh the last ACK time. 1081 * This is useful for a slave to ping back while loading a big 1082 * RDB file. */ 1083 if (querylen == 0 && c->flags & CLIENT_SLAVE) 1084 c->repl_ack_time = server.unixtime; 1085 1086 /* Leave data after the first line of the query in the buffer */ 1087 sdsrange(c->querybuf,querylen+2,-1); 1088 1089 /* Setup argv array on client structure */ 1090 if (argc) { 1091 if (c->argv) zfree(c->argv); 1092 c->argv = zmalloc(sizeof(robj*)*argc); 1093 } 1094 1095 /* Create redis objects for all arguments. */ 1096 for (c->argc = 0, j = 0; j < argc; j++) { 1097 if (sdslen(argv[j])) { 1098 c->argv[c->argc] = createObject(OBJ_STRING,argv[j]); 1099 c->argc++; 1100 } else { 1101 sdsfree(argv[j]); 1102 } 1103 } 1104 zfree(argv); 1105 return C_OK; 1106 } 1107 1108 /* Helper function. Trims query buffer to make the function that processes 1109 * multi bulk requests idempotent. */ 1110 static void setProtocolError(client *c, int pos) { 1111 if (server.verbosity <= LL_VERBOSE) { 1112 sds client = catClientInfoString(sdsempty(),c); 1113 serverLog(LL_VERBOSE, 1114 "Protocol error from client: %s", client); 1115 sdsfree(client); 1116 } 1117 c->flags |= CLIENT_CLOSE_AFTER_REPLY; 1118 sdsrange(c->querybuf,pos,-1); 1119 } 1120 1121 int processMultibulkBuffer(client *c) { 1122 char *newline = NULL; 1123 int pos = 0, ok; 1124 long long ll; 1125 1126 if (c->multibulklen == 0) { 1127 /* The client should have been reset */ 1128 serverAssertWithInfo(c,NULL,c->argc == 0); 1129 1130 /* Multi bulk length cannot be read without a \r\n */ 1131 newline = strchr(c->querybuf,'\r'); 1132 if (newline == NULL) { 1133 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { 1134 addReplyError(c,"Protocol error: too big mbulk count string"); 1135 setProtocolError(c,0); 1136 } 1137 return C_ERR; 1138 } 1139 1140 /* Buffer should also contain \n */ 1141 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) 1142 return C_ERR; 1143 1144 /* We know for sure there is a whole line since newline != NULL, 1145 * so go ahead and find out the multi bulk length. */ 1146 serverAssertWithInfo(c,NULL,c->querybuf[0] == '*'); 1147 ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); 1148 if (!ok || ll > 1024*1024) { 1149 addReplyError(c,"Protocol error: invalid multibulk length"); 1150 setProtocolError(c,pos); 1151 return C_ERR; 1152 } 1153 1154 pos = (newline-c->querybuf)+2; 1155 if (ll <= 0) { 1156 sdsrange(c->querybuf,pos,-1); 1157 return C_OK; 1158 } 1159 1160 c->multibulklen = ll; 1161 1162 /* Setup argv array on client structure */ 1163 if (c->argv) zfree(c->argv); 1164 c->argv = zmalloc(sizeof(robj*)*c->multibulklen); 1165 } 1166 1167 serverAssertWithInfo(c,NULL,c->multibulklen > 0); 1168 while(c->multibulklen) { 1169 /* Read bulk length if unknown */ 1170 if (c->bulklen == -1) { 1171 newline = strchr(c->querybuf+pos,'\r'); 1172 if (newline == NULL) { 1173 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { 1174 addReplyError(c, 1175 "Protocol error: too big bulk count string"); 1176 setProtocolError(c,0); 1177 return C_ERR; 1178 } 1179 break; 1180 } 1181 1182 /* Buffer should also contain \n */ 1183 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) 1184 break; 1185 1186 if (c->querybuf[pos] != '$') { 1187 addReplyErrorFormat(c, 1188 "Protocol error: expected '$', got '%c'", 1189 c->querybuf[pos]); 1190 setProtocolError(c,pos); 1191 return C_ERR; 1192 } 1193 1194 ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); 1195 if (!ok || ll < 0 || ll > 512*1024*1024) { 1196 addReplyError(c,"Protocol error: invalid bulk length"); 1197 setProtocolError(c,pos); 1198 return C_ERR; 1199 } 1200 1201 pos += newline-(c->querybuf+pos)+2; 1202 if (ll >= PROTO_MBULK_BIG_ARG) { 1203 size_t qblen; 1204 1205 /* If we are going to read a large object from network 1206 * try to make it likely that it will start at c->querybuf 1207 * boundary so that we can optimize object creation 1208 * avoiding a large copy of data. */ 1209 sdsrange(c->querybuf,pos,-1); 1210 pos = 0; 1211 qblen = sdslen(c->querybuf); 1212 /* Hint the sds library about the amount of bytes this string is 1213 * going to contain. */ 1214 if (qblen < (size_t)ll+2) 1215 c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-qblen); 1216 } 1217 c->bulklen = ll; 1218 } 1219 1220 /* Read bulk argument */ 1221 if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) { 1222 /* Not enough data (+2 == trailing \r\n) */ 1223 break; 1224 } else { 1225 /* Optimization: if the buffer contains JUST our bulk element 1226 * instead of creating a new object by *copying* the sds we 1227 * just use the current sds string. */ 1228 if (pos == 0 && 1229 c->bulklen >= PROTO_MBULK_BIG_ARG && 1230 (signed) sdslen(c->querybuf) == c->bulklen+2) 1231 { 1232 c->argv[c->argc++] = createObject(OBJ_STRING,c->querybuf); 1233 sdsIncrLen(c->querybuf,-2); /* remove CRLF */ 1234 /* Assume that if we saw a fat argument we'll see another one 1235 * likely... */ 1236 c->querybuf = sdsnewlen(NULL,c->bulklen+2); 1237 sdsclear(c->querybuf); 1238 pos = 0; 1239 } else { 1240 c->argv[c->argc++] = 1241 createStringObject(c->querybuf+pos,c->bulklen); 1242 pos += c->bulklen+2; 1243 } 1244 c->bulklen = -1; 1245 c->multibulklen--; 1246 } 1247 } 1248 1249 /* Trim to pos */ 1250 if (pos) sdsrange(c->querybuf,pos,-1); 1251 1252 /* We're done when c->multibulk == 0 */ 1253 if (c->multibulklen == 0) return C_OK; 1254 1255 /* Still not read to process the command */ 1256 return C_ERR; 1257 } 1258 1259 void processInputBuffer(client *c) { 1260 server.current_client = c; 1261 /* Keep processing while there is something in the input buffer */ 1262 while(sdslen(c->querybuf)) { 1263 /* Return if clients are paused. */ 1264 if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break; 1265 1266 /* Immediately abort if the client is in the middle of something. */ 1267 if (c->flags & CLIENT_BLOCKED) break; 1268 1269 /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is 1270 * written to the client. Make sure to not let the reply grow after 1271 * this flag has been set (i.e. don't process more commands). */ 1272 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) break; 1273 1274 /* Determine request type when unknown. */ 1275 if (!c->reqtype) { 1276 if (c->querybuf[0] == '*') { 1277 c->reqtype = PROTO_REQ_MULTIBULK; 1278 } else { 1279 c->reqtype = PROTO_REQ_INLINE; 1280 } 1281 } 1282 1283 if (c->reqtype == PROTO_REQ_INLINE) { 1284 if (processInlineBuffer(c) != C_OK) break; 1285 } else if (c->reqtype == PROTO_REQ_MULTIBULK) { 1286 if (processMultibulkBuffer(c) != C_OK) break; 1287 } else { 1288 serverPanic("Unknown request type"); 1289 } 1290 1291 /* Multibulk processing could see a <= 0 length. */ 1292 if (c->argc == 0) { 1293 resetClient(c); 1294 } else { 1295 /* Only reset the client when the command was executed. */ 1296 if (processCommand(c) == C_OK) 1297 resetClient(c); 1298 /* freeMemoryIfNeeded may flush slave output buffers. This may result 1299 * into a slave, that may be the active client, to be freed. */ 1300 if (server.current_client == NULL) break; 1301 } 1302 } 1303 server.current_client = NULL; 1304 } 1305 1306 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { 1307 client *c = (client*) privdata; 1308 int nread, readlen; 1309 size_t qblen; 1310 UNUSED(el); 1311 UNUSED(mask); 1312 1313 readlen = PROTO_IOBUF_LEN; 1314 /* If this is a multi bulk request, and we are processing a bulk reply 1315 * that is large enough, try to maximize the probability that the query 1316 * buffer contains exactly the SDS string representing the object, even 1317 * at the risk of requiring more read(2) calls. This way the function 1318 * processMultiBulkBuffer() can avoid copying buffers to create the 1319 * Redis Object representing the argument. */ 1320 if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 1321 && c->bulklen >= PROTO_MBULK_BIG_ARG) 1322 { 1323 int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); 1324 1325 if (remaining < readlen) readlen = remaining; 1326 } 1327 1328 qblen = sdslen(c->querybuf); 1329 if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; 1330 c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); 1331 nread = read(fd, c->querybuf+qblen, readlen); 1332 if (nread == -1) { 1333 if (errno == EAGAIN) { 1334 return; 1335 } else { 1336 serverLog(LL_VERBOSE, "Reading from client: %s",strerror(errno)); 1337 freeClient(c); 1338 return; 1339 } 1340 } else if (nread == 0) { 1341 serverLog(LL_VERBOSE, "Client closed connection"); 1342 freeClient(c); 1343 return; 1344 } 1345 1346 sdsIncrLen(c->querybuf,nread); 1347 c->lastinteraction = server.unixtime; 1348 if (c->flags & CLIENT_MASTER) c->reploff += nread; 1349 server.stat_net_input_bytes += nread; 1350 if (sdslen(c->querybuf) > server.client_max_querybuf_len) { 1351 sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); 1352 1353 bytes = sdscatrepr(bytes,c->querybuf,64); 1354 serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); 1355 sdsfree(ci); 1356 sdsfree(bytes); 1357 freeClient(c); 1358 return; 1359 } 1360 processInputBuffer(c); 1361 } 1362 1363 void getClientsMaxBuffers(unsigned long *longest_output_list, 1364 unsigned long *biggest_input_buffer) { 1365 client *c; 1366 listNode *ln; 1367 listIter li; 1368 unsigned long lol = 0, bib = 0; 1369 1370 listRewind(server.clients,&li); 1371 while ((ln = listNext(&li)) != NULL) { 1372 c = listNodeValue(ln); 1373 1374 if (listLength(c->reply) > lol) lol = listLength(c->reply); 1375 if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf); 1376 } 1377 *longest_output_list = lol; 1378 *biggest_input_buffer = bib; 1379 } 1380 1381 /* A Redis "Peer ID" is a colon separated ip:port pair. 1382 * For IPv4 it's in the form x.y.z.k:port, example: "127.0.0.1:1234". 1383 * For IPv6 addresses we use [] around the IP part, like in "[::1]:1234". 1384 * For Unix sockets we use path:0, like in "/tmp/redis:0". 1385 * 1386 * A Peer ID always fits inside a buffer of NET_PEER_ID_LEN bytes, including 1387 * the null term. 1388 * 1389 * On failure the function still populates 'peerid' with the "?:0" string 1390 * in case you want to relax error checking or need to display something 1391 * anyway (see anetPeerToString implementation for more info). */ 1392 void genClientPeerId(client *client, char *peerid, 1393 size_t peerid_len) { 1394 if (client->flags & CLIENT_UNIX_SOCKET) { 1395 /* Unix socket client. */ 1396 snprintf(peerid,peerid_len,"%s:0",server.unixsocket); 1397 } else { 1398 /* TCP client. */ 1399 anetFormatPeer(client->fd,peerid,peerid_len); 1400 } 1401 } 1402 1403 /* This function returns the client peer id, by creating and caching it 1404 * if client->peerid is NULL, otherwise returning the cached value. 1405 * The Peer ID never changes during the life of the client, however it 1406 * is expensive to compute. */ 1407 char *getClientPeerId(client *c) { 1408 char peerid[NET_PEER_ID_LEN]; 1409 1410 if (c->peerid == NULL) { 1411 genClientPeerId(c,peerid,sizeof(peerid)); 1412 c->peerid = sdsnew(peerid); 1413 } 1414 return c->peerid; 1415 } 1416 1417 /* Concatenate a string representing the state of a client in an human 1418 * readable format, into the sds string 's'. */ 1419 sds catClientInfoString(sds s, client *client) { 1420 char flags[16], events[3], *p; 1421 int emask; 1422 1423 p = flags; 1424 if (client->flags & CLIENT_SLAVE) { 1425 if (client->flags & CLIENT_MONITOR) 1426 *p++ = 'O'; 1427 else 1428 *p++ = 'S'; 1429 } 1430 if (client->flags & CLIENT_MASTER) *p++ = 'M'; 1431 if (client->flags & CLIENT_MULTI) *p++ = 'x'; 1432 if (client->flags & CLIENT_BLOCKED) *p++ = 'b'; 1433 if (client->flags & CLIENT_DIRTY_CAS) *p++ = 'd'; 1434 if (client->flags & CLIENT_CLOSE_AFTER_REPLY) *p++ = 'c'; 1435 if (client->flags & CLIENT_UNBLOCKED) *p++ = 'u'; 1436 if (client->flags & CLIENT_CLOSE_ASAP) *p++ = 'A'; 1437 if (client->flags & CLIENT_UNIX_SOCKET) *p++ = 'U'; 1438 if (client->flags & CLIENT_READONLY) *p++ = 'r'; 1439 if (p == flags) *p++ = 'N'; 1440 *p++ = '\0'; 1441 1442 emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd); 1443 p = events; 1444 if (emask & AE_READABLE) *p++ = 'r'; 1445 if (emask & AE_WRITABLE) *p++ = 'w'; 1446 *p = '\0'; 1447 return sdscatfmt(s, 1448 "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", 1449 (unsigned long long) client->id, 1450 getClientPeerId(client), 1451 client->fd, 1452 client->name ? (char*)client->name->ptr : "", 1453 (long long)(server.unixtime - client->ctime), 1454 (long long)(server.unixtime - client->lastinteraction), 1455 flags, 1456 client->db->id, 1457 (int) dictSize(client->pubsub_channels), 1458 (int) listLength(client->pubsub_patterns), 1459 (client->flags & CLIENT_MULTI) ? client->mstate.count : -1, 1460 (unsigned long long) sdslen(client->querybuf), 1461 (unsigned long long) sdsavail(client->querybuf), 1462 (unsigned long long) client->bufpos, 1463 (unsigned long long) listLength(client->reply), 1464 (unsigned long long) getClientOutputBufferMemoryUsage(client), 1465 events, 1466 client->lastcmd ? client->lastcmd->name : "NULL"); 1467 } 1468 1469 sds getAllClientsInfoString(void) { 1470 listNode *ln; 1471 listIter li; 1472 client *client; 1473 sds o = sdsnewlen(NULL,200*listLength(server.clients)); 1474 sdsclear(o); 1475 listRewind(server.clients,&li); 1476 while ((ln = listNext(&li)) != NULL) { 1477 client = listNodeValue(ln); 1478 o = catClientInfoString(o,client); 1479 o = sdscatlen(o,"\n",1); 1480 } 1481 return o; 1482 } 1483 1484 void clientCommand(client *c) { 1485 listNode *ln; 1486 listIter li; 1487 client *client; 1488 1489 if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { 1490 /* CLIENT LIST */ 1491 sds o = getAllClientsInfoString(); 1492 addReplyBulkCBuffer(c,o,sdslen(o)); 1493 sdsfree(o); 1494 } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) { 1495 /* CLIENT REPLY ON|OFF|SKIP */ 1496 if (!strcasecmp(c->argv[2]->ptr,"on")) { 1497 c->flags &= ~(CLIENT_REPLY_SKIP|CLIENT_REPLY_OFF); 1498 addReply(c,shared.ok); 1499 } else if (!strcasecmp(c->argv[2]->ptr,"off")) { 1500 c->flags |= CLIENT_REPLY_OFF; 1501 } else if (!strcasecmp(c->argv[2]->ptr,"skip")) { 1502 if (!(c->flags & CLIENT_REPLY_OFF)) 1503 c->flags |= CLIENT_REPLY_SKIP_NEXT; 1504 } else { 1505 addReply(c,shared.syntaxerr); 1506 return; 1507 } 1508 } else if (!strcasecmp(c->argv[1]->ptr,"kill")) { 1509 /* CLIENT KILL <ip:port> 1510 * CLIENT KILL <option> [value] ... <option> [value] */ 1511 char *addr = NULL; 1512 int type = -1; 1513 uint64_t id = 0; 1514 int skipme = 1; 1515 int killed = 0, close_this_client = 0; 1516 1517 if (c->argc == 3) { 1518 /* Old style syntax: CLIENT KILL <addr> */ 1519 addr = c->argv[2]->ptr; 1520 skipme = 0; /* With the old form, you can kill yourself. */ 1521 } else if (c->argc > 3) { 1522 int i = 2; /* Next option index. */ 1523 1524 /* New style syntax: parse options. */ 1525 while(i < c->argc) { 1526 int moreargs = c->argc > i+1; 1527 1528 if (!strcasecmp(c->argv[i]->ptr,"id") && moreargs) { 1529 long long tmp; 1530 1531 if (getLongLongFromObjectOrReply(c,c->argv[i+1],&tmp,NULL) 1532 != C_OK) return; 1533 id = tmp; 1534 } else if (!strcasecmp(c->argv[i]->ptr,"type") && moreargs) { 1535 type = getClientTypeByName(c->argv[i+1]->ptr); 1536 if (type == -1) { 1537 addReplyErrorFormat(c,"Unknown client type '%s'", 1538 (char*) c->argv[i+1]->ptr); 1539 return; 1540 } 1541 } else if (!strcasecmp(c->argv[i]->ptr,"addr") && moreargs) { 1542 addr = c->argv[i+1]->ptr; 1543 } else if (!strcasecmp(c->argv[i]->ptr,"skipme") && moreargs) { 1544 if (!strcasecmp(c->argv[i+1]->ptr,"yes")) { 1545 skipme = 1; 1546 } else if (!strcasecmp(c->argv[i+1]->ptr,"no")) { 1547 skipme = 0; 1548 } else { 1549 addReply(c,shared.syntaxerr); 1550 return; 1551 } 1552 } else { 1553 addReply(c,shared.syntaxerr); 1554 return; 1555 } 1556 i += 2; 1557 } 1558 } else { 1559 addReply(c,shared.syntaxerr); 1560 return; 1561 } 1562 1563 /* Iterate clients killing all the matching clients. */ 1564 listRewind(server.clients,&li); 1565 while ((ln = listNext(&li)) != NULL) { 1566 client = listNodeValue(ln); 1567 if (addr && strcmp(getClientPeerId(client),addr) != 0) continue; 1568 if (type != -1 && getClientType(client) != type) continue; 1569 if (id != 0 && client->id != id) continue; 1570 if (c == client && skipme) continue; 1571 1572 /* Kill it. */ 1573 if (c == client) { 1574 close_this_client = 1; 1575 } else { 1576 freeClient(client); 1577 } 1578 killed++; 1579 } 1580 1581 /* Reply according to old/new format. */ 1582 if (c->argc == 3) { 1583 if (killed == 0) 1584 addReplyError(c,"No such client"); 1585 else 1586 addReply(c,shared.ok); 1587 } else { 1588 addReplyLongLong(c,killed); 1589 } 1590 1591 /* If this client has to be closed, flag it as CLOSE_AFTER_REPLY 1592 * only after we queued the reply to its output buffers. */ 1593 if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY; 1594 } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) { 1595 int j, len = sdslen(c->argv[2]->ptr); 1596 char *p = c->argv[2]->ptr; 1597 1598 /* Setting the client name to an empty string actually removes 1599 * the current name. */ 1600 if (len == 0) { 1601 if (c->name) decrRefCount(c->name); 1602 c->name = NULL; 1603 addReply(c,shared.ok); 1604 return; 1605 } 1606 1607 /* Otherwise check if the charset is ok. We need to do this otherwise 1608 * CLIENT LIST format will break. You should always be able to 1609 * split by space to get the different fields. */ 1610 for (j = 0; j < len; j++) { 1611 if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */ 1612 addReplyError(c, 1613 "Client names cannot contain spaces, " 1614 "newlines or special characters."); 1615 return; 1616 } 1617 } 1618 if (c->name) decrRefCount(c->name); 1619 c->name = c->argv[2]; 1620 incrRefCount(c->name); 1621 addReply(c,shared.ok); 1622 } else if (!strcasecmp(c->argv[1]->ptr,"getname") && c->argc == 2) { 1623 if (c->name) 1624 addReplyBulk(c,c->name); 1625 else 1626 addReply(c,shared.nullbulk); 1627 } else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) { 1628 long long duration; 1629 1630 if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration,UNIT_MILLISECONDS) 1631 != C_OK) return; 1632 pauseClients(duration); 1633 addReply(c,shared.ok); 1634 } else { 1635 addReplyError(c, "Syntax error, try CLIENT (LIST | KILL | GETNAME | SETNAME | PAUSE | REPLY)"); 1636 } 1637 } 1638 1639 /* Rewrite the command vector of the client. All the new objects ref count 1640 * is incremented. The old command vector is freed, and the old objects 1641 * ref count is decremented. */ 1642 void rewriteClientCommandVector(client *c, int argc, ...) { 1643 va_list ap; 1644 int j; 1645 robj **argv; /* The new argument vector */ 1646 1647 argv = zmalloc(sizeof(robj*)*argc); 1648 va_start(ap,argc); 1649 for (j = 0; j < argc; j++) { 1650 robj *a; 1651 1652 a = va_arg(ap, robj*); 1653 argv[j] = a; 1654 incrRefCount(a); 1655 } 1656 /* We free the objects in the original vector at the end, so we are 1657 * sure that if the same objects are reused in the new vector the 1658 * refcount gets incremented before it gets decremented. */ 1659 for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]); 1660 zfree(c->argv); 1661 /* Replace argv and argc with our new versions. */ 1662 c->argv = argv; 1663 c->argc = argc; 1664 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); 1665 serverAssertWithInfo(c,NULL,c->cmd != NULL); 1666 va_end(ap); 1667 } 1668 1669 /* Completely replace the client command vector with the provided one. */ 1670 void replaceClientCommandVector(client *c, int argc, robj **argv) { 1671 freeClientArgv(c); 1672 zfree(c->argv); 1673 c->argv = argv; 1674 c->argc = argc; 1675 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); 1676 serverAssertWithInfo(c,NULL,c->cmd != NULL); 1677 } 1678 1679 /* Rewrite a single item in the command vector. 1680 * The new val ref count is incremented, and the old decremented. 1681 * 1682 * It is possible to specify an argument over the current size of the 1683 * argument vector: in this case the array of objects gets reallocated 1684 * and c->argc set to the max value. However it's up to the caller to 1685 * 1686 * 1. Make sure there are no "holes" and all the arguments are set. 1687 * 2. If the original argument vector was longer than the one we 1688 * want to end with, it's up to the caller to set c->argc and 1689 * free the no longer used objects on c->argv. */ 1690 void rewriteClientCommandArgument(client *c, int i, robj *newval) { 1691 robj *oldval; 1692 1693 if (i >= c->argc) { 1694 c->argv = zrealloc(c->argv,sizeof(robj*)*(i+1)); 1695 c->argc = i+1; 1696 c->argv[i] = NULL; 1697 } 1698 oldval = c->argv[i]; 1699 c->argv[i] = newval; 1700 incrRefCount(newval); 1701 if (oldval) decrRefCount(oldval); 1702 1703 /* If this is the command name make sure to fix c->cmd. */ 1704 if (i == 0) { 1705 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); 1706 serverAssertWithInfo(c,NULL,c->cmd != NULL); 1707 } 1708 } 1709 1710 /* This function returns the number of bytes that Redis is virtually 1711 * using to store the reply still not read by the client. 1712 * It is "virtual" since the reply output list may contain objects that 1713 * are shared and are not really using additional memory. 1714 * 1715 * The function returns the total sum of the length of all the objects 1716 * stored in the output list, plus the memory used to allocate every 1717 * list node. The static reply buffer is not taken into account since it 1718 * is allocated anyway. 1719 * 1720 * Note: this function is very fast so can be called as many time as 1721 * the caller wishes. The main usage of this function currently is 1722 * enforcing the client output length limits. */ 1723 unsigned long getClientOutputBufferMemoryUsage(client *c) { 1724 unsigned long list_item_size = sizeof(listNode)+sizeof(robj); 1725 1726 return c->reply_bytes + (list_item_size*listLength(c->reply)); 1727 } 1728 1729 /* Get the class of a client, used in order to enforce limits to different 1730 * classes of clients. 1731 * 1732 * The function will return one of the following: 1733 * CLIENT_TYPE_NORMAL -> Normal client 1734 * CLIENT_TYPE_SLAVE -> Slave or client executing MONITOR command 1735 * CLIENT_TYPE_PUBSUB -> Client subscribed to Pub/Sub channels 1736 * CLIENT_TYPE_MASTER -> The client representing our replication master. 1737 */ 1738 int getClientType(client *c) { 1739 if (c->flags & CLIENT_MASTER) return CLIENT_TYPE_MASTER; 1740 if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) 1741 return CLIENT_TYPE_SLAVE; 1742 if (c->flags & CLIENT_PUBSUB) return CLIENT_TYPE_PUBSUB; 1743 return CLIENT_TYPE_NORMAL; 1744 } 1745 1746 int getClientTypeByName(char *name) { 1747 if (!strcasecmp(name,"normal")) return CLIENT_TYPE_NORMAL; 1748 else if (!strcasecmp(name,"slave")) return CLIENT_TYPE_SLAVE; 1749 else if (!strcasecmp(name,"pubsub")) return CLIENT_TYPE_PUBSUB; 1750 else if (!strcasecmp(name,"master")) return CLIENT_TYPE_MASTER; 1751 else return -1; 1752 } 1753 1754 char *getClientTypeName(int class) { 1755 switch(class) { 1756 case CLIENT_TYPE_NORMAL: return "normal"; 1757 case CLIENT_TYPE_SLAVE: return "slave"; 1758 case CLIENT_TYPE_PUBSUB: return "pubsub"; 1759 case CLIENT_TYPE_MASTER: return "master"; 1760 default: return NULL; 1761 } 1762 } 1763 1764 /* The function checks if the client reached output buffer soft or hard 1765 * limit, and also update the state needed to check the soft limit as 1766 * a side effect. 1767 * 1768 * Return value: non-zero if the client reached the soft or the hard limit. 1769 * Otherwise zero is returned. */ 1770 int checkClientOutputBufferLimits(client *c) { 1771 int soft = 0, hard = 0, class; 1772 unsigned long used_mem = getClientOutputBufferMemoryUsage(c); 1773 1774 class = getClientType(c); 1775 /* For the purpose of output buffer limiting, masters are handled 1776 * like normal clients. */ 1777 if (class == CLIENT_TYPE_MASTER) class = CLIENT_TYPE_NORMAL; 1778 1779 if (server.client_obuf_limits[class].hard_limit_bytes && 1780 used_mem >= server.client_obuf_limits[class].hard_limit_bytes) 1781 hard = 1; 1782 if (server.client_obuf_limits[class].soft_limit_bytes && 1783 used_mem >= server.client_obuf_limits[class].soft_limit_bytes) 1784 soft = 1; 1785 1786 /* We need to check if the soft limit is reached continuously for the 1787 * specified amount of seconds. */ 1788 if (soft) { 1789 if (c->obuf_soft_limit_reached_time == 0) { 1790 c->obuf_soft_limit_reached_time = server.unixtime; 1791 soft = 0; /* First time we see the soft limit reached */ 1792 } else { 1793 time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time; 1794 1795 if (elapsed <= 1796 server.client_obuf_limits[class].soft_limit_seconds) { 1797 soft = 0; /* The client still did not reached the max number of 1798 seconds for the soft limit to be considered 1799 reached. */ 1800 } 1801 } 1802 } else { 1803 c->obuf_soft_limit_reached_time = 0; 1804 } 1805 return soft || hard; 1806 } 1807 1808 /* Asynchronously close a client if soft or hard limit is reached on the 1809 * output buffer size. The caller can check if the client will be closed 1810 * checking if the client CLIENT_CLOSE_ASAP flag is set. 1811 * 1812 * Note: we need to close the client asynchronously because this function is 1813 * called from contexts where the client can't be freed safely, i.e. from the 1814 * lower level functions pushing data inside the client output buffers. */ 1815 void asyncCloseClientOnOutputBufferLimitReached(client *c) { 1816 serverAssert(c->reply_bytes < SIZE_MAX-(1024*64)); 1817 if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return; 1818 if (checkClientOutputBufferLimits(c)) { 1819 sds client = catClientInfoString(sdsempty(),c); 1820 1821 freeClientAsync(c); 1822 serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); 1823 sdsfree(client); 1824 } 1825 } 1826 1827 /* Helper function used by freeMemoryIfNeeded() in order to flush slaves 1828 * output buffers without returning control to the event loop. 1829 * This is also called by SHUTDOWN for a best-effort attempt to send 1830 * slaves the latest writes. */ 1831 void flushSlavesOutputBuffers(void) { 1832 listIter li; 1833 listNode *ln; 1834 1835 listRewind(server.slaves,&li); 1836 while((ln = listNext(&li))) { 1837 client *slave = listNodeValue(ln); 1838 int events; 1839 1840 /* Note that the following will not flush output buffers of slaves 1841 * in STATE_ONLINE but having put_online_on_ack set to true: in this 1842 * case the writable event is never installed, since the purpose 1843 * of put_online_on_ack is to postpone the moment it is installed. 1844 * This is what we want since slaves in this state should not receive 1845 * writes before the first ACK. */ 1846 events = aeGetFileEvents(server.el,slave->fd); 1847 if (events & AE_WRITABLE && 1848 slave->replstate == SLAVE_STATE_ONLINE && 1849 clientHasPendingReplies(slave)) 1850 { 1851 writeToClient(slave->fd,slave,0); 1852 } 1853 } 1854 } 1855 1856 /* Pause clients up to the specified unixtime (in ms). While clients 1857 * are paused no command is processed from clients, so the data set can't 1858 * change during that time. 1859 * 1860 * However while this function pauses normal and Pub/Sub clients, slaves are 1861 * still served, so this function can be used on server upgrades where it is 1862 * required that slaves process the latest bytes from the replication stream 1863 * before being turned to masters. 1864 * 1865 * This function is also internally used by Redis Cluster for the manual 1866 * failover procedure implemented by CLUSTER FAILOVER. 1867 * 1868 * The function always succeed, even if there is already a pause in progress. 1869 * In such a case, the pause is extended if the duration is more than the 1870 * time left for the previous duration. However if the duration is smaller 1871 * than the time left for the previous pause, no change is made to the 1872 * left duration. */ 1873 void pauseClients(mstime_t end) { 1874 if (!server.clients_paused || end > server.clients_pause_end_time) 1875 server.clients_pause_end_time = end; 1876 server.clients_paused = 1; 1877 } 1878 1879 /* Return non-zero if clients are currently paused. As a side effect the 1880 * function checks if the pause time was reached and clear it. */ 1881 int clientsArePaused(void) { 1882 if (server.clients_paused && 1883 server.clients_pause_end_time < server.mstime) 1884 { 1885 listNode *ln; 1886 listIter li; 1887 client *c; 1888 1889 server.clients_paused = 0; 1890 1891 /* Put all the clients in the unblocked clients queue in order to 1892 * force the re-processing of the input buffer if any. */ 1893 listRewind(server.clients,&li); 1894 while ((ln = listNext(&li)) != NULL) { 1895 c = listNodeValue(ln); 1896 1897 /* Don't touch slaves and blocked clients. The latter pending 1898 * requests be processed when unblocked. */ 1899 if (c->flags & (CLIENT_SLAVE|CLIENT_BLOCKED)) continue; 1900 c->flags |= CLIENT_UNBLOCKED; 1901 listAddNodeTail(server.unblocked_clients,c); 1902 } 1903 } 1904 return server.clients_paused; 1905 } 1906 1907 /* This function is called by Redis in order to process a few events from 1908 * time to time while blocked into some not interruptible operation. 1909 * This allows to reply to clients with the -LOADING error while loading the 1910 * data set at startup or after a full resynchronization with the master 1911 * and so forth. 1912 * 1913 * It calls the event loop in order to process a few events. Specifically we 1914 * try to call the event loop 4 times as long as we receive acknowledge that 1915 * some event was processed, in order to go forward with the accept, read, 1916 * write, close sequence needed to serve a client. 1917 * 1918 * The function returns the total number of events processed. */ 1919 int processEventsWhileBlocked(void) { 1920 int iterations = 4; /* See the function top-comment. */ 1921 int count = 0; 1922 while (iterations--) { 1923 int events = 0; 1924 events += aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); 1925 events += handleClientsWithPendingWrites(); 1926 if (!events) break; 1927 count += events; 1928 } 1929 return count; 1930 } 1931