selectDb(c,0);
c->fd = fd;
c->querybuf = sdsempty();
- c->newline = NULL;
+ c->reqtype = 0;
c->argc = 0;
c->argv = NULL;
+ c->multibulklen = 0;
c->bulklen = -1;
- c->multibulk = 0;
- c->mbargc = 0;
- c->mbargv = NULL;
c->sentlen = 0;
c->flags = 0;
c->lastinteraction = time(NULL);
static void freeClientArgv(redisClient *c) {
int j;
-
for (j = 0; j < c->argc; j++)
decrRefCount(c->argv[j]);
- for (j = 0; j < c->mbargc; j++)
- decrRefCount(c->mbargv[j]);
c->argc = 0;
- c->mbargc = 0;
}
void freeClient(redisClient *c) {
}
/* Release memory */
zfree(c->argv);
- zfree(c->mbargv);
freeClientMultiState(c);
zfree(c);
}
/* Close connection after entire reply has been sent. */
if (c->flags & REDIS_QUIT) freeClient(c);
+ if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);
}
}
/* resetClient prepare the client to process the next command */
void resetClient(redisClient *c) {
freeClientArgv(c);
+ c->reqtype = 0;
+ c->multibulklen = 0;
c->bulklen = -1;
- c->multibulk = 0;
- c->newline = NULL;
}
void closeTimedoutClients(void) {
}
}
-void processInputBuffer(redisClient *c) {
- int seeknewline = 0;
-
-again:
- /* Before to process the input buffer, make sure the client is not
- * waitig for a blocking operation such as BLPOP. Note that the first
- * iteration the client is never blocked, otherwise the processInputBuffer
- * would not be called at all, but after the execution of the first commands
- * in the input buffer the client may be blocked, and the "goto again"
- * will try to reiterate. The following line will make it return asap. */
- if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
-
- /* Never continue to process the input buffer after QUIT. After the output
- * buffer is flushed (with the OK), the connection will be dropped. */
- if (c->flags & REDIS_QUIT) return;
-
- if (seeknewline && c->bulklen == -1) c->newline = strchr(c->querybuf,'\n');
- seeknewline = 1;
- if (c->bulklen == -1) {
- /* Read the first line of the query */
- size_t querylen;
-
- if (c->newline) {
- char *p = c->newline;
- sds query, *argv;
- int argc, j;
-
- c->newline = NULL;
- query = c->querybuf;
- c->querybuf = sdsempty();
- querylen = 1+(p-(query));
- if (sdslen(query) > querylen) {
- /* leave data after the first line of the query in the buffer */
- c->querybuf = sdscatlen(c->querybuf,query+querylen,sdslen(query)-querylen);
- }
- *p = '\0'; /* remove "\n" */
- if (*(p-1) == '\r') *(p-1) = '\0'; /* and "\r" if any */
- sdsupdatelen(query);
-
- /* Now we can split the query in arguments */
- argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
- sdsfree(query);
-
- if (c->argv) zfree(c->argv);
- c->argv = zmalloc(sizeof(robj*)*argc);
-
- for (j = 0; j < argc; j++) {
- if (sdslen(argv[j])) {
- c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
- c->argc++;
- } else {
- sdsfree(argv[j]);
+int processInlineBuffer(redisClient *c) {
+ char *newline = strstr(c->querybuf,"\r\n");
+ int argc, j;
+ sds *argv;
+ size_t querylen;
+
+ /* Nothing to do without a \r\n */
+ if (newline == NULL)
+ return REDIS_ERR;
+
+ /* Split the input buffer up to the \r\n */
+ querylen = newline-(c->querybuf);
+ argv = sdssplitlen(c->querybuf,querylen," ",1,&argc);
+
+ /* Leave data after the first line of the query in the buffer */
+ c->querybuf = sdsrange(c->querybuf,querylen+2,-1);
+
+ /* Setup argv array on client structure */
+ if (c->argv) zfree(c->argv);
+ c->argv = zmalloc(sizeof(robj*)*argc);
+
+ /* Create redis objects for all arguments. */
+ for (c->argc = 0, j = 0; j < argc; j++) {
+ if (sdslen(argv[j])) {
+ c->argv[c->argc] = createObject(REDIS_STRING,argv[j]);
+ c->argc++;
+ } else {
+ sdsfree(argv[j]);
+ }
+ }
+ zfree(argv);
+ return REDIS_OK;
+}
+
+/* Helper function. Trims query buffer to make the function that processes
+ * multi bulk requests idempotent. */
+static void setProtocolError(redisClient *c, int pos) {
+ c->flags |= REDIS_CLOSE_AFTER_REPLY;
+ c->querybuf = sdsrange(c->querybuf,pos,-1);
+}
+
+int processMultibulkBuffer(redisClient *c) {
+ char *newline = NULL;
+ char *eptr;
+ int pos = 0, tolerr;
+ long bulklen;
+
+ if (c->multibulklen == 0) {
+ /* The client should have been reset */
+ redisAssert(c->argc == 0);
+
+ /* Multi bulk length cannot be read without a \r\n */
+ newline = strstr(c->querybuf,"\r\n");
+ if (newline == NULL)
+ return REDIS_ERR;
+
+ /* We know for sure there is a whole line since newline != NULL,
+ * so go ahead and find out the multi bulk length. */
+ redisAssert(c->querybuf[0] == '*');
+ c->multibulklen = strtol(c->querybuf+1,&eptr,10);
+ pos = (newline-c->querybuf)+2;
+ if (c->multibulklen <= 0) {
+ c->querybuf = sdsrange(c->querybuf,pos,-1);
+ return REDIS_OK;
+ }
+
+ /* Setup argv array on client structure */
+ if (c->argv) zfree(c->argv);
+ c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
+
+ /* Search new newline */
+ newline = strstr(c->querybuf+pos,"\r\n");
+ }
+
+ redisAssert(c->multibulklen > 0);
+ while(c->multibulklen) {
+ /* Read bulk length if unknown */
+ if (c->bulklen == -1) {
+ newline = strstr(c->querybuf+pos,"\r\n");
+ if (newline != NULL) {
+ if (c->querybuf[pos] != '$') {
+ addReplyErrorFormat(c,
+ "Protocol error: expected '$', got '%c'",
+ c->querybuf[pos]);
+ setProtocolError(c,pos);
+ return REDIS_ERR;
}
+
+ bulklen = strtol(c->querybuf+pos+1,&eptr,10);
+ tolerr = (eptr[0] != '\r');
+ if (tolerr || bulklen == LONG_MIN || bulklen == LONG_MAX ||
+ bulklen < 0 || bulklen > 1024*1024*1024)
+ {
+ addReplyError(c,"Protocol error: invalid bulk length");
+ setProtocolError(c,pos);
+ return REDIS_ERR;
+ }
+ pos += eptr-(c->querybuf+pos)+2;
+ c->bulklen = bulklen;
+ } else {
+ /* No newline in current buffer, so wait for more data */
+ break;
}
- zfree(argv);
- if (c->argc) {
- /* Execute the command. If the client is still valid
- * after processCommand() return and there is something
- * on the query buffer try to process the next command. */
- if (processCommand(c) && sdslen(c->querybuf)) goto again;
+ }
+
+ /* Read bulk argument */
+ if (sdslen(c->querybuf)-pos < (unsigned)(c->bulklen+2)) {
+ /* Not enough data (+2 == trailing \r\n) */
+ break;
+ } else {
+ c->argv[c->argc++] = createStringObject(c->querybuf+pos,c->bulklen);
+ pos += c->bulklen+2;
+ c->bulklen = -1;
+ c->multibulklen--;
+ }
+ }
+
+ /* Trim to pos */
+ c->querybuf = sdsrange(c->querybuf,pos,-1);
+
+ /* We're done when c->multibulk == 0 */
+ if (c->multibulklen == 0) {
+ return REDIS_OK;
+ }
+ return REDIS_ERR;
+}
+
+void processInputBuffer(redisClient *c) {
+ /* Keep processing while there is something in the input buffer */
+ while(sdslen(c->querybuf)) {
+ /* Before to process the input buffer, make sure the client is not
+ * waitig for a blocking operation such as BLPOP. Note that the first
+ * iteration the client is never blocked, otherwise the processInputBuffer
+ * would not be called at all, but after the execution of the first commands
+ * in the input buffer the client may be blocked, and the "goto again"
+ * will try to reiterate. The following line will make it return asap. */
+ if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
+
+ /* Never continue to process the input buffer after QUIT. After the output
+ * buffer is flushed (with the OK), the connection will be dropped. */
+ if (c->flags & REDIS_QUIT) return;
+
+ /* Determine request type when unknown. */
+ if (!c->reqtype) {
+ if (c->querybuf[0] == '*') {
+ c->reqtype = REDIS_REQ_MULTIBULK;
} else {
- /* Nothing to process, argc == 0. Just process the query
- * buffer if it's not empty or return to the caller */
- if (sdslen(c->querybuf)) goto again;
+ c->reqtype = REDIS_REQ_INLINE;
}
- return;
- } else if (sdslen(c->querybuf) >= REDIS_REQUEST_MAX_SIZE) {
- redisLog(REDIS_VERBOSE, "Client protocol error");
- freeClient(c);
- return;
}
- } else {
- /* Bulk read handling. Note that if we are at this point
- the client already sent a command terminated with a newline,
- we are reading the bulk data that is actually the last
- argument of the command. */
- int qbl = sdslen(c->querybuf);
-
- if (c->bulklen <= qbl) {
- /* Copy everything but the final CRLF as final argument */
- c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
- c->argc++;
- c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
- /* Process the command. If the client is still valid after
- * the processing and there is more data in the buffer
- * try to parse it. */
- if (processCommand(c) && sdslen(c->querybuf)) goto again;
- return;
+
+ if (c->reqtype == REDIS_REQ_INLINE) {
+ if (processInlineBuffer(c) != REDIS_OK) break;
+ } else if (c->reqtype == REDIS_REQ_MULTIBULK) {
+ if (processMultibulkBuffer(c) != REDIS_OK) break;
+ } else {
+ redisPanic("Unknown request type");
}
+
+ /* Multibulk processing could see a <= 0 length. */
+ if (c->argc > 0)
+ processCommand(c);
+ resetClient(c);
}
}
return;
}
if (nread) {
- size_t oldlen = sdslen(c->querybuf);
- c->querybuf = sdscatlen(c->querybuf, buf, nread);
+ c->querybuf = sdscatlen(c->querybuf,buf,nread);
c->lastinteraction = time(NULL);
- /* Scan this new piece of the query for the newline. We do this
- * here in order to make sure we perform this scan just one time
- * per piece of buffer, leading to an O(N) scan instead of O(N*N) */
- if (c->bulklen == -1 && c->newline == NULL)
- c->newline = strchr(c->querybuf+oldlen,'\n');
} else {
return;
}
int processCommand(redisClient *c) {
struct redisCommand *cmd;
- /* Handle the multi bulk command type. This is an alternative protocol
- * supported by Redis in order to receive commands that are composed of
- * multiple binary-safe "bulk" arguments. The latency of processing is
- * a bit higher but this allows things like multi-sets, so if this
- * protocol is used only for MSET and similar commands this is a big win. */
- if (c->multibulk == 0 && c->argc == 1 && ((char*)(c->argv[0]->ptr))[0] == '*') {
- c->multibulk = atoi(((char*)c->argv[0]->ptr)+1);
- if (c->multibulk <= 0) {
- resetClient(c);
- return 1;
- } else {
- decrRefCount(c->argv[c->argc-1]);
- c->argc--;
- return 1;
- }
- } else if (c->multibulk) {
- if (c->bulklen == -1) {
- if (((char*)c->argv[0]->ptr)[0] != '$') {
- addReplyError(c,"multi bulk protocol error");
- resetClient(c);
- return 1;
- } else {
- char *eptr;
- long bulklen = strtol(((char*)c->argv[0]->ptr)+1,&eptr,10);
- int perr = eptr[0] != '\0';
-
- decrRefCount(c->argv[0]);
- if (perr || bulklen == LONG_MIN || bulklen == LONG_MAX ||
- bulklen < 0 || bulklen > 1024*1024*1024)
- {
- c->argc--;
- addReplyError(c,"invalid bulk write count");
- resetClient(c);
- return 1;
- }
- c->argc--;
- c->bulklen = bulklen+2; /* add two bytes for CR+LF */
- return 1;
- }
- } else {
- c->mbargv = zrealloc(c->mbargv,(sizeof(robj*))*(c->mbargc+1));
- c->mbargv[c->mbargc] = c->argv[0];
- c->mbargc++;
- c->argc--;
- c->multibulk--;
- if (c->multibulk == 0) {
- robj **auxargv;
- int auxargc;
-
- /* Here we need to swap the multi-bulk argc/argv with the
- * normal argc/argv of the client structure. */
- auxargv = c->argv;
- c->argv = c->mbargv;
- c->mbargv = auxargv;
-
- auxargc = c->argc;
- c->argc = c->mbargc;
- c->mbargc = auxargc;
-
- /* We need to set bulklen to something different than -1
- * in order for the code below to process the command without
- * to try to read the last argument of a bulk command as
- * a special argument. */
- c->bulklen = 0;
- /* continue below and process the command */
- } else {
- c->bulklen = -1;
- return 1;
- }
- }
- }
- /* -- end of multi bulk commands processing -- */
-
/* The QUIT command is handled separately. Normal command procs will
* go through checking for replication and QUIT will cause trouble
* when FORCE_REPLICATION is enabled and would be implemented in
if (!strcasecmp(c->argv[0]->ptr,"quit")) {
c->flags |= REDIS_QUIT;
addReply(c,shared.ok);
- return 0;
+ return REDIS_ERR;
}
/* Now lookup the command and check ASAP about trivial error conditions
if (!cmd) {
addReplyErrorFormat(c,"unknown command '%s'",
(char*)c->argv[0]->ptr);
- resetClient(c);
- return 1;
+ return REDIS_OK;
} else if ((cmd->arity > 0 && cmd->arity != c->argc) ||
(c->argc < -cmd->arity)) {
addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
cmd->name);
- resetClient(c);
- return 1;
- } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
- /* This is a bulk command, we have to read the last argument yet. */
- char *eptr;
- long bulklen = strtol(c->argv[c->argc-1]->ptr,&eptr,10);
- int perr = eptr[0] != '\0';
-
- decrRefCount(c->argv[c->argc-1]);
- if (perr || bulklen == LONG_MAX || bulklen == LONG_MIN ||
- bulklen < 0 || bulklen > 1024*1024*1024)
- {
- c->argc--;
- addReplyError(c,"invalid bulk write count");
- resetClient(c);
- return 1;
- }
- c->argc--;
- c->bulklen = bulklen+2; /* add two bytes for CR+LF */
- /* It is possible that the bulk read is already in the
- * buffer. Check this condition and handle it accordingly.
- * This is just a fast path, alternative to call processInputBuffer().
- * It's a good idea since the code is small and this condition
- * happens most of the times. */
- if ((signed)sdslen(c->querybuf) >= c->bulklen) {
- c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
- c->argc++;
- c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
- } else {
- /* Otherwise return... there is to read the last argument
- * from the socket. */
- return 1;
- }
+ return REDIS_OK;
}
+
/* Let's try to encode the bulk object to save space. */
if (cmd->flags & REDIS_CMD_BULK)
c->argv[c->argc-1] = tryObjectEncoding(c->argv[c->argc-1]);
/* Check if the user is authenticated */
if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
addReplyError(c,"operation not permitted");
- resetClient(c);
- return 1;
+ return REDIS_OK;
}
/* Handle the maxmemory directive.
zmalloc_used_memory() > server.maxmemory)
{
addReplyError(c,"command not allowed when used memory > 'maxmemory'");
- resetClient(c);
- return 1;
+ return REDIS_OK;
}
/* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
cmd->proc != subscribeCommand && cmd->proc != unsubscribeCommand &&
cmd->proc != psubscribeCommand && cmd->proc != punsubscribeCommand) {
addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context");
- resetClient(c);
- return 1;
+ return REDIS_OK;
}
/* Exec the command */
blockClientOnSwappedKeys(c,cmd)) return 1;
call(c,cmd);
}
-
- /* Prepare the client for the next command */
- resetClient(c);
- return 1;
+ return REDIS_OK;
}
/*================================== Shutdown =============================== */