2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
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.
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.
33 static void setProtocolError(redisClient
*c
, int pos
);
35 /* To evaluate the output buffer size of a client we need to get size of
36 * allocated objects, however we can't used zmalloc_size() directly on sds
37 * strings because of the trick they use to work (the header is before the
38 * returned pointer), so we use this helper function. */
39 size_t zmalloc_size_sds(sds s
) {
40 return zmalloc_size(s
-sizeof(struct sdshdr
));
43 void *dupClientReplyValue(void *o
) {
44 incrRefCount((robj
*)o
);
48 int listMatchObjects(void *a
, void *b
) {
49 return equalStringObjects(a
,b
);
52 redisClient
*createClient(int fd
) {
53 redisClient
*c
= zmalloc(sizeof(redisClient
));
55 /* passing -1 as fd it is possible to create a non connected client.
56 * This is useful since all the Redis commands needs to be executed
57 * in the context of a client. When commands are executed in other
58 * contexts (for instance a Lua script) we need a non connected client. */
60 anetNonBlock(NULL
,fd
);
61 anetTcpNoDelay(NULL
,fd
);
62 if (aeCreateFileEvent(server
.el
,fd
,AE_READABLE
,
63 readQueryFromClient
, c
) == AE_ERR
)
74 c
->querybuf
= sdsempty();
79 c
->cmd
= c
->lastcmd
= NULL
;
84 c
->ctime
= c
->lastinteraction
= server
.unixtime
;
86 c
->replstate
= REDIS_REPL_NONE
;
87 c
->slave_listening_port
= 0;
88 c
->reply
= listCreate();
90 c
->obuf_soft_limit_reached_time
= 0;
91 listSetFreeMethod(c
->reply
,decrRefCount
);
92 listSetDupMethod(c
->reply
,dupClientReplyValue
);
93 c
->bpop
.keys
= dictCreate(&setDictType
,NULL
);
95 c
->bpop
.target
= NULL
;
96 c
->io_keys
= listCreate();
97 c
->watched_keys
= listCreate();
98 listSetFreeMethod(c
->io_keys
,decrRefCount
);
99 c
->pubsub_channels
= dictCreate(&setDictType
,NULL
);
100 c
->pubsub_patterns
= listCreate();
101 listSetFreeMethod(c
->pubsub_patterns
,decrRefCount
);
102 listSetMatchMethod(c
->pubsub_patterns
,listMatchObjects
);
103 if (fd
!= -1) listAddNodeTail(server
.clients
,c
);
104 initClientMultiState(c
);
108 /* This function is called every time we are going to transmit new data
109 * to the client. The behavior is the following:
111 * If the client should receive new data (normal clients will) the function
112 * returns REDIS_OK, and make sure to install the write handler in our event
113 * loop so that when the socket is writable new data gets written.
115 * If the client should not receive new data, because it is a fake client
116 * or a slave, or because the setup of the write handler failed, the function
119 * Typically gets called every time a reply is built, before adding more
120 * data to the clients output buffers. If the function returns REDIS_ERR no
121 * data should be appended to the output buffers. */
122 int prepareClientToWrite(redisClient
*c
) {
123 if (c
->flags
& REDIS_LUA_CLIENT
) return REDIS_OK
;
124 if (c
->fd
<= 0) return REDIS_ERR
; /* Fake client */
125 if (c
->bufpos
== 0 && listLength(c
->reply
) == 0 &&
126 (c
->replstate
== REDIS_REPL_NONE
||
127 c
->replstate
== REDIS_REPL_ONLINE
) &&
128 aeCreateFileEvent(server
.el
, c
->fd
, AE_WRITABLE
,
129 sendReplyToClient
, c
) == AE_ERR
) return REDIS_ERR
;
133 /* Create a duplicate of the last object in the reply list when
134 * it is not exclusively owned by the reply list. */
135 robj
*dupLastObjectIfNeeded(list
*reply
) {
138 redisAssert(listLength(reply
) > 0);
139 ln
= listLast(reply
);
140 cur
= listNodeValue(ln
);
141 if (cur
->refcount
> 1) {
142 new = dupStringObject(cur
);
144 listNodeValue(ln
) = new;
146 return listNodeValue(ln
);
149 /* -----------------------------------------------------------------------------
150 * Low level functions to add more data to output buffers.
151 * -------------------------------------------------------------------------- */
153 int _addReplyToBuffer(redisClient
*c
, char *s
, size_t len
) {
154 size_t available
= sizeof(c
->buf
)-c
->bufpos
;
156 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return REDIS_OK
;
158 /* If there already are entries in the reply list, we cannot
159 * add anything more to the static buffer. */
160 if (listLength(c
->reply
) > 0) return REDIS_ERR
;
162 /* Check that the buffer has enough space available for this string. */
163 if (len
> available
) return REDIS_ERR
;
165 memcpy(c
->buf
+c
->bufpos
,s
,len
);
170 void _addReplyObjectToList(redisClient
*c
, robj
*o
) {
173 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
175 if (listLength(c
->reply
) == 0) {
177 listAddNodeTail(c
->reply
,o
);
178 c
->reply_bytes
+= zmalloc_size_sds(o
->ptr
);
180 tail
= listNodeValue(listLast(c
->reply
));
182 /* Append to this object when possible. */
183 if (tail
->ptr
!= NULL
&&
184 sdslen(tail
->ptr
)+sdslen(o
->ptr
) <= REDIS_REPLY_CHUNK_BYTES
)
186 c
->reply_bytes
-= zmalloc_size_sds(tail
->ptr
);
187 tail
= dupLastObjectIfNeeded(c
->reply
);
188 tail
->ptr
= sdscatlen(tail
->ptr
,o
->ptr
,sdslen(o
->ptr
));
189 c
->reply_bytes
+= zmalloc_size_sds(tail
->ptr
);
192 listAddNodeTail(c
->reply
,o
);
193 c
->reply_bytes
+= zmalloc_size_sds(o
->ptr
);
196 asyncCloseClientOnOutputBufferLimitReached(c
);
199 /* This method takes responsibility over the sds. When it is no longer
200 * needed it will be free'd, otherwise it ends up in a robj. */
201 void _addReplySdsToList(redisClient
*c
, sds s
) {
204 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) {
209 if (listLength(c
->reply
) == 0) {
210 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,s
));
211 c
->reply_bytes
+= zmalloc_size_sds(s
);
213 tail
= listNodeValue(listLast(c
->reply
));
215 /* Append to this object when possible. */
216 if (tail
->ptr
!= NULL
&&
217 sdslen(tail
->ptr
)+sdslen(s
) <= REDIS_REPLY_CHUNK_BYTES
)
219 c
->reply_bytes
-= zmalloc_size_sds(tail
->ptr
);
220 tail
= dupLastObjectIfNeeded(c
->reply
);
221 tail
->ptr
= sdscatlen(tail
->ptr
,s
,sdslen(s
));
222 c
->reply_bytes
+= zmalloc_size_sds(tail
->ptr
);
225 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,s
));
226 c
->reply_bytes
+= zmalloc_size_sds(s
);
229 asyncCloseClientOnOutputBufferLimitReached(c
);
232 void _addReplyStringToList(redisClient
*c
, char *s
, size_t len
) {
235 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
237 if (listLength(c
->reply
) == 0) {
238 robj
*o
= createStringObject(s
,len
);
240 listAddNodeTail(c
->reply
,o
);
241 c
->reply_bytes
+= zmalloc_size_sds(o
->ptr
);
243 tail
= listNodeValue(listLast(c
->reply
));
245 /* Append to this object when possible. */
246 if (tail
->ptr
!= NULL
&&
247 sdslen(tail
->ptr
)+len
<= REDIS_REPLY_CHUNK_BYTES
)
249 c
->reply_bytes
-= zmalloc_size_sds(tail
->ptr
);
250 tail
= dupLastObjectIfNeeded(c
->reply
);
251 tail
->ptr
= sdscatlen(tail
->ptr
,s
,len
);
252 c
->reply_bytes
+= zmalloc_size_sds(tail
->ptr
);
254 robj
*o
= createStringObject(s
,len
);
256 listAddNodeTail(c
->reply
,o
);
257 c
->reply_bytes
+= zmalloc_size_sds(o
->ptr
);
260 asyncCloseClientOnOutputBufferLimitReached(c
);
263 /* -----------------------------------------------------------------------------
264 * Higher level functions to queue data on the client output buffer.
265 * The following functions are the ones that commands implementations will call.
266 * -------------------------------------------------------------------------- */
268 void addReply(redisClient
*c
, robj
*obj
) {
269 if (prepareClientToWrite(c
) != REDIS_OK
) return;
271 /* This is an important place where we can avoid copy-on-write
272 * when there is a saving child running, avoiding touching the
273 * refcount field of the object if it's not needed.
275 * If the encoding is RAW and there is room in the static buffer
276 * we'll be able to send the object to the client without
277 * messing with its page. */
278 if (obj
->encoding
== REDIS_ENCODING_RAW
) {
279 if (_addReplyToBuffer(c
,obj
->ptr
,sdslen(obj
->ptr
)) != REDIS_OK
)
280 _addReplyObjectToList(c
,obj
);
281 } else if (obj
->encoding
== REDIS_ENCODING_INT
) {
282 /* Optimization: if there is room in the static buffer for 32 bytes
283 * (more than the max chars a 64 bit integer can take as string) we
284 * avoid decoding the object and go for the lower level approach. */
285 if (listLength(c
->reply
) == 0 && (sizeof(c
->buf
) - c
->bufpos
) >= 32) {
289 len
= ll2string(buf
,sizeof(buf
),(long)obj
->ptr
);
290 if (_addReplyToBuffer(c
,buf
,len
) == REDIS_OK
)
292 /* else... continue with the normal code path, but should never
293 * happen actually since we verified there is room. */
295 obj
= getDecodedObject(obj
);
296 if (_addReplyToBuffer(c
,obj
->ptr
,sdslen(obj
->ptr
)) != REDIS_OK
)
297 _addReplyObjectToList(c
,obj
);
300 redisPanic("Wrong obj->encoding in addReply()");
304 void addReplySds(redisClient
*c
, sds s
) {
305 if (prepareClientToWrite(c
) != REDIS_OK
) {
306 /* The caller expects the sds to be free'd. */
310 if (_addReplyToBuffer(c
,s
,sdslen(s
)) == REDIS_OK
) {
313 /* This method free's the sds when it is no longer needed. */
314 _addReplySdsToList(c
,s
);
318 void addReplyString(redisClient
*c
, char *s
, size_t len
) {
319 if (prepareClientToWrite(c
) != REDIS_OK
) return;
320 if (_addReplyToBuffer(c
,s
,len
) != REDIS_OK
)
321 _addReplyStringToList(c
,s
,len
);
324 void addReplyErrorLength(redisClient
*c
, char *s
, size_t len
) {
325 addReplyString(c
,"-ERR ",5);
326 addReplyString(c
,s
,len
);
327 addReplyString(c
,"\r\n",2);
330 void addReplyError(redisClient
*c
, char *err
) {
331 addReplyErrorLength(c
,err
,strlen(err
));
334 void addReplyErrorFormat(redisClient
*c
, const char *fmt
, ...) {
338 sds s
= sdscatvprintf(sdsempty(),fmt
,ap
);
340 /* Make sure there are no newlines in the string, otherwise invalid protocol
343 for (j
= 0; j
< l
; j
++) {
344 if (s
[j
] == '\r' || s
[j
] == '\n') s
[j
] = ' ';
346 addReplyErrorLength(c
,s
,sdslen(s
));
350 void addReplyStatusLength(redisClient
*c
, char *s
, size_t len
) {
351 addReplyString(c
,"+",1);
352 addReplyString(c
,s
,len
);
353 addReplyString(c
,"\r\n",2);
356 void addReplyStatus(redisClient
*c
, char *status
) {
357 addReplyStatusLength(c
,status
,strlen(status
));
360 void addReplyStatusFormat(redisClient
*c
, const char *fmt
, ...) {
363 sds s
= sdscatvprintf(sdsempty(),fmt
,ap
);
365 addReplyStatusLength(c
,s
,sdslen(s
));
369 /* Adds an empty object to the reply list that will contain the multi bulk
370 * length, which is not known when this function is called. */
371 void *addDeferredMultiBulkLength(redisClient
*c
) {
372 /* Note that we install the write event here even if the object is not
373 * ready to be sent, since we are sure that before returning to the
374 * event loop setDeferredMultiBulkLength() will be called. */
375 if (prepareClientToWrite(c
) != REDIS_OK
) return NULL
;
376 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,NULL
));
377 return listLast(c
->reply
);
380 /* Populate the length object and try glueing it to the next chunk. */
381 void setDeferredMultiBulkLength(redisClient
*c
, void *node
, long length
) {
382 listNode
*ln
= (listNode
*)node
;
385 /* Abort when *node is NULL (see addDeferredMultiBulkLength). */
386 if (node
== NULL
) return;
388 len
= listNodeValue(ln
);
389 len
->ptr
= sdscatprintf(sdsempty(),"*%ld\r\n",length
);
390 c
->reply_bytes
+= zmalloc_size_sds(len
->ptr
);
391 if (ln
->next
!= NULL
) {
392 next
= listNodeValue(ln
->next
);
394 /* Only glue when the next node is non-NULL (an sds in this case) */
395 if (next
->ptr
!= NULL
) {
396 c
->reply_bytes
-= zmalloc_size_sds(len
->ptr
);
397 c
->reply_bytes
-= zmalloc_size_sds(next
->ptr
);
398 len
->ptr
= sdscatlen(len
->ptr
,next
->ptr
,sdslen(next
->ptr
));
399 c
->reply_bytes
+= zmalloc_size_sds(len
->ptr
);
400 listDelNode(c
->reply
,ln
->next
);
403 asyncCloseClientOnOutputBufferLimitReached(c
);
406 /* Add a duble as a bulk reply */
407 void addReplyDouble(redisClient
*c
, double d
) {
408 char dbuf
[128], sbuf
[128];
410 dlen
= snprintf(dbuf
,sizeof(dbuf
),"%.17g",d
);
411 slen
= snprintf(sbuf
,sizeof(sbuf
),"$%d\r\n%s\r\n",dlen
,dbuf
);
412 addReplyString(c
,sbuf
,slen
);
415 /* Add a long long as integer reply or bulk len / multi bulk count.
416 * Basically this is used to output <prefix><long long><crlf>. */
417 void addReplyLongLongWithPrefix(redisClient
*c
, long long ll
, char prefix
) {
421 /* Things like $3\r\n or *2\r\n are emitted very often by the protocol
422 * so we have a few shared objects to use if the integer is small
423 * like it is most of the times. */
424 if (prefix
== '*' && ll
< REDIS_SHARED_BULKHDR_LEN
) {
425 addReply(c
,shared
.mbulkhdr
[ll
]);
427 } else if (prefix
== '$' && ll
< REDIS_SHARED_BULKHDR_LEN
) {
428 addReply(c
,shared
.bulkhdr
[ll
]);
433 len
= ll2string(buf
+1,sizeof(buf
)-1,ll
);
436 addReplyString(c
,buf
,len
+3);
439 void addReplyLongLong(redisClient
*c
, long long ll
) {
441 addReply(c
,shared
.czero
);
443 addReply(c
,shared
.cone
);
445 addReplyLongLongWithPrefix(c
,ll
,':');
448 void addReplyMultiBulkLen(redisClient
*c
, long length
) {
449 addReplyLongLongWithPrefix(c
,length
,'*');
452 /* Create the length prefix of a bulk reply, example: $2234 */
453 void addReplyBulkLen(redisClient
*c
, robj
*obj
) {
456 if (obj
->encoding
== REDIS_ENCODING_RAW
) {
457 len
= sdslen(obj
->ptr
);
459 long n
= (long)obj
->ptr
;
461 /* Compute how many bytes will take this integer as a radix 10 string */
467 while((n
= n
/10) != 0) {
471 addReplyLongLongWithPrefix(c
,len
,'$');
474 /* Add a Redis Object as a bulk reply */
475 void addReplyBulk(redisClient
*c
, robj
*obj
) {
476 addReplyBulkLen(c
,obj
);
478 addReply(c
,shared
.crlf
);
481 /* Add a C buffer as bulk reply */
482 void addReplyBulkCBuffer(redisClient
*c
, void *p
, size_t len
) {
483 addReplyLongLongWithPrefix(c
,len
,'$');
484 addReplyString(c
,p
,len
);
485 addReply(c
,shared
.crlf
);
488 /* Add a C nul term string as bulk reply */
489 void addReplyBulkCString(redisClient
*c
, char *s
) {
491 addReply(c
,shared
.nullbulk
);
493 addReplyBulkCBuffer(c
,s
,strlen(s
));
497 /* Add a long long as a bulk reply */
498 void addReplyBulkLongLong(redisClient
*c
, long long ll
) {
502 len
= ll2string(buf
,64,ll
);
503 addReplyBulkCBuffer(c
,buf
,len
);
506 /* Copy 'src' client output buffers into 'dst' client output buffers.
507 * The function takes care of freeing the old output buffers of the
508 * destination client. */
509 void copyClientOutputBuffer(redisClient
*dst
, redisClient
*src
) {
510 listRelease(dst
->reply
);
511 dst
->reply
= listDup(src
->reply
);
512 memcpy(dst
->buf
,src
->buf
,src
->bufpos
);
513 dst
->bufpos
= src
->bufpos
;
514 dst
->reply_bytes
= src
->reply_bytes
;
517 static void acceptCommonHandler(int fd
, int flags
) {
519 if ((c
= createClient(fd
)) == NULL
) {
520 redisLog(REDIS_WARNING
,"Error allocating resources for the client");
521 close(fd
); /* May be already closed, just ignore errors */
524 /* If maxclient directive is set and this is one client more... close the
525 * connection. Note that we create the client instead to check before
526 * for this condition, since now the socket is already set in nonblocking
527 * mode and we can send an error for free using the Kernel I/O */
528 if (listLength(server
.clients
) > server
.maxclients
) {
529 char *err
= "-ERR max number of clients reached\r\n";
531 /* That's a best effort error message, don't check write errors */
532 if (write(c
->fd
,err
,strlen(err
)) == -1) {
533 /* Nothing to do, Just to avoid the warning... */
535 server
.stat_rejected_conn
++;
539 server
.stat_numconnections
++;
543 void acceptTcpHandler(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
548 REDIS_NOTUSED(privdata
);
550 cfd
= anetTcpAccept(server
.neterr
, fd
, cip
, &cport
);
552 redisLog(REDIS_WARNING
,"Accepting client connection: %s", server
.neterr
);
555 redisLog(REDIS_VERBOSE
,"Accepted %s:%d", cip
, cport
);
556 acceptCommonHandler(cfd
,0);
559 void acceptUnixHandler(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
563 REDIS_NOTUSED(privdata
);
565 cfd
= anetUnixAccept(server
.neterr
, fd
);
567 redisLog(REDIS_WARNING
,"Accepting client connection: %s", server
.neterr
);
570 redisLog(REDIS_VERBOSE
,"Accepted connection to %s", server
.unixsocket
);
571 acceptCommonHandler(cfd
,REDIS_UNIX_SOCKET
);
575 static void freeClientArgv(redisClient
*c
) {
577 for (j
= 0; j
< c
->argc
; j
++)
578 decrRefCount(c
->argv
[j
]);
583 /* Close all the slaves connections. This is useful in chained replication
584 * when we resync with our own master and want to force all our slaves to
585 * resync with us as well. */
586 void disconnectSlaves(void) {
587 while (listLength(server
.slaves
)) {
588 listNode
*ln
= listFirst(server
.slaves
);
589 freeClient((redisClient
*)ln
->value
);
593 void freeClient(redisClient
*c
) {
596 /* If this is marked as current client unset it */
597 if (server
.current_client
== c
) server
.current_client
= NULL
;
599 /* Note that if the client we are freeing is blocked into a blocking
600 * call, we have to set querybuf to NULL *before* to call
601 * unblockClientWaitingData() to avoid processInputBuffer() will get
602 * called. Also it is important to remove the file events after
603 * this, because this call adds the READABLE event. */
604 sdsfree(c
->querybuf
);
606 if (c
->flags
& REDIS_BLOCKED
)
607 unblockClientWaitingData(c
);
609 /* UNWATCH all the keys */
611 listRelease(c
->watched_keys
);
612 /* Unsubscribe from all the pubsub channels */
613 pubsubUnsubscribeAllChannels(c
,0);
614 pubsubUnsubscribeAllPatterns(c
,0);
615 dictRelease(c
->pubsub_channels
);
616 listRelease(c
->pubsub_patterns
);
617 /* Obvious cleanup */
618 aeDeleteFileEvent(server
.el
,c
->fd
,AE_READABLE
);
619 aeDeleteFileEvent(server
.el
,c
->fd
,AE_WRITABLE
);
620 listRelease(c
->reply
);
623 /* Remove from the list of clients */
624 ln
= listSearchKey(server
.clients
,c
);
625 redisAssert(ln
!= NULL
);
626 listDelNode(server
.clients
,ln
);
627 /* When client was just unblocked because of a blocking operation,
628 * remove it from the list with unblocked clients. */
629 if (c
->flags
& REDIS_UNBLOCKED
) {
630 ln
= listSearchKey(server
.unblocked_clients
,c
);
631 redisAssert(ln
!= NULL
);
632 listDelNode(server
.unblocked_clients
,ln
);
634 listRelease(c
->io_keys
);
635 /* Master/slave cleanup.
636 * Case 1: we lost the connection with a slave. */
637 if (c
->flags
& REDIS_SLAVE
) {
638 if (c
->replstate
== REDIS_REPL_SEND_BULK
&& c
->repldbfd
!= -1)
640 list
*l
= (c
->flags
& REDIS_MONITOR
) ? server
.monitors
: server
.slaves
;
641 ln
= listSearchKey(l
,c
);
642 redisAssert(ln
!= NULL
);
646 /* Case 2: we lost the connection with the master. */
647 if (c
->flags
& REDIS_MASTER
) {
648 server
.master
= NULL
;
649 server
.repl_state
= REDIS_REPL_CONNECT
;
650 server
.repl_down_since
= server
.unixtime
;
651 /* We lost connection with our master, force our slaves to resync
652 * with us as well to load the new data set.
654 * If server.masterhost is NULL the user called SLAVEOF NO ONE so
655 * slave resync is not needed. */
656 if (server
.masterhost
!= NULL
) disconnectSlaves();
659 /* If this client was scheduled for async freeing we need to remove it
661 if (c
->flags
& REDIS_CLOSE_ASAP
) {
662 ln
= listSearchKey(server
.clients_to_close
,c
);
663 redisAssert(ln
!= NULL
);
664 listDelNode(server
.clients_to_close
,ln
);
669 freeClientMultiState(c
);
673 /* Schedule a client to free it at a safe time in the serverCron() function.
674 * This function is useful when we need to terminate a client but we are in
675 * a context where calling freeClient() is not possible, because the client
676 * should be valid for the continuation of the flow of the program. */
677 void freeClientAsync(redisClient
*c
) {
678 if (c
->flags
& REDIS_CLOSE_ASAP
) return;
679 c
->flags
|= REDIS_CLOSE_ASAP
;
680 listAddNodeTail(server
.clients_to_close
,c
);
683 void freeClientsInAsyncFreeQueue(void) {
684 while (listLength(server
.clients_to_close
)) {
685 listNode
*ln
= listFirst(server
.clients_to_close
);
686 redisClient
*c
= listNodeValue(ln
);
688 c
->flags
&= ~REDIS_CLOSE_ASAP
;
690 listDelNode(server
.clients_to_close
,ln
);
694 void sendReplyToClient(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
695 redisClient
*c
= privdata
;
696 int nwritten
= 0, totwritten
= 0, objlen
;
702 while(c
->bufpos
> 0 || listLength(c
->reply
)) {
704 if (c
->flags
& REDIS_MASTER
) {
705 /* Don't reply to a master */
706 nwritten
= c
->bufpos
- c
->sentlen
;
708 nwritten
= write(fd
,c
->buf
+c
->sentlen
,c
->bufpos
-c
->sentlen
);
709 if (nwritten
<= 0) break;
711 c
->sentlen
+= nwritten
;
712 totwritten
+= nwritten
;
714 /* If the buffer was sent, set bufpos to zero to continue with
715 * the remainder of the reply. */
716 if (c
->sentlen
== c
->bufpos
) {
721 o
= listNodeValue(listFirst(c
->reply
));
722 objlen
= sdslen(o
->ptr
);
723 objmem
= zmalloc_size_sds(o
->ptr
);
726 listDelNode(c
->reply
,listFirst(c
->reply
));
730 if (c
->flags
& REDIS_MASTER
) {
731 /* Don't reply to a master */
732 nwritten
= objlen
- c
->sentlen
;
734 nwritten
= write(fd
, ((char*)o
->ptr
)+c
->sentlen
,objlen
-c
->sentlen
);
735 if (nwritten
<= 0) break;
737 c
->sentlen
+= nwritten
;
738 totwritten
+= nwritten
;
740 /* If we fully sent the object on head go to the next one */
741 if (c
->sentlen
== objlen
) {
742 listDelNode(c
->reply
,listFirst(c
->reply
));
744 c
->reply_bytes
-= objmem
;
747 /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT
748 * bytes, in a single threaded server it's a good idea to serve
749 * other clients as well, even if a very large request comes from
750 * super fast link that is always able to accept data (in real world
751 * scenario think about 'KEYS *' against the loopback interface).
753 * However if we are over the maxmemory limit we ignore that and
754 * just deliver as much data as it is possible to deliver. */
755 if (totwritten
> REDIS_MAX_WRITE_PER_EVENT
&&
756 (server
.maxmemory
== 0 ||
757 zmalloc_used_memory() < server
.maxmemory
)) break;
759 if (nwritten
== -1) {
760 if (errno
== EAGAIN
) {
763 redisLog(REDIS_VERBOSE
,
764 "Error writing to client: %s", strerror(errno
));
769 if (totwritten
> 0) c
->lastinteraction
= server
.unixtime
;
770 if (c
->bufpos
== 0 && listLength(c
->reply
) == 0) {
772 aeDeleteFileEvent(server
.el
,c
->fd
,AE_WRITABLE
);
774 /* Close connection after entire reply has been sent. */
775 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) freeClient(c
);
779 /* resetClient prepare the client to process the next command */
780 void resetClient(redisClient
*c
) {
785 /* We clear the ASKING flag as well if we are not inside a MULTI. */
786 if (!(c
->flags
& REDIS_MULTI
)) c
->flags
&= (~REDIS_ASKING
);
789 int processInlineBuffer(redisClient
*c
) {
790 char *newline
= strstr(c
->querybuf
,"\r\n");
795 /* Nothing to do without a \r\n */
796 if (newline
== NULL
) {
797 if (sdslen(c
->querybuf
) > REDIS_INLINE_MAX_SIZE
) {
798 addReplyError(c
,"Protocol error: too big inline request");
799 setProtocolError(c
,0);
804 /* Split the input buffer up to the \r\n */
805 querylen
= newline
-(c
->querybuf
);
806 argv
= sdssplitlen(c
->querybuf
,querylen
," ",1,&argc
);
808 /* Leave data after the first line of the query in the buffer */
809 c
->querybuf
= sdsrange(c
->querybuf
,querylen
+2,-1);
811 /* Setup argv array on client structure */
812 if (c
->argv
) zfree(c
->argv
);
813 c
->argv
= zmalloc(sizeof(robj
*)*argc
);
815 /* Create redis objects for all arguments. */
816 for (c
->argc
= 0, j
= 0; j
< argc
; j
++) {
817 if (sdslen(argv
[j
])) {
818 c
->argv
[c
->argc
] = createObject(REDIS_STRING
,argv
[j
]);
828 /* Helper function. Trims query buffer to make the function that processes
829 * multi bulk requests idempotent. */
830 static void setProtocolError(redisClient
*c
, int pos
) {
831 if (server
.verbosity
>= REDIS_VERBOSE
) {
832 sds client
= getClientInfoString(c
);
833 redisLog(REDIS_VERBOSE
,
834 "Protocol error from client: %s", client
);
837 c
->flags
|= REDIS_CLOSE_AFTER_REPLY
;
838 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
841 int processMultibulkBuffer(redisClient
*c
) {
842 char *newline
= NULL
;
846 if (c
->multibulklen
== 0) {
847 /* The client should have been reset */
848 redisAssertWithInfo(c
,NULL
,c
->argc
== 0);
850 /* Multi bulk length cannot be read without a \r\n */
851 newline
= strchr(c
->querybuf
,'\r');
852 if (newline
== NULL
) {
853 if (sdslen(c
->querybuf
) > REDIS_INLINE_MAX_SIZE
) {
854 addReplyError(c
,"Protocol error: too big mbulk count string");
855 setProtocolError(c
,0);
860 /* Buffer should also contain \n */
861 if (newline
-(c
->querybuf
) > ((signed)sdslen(c
->querybuf
)-2))
864 /* We know for sure there is a whole line since newline != NULL,
865 * so go ahead and find out the multi bulk length. */
866 redisAssertWithInfo(c
,NULL
,c
->querybuf
[0] == '*');
867 ok
= string2ll(c
->querybuf
+1,newline
-(c
->querybuf
+1),&ll
);
868 if (!ok
|| ll
> 1024*1024) {
869 addReplyError(c
,"Protocol error: invalid multibulk length");
870 setProtocolError(c
,pos
);
874 pos
= (newline
-c
->querybuf
)+2;
876 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
880 c
->multibulklen
= ll
;
882 /* Setup argv array on client structure */
883 if (c
->argv
) zfree(c
->argv
);
884 c
->argv
= zmalloc(sizeof(robj
*)*c
->multibulklen
);
887 redisAssertWithInfo(c
,NULL
,c
->multibulklen
> 0);
888 while(c
->multibulklen
) {
889 /* Read bulk length if unknown */
890 if (c
->bulklen
== -1) {
891 newline
= strchr(c
->querybuf
+pos
,'\r');
892 if (newline
== NULL
) {
893 if (sdslen(c
->querybuf
) > REDIS_INLINE_MAX_SIZE
) {
894 addReplyError(c
,"Protocol error: too big bulk count string");
895 setProtocolError(c
,0);
900 /* Buffer should also contain \n */
901 if (newline
-(c
->querybuf
) > ((signed)sdslen(c
->querybuf
)-2))
904 if (c
->querybuf
[pos
] != '$') {
905 addReplyErrorFormat(c
,
906 "Protocol error: expected '$', got '%c'",
908 setProtocolError(c
,pos
);
912 ok
= string2ll(c
->querybuf
+pos
+1,newline
-(c
->querybuf
+pos
+1),&ll
);
913 if (!ok
|| ll
< 0 || ll
> 512*1024*1024) {
914 addReplyError(c
,"Protocol error: invalid bulk length");
915 setProtocolError(c
,pos
);
919 pos
+= newline
-(c
->querybuf
+pos
)+2;
920 if (ll
>= REDIS_MBULK_BIG_ARG
) {
921 /* If we are going to read a large object from network
922 * try to make it likely that it will start at c->querybuf
923 * boundary so that we can optimized object creation
924 * avoiding a large copy of data. */
925 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
927 /* Hint the sds library about the amount of bytes this string is
928 * going to contain. */
929 c
->querybuf
= sdsMakeRoomFor(c
->querybuf
,ll
+2);
934 /* Read bulk argument */
935 if (sdslen(c
->querybuf
)-pos
< (unsigned)(c
->bulklen
+2)) {
936 /* Not enough data (+2 == trailing \r\n) */
939 /* Optimization: if the buffer contanins JUST our bulk element
940 * instead of creating a new object by *copying* the sds we
941 * just use the current sds string. */
943 c
->bulklen
>= REDIS_MBULK_BIG_ARG
&&
944 (signed) sdslen(c
->querybuf
) == c
->bulklen
+2)
946 c
->argv
[c
->argc
++] = createObject(REDIS_STRING
,c
->querybuf
);
947 sdsIncrLen(c
->querybuf
,-2); /* remove CRLF */
948 c
->querybuf
= sdsempty();
949 /* Assume that if we saw a fat argument we'll see another one
951 c
->querybuf
= sdsMakeRoomFor(c
->querybuf
,c
->bulklen
+2);
955 createStringObject(c
->querybuf
+pos
,c
->bulklen
);
964 if (pos
) c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
966 /* We're done when c->multibulk == 0 */
967 if (c
->multibulklen
== 0) return REDIS_OK
;
969 /* Still not read to process the command */
973 void processInputBuffer(redisClient
*c
) {
974 /* Keep processing while there is something in the input buffer */
975 while(sdslen(c
->querybuf
)) {
976 /* Immediately abort if the client is in the middle of something. */
977 if (c
->flags
& REDIS_BLOCKED
) return;
979 /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
980 * written to the client. Make sure to not let the reply grow after
981 * this flag has been set (i.e. don't process more commands). */
982 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
984 /* Determine request type when unknown. */
986 if (c
->querybuf
[0] == '*') {
987 c
->reqtype
= REDIS_REQ_MULTIBULK
;
989 c
->reqtype
= REDIS_REQ_INLINE
;
993 if (c
->reqtype
== REDIS_REQ_INLINE
) {
994 if (processInlineBuffer(c
) != REDIS_OK
) break;
995 } else if (c
->reqtype
== REDIS_REQ_MULTIBULK
) {
996 if (processMultibulkBuffer(c
) != REDIS_OK
) break;
998 redisPanic("Unknown request type");
1001 /* Multibulk processing could see a <= 0 length. */
1005 /* Only reset the client when the command was executed. */
1006 if (processCommand(c
) == REDIS_OK
)
1012 void readQueryFromClient(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
1013 redisClient
*c
= (redisClient
*) privdata
;
1017 REDIS_NOTUSED(mask
);
1019 server
.current_client
= c
;
1020 readlen
= REDIS_IOBUF_LEN
;
1021 /* If this is a multi bulk request, and we are processing a bulk reply
1022 * that is large enough, try to maximize the probability that the query
1023 * buffer contains exactly the SDS string representing the object, even
1024 * at the risk of requiring more read(2) calls. This way the function
1025 * processMultiBulkBuffer() can avoid copying buffers to create the
1026 * Redis Object representing the argument. */
1027 if (c
->reqtype
== REDIS_REQ_MULTIBULK
&& c
->multibulklen
&& c
->bulklen
!= -1
1028 && c
->bulklen
>= REDIS_MBULK_BIG_ARG
)
1030 int remaining
= (unsigned)(c
->bulklen
+2)-sdslen(c
->querybuf
);
1032 if (remaining
< readlen
) readlen
= remaining
;
1035 qblen
= sdslen(c
->querybuf
);
1036 if (c
->querybuf_peak
< qblen
) c
->querybuf_peak
= qblen
;
1037 c
->querybuf
= sdsMakeRoomFor(c
->querybuf
, readlen
);
1038 nread
= read(fd
, c
->querybuf
+qblen
, readlen
);
1040 if (errno
== EAGAIN
) {
1043 redisLog(REDIS_VERBOSE
, "Reading from client: %s",strerror(errno
));
1047 } else if (nread
== 0) {
1048 redisLog(REDIS_VERBOSE
, "Client closed connection");
1053 sdsIncrLen(c
->querybuf
,nread
);
1054 c
->lastinteraction
= server
.unixtime
;
1056 server
.current_client
= NULL
;
1059 if (sdslen(c
->querybuf
) > server
.client_max_querybuf_len
) {
1060 sds ci
= getClientInfoString(c
), bytes
= sdsempty();
1062 bytes
= sdscatrepr(bytes
,c
->querybuf
,64);
1063 redisLog(REDIS_WARNING
,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci
, bytes
);
1069 processInputBuffer(c
);
1070 server
.current_client
= NULL
;
1073 void getClientsMaxBuffers(unsigned long *longest_output_list
,
1074 unsigned long *biggest_input_buffer
) {
1078 unsigned long lol
= 0, bib
= 0;
1080 listRewind(server
.clients
,&li
);
1081 while ((ln
= listNext(&li
)) != NULL
) {
1082 c
= listNodeValue(ln
);
1084 if (listLength(c
->reply
) > lol
) lol
= listLength(c
->reply
);
1085 if (sdslen(c
->querybuf
) > bib
) bib
= sdslen(c
->querybuf
);
1087 *longest_output_list
= lol
;
1088 *biggest_input_buffer
= bib
;
1091 /* Turn a Redis client into an sds string representing its state. */
1092 sds
getClientInfoString(redisClient
*client
) {
1093 char ip
[32], flags
[16], events
[3], *p
;
1094 int port
= 0; /* initialized to zero for the unix socket case. */
1097 if (!(client
->flags
& REDIS_UNIX_SOCKET
))
1098 anetPeerToString(client
->fd
,ip
,&port
);
1100 if (client
->flags
& REDIS_SLAVE
) {
1101 if (client
->flags
& REDIS_MONITOR
)
1106 if (client
->flags
& REDIS_MASTER
) *p
++ = 'M';
1107 if (client
->flags
& REDIS_MULTI
) *p
++ = 'x';
1108 if (client
->flags
& REDIS_BLOCKED
) *p
++ = 'b';
1109 if (client
->flags
& REDIS_DIRTY_CAS
) *p
++ = 'd';
1110 if (client
->flags
& REDIS_CLOSE_AFTER_REPLY
) *p
++ = 'c';
1111 if (client
->flags
& REDIS_UNBLOCKED
) *p
++ = 'u';
1112 if (client
->flags
& REDIS_CLOSE_ASAP
) *p
++ = 'A';
1113 if (client
->flags
& REDIS_UNIX_SOCKET
) *p
++ = 'U';
1114 if (p
== flags
) *p
++ = 'N';
1117 emask
= client
->fd
== -1 ? 0 : aeGetFileEvents(server
.el
,client
->fd
);
1119 if (emask
& AE_READABLE
) *p
++ = 'r';
1120 if (emask
& AE_WRITABLE
) *p
++ = 'w';
1122 return sdscatprintf(sdsempty(),
1123 "addr=%s:%d fd=%d age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d multi=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
1124 (client
->flags
& REDIS_UNIX_SOCKET
) ? server
.unixsocket
: ip
,
1126 (long)(server
.unixtime
- client
->ctime
),
1127 (long)(server
.unixtime
- client
->lastinteraction
),
1130 (int) dictSize(client
->pubsub_channels
),
1131 (int) listLength(client
->pubsub_patterns
),
1132 (client
->flags
& REDIS_MULTI
) ? client
->mstate
.count
: -1,
1133 (unsigned long) sdslen(client
->querybuf
),
1134 (unsigned long) sdsavail(client
->querybuf
),
1135 (unsigned long) client
->bufpos
,
1136 (unsigned long) listLength(client
->reply
),
1137 getClientOutputBufferMemoryUsage(client
),
1139 client
->lastcmd
? client
->lastcmd
->name
: "NULL");
1142 sds
getAllClientsInfoString(void) {
1145 redisClient
*client
;
1148 listRewind(server
.clients
,&li
);
1149 while ((ln
= listNext(&li
)) != NULL
) {
1152 client
= listNodeValue(ln
);
1153 cs
= getClientInfoString(client
);
1154 o
= sdscatsds(o
,cs
);
1156 o
= sdscatlen(o
,"\n",1);
1161 void clientCommand(redisClient
*c
) {
1164 redisClient
*client
;
1166 if (!strcasecmp(c
->argv
[1]->ptr
,"list") && c
->argc
== 2) {
1167 sds o
= getAllClientsInfoString();
1168 addReplyBulkCBuffer(c
,o
,sdslen(o
));
1170 } else if (!strcasecmp(c
->argv
[1]->ptr
,"kill") && c
->argc
== 3) {
1171 listRewind(server
.clients
,&li
);
1172 while ((ln
= listNext(&li
)) != NULL
) {
1173 char ip
[32], addr
[64];
1176 client
= listNodeValue(ln
);
1177 if (anetPeerToString(client
->fd
,ip
,&port
) == -1) continue;
1178 snprintf(addr
,sizeof(addr
),"%s:%d",ip
,port
);
1179 if (strcmp(addr
,c
->argv
[2]->ptr
) == 0) {
1180 addReply(c
,shared
.ok
);
1182 client
->flags
|= REDIS_CLOSE_AFTER_REPLY
;
1189 addReplyError(c
,"No such client");
1191 addReplyError(c
, "Syntax error, try CLIENT (LIST | KILL ip:port)");
1195 /* Rewrite the command vector of the client. All the new objects ref count
1196 * is incremented. The old command vector is freed, and the old objects
1197 * ref count is decremented. */
1198 void rewriteClientCommandVector(redisClient
*c
, int argc
, ...) {
1201 robj
**argv
; /* The new argument vector */
1203 argv
= zmalloc(sizeof(robj
*)*argc
);
1205 for (j
= 0; j
< argc
; j
++) {
1208 a
= va_arg(ap
, robj
*);
1212 /* We free the objects in the original vector at the end, so we are
1213 * sure that if the same objects are reused in the new vector the
1214 * refcount gets incremented before it gets decremented. */
1215 for (j
= 0; j
< c
->argc
; j
++) decrRefCount(c
->argv
[j
]);
1217 /* Replace argv and argc with our new versions. */
1220 c
->cmd
= lookupCommand(c
->argv
[0]->ptr
);
1221 redisAssertWithInfo(c
,NULL
,c
->cmd
!= NULL
);
1225 /* Rewrite a single item in the command vector.
1226 * The new val ref count is incremented, and the old decremented. */
1227 void rewriteClientCommandArgument(redisClient
*c
, int i
, robj
*newval
) {
1230 redisAssertWithInfo(c
,NULL
,i
< c
->argc
);
1231 oldval
= c
->argv
[i
];
1232 c
->argv
[i
] = newval
;
1233 incrRefCount(newval
);
1234 decrRefCount(oldval
);
1236 /* If this is the command name make sure to fix c->cmd. */
1238 c
->cmd
= lookupCommand(c
->argv
[0]->ptr
);
1239 redisAssertWithInfo(c
,NULL
,c
->cmd
!= NULL
);
1243 /* This function returns the number of bytes that Redis is virtually
1244 * using to store the reply still not read by the client.
1245 * It is "virtual" since the reply output list may contain objects that
1246 * are shared and are not really using additional memory.
1248 * The function returns the total sum of the length of all the objects
1249 * stored in the output list, plus the memory used to allocate every
1250 * list node. The static reply buffer is not taken into account since it
1251 * is allocated anyway.
1253 * Note: this function is very fast so can be called as many time as
1254 * the caller wishes. The main usage of this function currently is
1255 * enforcing the client output length limits. */
1256 unsigned long getClientOutputBufferMemoryUsage(redisClient
*c
) {
1257 unsigned long list_item_size
= sizeof(listNode
)+sizeof(robj
);
1259 return c
->reply_bytes
+ (list_item_size
*listLength(c
->reply
));
1262 /* Get the class of a client, used in order to enforce limits to different
1263 * classes of clients.
1265 * The function will return one of the following:
1266 * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client
1267 * REDIS_CLIENT_LIMIT_CLASS_SLAVE -> Slave or client executing MONITOR command
1268 * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels
1270 int getClientLimitClass(redisClient
*c
) {
1271 if (c
->flags
& REDIS_SLAVE
) return REDIS_CLIENT_LIMIT_CLASS_SLAVE
;
1272 if (dictSize(c
->pubsub_channels
) || listLength(c
->pubsub_patterns
))
1273 return REDIS_CLIENT_LIMIT_CLASS_PUBSUB
;
1274 return REDIS_CLIENT_LIMIT_CLASS_NORMAL
;
1277 int getClientLimitClassByName(char *name
) {
1278 if (!strcasecmp(name
,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL
;
1279 else if (!strcasecmp(name
,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE
;
1280 else if (!strcasecmp(name
,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB
;
1284 char *getClientLimitClassName(int class) {
1286 case REDIS_CLIENT_LIMIT_CLASS_NORMAL
: return "normal";
1287 case REDIS_CLIENT_LIMIT_CLASS_SLAVE
: return "slave";
1288 case REDIS_CLIENT_LIMIT_CLASS_PUBSUB
: return "pubsub";
1289 default: return NULL
;
1293 /* The function checks if the client reached output buffer soft or hard
1294 * limit, and also update the state needed to check the soft limit as
1297 * Return value: non-zero if the client reached the soft or the hard limit.
1298 * Otherwise zero is returned. */
1299 int checkClientOutputBufferLimits(redisClient
*c
) {
1300 int soft
= 0, hard
= 0, class;
1301 unsigned long used_mem
= getClientOutputBufferMemoryUsage(c
);
1303 class = getClientLimitClass(c
);
1304 if (server
.client_obuf_limits
[class].hard_limit_bytes
&&
1305 used_mem
>= server
.client_obuf_limits
[class].hard_limit_bytes
)
1307 if (server
.client_obuf_limits
[class].soft_limit_bytes
&&
1308 used_mem
>= server
.client_obuf_limits
[class].soft_limit_bytes
)
1311 /* We need to check if the soft limit is reached continuously for the
1312 * specified amount of seconds. */
1314 if (c
->obuf_soft_limit_reached_time
== 0) {
1315 c
->obuf_soft_limit_reached_time
= server
.unixtime
;
1316 soft
= 0; /* First time we see the soft limit reached */
1318 time_t elapsed
= server
.unixtime
- c
->obuf_soft_limit_reached_time
;
1321 server
.client_obuf_limits
[class].soft_limit_seconds
) {
1322 soft
= 0; /* The client still did not reached the max number of
1323 seconds for the soft limit to be considered
1328 c
->obuf_soft_limit_reached_time
= 0;
1330 return soft
|| hard
;
1333 /* Asynchronously close a client if soft or hard limit is reached on the
1334 * output buffer size. The caller can check if the client will be closed
1335 * checking if the client REDIS_CLOSE_ASAP flag is set.
1337 * Note: we need to close the client asynchronously because this function is
1338 * called from contexts where the client can't be freed safely, i.e. from the
1339 * lower level functions pushing data inside the client output buffers. */
1340 void asyncCloseClientOnOutputBufferLimitReached(redisClient
*c
) {
1341 redisAssert(c
->reply_bytes
< ULONG_MAX
-(1024*64));
1342 if (c
->reply_bytes
== 0 || c
->flags
& REDIS_CLOSE_ASAP
) return;
1343 if (checkClientOutputBufferLimits(c
)) {
1344 sds client
= getClientInfoString(c
);
1347 redisLog(REDIS_WARNING
,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client
);
1352 /* Helper function used by freeMemoryIfNeeded() in order to flush slaves
1353 * output buffers without returning control to the event loop. */
1354 void flushSlavesOutputBuffers(void) {
1358 listRewind(server
.slaves
,&li
);
1359 while((ln
= listNext(&li
))) {
1360 redisClient
*slave
= listNodeValue(ln
);
1363 events
= aeGetFileEvents(server
.el
,slave
->fd
);
1364 if (events
& AE_WRITABLE
&&
1365 slave
->replstate
== REDIS_REPL_ONLINE
&&
1366 listLength(slave
->reply
))
1368 sendReplyToClient(server
.el
,slave
->fd
,slave
,0);