Simplified RIFL is implemented with AOF recovery support.

Author: seojinpark

redis.conf

@@ -620,8 +620,8 @@ appendfilename "appendonly.aof"
 #
 # If unsure, use "everysec".
 
- appendfsync always
-# appendfsync everysec
+# appendfsync always
+ appendfsync everysec
 # appendfsync no
 
 # When the AOF fsync policy is set to always or everysec, and a background

src/Makefile

@@ -127,7 +127,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server
 REDIS_SENTINEL_NAME=redis-sentinel
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o geo.o
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rifl.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o geo.o
 REDIS_GEOHASH_OBJ=../deps/geohash-int/geohash.o ../deps/geohash-int/geohash_helper.o
 REDIS_CLI_NAME=redis-cli
 REDIS_CLI_OBJ=anet.o adlist.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o

src/Makefile.dep

@@ -132,7 +132,7 @@ sentinel.o: sentinel.c server.h fmacros.h config.h solarisfixes.h \
 server.o: server.c server.h fmacros.h config.h solarisfixes.h \
  ../deps/lua/src/lua.h ../deps/lua/src/luaconf.h ae.h sds.h dict.h \
  adlist.h zmalloc.h anet.h ziplist.h intset.h version.h util.h latency.h \
- sparkline.h quicklist.h zipmap.h sha1.h endianconv.h crc64.h rdb.h rio.h \
+ sparkline.h quicklist.h zipmap.h sha1.h endianconv.h crc64.h rdb.h rifl.h rio.h \
  cluster.h slowlog.h bio.h asciilogo.h
 setproctitle.o: setproctitle.c
 sha1.o: sha1.c solarisfixes.h sha1.h config.h

src/aof.c

@@ -29,6 +29,7 @@
 
 #include "server.h"
 #include "bio.h"
+#include "rifl.h"
 #include "rio.h"
 
 #include <signal.h>
@@ -695,6 +696,14 @@ int loadAppendOnlyFile(char *filename) {
             exit(1);
         }
 
+        // TODO(seojin): RIFL here?
+        if (cmd->flags & CMD_AT_MOST_ONCE) {
+            long long clientId, requestId;
+            getLongLongFromObject(argv[argc-2], &clientId);
+            getLongLongFromObject(argv[argc-1], &requestId);
+            riflCheckDuplicate(clientId, requestId);
+        }
+
         /* Run the command in the context of a fake client */
         cmd->proc(fakeClient);
 
@@ -1069,11 +1078,13 @@ int rewriteAppendOnlyFile(char *filename) {
             /* Save the key and associated value */
             if (o->type == OBJ_STRING) {
                 /* Emit a SET command */
-                char cmd[]="*3\r\n$3\r\nSET\r\n";
+                char cmd[]="*5\r\n$3\r\nSET\r\n";
                 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                 /* Key and value */
                 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
                 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
+                char rifl[]="$1\r\n0\r\n$1\r\n0\r\n"; // fill dummy now.
+                if (rioWrite(&aof,rifl,sizeof(rifl)-1) == 0) goto werr;
             } else if (o->type == OBJ_LIST) {
                 if (rewriteListObject(&aof,&key,o) == 0) goto werr;
             } else if (o->type == OBJ_SET) {
@@ -1102,6 +1113,20 @@ int rewriteAppendOnlyFile(char *filename) {
         di = NULL;
     }
 
+    /* Now, sweep RIFL table and write SET log accordingly. */
+    // TODO(seojin): Make new command. RIFLCR clientId rpcId
+    long long clientId;
+    for (clientId = riflGetNext(-1); clientId != -1; clientId = riflGetNext(clientId)) {
+        long long requestId = riflGetprocessedRpcId(clientId);
+        /* Emit a SET command */
+        char cmd[]="*5\r\n$3\r\nSET\r\n$5\r\n_rifl\r\n$1\r\n0\r\n";
+        if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
+        /* Key and value */
+        if (rioWriteBulkLongLong(&aof, clientId) == 0) goto werr;
+        if (rioWriteBulkLongLong(&aof, requestId) == 0) goto werr;
+    }
+
+
     /* Do an initial slow fsync here while the parent is still sending
      * data, in order to make the next final fsync faster. */
     if (fflush(fp) == EOF) goto werr;

src/rifl.c

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2017 Stanford University.
+ * All rights reserved.
+ */
+
+#include "server.h"
+#include "redisassert.h"
+
+/*================================= Globals ================================= */
+
+/* Global vars */
+#define RIFL_TABLE_SIZE 1048576 /* 1024*1024. This must be power of two. */
+
+/* 0th index is not used. */
+long long processedRpcIds[RIFL_TABLE_SIZE] = {0, };
+long long clientIds[RIFL_TABLE_SIZE] = {0, };
+int bitmask = RIFL_TABLE_SIZE - 1;
+bool witnessRecoveryMode = false; /* Don't bump processedRpcId while recovery */
+
+/*================================= Functions =============================== */
+bool riflCheckClientIdOk(client *c) {
+    if (clientIds[c->clientId & bitmask]) {
+        return clientIds[c->clientId & bitmask] == c->clientId;
+    }
+    clientIds[c->clientId & bitmask] = c->clientId;
+    return true;
+}
+
+bool riflCheckDuplicate(long long clientId, long long requestId) {
+    if (clientId == 0) {
+        assert(requestId == 0);
+        return false;
+    }
+
+    int index = clientId & bitmask;
+    if (processedRpcIds[index] >= requestId) {
+        return true;
+    }
+    if (!witnessRecoveryMode) {
+        processedRpcIds[index] = requestId;
+    }
+    return false;
+}
+
+void riflStartRecoveryByWitness() {
+    witnessRecoveryMode = true;
+}
+
+void riflEndRecoveryByWitness() {
+    witnessRecoveryMode = false;
+}
+
+
+long long riflGetNext(long long clientId) {
+    int index = (clientId + 1) & bitmask;
+    while (index < RIFL_TABLE_SIZE) {
+        if (clientIds[index]) {
+            return clientIds[index];
+        }
+        ++index;
+    }
+    return -1; // End of table.
+}
+
+long long riflGetprocessedRpcId(long long clientId) {
+    int index = clientId & bitmask;
+    assert(clientIds[index] == clientId);
+    return processedRpcIds[index];
+}

src/rifl.h

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2017 Stanford University.
+ * All rights reserved.
+ */
+
+#ifndef __RIFL_H
+#define __RIFL_H
+
+#include <stdio.h>
+
+/* TBD: include only necessary headers. */
+#include "server.h"
+
+/*
+ * Assumption: client only sends RPCs in order.
+ * We can assume this since Redis uses TCP socket.
+ */
+bool riflCheckClientIdOk(client *c);
+bool riflCheckDuplicate(long long clientId, long long requestId);
+void riflStartRecoveryByWitness();
+void riflEndRecoveryByWitness();
+
+/* Used for AOF rewrite. */
+long long riflGetNext(long long clientId);
+long long riflGetprocessedRpcId(long long clientId);
+
+#endif

src/server.c

@@ -32,6 +32,7 @@
 #include "slowlog.h"
 #include "bio.h"
 #include "latency.h"
+#include "rifl.h"
 
 #include <time.h>
 #include <signal.h>
@@ -124,7 +125,7 @@ struct redisServer server; /* server global state */
  */
 struct redisCommand redisCommandTable[] = {
     {"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},
-    {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},
+    {"set",setCommand,-5,"wmO",0,NULL,1,1,1,0,0},
     {"setnx",setnxCommand,3,"wmF",0,NULL,1,1,1,0,0},
     {"setex",setexCommand,4,"wm",0,NULL,1,1,1,0,0},
     {"psetex",psetexCommand,4,"wm",0,NULL,1,1,1,0,0},
@@ -1446,6 +1447,8 @@ void createSharedObjects(void) {
      * string in string comparisons for the ZRANGEBYLEX command. */
     shared.minstring = createStringObject("minstring",9);
     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"));
 }
 
 void initServerConfig(void) {
@@ -2027,6 +2030,7 @@ void populateCommandTable(void) {
             case 'M': c->flags |= CMD_SKIP_MONITOR; break;
             case 'k': c->flags |= CMD_ASKING; break;
             case 'F': c->flags |= CMD_FAST; break;
+            case 'O': c->flags |= CMD_AT_MOST_ONCE; break;
             default: serverPanic("Unsupported command flag"); break;
             }
             f++;
@@ -2249,6 +2253,19 @@ void call(client *c, int flags) {
     /* Call the command. */
     dirty = server.dirty;
     start = ustime();
+    // TODO(seojin): Add RIFL check.
+    if (c->cmd->flags & CMD_AT_MOST_ONCE) {
+        getLongLongFromObject(c->argv[c->argc-2], &c->clientId);
+        getLongLongFromObject(c->argv[c->argc-1], &c->requestId);
+        if (!riflCheckClientIdOk(c)) {
+            addReply(c, shared.riflClientIdCollision);
+            return;
+        }
+        if (riflCheckDuplicate(c->clientId, c->requestId)){
+            addReply(c, shared.riflDuplicate);
+            return;
+        }
+    }
     c->cmd->proc(c);
     duration = ustime()-start;
     dirty = server.dirty-dirty;
@@ -2752,6 +2769,7 @@ void addReplyCommand(client *c, struct redisCommand *cmd) {
         flagcount += addReplyCommandFlag(c,cmd,CMD_SKIP_MONITOR, "skip_monitor");
         flagcount += addReplyCommandFlag(c,cmd,CMD_ASKING, "asking");
         flagcount += addReplyCommandFlag(c,cmd,CMD_FAST, "fast");
+        flagcount += addReplyCommandFlag(c,cmd,CMD_AT_MOST_ONCE, "execute_at_most_once");
         if (cmd->getkeys_proc) {
             addReplyStatus(c, "movablekeys");
             flagcount += 1;

src/server.h

@@ -187,6 +187,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CMD_SKIP_MONITOR 2048         /* "M" flag */
 #define CMD_ASKING 4096               /* "k" flag */
 #define CMD_FAST 8192                 /* "F" flag */
+#define CMD_AT_MOST_ONCE 16384        /* "O" flag (safe to retry) */
 
 /* Object types */
 #define OBJ_STRING 0
@@ -609,6 +610,8 @@ typedef struct client {
     dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
     list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
     sds peerid;             /* Cached peer ID. */
+    long long clientId;     /* RIFL client id. */
+    long long requestId;    /* RIFL request sequence number of current request.*/
 
     /* Response buffer */
     int bufpos;
@@ -629,6 +632,7 @@ struct sharedObjectsStruct {
     *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
     *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *rpop, *lpop,
     *lpush, *emptyscan, *minstring, *maxstring,
+    *riflDuplicate, *riflClientIdCollision,
     *select[PROTO_SHARED_SELECT_CMDS],
     *integers[OBJ_SHARED_INTEGERS],
     *mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */

src/t_string.c

@@ -99,9 +99,10 @@ void setCommand(client *c) {
     int unit = UNIT_SECONDS;
     int flags = OBJ_SET_NO_FLAGS;
 
-    for (j = 3; j < c->argc; j++) {
+    // exclude last 3 RIFL arguments from original Redis argument parser.
+    for (j = 3; j < c->argc - 3; j++) {
         char *a = c->argv[j]->ptr;
-        robj *next = (j == c->argc-1) ? NULL : c->argv[j+1];
+        robj *next = (j == c->argc-1 - 3) ? NULL : c->argv[j+1];
 
         if ((a[0] == 'n' || a[0] == 'N') &&
             (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&