4 void *dupClientReplyValue(void *o
) {
5 incrRefCount((robj
*)o
);
9 int listMatchObjects(void *a
, void *b
) {
10 return equalStringObjects(a
,b
);
13 redisClient
*createClient(int fd
) {
14 redisClient
*c
= zmalloc(sizeof(redisClient
));
17 /* passing -1 as fd it is possible to create a non connected client.
18 * This is useful since all the Redis commands needs to be executed
19 * in the context of a client. When commands are executed in other
20 * contexts (for instance a Lua script) we need a non connected client. */
22 anetNonBlock(NULL
,fd
);
23 anetTcpNoDelay(NULL
,fd
);
24 if (aeCreateFileEvent(server
.el
,fd
,AE_READABLE
,
25 readQueryFromClient
, c
) == AE_ERR
)
35 c
->querybuf
= sdsempty();
44 c
->lastinteraction
= time(NULL
);
46 c
->replstate
= REDIS_REPL_NONE
;
47 c
->reply
= listCreate();
48 listSetFreeMethod(c
->reply
,decrRefCount
);
49 listSetDupMethod(c
->reply
,dupClientReplyValue
);
53 c
->bpop
.target
= NULL
;
54 c
->io_keys
= listCreate();
55 c
->watched_keys
= listCreate();
56 listSetFreeMethod(c
->io_keys
,decrRefCount
);
57 c
->pubsub_channels
= dictCreate(&setDictType
,NULL
);
58 c
->pubsub_patterns
= listCreate();
59 listSetFreeMethod(c
->pubsub_patterns
,decrRefCount
);
60 listSetMatchMethod(c
->pubsub_patterns
,listMatchObjects
);
61 if (fd
!= -1) listAddNodeTail(server
.clients
,c
);
62 initClientMultiState(c
);
66 /* Set the event loop to listen for write events on the client's socket.
67 * Typically gets called every time a reply is built. */
68 int _installWriteEvent(redisClient
*c
) {
69 if (c
->flags
& REDIS_LUA_CLIENT
) return REDIS_OK
;
70 if (c
->fd
<= 0) return REDIS_ERR
;
71 if (c
->bufpos
== 0 && listLength(c
->reply
) == 0 &&
72 (c
->replstate
== REDIS_REPL_NONE
||
73 c
->replstate
== REDIS_REPL_ONLINE
) &&
74 aeCreateFileEvent(server
.el
, c
->fd
, AE_WRITABLE
,
75 sendReplyToClient
, c
) == AE_ERR
) return REDIS_ERR
;
79 /* Create a duplicate of the last object in the reply list when
80 * it is not exclusively owned by the reply list. */
81 robj
*dupLastObjectIfNeeded(list
*reply
) {
84 redisAssert(listLength(reply
) > 0);
86 cur
= listNodeValue(ln
);
87 if (cur
->refcount
> 1) {
88 new = dupStringObject(cur
);
90 listNodeValue(ln
) = new;
92 return listNodeValue(ln
);
95 /* -----------------------------------------------------------------------------
96 * Low level functions to add more data to output buffers.
97 * -------------------------------------------------------------------------- */
99 int _addReplyToBuffer(redisClient
*c
, char *s
, size_t len
) {
100 size_t available
= sizeof(c
->buf
)-c
->bufpos
;
102 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return REDIS_OK
;
104 /* If there already are entries in the reply list, we cannot
105 * add anything more to the static buffer. */
106 if (listLength(c
->reply
) > 0) return REDIS_ERR
;
108 /* Check that the buffer has enough space available for this string. */
109 if (len
> available
) return REDIS_ERR
;
111 memcpy(c
->buf
+c
->bufpos
,s
,len
);
116 void _addReplyObjectToList(redisClient
*c
, robj
*o
) {
119 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
121 if (listLength(c
->reply
) == 0) {
123 listAddNodeTail(c
->reply
,o
);
125 tail
= listNodeValue(listLast(c
->reply
));
127 /* Append to this object when possible. */
128 if (tail
->ptr
!= NULL
&&
129 sdslen(tail
->ptr
)+sdslen(o
->ptr
) <= REDIS_REPLY_CHUNK_BYTES
)
131 tail
= dupLastObjectIfNeeded(c
->reply
);
132 tail
->ptr
= sdscatlen(tail
->ptr
,o
->ptr
,sdslen(o
->ptr
));
135 listAddNodeTail(c
->reply
,o
);
140 /* This method takes responsibility over the sds. When it is no longer
141 * needed it will be free'd, otherwise it ends up in a robj. */
142 void _addReplySdsToList(redisClient
*c
, sds s
) {
145 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) {
150 if (listLength(c
->reply
) == 0) {
151 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,s
));
153 tail
= listNodeValue(listLast(c
->reply
));
155 /* Append to this object when possible. */
156 if (tail
->ptr
!= NULL
&&
157 sdslen(tail
->ptr
)+sdslen(s
) <= REDIS_REPLY_CHUNK_BYTES
)
159 tail
= dupLastObjectIfNeeded(c
->reply
);
160 tail
->ptr
= sdscatlen(tail
->ptr
,s
,sdslen(s
));
163 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,s
));
168 void _addReplyStringToList(redisClient
*c
, char *s
, size_t len
) {
171 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
173 if (listLength(c
->reply
) == 0) {
174 listAddNodeTail(c
->reply
,createStringObject(s
,len
));
176 tail
= listNodeValue(listLast(c
->reply
));
178 /* Append to this object when possible. */
179 if (tail
->ptr
!= NULL
&&
180 sdslen(tail
->ptr
)+len
<= REDIS_REPLY_CHUNK_BYTES
)
182 tail
= dupLastObjectIfNeeded(c
->reply
);
183 tail
->ptr
= sdscatlen(tail
->ptr
,s
,len
);
185 listAddNodeTail(c
->reply
,createStringObject(s
,len
));
190 /* -----------------------------------------------------------------------------
191 * Higher level functions to queue data on the client output buffer.
192 * The following functions are the ones that commands implementations will call.
193 * -------------------------------------------------------------------------- */
195 void addReply(redisClient
*c
, robj
*obj
) {
196 if (_installWriteEvent(c
) != REDIS_OK
) return;
198 /* This is an important place where we can avoid copy-on-write
199 * when there is a saving child running, avoiding touching the
200 * refcount field of the object if it's not needed.
202 * If the encoding is RAW and there is room in the static buffer
203 * we'll be able to send the object to the client without
204 * messing with its page. */
205 if (obj
->encoding
== REDIS_ENCODING_RAW
) {
206 if (_addReplyToBuffer(c
,obj
->ptr
,sdslen(obj
->ptr
)) != REDIS_OK
)
207 _addReplyObjectToList(c
,obj
);
209 /* FIXME: convert the long into string and use _addReplyToBuffer()
210 * instead of calling getDecodedObject. As this place in the
211 * code is too performance critical. */
212 obj
= getDecodedObject(obj
);
213 if (_addReplyToBuffer(c
,obj
->ptr
,sdslen(obj
->ptr
)) != REDIS_OK
)
214 _addReplyObjectToList(c
,obj
);
219 void addReplySds(redisClient
*c
, sds s
) {
220 if (_installWriteEvent(c
) != REDIS_OK
) {
221 /* The caller expects the sds to be free'd. */
225 if (_addReplyToBuffer(c
,s
,sdslen(s
)) == REDIS_OK
) {
228 /* This method free's the sds when it is no longer needed. */
229 _addReplySdsToList(c
,s
);
233 void addReplyString(redisClient
*c
, char *s
, size_t len
) {
234 if (_installWriteEvent(c
) != REDIS_OK
) return;
235 if (_addReplyToBuffer(c
,s
,len
) != REDIS_OK
)
236 _addReplyStringToList(c
,s
,len
);
239 void _addReplyError(redisClient
*c
, char *s
, size_t len
) {
240 addReplyString(c
,"-ERR ",5);
241 addReplyString(c
,s
,len
);
242 addReplyString(c
,"\r\n",2);
245 void addReplyError(redisClient
*c
, char *err
) {
246 _addReplyError(c
,err
,strlen(err
));
249 void addReplyErrorFormat(redisClient
*c
, const char *fmt
, ...) {
253 sds s
= sdscatvprintf(sdsempty(),fmt
,ap
);
255 /* Make sure there are no newlines in the string, otherwise invalid protocol
258 for (j
= 0; j
< l
; j
++) {
259 if (s
[j
] == '\r' || s
[j
] == '\n') s
[j
] = ' ';
261 _addReplyError(c
,s
,sdslen(s
));
265 void _addReplyStatus(redisClient
*c
, char *s
, size_t len
) {
266 addReplyString(c
,"+",1);
267 addReplyString(c
,s
,len
);
268 addReplyString(c
,"\r\n",2);
271 void addReplyStatus(redisClient
*c
, char *status
) {
272 _addReplyStatus(c
,status
,strlen(status
));
275 void addReplyStatusFormat(redisClient
*c
, const char *fmt
, ...) {
278 sds s
= sdscatvprintf(sdsempty(),fmt
,ap
);
280 _addReplyStatus(c
,s
,sdslen(s
));
284 /* Adds an empty object to the reply list that will contain the multi bulk
285 * length, which is not known when this function is called. */
286 void *addDeferredMultiBulkLength(redisClient
*c
) {
287 /* Note that we install the write event here even if the object is not
288 * ready to be sent, since we are sure that before returning to the
289 * event loop setDeferredMultiBulkLength() will be called. */
290 if (_installWriteEvent(c
) != REDIS_OK
) return NULL
;
291 listAddNodeTail(c
->reply
,createObject(REDIS_STRING
,NULL
));
292 return listLast(c
->reply
);
295 /* Populate the length object and try glueing it to the next chunk. */
296 void setDeferredMultiBulkLength(redisClient
*c
, void *node
, long length
) {
297 listNode
*ln
= (listNode
*)node
;
300 /* Abort when *node is NULL (see addDeferredMultiBulkLength). */
301 if (node
== NULL
) return;
303 len
= listNodeValue(ln
);
304 len
->ptr
= sdscatprintf(sdsempty(),"*%ld\r\n",length
);
305 if (ln
->next
!= NULL
) {
306 next
= listNodeValue(ln
->next
);
308 /* Only glue when the next node is non-NULL (an sds in this case) */
309 if (next
->ptr
!= NULL
) {
310 len
->ptr
= sdscatlen(len
->ptr
,next
->ptr
,sdslen(next
->ptr
));
311 listDelNode(c
->reply
,ln
->next
);
316 /* Add a duble as a bulk reply */
317 void addReplyDouble(redisClient
*c
, double d
) {
318 char dbuf
[128], sbuf
[128];
320 dlen
= snprintf(dbuf
,sizeof(dbuf
),"%.17g",d
);
321 slen
= snprintf(sbuf
,sizeof(sbuf
),"$%d\r\n%s\r\n",dlen
,dbuf
);
322 addReplyString(c
,sbuf
,slen
);
325 /* Add a long long as integer reply or bulk len / multi bulk count.
326 * Basically this is used to output <prefix><long long><crlf>. */
327 void _addReplyLongLong(redisClient
*c
, long long ll
, char prefix
) {
331 len
= ll2string(buf
+1,sizeof(buf
)-1,ll
);
334 addReplyString(c
,buf
,len
+3);
337 void addReplyLongLong(redisClient
*c
, long long ll
) {
339 addReply(c
,shared
.czero
);
341 addReply(c
,shared
.cone
);
343 _addReplyLongLong(c
,ll
,':');
346 void addReplyMultiBulkLen(redisClient
*c
, long length
) {
347 _addReplyLongLong(c
,length
,'*');
350 /* Create the length prefix of a bulk reply, example: $2234 */
351 void addReplyBulkLen(redisClient
*c
, robj
*obj
) {
354 if (obj
->encoding
== REDIS_ENCODING_RAW
) {
355 len
= sdslen(obj
->ptr
);
357 long n
= (long)obj
->ptr
;
359 /* Compute how many bytes will take this integer as a radix 10 string */
365 while((n
= n
/10) != 0) {
369 _addReplyLongLong(c
,len
,'$');
372 /* Add a Redis Object as a bulk reply */
373 void addReplyBulk(redisClient
*c
, robj
*obj
) {
374 addReplyBulkLen(c
,obj
);
376 addReply(c
,shared
.crlf
);
379 /* Add a C buffer as bulk reply */
380 void addReplyBulkCBuffer(redisClient
*c
, void *p
, size_t len
) {
381 _addReplyLongLong(c
,len
,'$');
382 addReplyString(c
,p
,len
);
383 addReply(c
,shared
.crlf
);
386 /* Add a C nul term string as bulk reply */
387 void addReplyBulkCString(redisClient
*c
, char *s
) {
389 addReply(c
,shared
.nullbulk
);
391 addReplyBulkCBuffer(c
,s
,strlen(s
));
395 /* Add a long long as a bulk reply */
396 void addReplyBulkLongLong(redisClient
*c
, long long ll
) {
400 len
= ll2string(buf
,64,ll
);
401 addReplyBulkCBuffer(c
,buf
,len
);
404 static void acceptCommonHandler(int fd
) {
406 if ((c
= createClient(fd
)) == NULL
) {
407 redisLog(REDIS_WARNING
,"Error allocating resoures for the client");
408 close(fd
); /* May be already closed, just ingore errors */
411 /* If maxclient directive is set and this is one client more... close the
412 * connection. Note that we create the client instead to check before
413 * for this condition, since now the socket is already set in nonblocking
414 * mode and we can send an error for free using the Kernel I/O */
415 if (listLength(server
.clients
) > server
.maxclients
) {
416 char *err
= "-ERR max number of clients reached\r\n";
418 /* That's a best effort error message, don't check write errors */
419 if (write(c
->fd
,err
,strlen(err
)) == -1) {
420 /* Nothing to do, Just to avoid the warning... */
425 server
.stat_numconnections
++;
428 void acceptTcpHandler(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
433 REDIS_NOTUSED(privdata
);
435 cfd
= anetTcpAccept(server
.neterr
, fd
, cip
, &cport
);
437 redisLog(REDIS_WARNING
,"Accepting client connection: %s", server
.neterr
);
440 redisLog(REDIS_VERBOSE
,"Accepted %s:%d", cip
, cport
);
441 acceptCommonHandler(cfd
);
444 void acceptUnixHandler(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
448 REDIS_NOTUSED(privdata
);
450 cfd
= anetUnixAccept(server
.neterr
, fd
);
452 redisLog(REDIS_WARNING
,"Accepting client connection: %s", server
.neterr
);
455 redisLog(REDIS_VERBOSE
,"Accepted connection to %s", server
.unixsocket
);
456 acceptCommonHandler(cfd
);
460 static void freeClientArgv(redisClient
*c
) {
462 for (j
= 0; j
< c
->argc
; j
++)
463 decrRefCount(c
->argv
[j
]);
468 void freeClient(redisClient
*c
) {
471 /* Note that if the client we are freeing is blocked into a blocking
472 * call, we have to set querybuf to NULL *before* to call
473 * unblockClientWaitingData() to avoid processInputBuffer() will get
474 * called. Also it is important to remove the file events after
475 * this, because this call adds the READABLE event. */
476 sdsfree(c
->querybuf
);
478 if (c
->flags
& REDIS_BLOCKED
)
479 unblockClientWaitingData(c
);
481 /* UNWATCH all the keys */
483 listRelease(c
->watched_keys
);
484 /* Unsubscribe from all the pubsub channels */
485 pubsubUnsubscribeAllChannels(c
,0);
486 pubsubUnsubscribeAllPatterns(c
,0);
487 dictRelease(c
->pubsub_channels
);
488 listRelease(c
->pubsub_patterns
);
489 /* Obvious cleanup */
490 aeDeleteFileEvent(server
.el
,c
->fd
,AE_READABLE
);
491 aeDeleteFileEvent(server
.el
,c
->fd
,AE_WRITABLE
);
492 listRelease(c
->reply
);
495 /* Remove from the list of clients */
496 ln
= listSearchKey(server
.clients
,c
);
497 redisAssert(ln
!= NULL
);
498 listDelNode(server
.clients
,ln
);
499 /* When client was just unblocked because of a blocking operation,
500 * remove it from the list with unblocked clients. */
501 if (c
->flags
& REDIS_UNBLOCKED
) {
502 ln
= listSearchKey(server
.unblocked_clients
,c
);
503 redisAssert(ln
!= NULL
);
504 listDelNode(server
.unblocked_clients
,ln
);
506 listRelease(c
->io_keys
);
507 /* Master/slave cleanup.
508 * Case 1: we lost the connection with a slave. */
509 if (c
->flags
& REDIS_SLAVE
) {
510 if (c
->replstate
== REDIS_REPL_SEND_BULK
&& c
->repldbfd
!= -1)
512 list
*l
= (c
->flags
& REDIS_MONITOR
) ? server
.monitors
: server
.slaves
;
513 ln
= listSearchKey(l
,c
);
514 redisAssert(ln
!= NULL
);
518 /* Case 2: we lost the connection with the master. */
519 if (c
->flags
& REDIS_MASTER
) {
520 server
.master
= NULL
;
521 server
.replstate
= REDIS_REPL_CONNECT
;
522 server
.repl_down_since
= time(NULL
);
523 /* Since we lost the connection with the master, we should also
524 * close the connection with all our slaves if we have any, so
525 * when we'll resync with the master the other slaves will sync again
526 * with us as well. Note that also when the slave is not connected
527 * to the master it will keep refusing connections by other slaves.
529 * We do this only if server.masterhost != NULL. If it is NULL this
530 * means the user called SLAVEOF NO ONE and we are freeing our
531 * link with the master, so no need to close link with slaves. */
532 if (server
.masterhost
!= NULL
) {
533 while (listLength(server
.slaves
)) {
534 ln
= listFirst(server
.slaves
);
535 freeClient((redisClient
*)ln
->value
);
541 freeClientMultiState(c
);
545 void sendReplyToClient(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
546 redisClient
*c
= privdata
;
547 int nwritten
= 0, totwritten
= 0, objlen
;
552 while(c
->bufpos
> 0 || listLength(c
->reply
)) {
554 if (c
->flags
& REDIS_MASTER
) {
555 /* Don't reply to a master */
556 nwritten
= c
->bufpos
- c
->sentlen
;
558 nwritten
= write(fd
,c
->buf
+c
->sentlen
,c
->bufpos
-c
->sentlen
);
559 if (nwritten
<= 0) break;
561 c
->sentlen
+= nwritten
;
562 totwritten
+= nwritten
;
564 /* If the buffer was sent, set bufpos to zero to continue with
565 * the remainder of the reply. */
566 if (c
->sentlen
== c
->bufpos
) {
571 o
= listNodeValue(listFirst(c
->reply
));
572 objlen
= sdslen(o
->ptr
);
575 listDelNode(c
->reply
,listFirst(c
->reply
));
579 if (c
->flags
& REDIS_MASTER
) {
580 /* Don't reply to a master */
581 nwritten
= objlen
- c
->sentlen
;
583 nwritten
= write(fd
, ((char*)o
->ptr
)+c
->sentlen
,objlen
-c
->sentlen
);
584 if (nwritten
<= 0) break;
586 c
->sentlen
+= nwritten
;
587 totwritten
+= nwritten
;
589 /* If we fully sent the object on head go to the next one */
590 if (c
->sentlen
== objlen
) {
591 listDelNode(c
->reply
,listFirst(c
->reply
));
595 /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
596 * bytes, in a single threaded server it's a good idea to serve
597 * other clients as well, even if a very large request comes from
598 * super fast link that is always able to accept data (in real world
599 * scenario think about 'KEYS *' against the loopback interfae) */
600 if (totwritten
> REDIS_MAX_WRITE_PER_EVENT
) break;
602 if (nwritten
== -1) {
603 if (errno
== EAGAIN
) {
606 redisLog(REDIS_VERBOSE
,
607 "Error writing to client: %s", strerror(errno
));
612 if (totwritten
> 0) c
->lastinteraction
= time(NULL
);
613 if (c
->bufpos
== 0 && listLength(c
->reply
) == 0) {
615 aeDeleteFileEvent(server
.el
,c
->fd
,AE_WRITABLE
);
617 /* Close connection after entire reply has been sent. */
618 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) freeClient(c
);
622 /* resetClient prepare the client to process the next command */
623 void resetClient(redisClient
*c
) {
628 /* We clear the ASKING flag as well if we are not inside a MULTI. */
629 if (!(c
->flags
& REDIS_MULTI
)) c
->flags
&= (~REDIS_ASKING
);
632 void closeTimedoutClients(void) {
635 time_t now
= time(NULL
);
638 listRewind(server
.clients
,&li
);
639 while ((ln
= listNext(&li
)) != NULL
) {
640 c
= listNodeValue(ln
);
641 if (server
.maxidletime
&&
642 !(c
->flags
& REDIS_SLAVE
) && /* no timeout for slaves */
643 !(c
->flags
& REDIS_MASTER
) && /* no timeout for masters */
644 !(c
->flags
& REDIS_BLOCKED
) && /* no timeout for BLPOP */
645 dictSize(c
->pubsub_channels
) == 0 && /* no timeout for pubsub */
646 listLength(c
->pubsub_patterns
) == 0 &&
647 (now
- c
->lastinteraction
> server
.maxidletime
))
649 redisLog(REDIS_VERBOSE
,"Closing idle client");
651 } else if (c
->flags
& REDIS_BLOCKED
) {
652 if (c
->bpop
.timeout
!= 0 && c
->bpop
.timeout
< now
) {
653 addReply(c
,shared
.nullmultibulk
);
654 unblockClientWaitingData(c
);
660 int processInlineBuffer(redisClient
*c
) {
661 char *newline
= strstr(c
->querybuf
,"\r\n");
666 /* Nothing to do without a \r\n */
670 /* Split the input buffer up to the \r\n */
671 querylen
= newline
-(c
->querybuf
);
672 argv
= sdssplitlen(c
->querybuf
,querylen
," ",1,&argc
);
674 /* Leave data after the first line of the query in the buffer */
675 c
->querybuf
= sdsrange(c
->querybuf
,querylen
+2,-1);
677 /* Setup argv array on client structure */
678 if (c
->argv
) zfree(c
->argv
);
679 c
->argv
= zmalloc(sizeof(robj
*)*argc
);
681 /* Create redis objects for all arguments. */
682 for (c
->argc
= 0, j
= 0; j
< argc
; j
++) {
683 if (sdslen(argv
[j
])) {
684 c
->argv
[c
->argc
] = createObject(REDIS_STRING
,argv
[j
]);
694 /* Helper function. Trims query buffer to make the function that processes
695 * multi bulk requests idempotent. */
696 static void setProtocolError(redisClient
*c
, int pos
) {
697 c
->flags
|= REDIS_CLOSE_AFTER_REPLY
;
698 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
701 int processMultibulkBuffer(redisClient
*c
) {
702 char *newline
= NULL
;
706 if (c
->multibulklen
== 0) {
707 /* The client should have been reset */
708 redisAssertWithInfo(c
,NULL
,c
->argc
== 0);
710 /* Multi bulk length cannot be read without a \r\n */
711 newline
= strchr(c
->querybuf
,'\r');
715 /* Buffer should also contain \n */
716 if (newline
-(c
->querybuf
) > ((signed)sdslen(c
->querybuf
)-2))
719 /* We know for sure there is a whole line since newline != NULL,
720 * so go ahead and find out the multi bulk length. */
721 redisAssertWithInfo(c
,NULL
,c
->querybuf
[0] == '*');
722 ok
= string2ll(c
->querybuf
+1,newline
-(c
->querybuf
+1),&ll
);
723 if (!ok
|| ll
> 1024*1024) {
724 addReplyError(c
,"Protocol error: invalid multibulk length");
725 setProtocolError(c
,pos
);
729 pos
= (newline
-c
->querybuf
)+2;
731 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
735 c
->multibulklen
= ll
;
737 /* Setup argv array on client structure */
738 if (c
->argv
) zfree(c
->argv
);
739 c
->argv
= zmalloc(sizeof(robj
*)*c
->multibulklen
);
742 redisAssertWithInfo(c
,NULL
,c
->multibulklen
> 0);
743 while(c
->multibulklen
) {
744 /* Read bulk length if unknown */
745 if (c
->bulklen
== -1) {
746 newline
= strchr(c
->querybuf
+pos
,'\r');
750 /* Buffer should also contain \n */
751 if (newline
-(c
->querybuf
) > ((signed)sdslen(c
->querybuf
)-2))
754 if (c
->querybuf
[pos
] != '$') {
755 addReplyErrorFormat(c
,
756 "Protocol error: expected '$', got '%c'",
758 setProtocolError(c
,pos
);
762 ok
= string2ll(c
->querybuf
+pos
+1,newline
-(c
->querybuf
+pos
+1),&ll
);
763 if (!ok
|| ll
< 0 || ll
> 512*1024*1024) {
764 addReplyError(c
,"Protocol error: invalid bulk length");
765 setProtocolError(c
,pos
);
769 pos
+= newline
-(c
->querybuf
+pos
)+2;
773 /* Read bulk argument */
774 if (sdslen(c
->querybuf
)-pos
< (unsigned)(c
->bulklen
+2)) {
775 /* Not enough data (+2 == trailing \r\n) */
778 c
->argv
[c
->argc
++] = createStringObject(c
->querybuf
+pos
,c
->bulklen
);
786 c
->querybuf
= sdsrange(c
->querybuf
,pos
,-1);
788 /* We're done when c->multibulk == 0 */
789 if (c
->multibulklen
== 0) {
795 void processInputBuffer(redisClient
*c
) {
796 /* Keep processing while there is something in the input buffer */
797 while(sdslen(c
->querybuf
)) {
798 /* Immediately abort if the client is in the middle of something. */
799 if (c
->flags
& REDIS_BLOCKED
) return;
801 /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
802 * written to the client. Make sure to not let the reply grow after
803 * this flag has been set (i.e. don't process more commands). */
804 if (c
->flags
& REDIS_CLOSE_AFTER_REPLY
) return;
806 /* Determine request type when unknown. */
808 if (c
->querybuf
[0] == '*') {
809 c
->reqtype
= REDIS_REQ_MULTIBULK
;
811 c
->reqtype
= REDIS_REQ_INLINE
;
815 if (c
->reqtype
== REDIS_REQ_INLINE
) {
816 if (processInlineBuffer(c
) != REDIS_OK
) break;
817 } else if (c
->reqtype
== REDIS_REQ_MULTIBULK
) {
818 if (processMultibulkBuffer(c
) != REDIS_OK
) break;
820 redisPanic("Unknown request type");
823 /* Multibulk processing could see a <= 0 length. */
827 /* Only reset the client when the command was executed. */
828 if (processCommand(c
) == REDIS_OK
)
834 void readQueryFromClient(aeEventLoop
*el
, int fd
, void *privdata
, int mask
) {
835 redisClient
*c
= (redisClient
*) privdata
;
836 char buf
[REDIS_IOBUF_LEN
];
841 nread
= read(fd
, buf
, REDIS_IOBUF_LEN
);
843 if (errno
== EAGAIN
) {
846 redisLog(REDIS_VERBOSE
, "Reading from client: %s",strerror(errno
));
850 } else if (nread
== 0) {
851 redisLog(REDIS_VERBOSE
, "Client closed connection");
856 c
->querybuf
= sdscatlen(c
->querybuf
,buf
,nread
);
857 c
->lastinteraction
= time(NULL
);
861 processInputBuffer(c
);
864 void getClientsMaxBuffers(unsigned long *longest_output_list
,
865 unsigned long *biggest_input_buffer
) {
869 unsigned long lol
= 0, bib
= 0;
871 listRewind(server
.clients
,&li
);
872 while ((ln
= listNext(&li
)) != NULL
) {
873 c
= listNodeValue(ln
);
875 if (listLength(c
->reply
) > lol
) lol
= listLength(c
->reply
);
876 if (sdslen(c
->querybuf
) > bib
) bib
= sdslen(c
->querybuf
);
878 *longest_output_list
= lol
;
879 *biggest_input_buffer
= bib
;
882 void clientCommand(redisClient
*c
) {
887 if (!strcasecmp(c
->argv
[1]->ptr
,"list") && c
->argc
== 2) {
889 time_t now
= time(NULL
);
891 listRewind(server
.clients
,&li
);
892 while ((ln
= listNext(&li
)) != NULL
) {
893 char ip
[32], flags
[16], *p
;
896 client
= listNodeValue(ln
);
897 if (anetPeerToString(client
->fd
,ip
,&port
) == -1) continue;
899 if (client
->flags
& REDIS_SLAVE
) {
900 if (client
->flags
& REDIS_MONITOR
)
905 if (client
->flags
& REDIS_MASTER
) *p
++ = 'M';
906 if (p
== flags
) *p
++ = 'N';
907 if (client
->flags
& REDIS_MULTI
) *p
++ = 'x';
908 if (client
->flags
& REDIS_BLOCKED
) *p
++ = 'b';
909 if (client
->flags
& REDIS_DIRTY_CAS
) *p
++ = 'd';
910 if (client
->flags
& REDIS_CLOSE_AFTER_REPLY
) *p
++ = 'c';
911 if (client
->flags
& REDIS_UNBLOCKED
) *p
++ = 'u';
914 "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d\n",
916 (long)(now
- client
->lastinteraction
),
919 (int) dictSize(client
->pubsub_channels
),
920 (int) listLength(client
->pubsub_patterns
));
922 addReplyBulkCBuffer(c
,o
,sdslen(o
));
924 } else if (!strcasecmp(c
->argv
[1]->ptr
,"kill") && c
->argc
== 3) {
925 listRewind(server
.clients
,&li
);
926 while ((ln
= listNext(&li
)) != NULL
) {
927 char ip
[32], addr
[64];
930 client
= listNodeValue(ln
);
931 if (anetPeerToString(client
->fd
,ip
,&port
) == -1) continue;
932 snprintf(addr
,sizeof(addr
),"%s:%d",ip
,port
);
933 if (strcmp(addr
,c
->argv
[2]->ptr
) == 0) {
934 addReply(c
,shared
.ok
);
936 client
->flags
|= REDIS_CLOSE_AFTER_REPLY
;
943 addReplyError(c
,"No such client");
945 addReplyError(c
, "Syntax error, try CLIENT (LIST | KILL ip:port)");
949 /* Rewrite the command vector of the client. All the new objects ref count
950 * is incremented. The old command vector is freed, and the old objects
951 * ref count is decremented. */
952 void rewriteClientCommandVector(redisClient
*c
, int argc
, ...) {
955 robj
**argv
; /* The new argument vector */
957 argv
= zmalloc(sizeof(robj
*)*argc
);
959 for (j
= 0; j
< argc
; j
++) {
962 a
= va_arg(ap
, robj
*);
966 /* We free the objects in the original vector at the end, so we are
967 * sure that if the same objects are reused in the new vector the
968 * refcount gets incremented before it gets decremented. */
969 for (j
= 0; j
< c
->argc
; j
++) decrRefCount(c
->argv
[j
]);
971 /* Replace argv and argc with our new versions. */
974 c
->cmd
= lookupCommand(c
->argv
[0]->ptr
);
975 redisAssertWithInfo(c
,NULL
,c
->cmd
!= NULL
);
979 /* Rewrite a single item in the command vector.
980 * The new val ref count is incremented, and the old decremented. */
981 void rewriteClientCommandArgument(redisClient
*c
, int i
, robj
*newval
) {
984 redisAssertWithInfo(c
,NULL
,i
< c
->argc
);
987 incrRefCount(newval
);
988 decrRefCount(oldval
);
990 /* If this is the command name make sure to fix c->cmd. */
992 c
->cmd
= lookupCommand(c
->argv
[0]->ptr
);
993 redisAssertWithInfo(c
,NULL
,c
->cmd
!= NULL
);