Supports client-driven retry recovery for SET.

Author: seojinpark

deps/hiredis/hiredis.c

@@ -518,6 +518,9 @@ static int processItem(redisReader *r) {
             case '+':
                 cur->type = REDIS_REPLY_STATUS;
                 break;
+            case '@':
+                cur->type = REDIS_REPLY_STATUS;
+                break;
             case ':':
                 cur->type = REDIS_REPLY_INTEGER;
                 break;

redis.conf

@@ -199,9 +199,9 @@ databases 16
 #   save ""
 
 save ""
-save 900 1
-save 300 10
-save 60 10000
+#save 900 1
+#save 300 10
+#save 60 10000
 
 # By default Redis will stop accepting writes if RDB snapshots are enabled
 # (at least one save point) and the latest background save failed.
@@ -621,8 +621,8 @@ appendfilename "appendonly.aof"
 # If unsure, use "everysec".
 
 # appendfsync always
-# appendfsync everysec
- appendfsync no
+ appendfsync everysec
+# appendfsync no
 
 # When the AOF fsync policy is set to always or everysec, and a background
 # saving process (a background save or AOF log background rewriting) is

src/aof.c

@@ -702,6 +702,8 @@ int loadAppendOnlyFile(char *filename) {
             long long clientId, requestId;
             getLongLongFromObject(argv[argc-2], &clientId);
             getLongLongFromObject(argv[argc-1], &requestId);
+            fakeClient->clientId = clientId;
+            riflCheckClientIdOk(fakeClient);
             riflCheckDuplicate(clientId, requestId);
         }
 

src/db.c

@@ -60,9 +60,9 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
         }
         // TODO(seojin): verify every path needs this...
         if (!(flags & LOOKUP_NOTOUCH) && de->lastModOpNum > server.aof_last_fsync_opNum) {
-            serverLog(LL_NOTICE, "Redis read is reading non-durable data. "
-                    "It is blocked by fsync. ModifiedBy: %lld, Synced: %lld, CurrentOp: %lld",
-                    de->lastModOpNum, server.aof_last_fsync_opNum, server.currentOpNum);
+//            serverLog(LL_NOTICE, "Redis read is reading non-durable data. "
+//                    "It is blocked by fsync. ModifiedBy: %lld, Synced: %lld, CurrentOp: %lld",
+//                    de->lastModOpNum, server.aof_last_fsync_opNum, server.currentOpNum);
             flushAppendOnlyFile(1);
             aof_fsync(server.aof_fd);
             server.aof_last_fsync_opNum = server.currentOpNum - 1;

src/networking.c

@@ -613,7 +613,7 @@ int clientHasPendingReplies(client *c) {
 }
 
 #define MAX_ACCEPTS_PER_CALL 1000
-static void acceptCommonHandler(int fd, int flags, char *ip) {
+static void acceptCommonHandler(int fd, int flags, char *ip, bool isRecovery) {
     client *c;
     if ((c = createClient(fd)) == NULL) {
         serverLog(LL_WARNING,
@@ -622,6 +622,30 @@ static void acceptCommonHandler(int fd, int flags, char *ip) {
         close(fd); /* May be already closed, just ignore errors */
         return;
     }
+    c->isRecovery = isRecovery;
+
+    if (isRecovery && server.serverState >= SERVER_STATE_NORMAL) {
+        // Trial of recovery was too late. Server already served normal requests.
+        char *err = "-ERR server is already in normal mode. Too late for replay.\r\n";
+        if (write(c->fd,err,strlen(err)) == -1) {
+        }
+        server.stat_rejected_conn++;
+        freeClient(c);
+        return;
+    }
+
+    if (!isRecovery && server.serverState <= SERVER_STATE_ACCEPTING_REPLAY) {
+//        char *err = "-RETRY server is not ready.\r\n";
+//        if (write(c->fd,err,strlen(err)) == -1) {
+//        }
+        // Don't send any message. Sending random text causes protocol error
+        // instead of connection_error, which can be automatically handled
+        // and retry later.
+        server.stat_rejected_conn++;
+        freeClient(c);
+        return;
+    }
+
     /* If maxclient directive is set and this is one client more... close the
      * connection. Note that we create the client instead to check before
      * for this condition, since now the socket is already set in non-blocking
@@ -699,7 +723,27 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
             return;
         }
         serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
-        acceptCommonHandler(cfd,0,cip);
+        acceptCommonHandler(cfd,0,cip,false);
+    }
+}
+
+void acceptTcpHandler4replay(aeEventLoop *el, int fd, void *privdata, int mask) {
+    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
+    char cip[NET_IP_STR_LEN];
+    UNUSED(el);
+    UNUSED(mask);
+    UNUSED(privdata);
+
+    while(max--) {
+        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
+        if (cfd == ANET_ERR) {
+            if (errno != EWOULDBLOCK)
+                serverLog(LL_WARNING,
+                    "Accepting client connection for recovery: %s", server.neterr);
+            return;
+        }
+        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
+        acceptCommonHandler(cfd,0,cip,true);
     }
 }
 
@@ -718,7 +762,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
             return;
         }
         serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
-        acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL);
+        acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL,false);
     }
 }
 
@@ -1360,6 +1404,28 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         freeClient(c);
         return;
     }
+
+    if (c->isRecovery && server.serverState >= SERVER_STATE_NORMAL) {
+        // Trial of recovery was too late. Server already served normal requests.
+        char *err = "-ERR server is already in normal mode. Too late for replay.\r\n";
+        if (write(c->fd,err,strlen(err)) == -1) {
+        }
+        freeClient(c);
+        return;
+    }
+
+    if (!c->isRecovery && server.serverState <= SERVER_STATE_ACCEPTING_REPLAY) {
+        serverLog(LL_WARNING,"Non-recovery connection was accepted while recovery. Dying..");
+        char *err = "-RETRY server is not ready.\r\n";
+        if (write(c->fd,err,strlen(err)) == -1) {
+        }
+        resetClient(c);
+        return;
+//        server.stat_rejected_conn++;
+//        freeClient(c);
+//        exit(1);
+    }
+
     processInputBuffer(c);
 }
 

src/rifl.c

@@ -34,6 +34,7 @@ bool riflCheckDuplicate(long long clientId, long long requestId) {
 
     int index = clientId & bitmask;
     if (processedRpcIds[index] >= requestId) {
+//        serverLog(LL_NOTICE,"RIFL found duplicate. ClientId: %lld, requestId: %lld, lastRpcId: %lld", clientIds[index], requestId, processedRpcIds[index]);
         return true;
     }
     if (!witnessRecoveryMode) {
@@ -42,15 +43,25 @@ bool riflCheckDuplicate(long long clientId, long long requestId) {
     return false;
 }
 
+void riflPrintData() {
+    serverLog(LL_NOTICE,"RIFL Table dump after recovery.");
+    for (int i = 0; i < RIFL_TABLE_SIZE; ++i) {
+        if (clientIds[i] != 0) {
+            serverLog(LL_NOTICE,"ClientId: %lld, lastRpcId: %lld", clientIds[i], processedRpcIds[i]);
+        }
+    }
+}
+
 void riflStartRecoveryByWitness() {
     witnessRecoveryMode = true;
+    riflPrintData();
 }
 
 void riflEndRecoveryByWitness() {
     witnessRecoveryMode = false;
+    riflPrintData();
 }
 
-
 long long riflGetNext(long long clientId) {
     int index = (clientId + 1) & bitmask;
     while (index < RIFL_TABLE_SIZE) {

src/server.c

@@ -1289,6 +1289,18 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
         migrateCloseTimedoutSockets();
     }
 
+    run_with_period(1000) {
+        /* If server haven't received client's replay request for a while,
+         * assume completion of recover and start normal processing. */
+        if (server.serverState == SERVER_STATE_ACCEPTING_REPLAY &&
+                server.unixtime - server.last_client_replay > SECONDS_WAITING_REPLAY) {
+            server.serverState = SERVER_STATE_NORMAL;
+            riflEndRecoveryByWitness();
+            serverLog(LL_NOTICE,"Recovery finished (recovered %lld operations). "
+                    "Started to take normal requests.", server.currentOpNum);
+        }
+    }
+
     /* Start a scheduled BGSAVE if the corresponding flag is set. This is
      * useful when we are forced to postpone a BGSAVE because an AOF
      * rewrite is in progress.
@@ -1449,6 +1461,7 @@ void createSharedObjects(void) {
     shared.maxstring = createStringObject("maxstring",9);
     shared.riflDuplicate = createObject(OBJ_STRING,sdsnew("+OK (RIFL duplicate)\r\n"));
     shared.riflClientIdCollision = createObject(OBJ_STRING,sdsnew("-ERR (RIFL clientId collision)\r\n"));
+    shared.unsyncedOk = createObject(OBJ_STRING,sdsnew("@OK "));
 }
 
 void initServerConfig(void) {
@@ -1462,6 +1475,7 @@ void initServerConfig(void) {
     server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
     server.currentOpNum = 0;
     server.port = CONFIG_DEFAULT_SERVER_PORT;
+    server.portForRecovery = CONFIG_DEFAULT_RECOVERY_PORT;
     server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
     server.bindaddr_count = 0;
     server.unixsocket = NULL;
@@ -1901,6 +1915,10 @@ void initServer(void) {
         listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
         exit(1);
 
+    if (server.portForRecovery != 0 &&
+        listenToPort(server.portForRecovery,server.ipfd4replay,&server.ipfd4replay_count) == C_ERR)
+        exit(1);
+
     /* Open the listening Unix domain socket. */
     if (server.unixsocket != NULL) {
         unlink(server.unixsocket); /* don't care if this fails */
@@ -1935,6 +1953,7 @@ void initServer(void) {
     listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
     listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
     server.cronloops = 0;
+    server.serverState = SERVER_STATE_ACCEPTING_REPLAY;
     server.rdb_child_pid = -1;
     server.aof_child_pid = -1;
     server.rdb_child_type = RDB_CHILD_TYPE_NONE;
@@ -1956,6 +1975,7 @@ void initServer(void) {
     server.aof_last_write_errno = 0;
     server.repl_good_slaves_count = 0;
     updateCachedTime();
+    server.last_client_replay = server.unixtime + 5; // give extra 5 secs for bootstrapping.
 
     /* Create the serverCron() time event, that's our main way to process
      * background operations. */
@@ -1974,6 +1994,14 @@ void initServer(void) {
                     "Unrecoverable error creating server.ipfd file event.");
             }
     }
+    for (j = 0; j < server.ipfd4replay_count; j++) {
+        if (aeCreateFileEvent(server.el, server.ipfd4replay[j], AE_READABLE,
+            acceptTcpHandler4replay,NULL) == AE_ERR)
+            {
+                serverPanic(
+                    "Unrecoverable error creating server.ipfd4replay file event.");
+            }
+    }
     if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
         acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");
 
@@ -2268,7 +2296,9 @@ void call(client *c, int flags) {
             return;
         }
     }
-    ++server.currentOpNum;
+    if (c->cmd->proc != selectCommand) {
+        ++server.currentOpNum;
+    }
     c->cmd->proc(c);
     duration = ustime()-start;
     dirty = server.dirty-dirty;
@@ -4126,6 +4156,9 @@ int main(int argc, char **argv) {
         sentinelIsRunning();
     }
 
+    // Call this only after AOF recovery.
+    riflStartRecoveryByWitness();
+
     /* Warning the user about suspicious maxmemory setting. */
     if (server.maxmemory > 0 && server.maxmemory < 1024*1024) {
         serverLog(LL_WARNING,"WARNING: You specified a maxmemory value that is less than 1MB (current value is %llu bytes). Are you sure this is what you really want?", server.maxmemory);

src/server.h

@@ -80,6 +80,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CONFIG_MIN_HZ            1
 #define CONFIG_MAX_HZ            500
 #define CONFIG_DEFAULT_SERVER_PORT        6379    /* TCP port */
+#define CONFIG_DEFAULT_RECOVERY_PORT      6380    /* TCP port for recovery*/
 #define CONFIG_DEFAULT_TCP_BACKLOG       511     /* TCP listen backlog */
 #define CONFIG_DEFAULT_CLIENT_TIMEOUT       0       /* default client timeout: infinite */
 #define CONFIG_DEFAULT_DBNUM     16
@@ -439,6 +440,14 @@ typedef long long mstime_t; /* millisecond time type. */
 #define NOTIFY_EVICTED (1<<9)     /* e */
 #define NOTIFY_ALL (NOTIFY_GENERIC | NOTIFY_STRING | NOTIFY_LIST | NOTIFY_SET | NOTIFY_HASH | NOTIFY_ZSET | NOTIFY_EXPIRED | NOTIFY_EVICTED)      /* A */
 
+/* Server's opration state. */
+#define SERVER_STATE_ACCEPTING_REPLAY 1   /* Server only takes client's replay requests. */
+#define SERVER_STATE_NORMAL 2           /* Server takes normal request now. */
+
+#define SECONDS_WAITING_REPLAY 10       /* server waits extra seconds after last
+                                           client replay before swithing from
+                        SERVER_STATE_ACCEPTING_REPLAY to SERVER_STATE_NORMAL. */
+
 /* Get the first bind addr or NULL */
 #define NET_FIRST_BIND_ADDR (server.bindaddr_count ? server.bindaddr[0] : NULL)
 
@@ -613,7 +622,7 @@ typedef struct client {
     sds peerid;             /* Cached peer ID. */
     long long clientId;     /* RIFL client id. */
     long long requestId;    /* RIFL request sequence number of current request.*/
-
+    bool isRecovery;        /* Indicates this connection is for recovery. */
     /* Response buffer */
     int bufpos;
     char buf[PROTO_REPLY_CHUNK_BYTES];
@@ -633,7 +642,7 @@ struct sharedObjectsStruct {
     *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
     *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *rpop, *lpop,
     *lpush, *emptyscan, *minstring, *maxstring,
-    *riflDuplicate, *riflClientIdCollision,
+    *riflDuplicate, *riflClientIdCollision, *unsyncedOk,
     *select[PROTO_SHARED_SELECT_CMDS],
     *integers[OBJ_SHARED_INTEGERS],
     *mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */
@@ -727,15 +736,20 @@ struct redisServer {
     char runid[CONFIG_RUN_ID_SIZE+1];  /* ID always different at every exec. */
     int sentinel_mode;          /* True if this instance is a Sentinel. */
     long long currentOpNum;     /* Operation number that we are working on. */
+    int serverState;            /* State of server (accepting recovery, normal).*/
+    time_t last_client_replay;  /* Unix time at which clients stopped sending replay */
     /* Networking */
     int port;                   /* TCP listening port */
+    int portForRecovery;        /* TCP listening port for client replay */
     int tcp_backlog;            /* TCP listen() backlog */
     char *bindaddr[CONFIG_BINDADDR_MAX]; /* Addresses we should bind to */
     int bindaddr_count;         /* Number of addresses in server.bindaddr[] */
     char *unixsocket;           /* UNIX socket path */
     mode_t unixsocketperm;      /* UNIX socket permission */
     int ipfd[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
     int ipfd_count;             /* Used slots in ipfd[] */
+    int ipfd4replay[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
+    int ipfd4replay_count;             /* Used slots in ipfd[] */
     int sofd;                   /* Unix socket file descriptor */
     int cfd[CONFIG_BINDADDR_MAX];/* Cluster bus listening socket */
     int cfd_count;              /* Used slots in cfd[] */
@@ -1107,6 +1121,7 @@ void setDeferredMultiBulkLength(client *c, void *node, long length);
 void processInputBuffer(client *c);
 void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
+void acceptTcpHandler4replay(aeEventLoop *el, int fd, void *privdata, int mask);
 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
 void addReplyBulk(client *c, robj *obj);

src/t_string.c

@@ -89,7 +89,14 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
     notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
     if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
         "expire",key,c->db->id);
-    addReply(c, ok_reply ? ok_reply : shared.ok);
+
+//    addReply(c, ok_reply ? ok_reply : shared.ok);
+    if (ok_reply) {
+        addReply(c, ok_reply);
+    } else {
+        sds s = sdsempty();
+        addReplySds(c, sdscatfmt(s, "%S %U %U\r\n", shared.unsyncedOk->ptr, server.currentOpNum, server.aof_last_fsync_opNum));
+    }
 }
 
 /* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */
@@ -99,8 +106,8 @@ void setCommand(client *c) {
     int unit = UNIT_SECONDS;
     int flags = OBJ_SET_NO_FLAGS;
 
-    // exclude last 3 RIFL arguments from original Redis argument parser.
-    for (j = 3; j < c->argc - 3; j++) {
+    // exclude last 2 RIFL arguments from original Redis argument parser.
+    for (j = 3; j < c->argc - 2; j++) {
         char *a = c->argv[j]->ptr;
         robj *next = (j == c->argc-1 - 3) ? NULL : c->argv[j+1];