Merge branch 'master' into brpoplpush

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 10273cc1100e3de57cdb34a2664a36b07ba2dab7..7d14ac53f18bfc8c60ef457ff3b4c6383e3d59da 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -41,8 +41,10 @@ redisClient *createClient(int fd) {
     c->reply = listCreate();
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     c->reply = listCreate();
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);

-    c->blocking_keys = NULL;
-    c->blocking_keys_num = 0;
+    c->bpop.keys = NULL;
+    c->bpop.count = 0;
+    c->bpop.timeout = 0;
+    c->bpop.target = NULL;

     c->io_keys = listCreate();
     c->watched_keys = listCreate();
     listSetFreeMethod(c->io_keys,decrRefCount);
     c->io_keys = listCreate();
     c->watched_keys = listCreate();
     listSetFreeMethod(c->io_keys,decrRefCount);


@@ -339,23 +341,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;
     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");
         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
         return;
     }
     /* If maxclient directive is set and this is one client more... close the


@@ -375,6 +365,38 @@ void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     server.stat_numconnections++;
 }
 
     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++)
 static void freeClientArgv(redisClient *c) {
     int j;
     for (j = 0; j < c->argc; j++)


@@ -447,6 +469,7 @@ void freeClient(redisClient *c) {
     /* Case 2: we lost the connection with the master. */
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
     /* 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
         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


@@ -656,7 +679,7 @@ void closeTimedoutClients(void) {
             redisLog(REDIS_VERBOSE,"Closing idle client");
             freeClient(c);
         } else if (c->flags & REDIS_BLOCKED) {
             redisLog(REDIS_VERBOSE,"Closing idle client");
             freeClient(c);
         } else if (c->flags & REDIS_BLOCKED) {

-            if (c->blockingto != 0 && c->blockingto < now) {
+            if (c->bpop.timeout != 0 && c->bpop.timeout < now) {

                 addReply(c,shared.nullmultibulk);
                 unblockClientWaitingData(c);
             }
                 addReply(c,shared.nullmultibulk);
                 unblockClientWaitingData(c);
             }


@@ -798,12 +821,7 @@ int processMultibulkBuffer(redisClient *c) {
 void processInputBuffer(redisClient *c) {
     /* Keep processing while there is something in the input buffer */
     while(sdslen(c->querybuf)) {
 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;
 
         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
         if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
 
         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is