non blocking loading of the DB / AOF with informations and ETA in INFO output

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index cc4c9341d6b2ad6d4bb2323e1a8c6a59ffe7ec7e..634e2107c0fcedb288977daf562f53bb893b8c75 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -55,7 +55,12 @@ redisClient *createClient(int fd) {
     return 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) {
+    /* When CLOSE_AFTER_REPLY is set, no more replies may be added! */
+    redisAssert(!(c->flags & REDIS_CLOSE_AFTER_REPLY));
+
     if (c->fd <= 0) return REDIS_ERR;
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
@@ -334,23 +339,11 @@ void addReplyBulkCString(redisClient *c, char *s) {
     }
 }
 
-void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
-    int cport, cfd;
-    char cip[128];
+static void acceptCommonHandler(int fd) {
     redisClient *c;
-    REDIS_NOTUSED(el);
-    REDIS_NOTUSED(mask);
-    REDIS_NOTUSED(privdata);
-
-    cfd = anetAccept(server.neterr, fd, cip, &cport);
-    if (cfd == AE_ERR) {
-        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
-        return;
-    }
-    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
-    if ((c = createClient(cfd)) == NULL) {
+    if ((c = createClient(fd)) == NULL) {
         redisLog(REDIS_WARNING,"Error allocating resoures for the client");
-        close(cfd); /* May be already closed, just ingore errors */
+        close(fd); /* May be already closed, just ingore errors */
         return;
     }
     /* If maxclient directive is set and this is one client more... close the
@@ -370,6 +363,38 @@ void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     server.stat_numconnections++;
 }
 
+void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
+    int cport, cfd;
+    char cip[128];
+    REDIS_NOTUSED(el);
+    REDIS_NOTUSED(mask);
+    REDIS_NOTUSED(privdata);
+
+    cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
+    if (cfd == AE_ERR) {
+        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+        return;
+    }
+    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
+    acceptCommonHandler(cfd);
+}
+
+void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
+    int cfd;
+    REDIS_NOTUSED(el);
+    REDIS_NOTUSED(mask);
+    REDIS_NOTUSED(privdata);
+
+    cfd = anetUnixAccept(server.neterr, fd);
+    if (cfd == AE_ERR) {
+        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+        return;
+    }
+    redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
+    acceptCommonHandler(cfd);
+}
+
+
 static void freeClientArgv(redisClient *c) {
     int j;
     for (j = 0; j < c->argc; j++)
@@ -442,6 +467,7 @@ void freeClient(redisClient *c) {
     /* Case 2: we lost the connection with the master. */
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
+        /* FIXME */
         server.replstate = REDIS_REPL_CONNECT;
         /* Since we lost the connection with the master, we should also
          * close the connection with all our slaves if we have any, so
@@ -541,7 +567,6 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
 
         /* Close connection after entire reply has been sent. */
-        if (c->flags & REDIS_QUIT) freeClient(c);
         if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);
     }
 }
@@ -724,6 +749,10 @@ int processMultibulkBuffer(redisClient *c) {
         if (c->multibulklen <= 0) {
             c->querybuf = sdsrange(c->querybuf,pos,-1);
             return REDIS_OK;
+        } else if (c->multibulklen > 1024*1024) {
+            addReplyError(c,"Protocol error: invalid multibulk length");
+            setProtocolError(c,pos);
+            return REDIS_ERR;
         }
 
         /* Setup argv array on client structure */
@@ -790,17 +819,13 @@ int processMultibulkBuffer(redisClient *c) {
 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. */
+        /* Immediately abort if the client is in the middle of something. */
         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;
+        /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
+         * written to the client. Make sure to not let the reply grow after
+         * this flag has been set (i.e. don't process more commands). */
+        if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
 
         /* Determine request type when unknown. */
         if (!c->reqtype) {
@@ -820,9 +845,13 @@ void processInputBuffer(redisClient *c) {
         }
 
         /* Multibulk processing could see a <= 0 length. */
-        if (c->argc > 0)
-            processCommand(c);
-        resetClient(c);
+        if (c->argc == 0) {
+            resetClient(c);
+        } else {
+            /* Only reset the client when the command was executed. */
+            if (processCommand(c) == REDIS_OK)
+                resetClient(c);
+        }
     }
 }