Lines Matching refs:c
34 static void setProtocolError(client *c, int pos);
65 client *c = zmalloc(sizeof(client)); in createClient() local
77 readQueryFromClient, c) == AE_ERR) in createClient()
80 zfree(c); in createClient()
85 selectDb(c,0); in createClient()
86 c->id = server.next_client_id++; in createClient()
87 c->fd = fd; in createClient()
88 c->name = NULL; in createClient()
89 c->bufpos = 0; in createClient()
90 c->querybuf = sdsempty(); in createClient()
91 c->querybuf_peak = 0; in createClient()
92 c->reqtype = 0; in createClient()
93 c->argc = 0; in createClient()
94 c->argv = NULL; in createClient()
95 c->cmd = c->lastcmd = NULL; in createClient()
96 c->multibulklen = 0; in createClient()
97 c->bulklen = -1; in createClient()
98 c->sentlen = 0; in createClient()
99 c->flags = 0; in createClient()
100 c->ctime = c->lastinteraction = server.unixtime; in createClient()
101 c->authenticated = 0; in createClient()
102 c->replstate = REPL_STATE_NONE; in createClient()
103 c->repl_put_online_on_ack = 0; in createClient()
104 c->reploff = 0; in createClient()
105 c->repl_ack_off = 0; in createClient()
106 c->repl_ack_time = 0; in createClient()
107 c->slave_listening_port = 0; in createClient()
108 c->slave_ip[0] = '\0'; in createClient()
109 c->slave_capa = SLAVE_CAPA_NONE; in createClient()
110 c->reply = listCreate(); in createClient()
111 c->reply_bytes = 0; in createClient()
112 c->obuf_soft_limit_reached_time = 0; in createClient()
113 listSetFreeMethod(c->reply,decrRefCountVoid); in createClient()
114 listSetDupMethod(c->reply,dupClientReplyValue); in createClient()
115 c->btype = BLOCKED_NONE; in createClient()
116 c->bpop.timeout = 0; in createClient()
117 c->bpop.keys = dictCreate(&setDictType,NULL); in createClient()
118 c->bpop.target = NULL; in createClient()
119 c->bpop.numreplicas = 0; in createClient()
120 c->bpop.reploffset = 0; in createClient()
121 c->woff = 0; in createClient()
122 c->watched_keys = listCreate(); in createClient()
123 c->pubsub_channels = dictCreate(&setDictType,NULL); in createClient()
124 c->pubsub_patterns = listCreate(); in createClient()
125 c->peerid = NULL; in createClient()
126 listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid); in createClient()
127 listSetMatchMethod(c->pubsub_patterns,listMatchObjects); in createClient()
128 if (fd != -1) listAddNodeTail(server.clients,c); in createClient()
129 initClientMultiState(c); in createClient()
130 return c; in createClient()
155 int prepareClientToWrite(client *c) { in prepareClientToWrite() argument
158 if (c->flags & CLIENT_LUA) return C_OK; in prepareClientToWrite()
161 if (c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) return C_ERR; in prepareClientToWrite()
165 if ((c->flags & CLIENT_MASTER) && in prepareClientToWrite()
166 !(c->flags & CLIENT_MASTER_FORCE_REPLY)) return C_ERR; in prepareClientToWrite()
168 if (c->fd <= 0) return C_ERR; /* Fake client for AOF loading. */ in prepareClientToWrite()
174 if (!clientHasPendingReplies(c) && in prepareClientToWrite()
175 !(c->flags & CLIENT_PENDING_WRITE) && in prepareClientToWrite()
176 (c->replstate == REPL_STATE_NONE || in prepareClientToWrite()
177 (c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack))) in prepareClientToWrite()
185 c->flags |= CLIENT_PENDING_WRITE; in prepareClientToWrite()
186 listAddNodeHead(server.clients_pending_write,c); in prepareClientToWrite()
213 int _addReplyToBuffer(client *c, const char *s, size_t len) { in _addReplyToBuffer() argument
214 size_t available = sizeof(c->buf)-c->bufpos; in _addReplyToBuffer()
216 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return C_OK; in _addReplyToBuffer()
220 if (listLength(c->reply) > 0) return C_ERR; in _addReplyToBuffer()
225 memcpy(c->buf+c->bufpos,s,len); in _addReplyToBuffer()
226 c->bufpos+=len; in _addReplyToBuffer()
230 void _addReplyObjectToList(client *c, robj *o) { in _addReplyObjectToList() argument
233 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return; in _addReplyObjectToList()
235 if (listLength(c->reply) == 0) { in _addReplyObjectToList()
237 listAddNodeTail(c->reply,o); in _addReplyObjectToList()
238 c->reply_bytes += getStringObjectSdsUsedMemory(o); in _addReplyObjectToList()
240 tail = listNodeValue(listLast(c->reply)); in _addReplyObjectToList()
247 c->reply_bytes -= sdsZmallocSize(tail->ptr); in _addReplyObjectToList()
248 tail = dupLastObjectIfNeeded(c->reply); in _addReplyObjectToList()
250 c->reply_bytes += sdsZmallocSize(tail->ptr); in _addReplyObjectToList()
253 listAddNodeTail(c->reply,o); in _addReplyObjectToList()
254 c->reply_bytes += getStringObjectSdsUsedMemory(o); in _addReplyObjectToList()
257 asyncCloseClientOnOutputBufferLimitReached(c); in _addReplyObjectToList()
262 void _addReplySdsToList(client *c, sds s) { in _addReplySdsToList() argument
265 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) { in _addReplySdsToList()
270 if (listLength(c->reply) == 0) { in _addReplySdsToList()
271 listAddNodeTail(c->reply,createObject(OBJ_STRING,s)); in _addReplySdsToList()
272 c->reply_bytes += sdsZmallocSize(s); in _addReplySdsToList()
274 tail = listNodeValue(listLast(c->reply)); in _addReplySdsToList()
280 c->reply_bytes -= sdsZmallocSize(tail->ptr); in _addReplySdsToList()
281 tail = dupLastObjectIfNeeded(c->reply); in _addReplySdsToList()
283 c->reply_bytes += sdsZmallocSize(tail->ptr); in _addReplySdsToList()
286 listAddNodeTail(c->reply,createObject(OBJ_STRING,s)); in _addReplySdsToList()
287 c->reply_bytes += sdsZmallocSize(s); in _addReplySdsToList()
290 asyncCloseClientOnOutputBufferLimitReached(c); in _addReplySdsToList()
293 void _addReplyStringToList(client *c, const char *s, size_t len) { in _addReplyStringToList() argument
296 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return; in _addReplyStringToList()
298 if (listLength(c->reply) == 0) { in _addReplyStringToList()
301 listAddNodeTail(c->reply,o); in _addReplyStringToList()
302 c->reply_bytes += getStringObjectSdsUsedMemory(o); in _addReplyStringToList()
304 tail = listNodeValue(listLast(c->reply)); in _addReplyStringToList()
310 c->reply_bytes -= sdsZmallocSize(tail->ptr); in _addReplyStringToList()
311 tail = dupLastObjectIfNeeded(c->reply); in _addReplyStringToList()
313 c->reply_bytes += sdsZmallocSize(tail->ptr); in _addReplyStringToList()
317 listAddNodeTail(c->reply,o); in _addReplyStringToList()
318 c->reply_bytes += getStringObjectSdsUsedMemory(o); in _addReplyStringToList()
321 asyncCloseClientOnOutputBufferLimitReached(c); in _addReplyStringToList()
329 void addReply(client *c, robj *obj) { in addReply() argument
330 if (prepareClientToWrite(c) != C_OK) return; in addReply()
340 if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK) in addReply()
341 _addReplyObjectToList(c,obj); in addReply()
346 if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) { in addReply()
351 if (_addReplyToBuffer(c,buf,len) == C_OK) in addReply()
357 if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK) in addReply()
358 _addReplyObjectToList(c,obj); in addReply()
365 void addReplySds(client *c, sds s) { in addReplySds() argument
366 if (prepareClientToWrite(c) != C_OK) { in addReplySds()
371 if (_addReplyToBuffer(c,s,sdslen(s)) == C_OK) { in addReplySds()
375 _addReplySdsToList(c,s); in addReplySds()
379 void addReplyString(client *c, const char *s, size_t len) { in addReplyString() argument
380 if (prepareClientToWrite(c) != C_OK) return; in addReplyString()
381 if (_addReplyToBuffer(c,s,len) != C_OK) in addReplyString()
382 _addReplyStringToList(c,s,len); in addReplyString()
385 void addReplyErrorLength(client *c, const char *s, size_t len) { in addReplyErrorLength() argument
386 addReplyString(c,"-ERR ",5); in addReplyErrorLength()
387 addReplyString(c,s,len); in addReplyErrorLength()
388 addReplyString(c,"\r\n",2); in addReplyErrorLength()
391 void addReplyError(client *c, const char *err) { in addReplyError() argument
392 addReplyErrorLength(c,err,strlen(err)); in addReplyError()
395 void addReplyErrorFormat(client *c, const char *fmt, ...) { in addReplyErrorFormat() argument
407 addReplyErrorLength(c,s,sdslen(s)); in addReplyErrorFormat()
411 void addReplyStatusLength(client *c, const char *s, size_t len) { in addReplyStatusLength() argument
412 addReplyString(c,"+",1); in addReplyStatusLength()
413 addReplyString(c,s,len); in addReplyStatusLength()
414 addReplyString(c,"\r\n",2); in addReplyStatusLength()
417 void addReplyStatus(client *c, const char *status) { in addReplyStatus() argument
418 addReplyStatusLength(c,status,strlen(status)); in addReplyStatus()
421 void addReplyStatusFormat(client *c, const char *fmt, ...) { in addReplyStatusFormat() argument
426 addReplyStatusLength(c,s,sdslen(s)); in addReplyStatusFormat()
432 void *addDeferredMultiBulkLength(client *c) { in addDeferredMultiBulkLength() argument
436 if (prepareClientToWrite(c) != C_OK) return NULL; in addDeferredMultiBulkLength()
437 listAddNodeTail(c->reply,createObject(OBJ_STRING,NULL)); in addDeferredMultiBulkLength()
438 return listLast(c->reply); in addDeferredMultiBulkLength()
442 void setDeferredMultiBulkLength(client *c, void *node, long length) { in setDeferredMultiBulkLength() argument
452 c->reply_bytes += sdsZmallocSize(len->ptr); in setDeferredMultiBulkLength()
458 c->reply_bytes -= sdsZmallocSize(len->ptr); in setDeferredMultiBulkLength()
459 c->reply_bytes -= getStringObjectSdsUsedMemory(next); in setDeferredMultiBulkLength()
461 c->reply_bytes += sdsZmallocSize(len->ptr); in setDeferredMultiBulkLength()
462 listDelNode(c->reply,ln->next); in setDeferredMultiBulkLength()
465 asyncCloseClientOnOutputBufferLimitReached(c); in setDeferredMultiBulkLength()
469 void addReplyDouble(client *c, double d) { in addReplyDouble() argument
475 addReplyBulkCString(c, d > 0 ? "inf" : "-inf"); in addReplyDouble()
479 addReplyString(c,sbuf,slen); in addReplyDouble()
486 void addReplyHumanLongDouble(client *c, long double d) { in addReplyHumanLongDouble() argument
488 addReplyBulk(c,o); in addReplyHumanLongDouble()
494 void addReplyLongLongWithPrefix(client *c, long long ll, char prefix) { in addReplyLongLongWithPrefix() argument
502 addReply(c,shared.mbulkhdr[ll]); in addReplyLongLongWithPrefix()
505 addReply(c,shared.bulkhdr[ll]); in addReplyLongLongWithPrefix()
513 addReplyString(c,buf,len+3); in addReplyLongLongWithPrefix()
516 void addReplyLongLong(client *c, long long ll) { in addReplyLongLong() argument
518 addReply(c,shared.czero); in addReplyLongLong()
520 addReply(c,shared.cone); in addReplyLongLong()
522 addReplyLongLongWithPrefix(c,ll,':'); in addReplyLongLong()
525 void addReplyMultiBulkLen(client *c, long length) { in addReplyMultiBulkLen() argument
527 addReply(c,shared.mbulkhdr[length]); in addReplyMultiBulkLen()
529 addReplyLongLongWithPrefix(c,length,'*'); in addReplyMultiBulkLen()
533 void addReplyBulkLen(client *c, robj *obj) { in addReplyBulkLen() argument
553 addReply(c,shared.bulkhdr[len]); in addReplyBulkLen()
555 addReplyLongLongWithPrefix(c,len,'$'); in addReplyBulkLen()
559 void addReplyBulk(client *c, robj *obj) { in addReplyBulk() argument
560 addReplyBulkLen(c,obj); in addReplyBulk()
561 addReply(c,obj); in addReplyBulk()
562 addReply(c,shared.crlf); in addReplyBulk()
566 void addReplyBulkCBuffer(client *c, const void *p, size_t len) { in addReplyBulkCBuffer() argument
567 addReplyLongLongWithPrefix(c,len,'$'); in addReplyBulkCBuffer()
568 addReplyString(c,p,len); in addReplyBulkCBuffer()
569 addReply(c,shared.crlf); in addReplyBulkCBuffer()
573 void addReplyBulkSds(client *c, sds s) { in addReplyBulkSds() argument
574 addReplySds(c,sdscatfmt(sdsempty(),"$%u\r\n", in addReplyBulkSds()
576 addReplySds(c,s); in addReplyBulkSds()
577 addReply(c,shared.crlf); in addReplyBulkSds()
581 void addReplyBulkCString(client *c, const char *s) { in addReplyBulkCString() argument
583 addReply(c,shared.nullbulk); in addReplyBulkCString()
585 addReplyBulkCBuffer(c,s,strlen(s)); in addReplyBulkCString()
590 void addReplyBulkLongLong(client *c, long long ll) { in addReplyBulkLongLong() argument
595 addReplyBulkCBuffer(c,buf,len); in addReplyBulkLongLong()
611 int clientHasPendingReplies(client *c) { in clientHasPendingReplies() argument
612 return c->bufpos || listLength(c->reply); in clientHasPendingReplies()
617 client *c; in acceptCommonHandler() local
618 if ((c = createClient(fd)) == NULL) { in acceptCommonHandler()
633 if (write(c->fd,err,strlen(err)) == -1) { in acceptCommonHandler()
637 freeClient(c); in acceptCommonHandler()
673 if (write(c->fd,err,strlen(err)) == -1) { in acceptCommonHandler()
677 freeClient(c); in acceptCommonHandler()
683 c->flags |= flags; in acceptCommonHandler()
725 static void freeClientArgv(client *c) { in freeClientArgv() argument
727 for (j = 0; j < c->argc; j++) in freeClientArgv()
728 decrRefCount(c->argv[j]); in freeClientArgv()
729 c->argc = 0; in freeClientArgv()
730 c->cmd = NULL; in freeClientArgv()
746 void unlinkClient(client *c) { in unlinkClient() argument
750 if (server.current_client == c) server.current_client = NULL; in unlinkClient()
755 if (c->fd != -1) { in unlinkClient()
757 ln = listSearchKey(server.clients,c); in unlinkClient()
762 aeDeleteFileEvent(server.el,c->fd,AE_READABLE); in unlinkClient()
763 aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); in unlinkClient()
764 close(c->fd); in unlinkClient()
765 c->fd = -1; in unlinkClient()
769 if (c->flags & CLIENT_PENDING_WRITE) { in unlinkClient()
770 ln = listSearchKey(server.clients_pending_write,c); in unlinkClient()
773 c->flags &= ~CLIENT_PENDING_WRITE; in unlinkClient()
778 if (c->flags & CLIENT_UNBLOCKED) { in unlinkClient()
779 ln = listSearchKey(server.unblocked_clients,c); in unlinkClient()
782 c->flags &= ~CLIENT_UNBLOCKED; in unlinkClient()
786 void freeClient(client *c) { in freeClient() argument
794 if (server.master && c->flags & CLIENT_MASTER) { in freeClient()
796 if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY| in freeClient()
801 replicationCacheMaster(c); in freeClient()
807 if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) { in freeClient()
809 replicationGetSlaveName(c)); in freeClient()
813 sdsfree(c->querybuf); in freeClient()
814 c->querybuf = NULL; in freeClient()
817 if (c->flags & CLIENT_BLOCKED) unblockClient(c); in freeClient()
818 dictRelease(c->bpop.keys); in freeClient()
821 unwatchAllKeys(c); in freeClient()
822 listRelease(c->watched_keys); in freeClient()
825 pubsubUnsubscribeAllChannels(c,0); in freeClient()
826 pubsubUnsubscribeAllPatterns(c,0); in freeClient()
827 dictRelease(c->pubsub_channels); in freeClient()
828 listRelease(c->pubsub_patterns); in freeClient()
831 listRelease(c->reply); in freeClient()
832 freeClientArgv(c); in freeClient()
837 unlinkClient(c); in freeClient()
841 if (c->flags & CLIENT_SLAVE) { in freeClient()
842 if (c->replstate == SLAVE_STATE_SEND_BULK) { in freeClient()
843 if (c->repldbfd != -1) close(c->repldbfd); in freeClient()
844 if (c->replpreamble) sdsfree(c->replpreamble); in freeClient()
846 list *l = (c->flags & CLIENT_MONITOR) ? server.monitors : server.slaves; in freeClient()
847 ln = listSearchKey(l,c); in freeClient()
853 if (c->flags & CLIENT_SLAVE && listLength(server.slaves) == 0) in freeClient()
860 if (c->flags & CLIENT_MASTER) replicationHandleMasterDisconnection(); in freeClient()
864 if (c->flags & CLIENT_CLOSE_ASAP) { in freeClient()
865 ln = listSearchKey(server.clients_to_close,c); in freeClient()
872 if (c->name) decrRefCount(c->name); in freeClient()
873 zfree(c->argv); in freeClient()
874 freeClientMultiState(c); in freeClient()
875 sdsfree(c->peerid); in freeClient()
876 zfree(c); in freeClient()
883 void freeClientAsync(client *c) { in freeClientAsync() argument
884 if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return; in freeClientAsync()
885 c->flags |= CLIENT_CLOSE_ASAP; in freeClientAsync()
886 listAddNodeTail(server.clients_to_close,c); in freeClientAsync()
892 client *c = listNodeValue(ln); in freeClientsInAsyncFreeQueue() local
894 c->flags &= ~CLIENT_CLOSE_ASAP; in freeClientsInAsyncFreeQueue()
895 freeClient(c); in freeClientsInAsyncFreeQueue()
902 int writeToClient(int fd, client *c, int handler_installed) { in writeToClient() argument
908 while(clientHasPendingReplies(c)) { in writeToClient()
909 if (c->bufpos > 0) { in writeToClient()
910 nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen); in writeToClient()
912 c->sentlen += nwritten; in writeToClient()
917 if ((int)c->sentlen == c->bufpos) { in writeToClient()
918 c->bufpos = 0; in writeToClient()
919 c->sentlen = 0; in writeToClient()
922 o = listNodeValue(listFirst(c->reply)); in writeToClient()
927 listDelNode(c->reply,listFirst(c->reply)); in writeToClient()
928 c->reply_bytes -= objmem; in writeToClient()
932 nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen); in writeToClient()
934 c->sentlen += nwritten; in writeToClient()
938 if (c->sentlen == objlen) { in writeToClient()
939 listDelNode(c->reply,listFirst(c->reply)); in writeToClient()
940 c->sentlen = 0; in writeToClient()
941 c->reply_bytes -= objmem; in writeToClient()
963 freeClient(c); in writeToClient()
972 if (!(c->flags & CLIENT_MASTER)) c->lastinteraction = server.unixtime; in writeToClient()
974 if (!clientHasPendingReplies(c)) { in writeToClient()
975 c->sentlen = 0; in writeToClient()
976 if (handler_installed) aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); in writeToClient()
979 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) { in writeToClient()
980 freeClient(c); in writeToClient()
1005 client *c = listNodeValue(ln); in handleClientsWithPendingWrites() local
1006 c->flags &= ~CLIENT_PENDING_WRITE; in handleClientsWithPendingWrites()
1010 if (writeToClient(c->fd,c,0) == C_ERR) continue; in handleClientsWithPendingWrites()
1014 if (clientHasPendingReplies(c) && in handleClientsWithPendingWrites()
1015 aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, in handleClientsWithPendingWrites()
1016 sendReplyToClient, c) == AE_ERR) in handleClientsWithPendingWrites()
1018 freeClientAsync(c); in handleClientsWithPendingWrites()
1025 void resetClient(client *c) { in resetClient() argument
1026 redisCommandProc *prevcmd = c->cmd ? c->cmd->proc : NULL; in resetClient()
1028 freeClientArgv(c); in resetClient()
1029 c->reqtype = 0; in resetClient()
1030 c->multibulklen = 0; in resetClient()
1031 c->bulklen = -1; in resetClient()
1035 if (!(c->flags & CLIENT_MULTI) && prevcmd != askingCommand) in resetClient()
1036 c->flags &= ~CLIENT_ASKING; in resetClient()
1041 c->flags &= ~CLIENT_REPLY_SKIP; in resetClient()
1042 if (c->flags & CLIENT_REPLY_SKIP_NEXT) { in resetClient()
1043 c->flags |= CLIENT_REPLY_SKIP; in resetClient()
1044 c->flags &= ~CLIENT_REPLY_SKIP_NEXT; in resetClient()
1048 int processInlineBuffer(client *c) { in processInlineBuffer() argument
1055 newline = strchr(c->querybuf,'\n'); in processInlineBuffer()
1059 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { in processInlineBuffer()
1060 addReplyError(c,"Protocol error: too big inline request"); in processInlineBuffer()
1061 setProtocolError(c,0); in processInlineBuffer()
1067 if (newline && newline != c->querybuf && *(newline-1) == '\r') in processInlineBuffer()
1071 querylen = newline-(c->querybuf); in processInlineBuffer()
1072 aux = sdsnewlen(c->querybuf,querylen); in processInlineBuffer()
1076 addReplyError(c,"Protocol error: unbalanced quotes in request"); in processInlineBuffer()
1077 setProtocolError(c,0); in processInlineBuffer()
1084 if (querylen == 0 && c->flags & CLIENT_SLAVE) in processInlineBuffer()
1085 c->repl_ack_time = server.unixtime; in processInlineBuffer()
1088 sdsrange(c->querybuf,querylen+2,-1); in processInlineBuffer()
1092 if (c->argv) zfree(c->argv); in processInlineBuffer()
1093 c->argv = zmalloc(sizeof(robj*)*argc); in processInlineBuffer()
1097 for (c->argc = 0, j = 0; j < argc; j++) { in processInlineBuffer()
1099 c->argv[c->argc] = createObject(OBJ_STRING,argv[j]); in processInlineBuffer()
1100 c->argc++; in processInlineBuffer()
1111 static void setProtocolError(client *c, int pos) { in setProtocolError() argument
1113 sds client = catClientInfoString(sdsempty(),c); in setProtocolError()
1118 c->flags |= CLIENT_CLOSE_AFTER_REPLY; in setProtocolError()
1119 sdsrange(c->querybuf,pos,-1); in setProtocolError()
1122 int processMultibulkBuffer(client *c) { in processMultibulkBuffer() argument
1127 if (c->multibulklen == 0) { in processMultibulkBuffer()
1129 serverAssertWithInfo(c,NULL,c->argc == 0); in processMultibulkBuffer()
1132 newline = strchr(c->querybuf,'\r'); in processMultibulkBuffer()
1134 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { in processMultibulkBuffer()
1135 addReplyError(c,"Protocol error: too big mbulk count string"); in processMultibulkBuffer()
1136 setProtocolError(c,0); in processMultibulkBuffer()
1142 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) in processMultibulkBuffer()
1147 serverAssertWithInfo(c,NULL,c->querybuf[0] == '*'); in processMultibulkBuffer()
1148 ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); in processMultibulkBuffer()
1150 addReplyError(c,"Protocol error: invalid multibulk length"); in processMultibulkBuffer()
1151 setProtocolError(c,pos); in processMultibulkBuffer()
1155 pos = (newline-c->querybuf)+2; in processMultibulkBuffer()
1157 sdsrange(c->querybuf,pos,-1); in processMultibulkBuffer()
1161 c->multibulklen = ll; in processMultibulkBuffer()
1164 if (c->argv) zfree(c->argv); in processMultibulkBuffer()
1165 c->argv = zmalloc(sizeof(robj*)*c->multibulklen); in processMultibulkBuffer()
1168 serverAssertWithInfo(c,NULL,c->multibulklen > 0); in processMultibulkBuffer()
1169 while(c->multibulklen) { in processMultibulkBuffer()
1171 if (c->bulklen == -1) { in processMultibulkBuffer()
1172 newline = strchr(c->querybuf+pos,'\r'); in processMultibulkBuffer()
1174 if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) { in processMultibulkBuffer()
1175 addReplyError(c, in processMultibulkBuffer()
1177 setProtocolError(c,0); in processMultibulkBuffer()
1184 if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2)) in processMultibulkBuffer()
1187 if (c->querybuf[pos] != '$') { in processMultibulkBuffer()
1188 addReplyErrorFormat(c, in processMultibulkBuffer()
1190 c->querybuf[pos]); in processMultibulkBuffer()
1191 setProtocolError(c,pos); in processMultibulkBuffer()
1195 ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); in processMultibulkBuffer()
1197 addReplyError(c,"Protocol error: invalid bulk length"); in processMultibulkBuffer()
1198 setProtocolError(c,pos); in processMultibulkBuffer()
1202 pos += newline-(c->querybuf+pos)+2; in processMultibulkBuffer()
1210 sdsrange(c->querybuf,pos,-1); in processMultibulkBuffer()
1212 qblen = sdslen(c->querybuf); in processMultibulkBuffer()
1216 c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-qblen); in processMultibulkBuffer()
1218 c->bulklen = ll; in processMultibulkBuffer()
1222 if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) { in processMultibulkBuffer()
1230 c->bulklen >= PROTO_MBULK_BIG_ARG && in processMultibulkBuffer()
1231 (signed) sdslen(c->querybuf) == c->bulklen+2) in processMultibulkBuffer()
1233 c->argv[c->argc++] = createObject(OBJ_STRING,c->querybuf); in processMultibulkBuffer()
1234 sdsIncrLen(c->querybuf,-2); /* remove CRLF */ in processMultibulkBuffer()
1237 c->querybuf = sdsnewlen(NULL,c->bulklen+2); in processMultibulkBuffer()
1238 sdsclear(c->querybuf); in processMultibulkBuffer()
1241 c->argv[c->argc++] = in processMultibulkBuffer()
1242 createStringObject(c->querybuf+pos,c->bulklen); in processMultibulkBuffer()
1243 pos += c->bulklen+2; in processMultibulkBuffer()
1245 c->bulklen = -1; in processMultibulkBuffer()
1246 c->multibulklen--; in processMultibulkBuffer()
1251 if (pos) sdsrange(c->querybuf,pos,-1); in processMultibulkBuffer()
1254 if (c->multibulklen == 0) return C_OK; in processMultibulkBuffer()
1260 void processInputBuffer(client *c) { in processInputBuffer() argument
1261 server.current_client = c; in processInputBuffer()
1263 while(sdslen(c->querybuf)) { in processInputBuffer()
1265 if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break; in processInputBuffer()
1268 if (c->flags & CLIENT_BLOCKED) break; in processInputBuffer()
1273 if (c->flags & CLIENT_CLOSE_AFTER_REPLY) break; in processInputBuffer()
1276 if (!c->reqtype) { in processInputBuffer()
1277 if (c->querybuf[0] == '*') { in processInputBuffer()
1278 c->reqtype = PROTO_REQ_MULTIBULK; in processInputBuffer()
1280 c->reqtype = PROTO_REQ_INLINE; in processInputBuffer()
1284 if (c->reqtype == PROTO_REQ_INLINE) { in processInputBuffer()
1285 if (processInlineBuffer(c) != C_OK) break; in processInputBuffer()
1286 } else if (c->reqtype == PROTO_REQ_MULTIBULK) { in processInputBuffer()
1287 if (processMultibulkBuffer(c) != C_OK) break; in processInputBuffer()
1293 if (c->argc == 0) { in processInputBuffer()
1294 resetClient(c); in processInputBuffer()
1297 if (processCommand(c) == C_OK) in processInputBuffer()
1298 resetClient(c); in processInputBuffer()
1308 client *c = (client*) privdata; in readQueryFromClient() local
1321 if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 in readQueryFromClient()
1322 && c->bulklen >= PROTO_MBULK_BIG_ARG) in readQueryFromClient()
1324 int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); in readQueryFromClient()
1329 qblen = sdslen(c->querybuf); in readQueryFromClient()
1330 if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; in readQueryFromClient()
1331 c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); in readQueryFromClient()
1332 nread = read(fd, c->querybuf+qblen, readlen); in readQueryFromClient()
1338 freeClient(c); in readQueryFromClient()
1343 freeClient(c); in readQueryFromClient()
1347 sdsIncrLen(c->querybuf,nread); in readQueryFromClient()
1348 c->lastinteraction = server.unixtime; in readQueryFromClient()
1349 if (c->flags & CLIENT_MASTER) c->reploff += nread; in readQueryFromClient()
1351 if (sdslen(c->querybuf) > server.client_max_querybuf_len) { in readQueryFromClient()
1352 sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); in readQueryFromClient()
1354 bytes = sdscatrepr(bytes,c->querybuf,64); in readQueryFromClient()
1358 freeClient(c); in readQueryFromClient()
1361 processInputBuffer(c); in readQueryFromClient()
1366 client *c; in getClientsMaxBuffers() local
1373 c = listNodeValue(ln); in getClientsMaxBuffers()
1375 if (listLength(c->reply) > lol) lol = listLength(c->reply); in getClientsMaxBuffers()
1376 if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf); in getClientsMaxBuffers()
1408 char *getClientPeerId(client *c) { in getClientPeerId() argument
1411 if (c->peerid == NULL) { in getClientPeerId()
1412 genClientPeerId(c,peerid,sizeof(peerid)); in getClientPeerId()
1413 c->peerid = sdsnew(peerid); in getClientPeerId()
1415 return c->peerid; in getClientPeerId()
1485 void clientCommand(client *c) { in clientCommand() argument
1490 if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { in clientCommand()
1493 addReplyBulkCBuffer(c,o,sdslen(o)); in clientCommand()
1495 } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) { in clientCommand()
1497 if (!strcasecmp(c->argv[2]->ptr,"on")) { in clientCommand()
1498 c->flags &= ~(CLIENT_REPLY_SKIP|CLIENT_REPLY_OFF); in clientCommand()
1499 addReply(c,shared.ok); in clientCommand()
1500 } else if (!strcasecmp(c->argv[2]->ptr,"off")) { in clientCommand()
1501 c->flags |= CLIENT_REPLY_OFF; in clientCommand()
1502 } else if (!strcasecmp(c->argv[2]->ptr,"skip")) { in clientCommand()
1503 if (!(c->flags & CLIENT_REPLY_OFF)) in clientCommand()
1504 c->flags |= CLIENT_REPLY_SKIP_NEXT; in clientCommand()
1506 addReply(c,shared.syntaxerr); in clientCommand()
1509 } else if (!strcasecmp(c->argv[1]->ptr,"kill")) { in clientCommand()
1518 if (c->argc == 3) { in clientCommand()
1520 addr = c->argv[2]->ptr; in clientCommand()
1522 } else if (c->argc > 3) { in clientCommand()
1526 while(i < c->argc) { in clientCommand()
1527 int moreargs = c->argc > i+1; in clientCommand()
1529 if (!strcasecmp(c->argv[i]->ptr,"id") && moreargs) { in clientCommand()
1532 if (getLongLongFromObjectOrReply(c,c->argv[i+1],&tmp,NULL) in clientCommand()
1535 } else if (!strcasecmp(c->argv[i]->ptr,"type") && moreargs) { in clientCommand()
1536 type = getClientTypeByName(c->argv[i+1]->ptr); in clientCommand()
1538 addReplyErrorFormat(c,"Unknown client type '%s'", in clientCommand()
1539 (char*) c->argv[i+1]->ptr); in clientCommand()
1542 } else if (!strcasecmp(c->argv[i]->ptr,"addr") && moreargs) { in clientCommand()
1543 addr = c->argv[i+1]->ptr; in clientCommand()
1544 } else if (!strcasecmp(c->argv[i]->ptr,"skipme") && moreargs) { in clientCommand()
1545 if (!strcasecmp(c->argv[i+1]->ptr,"yes")) { in clientCommand()
1547 } else if (!strcasecmp(c->argv[i+1]->ptr,"no")) { in clientCommand()
1550 addReply(c,shared.syntaxerr); in clientCommand()
1554 addReply(c,shared.syntaxerr); in clientCommand()
1560 addReply(c,shared.syntaxerr); in clientCommand()
1571 if (c == client && skipme) continue; in clientCommand()
1574 if (c == client) { in clientCommand()
1583 if (c->argc == 3) { in clientCommand()
1585 addReplyError(c,"No such client"); in clientCommand()
1587 addReply(c,shared.ok); in clientCommand()
1589 addReplyLongLong(c,killed); in clientCommand()
1594 if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY; in clientCommand()
1595 } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) { in clientCommand()
1596 int j, len = sdslen(c->argv[2]->ptr); in clientCommand()
1597 char *p = c->argv[2]->ptr; in clientCommand()
1602 if (c->name) decrRefCount(c->name); in clientCommand()
1603 c->name = NULL; in clientCommand()
1604 addReply(c,shared.ok); in clientCommand()
1613 addReplyError(c, in clientCommand()
1619 if (c->name) decrRefCount(c->name); in clientCommand()
1620 c->name = c->argv[2]; in clientCommand()
1621 incrRefCount(c->name); in clientCommand()
1622 addReply(c,shared.ok); in clientCommand()
1623 } else if (!strcasecmp(c->argv[1]->ptr,"getname") && c->argc == 2) { in clientCommand()
1624 if (c->name) in clientCommand()
1625 addReplyBulk(c,c->name); in clientCommand()
1627 addReply(c,shared.nullbulk); in clientCommand()
1628 } else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) { in clientCommand()
1631 if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration,UNIT_MILLISECONDS) in clientCommand()
1634 addReply(c,shared.ok); in clientCommand()
1636 … addReplyError(c, "Syntax error, try CLIENT (LIST | KILL | GETNAME | SETNAME | PAUSE | REPLY)"); in clientCommand()
1643 void rewriteClientCommandVector(client *c, int argc, ...) { in rewriteClientCommandVector() argument
1660 for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]); in rewriteClientCommandVector()
1661 zfree(c->argv); in rewriteClientCommandVector()
1663 c->argv = argv; in rewriteClientCommandVector()
1664 c->argc = argc; in rewriteClientCommandVector()
1665 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); in rewriteClientCommandVector()
1666 serverAssertWithInfo(c,NULL,c->cmd != NULL); in rewriteClientCommandVector()
1671 void replaceClientCommandVector(client *c, int argc, robj **argv) { in replaceClientCommandVector() argument
1672 freeClientArgv(c); in replaceClientCommandVector()
1673 zfree(c->argv); in replaceClientCommandVector()
1674 c->argv = argv; in replaceClientCommandVector()
1675 c->argc = argc; in replaceClientCommandVector()
1676 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); in replaceClientCommandVector()
1677 serverAssertWithInfo(c,NULL,c->cmd != NULL); in replaceClientCommandVector()
1691 void rewriteClientCommandArgument(client *c, int i, robj *newval) { in rewriteClientCommandArgument() argument
1694 if (i >= c->argc) { in rewriteClientCommandArgument()
1695 c->argv = zrealloc(c->argv,sizeof(robj*)*(i+1)); in rewriteClientCommandArgument()
1696 c->argc = i+1; in rewriteClientCommandArgument()
1697 c->argv[i] = NULL; in rewriteClientCommandArgument()
1699 oldval = c->argv[i]; in rewriteClientCommandArgument()
1700 c->argv[i] = newval; in rewriteClientCommandArgument()
1706 c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); in rewriteClientCommandArgument()
1707 serverAssertWithInfo(c,NULL,c->cmd != NULL); in rewriteClientCommandArgument()
1724 unsigned long getClientOutputBufferMemoryUsage(client *c) { in getClientOutputBufferMemoryUsage() argument
1727 return c->reply_bytes + (list_item_size*listLength(c->reply)); in getClientOutputBufferMemoryUsage()
1739 int getClientType(client *c) { in getClientType() argument
1740 if (c->flags & CLIENT_MASTER) return CLIENT_TYPE_MASTER; in getClientType()
1741 if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) in getClientType()
1743 if (c->flags & CLIENT_PUBSUB) return CLIENT_TYPE_PUBSUB; in getClientType()
1771 int checkClientOutputBufferLimits(client *c) { in checkClientOutputBufferLimits() argument
1773 unsigned long used_mem = getClientOutputBufferMemoryUsage(c); in checkClientOutputBufferLimits()
1775 class = getClientType(c); in checkClientOutputBufferLimits()
1790 if (c->obuf_soft_limit_reached_time == 0) { in checkClientOutputBufferLimits()
1791 c->obuf_soft_limit_reached_time = server.unixtime; in checkClientOutputBufferLimits()
1794 time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time; in checkClientOutputBufferLimits()
1804 c->obuf_soft_limit_reached_time = 0; in checkClientOutputBufferLimits()
1816 void asyncCloseClientOnOutputBufferLimitReached(client *c) { in asyncCloseClientOnOutputBufferLimitReached() argument
1817 serverAssert(c->reply_bytes < SIZE_MAX-(1024*64)); in asyncCloseClientOnOutputBufferLimitReached()
1818 if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return; in asyncCloseClientOnOutputBufferLimitReached()
1819 if (checkClientOutputBufferLimits(c)) { in asyncCloseClientOnOutputBufferLimitReached()
1820 sds client = catClientInfoString(sdsempty(),c); in asyncCloseClientOnOutputBufferLimitReached()
1822 freeClientAsync(c); in asyncCloseClientOnOutputBufferLimitReached()
1888 client *c; in clientsArePaused() local
1896 c = listNodeValue(ln); in clientsArePaused()
1900 if (c->flags & (CLIENT_SLAVE|CLIENT_BLOCKED)) continue; in clientsArePaused()
1901 c->flags |= CLIENT_UNBLOCKED; in clientsArePaused()
1902 listAddNodeTail(server.unblocked_clients,c); in clientsArePaused()