Merge pull request #118 from dvirsky/unstable

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 8f2e6d8f49b7596d725c865c374435cbadd5a500..f14b4ef8cec2a371fb4dce84402893ee946ffd63 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -14,14 +14,20 @@ redisClient *createClient(int fd) {
     redisClient *c = zmalloc(sizeof(redisClient));
     c->bufpos = 0;
 
     redisClient *c = zmalloc(sizeof(redisClient));
     c->bufpos = 0;
 

-    anetNonBlock(NULL,fd);
-    anetTcpNoDelay(NULL,fd);
-    if (aeCreateFileEvent(server.el,fd,AE_READABLE,
-        readQueryFromClient, c) == AE_ERR)
-    {
-        close(fd);
-        zfree(c);
-        return NULL;
+    /* passing -1 as fd it is possible to create a non connected client.
+     * This is useful since all the Redis commands needs to be executed
+     * in the context of a client. When commands are executed in other
+     * contexts (for instance a Lua script) we need a non connected client. */
+    if (fd != -1) {
+        anetNonBlock(NULL,fd);
+        anetTcpNoDelay(NULL,fd);
+        if (aeCreateFileEvent(server.el,fd,AE_READABLE,
+            readQueryFromClient, c) == AE_ERR)
+        {
+            close(fd);
+            zfree(c);
+            return NULL;
+        }

     }
 
     selectDb(c,0);
     }
 
     selectDb(c,0);


@@ -30,6 +36,7 @@ redisClient *createClient(int fd) {
     c->reqtype = 0;
     c->argc = 0;
     c->argv = NULL;
     c->reqtype = 0;
     c->argc = 0;
     c->argv = NULL;

+    c->cmd = NULL;

     c->multibulklen = 0;
     c->bulklen = -1;
     c->sentlen = 0;
     c->multibulklen = 0;
     c->bulklen = -1;
     c->sentlen = 0;


@@ -51,7 +58,7 @@ redisClient *createClient(int fd) {
     c->pubsub_patterns = listCreate();
     listSetFreeMethod(c->pubsub_patterns,decrRefCount);
     listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
     c->pubsub_patterns = listCreate();
     listSetFreeMethod(c->pubsub_patterns,decrRefCount);
     listSetMatchMethod(c->pubsub_patterns,listMatchObjects);

-    listAddNodeTail(server.clients,c);
+    if (fd != -1) listAddNodeTail(server.clients,c);

     initClientMultiState(c);
     return c;
 }
     initClientMultiState(c);
     return c;
 }


@@ -59,6 +66,7 @@ redisClient *createClient(int fd) {
 /* Set the event loop to listen for write events on the client's socket.
  * Typically gets called every time a reply is built. */
 int _installWriteEvent(redisClient *c) {
 /* Set the event loop to listen for write events on the client's socket.
  * Typically gets called every time a reply is built. */
 int _installWriteEvent(redisClient *c) {

+    if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;

     if (c->fd <= 0) return REDIS_ERR;
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
     if (c->fd <= 0) return REDIS_ERR;
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||


@@ -239,10 +247,17 @@ void addReplyError(redisClient *c, char *err) {
 }
 
 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
 }
 
 void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {

+    size_t l, j;

     va_list ap;
     va_start(ap,fmt);
     sds s = sdscatvprintf(sdsempty(),fmt,ap);
     va_end(ap);
     va_list ap;
     va_start(ap,fmt);
     sds s = sdscatvprintf(sdsempty(),fmt,ap);
     va_end(ap);

+    /* Make sure there are no newlines in the string, otherwise invalid protocol
+     * is emitted. */
+    l = sdslen(s);
+    for (j = 0; j < l; j++) {
+        if (s[j] == '\r' || s[j] == '\n') s[j] = ' ';
+    }

     _addReplyError(c,s,sdslen(s));
     sdsfree(s);
 }
     _addReplyError(c,s,sdslen(s));
     sdsfree(s);
 }


@@ -447,6 +462,7 @@ static void freeClientArgv(redisClient *c) {
     for (j = 0; j < c->argc; j++)
         decrRefCount(c->argv[j]);
     c->argc = 0;
     for (j = 0; j < c->argc; j++)
         decrRefCount(c->argv[j]);
     c->argc = 0;

+    c->cmd = NULL;

 }
 
 void freeClient(redisClient *c) {
 }
 
 void freeClient(redisClient *c) {


@@ -487,25 +503,6 @@ void freeClient(redisClient *c) {
         redisAssert(ln != NULL);
         listDelNode(server.unblocked_clients,ln);
     }
         redisAssert(ln != NULL);
         listDelNode(server.unblocked_clients,ln);
     }

-    /* Remove from the list of clients waiting for swapped keys, or ready
-     * to be restarted, but not yet woken up again. */
-    if (c->flags & REDIS_IO_WAIT) {
-        redisAssert(server.ds_enabled);
-        if (listLength(c->io_keys) == 0) {
-            ln = listSearchKey(server.io_ready_clients,c);
-
-            /* When this client is waiting to be woken up (REDIS_IO_WAIT),
-             * it should be present in the list io_ready_clients */
-            redisAssert(ln != NULL);
-            listDelNode(server.io_ready_clients,ln);
-        } else {
-            while (listLength(c->io_keys)) {
-                ln = listFirst(c->io_keys);
-                dontWaitForSwappedKey(c,ln->value);
-            }
-        }
-        server.cache_blocked_clients--;
-    }

     listRelease(c->io_keys);
     /* Master/slave cleanup.
      * Case 1: we lost the connection with a slave. */
     listRelease(c->io_keys);
     /* Master/slave cleanup.
      * Case 1: we lost the connection with a slave. */


@@ -613,7 +610,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         }
     }
     if (totwritten > 0) c->lastinteraction = time(NULL);
         }
     }
     if (totwritten > 0) c->lastinteraction = time(NULL);

-    if (listLength(c->reply) == 0) {
+    if (c->bufpos == 0 && listLength(c->reply) == 0) {

         c->sentlen = 0;
         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
 
         c->sentlen = 0;
         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
 


@@ -706,7 +703,7 @@ int processMultibulkBuffer(redisClient *c) {
 
     if (c->multibulklen == 0) {
         /* The client should have been reset */
 
     if (c->multibulklen == 0) {
         /* The client should have been reset */

-        redisAssert(c->argc == 0);
+        redisAssertWithInfo(c,NULL,c->argc == 0);

 
         /* Multi bulk length cannot be read without a \r\n */
         newline = strchr(c->querybuf,'\r');
 
         /* Multi bulk length cannot be read without a \r\n */
         newline = strchr(c->querybuf,'\r');


@@ -719,7 +716,7 @@ int processMultibulkBuffer(redisClient *c) {
 
         /* We know for sure there is a whole line since newline != NULL,
          * so go ahead and find out the multi bulk length. */
 
         /* 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] == '*');
+        redisAssertWithInfo(c,NULL,c->querybuf[0] == '*');

         ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
         if (!ok || ll > 1024*1024) {
             addReplyError(c,"Protocol error: invalid multibulk length");
         ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
         if (!ok || ll > 1024*1024) {
             addReplyError(c,"Protocol error: invalid multibulk length");


@@ -740,7 +737,7 @@ int processMultibulkBuffer(redisClient *c) {
         c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
     }
 
         c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
     }
 

-    redisAssert(c->multibulklen > 0);
+    redisAssertWithInfo(c,NULL,c->multibulklen > 0);

     while(c->multibulklen) {
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {
     while(c->multibulklen) {
         /* Read bulk length if unknown */
         if (c->bulklen == -1) {


@@ -797,7 +794,7 @@ void processInputBuffer(redisClient *c) {
     /* Keep processing while there is something in the input buffer */
     while(sdslen(c->querybuf)) {
         /* Immediately abort if the client is in the middle of something. */
     /* Keep processing while there is something in the input buffer */
     while(sdslen(c->querybuf)) {
         /* Immediately abort if the client is in the middle of something. */

-        if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
+        if (c->flags & REDIS_BLOCKED) return;

 
         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
          * written to the client. Make sure to not let the reply grow after
 
         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
          * written to the client. Make sure to not let the reply grow after


@@ -907,7 +904,6 @@ void clientCommand(redisClient *c) {
             if (p == flags) *p++ = 'N';
             if (client->flags & REDIS_MULTI) *p++ = 'x';
             if (client->flags & REDIS_BLOCKED) *p++ = 'b';
             if (p == flags) *p++ = 'N';
             if (client->flags & REDIS_MULTI) *p++ = 'x';
             if (client->flags & REDIS_BLOCKED) *p++ = 'b';

-            if (client->flags & REDIS_IO_WAIT) *p++ = 'i';

             if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
             if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
             if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
             if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
             if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
             if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';


@@ -948,6 +944,9 @@ void clientCommand(redisClient *c) {
     }
 }
 
     }
 }
 

+/* Rewrite the command vector of the client. All the new objects ref count
+ * is incremented. The old command vector is freed, and the old objects
+ * ref count is decremented. */

 void rewriteClientCommandVector(redisClient *c, int argc, ...) {
     va_list ap;
     int j;
 void rewriteClientCommandVector(redisClient *c, int argc, ...) {
     va_list ap;
     int j;


@@ -970,5 +969,25 @@ void rewriteClientCommandVector(redisClient *c, int argc, ...) {
     /* Replace argv and argc with our new versions. */
     c->argv = argv;
     c->argc = argc;
     /* Replace argv and argc with our new versions. */
     c->argv = argv;
     c->argc = argc;

+    c->cmd = lookupCommand(c->argv[0]->ptr);
+    redisAssertWithInfo(c,NULL,c->cmd != NULL);

     va_end(ap);
 }
     va_end(ap);
 }

+
+/* Rewrite a single item in the command vector.
+ * The new val ref count is incremented, and the old decremented. */
+void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
+    robj *oldval;
+   
+    redisAssertWithInfo(c,NULL,i < c->argc);
+    oldval = c->argv[i];
+    c->argv[i] = newval;
+    incrRefCount(newval);
+    decrRefCount(oldval);
+
+    /* If this is the command name make sure to fix c->cmd. */
+    if (i == 0) {
+        c->cmd = lookupCommand(c->argv[0]->ptr);
+        redisAssertWithInfo(c,NULL,c->cmd != NULL);
+    }
+}