Fixed throughput, lpush, 64base encode, temp hack for GC stale values.

- Improved throughput by delaying and batching fsync. (Just mark that
        fsync is necessary before responding back to client. Fsync once per
        eventloop iteration.)
- Output throughput info to log.
- Implemented CGAR on lpush. (!!! Not tested yet)
- Switched to 64-base encoding for serialization of
    clientId, rpcId, opNum, and syncNum.
- Temporary hack (timeout based) for cleaning stale witness entries.

Author: seojinpark

src/aof.c

@@ -432,7 +432,7 @@ void flushAppendOnlyFile(int force) {
             return;
 
     /* Perform the fsync if needed. */
-    if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
+    if (server.aof_fsync == AOF_FSYNC_ALWAYS || server.must_aof_fsync) {
         /* aof_fsync is defined as fdatasync() for Linux in order to avoid
          * flushing metadata. */
         latencyStartMonitor(latency);
@@ -446,6 +446,7 @@ void flushAppendOnlyFile(int force) {
         if (!sync_in_progress) aof_background_fsync(server.aof_fd);
         server.aof_last_fsync = server.unixtime;
     }
+    server.must_aof_fsync = false;
 }
 
 sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) {
@@ -700,8 +701,9 @@ int loadAppendOnlyFile(char *filename) {
         // 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);
+            getLongLongFromObjectInBase64(argv[argc-2], &clientId);
+            getLongLongFromObjectInBase64(argv[argc-1], &requestId);
+
             fakeClient->clientId = clientId;
             riflCheckClientIdOk(fakeClient);
             riflCheckDuplicate(clientId, requestId);

src/db.c

@@ -29,6 +29,7 @@
 
 #include "server.h"
 #include "cluster.h"
+#include "witnessTracker.h"
 
 #include <signal.h>
 #include <ctype.h>
@@ -60,12 +61,16 @@ 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) {
+            server.must_aof_fsync = true;
 //            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;
+//            flushAppendOnlyFile(1);
+//            aof_fsync(server.aof_fd);
+//            server.aof_last_fsync_opNum = server.currentOpNum - 1;
+            //scheduleFsyncAndWitnessGc();
+            // TODO(seojin): move these into after handling all clients.
+            // so that we only fsync once per loop.
         }
         return val;
     } else {

src/networking.c

@@ -29,6 +29,7 @@
 
 #include "server.h"
 #include <sys/uio.h>
+#include <sys/time.h>
 #include <math.h>
 
 static void setProtocolError(client *c, int pos);
@@ -382,6 +383,31 @@ void addReplyString(client *c, const char *s, size_t len) {
         _addReplyStringToList(c,s,len);
 }
 
+void addReplyOkCgar(client *c) {//, long long arg1, long long arg2) {
+    char reply[31]; // "@OK " + two 64-bit ints encoded in 64-base + "\r\n\0".
+    bzero(reply, 31);
+    memcpy(reply, shared.unsyncedOk->ptr, 4);
+    int offset = 4;
+    int appended = ulltoa64(reply + offset, 28 - offset, server.currentOpNum);
+    if (appended == 0) {
+        serverLog(LL_WARNING,"Error encoding currentOpNum (%lld) to ASCII."
+                " Exiting.",server.currentOpNum);
+            exit(1);
+    }
+    offset += appended;
+    reply[offset++] = ' ';
+    appended = ulltoa64(reply + offset, 28 - offset, server.aof_last_fsync_opNum);
+    if (appended == 0) {
+        serverLog(LL_WARNING,"Error encoding currentOpNum (%lld) to ASCII."
+                " Exiting.",server.currentOpNum);
+            exit(1);
+    }
+    offset += appended;
+    memcpy(reply + offset, "\r\n", 2);
+//    serverLog(LL_NOTICE,"Reply is constructed: %s (length: %d)", reply, offset);
+    addReplyString(c, reply, offset + 2);
+}
+
 void addReplyErrorLength(client *c, const char *s, size_t len) {
     addReplyString(c,"-ERR ",5);
     addReplyString(c,s,len);
@@ -705,6 +731,19 @@ static void acceptCommonHandler(int fd, int flags, char *ip, bool isRecovery) {
 
     server.stat_numconnections++;
     c->flags |= flags;
+
+    // For throughput benchmark, print the throughput from last new connection.
+    struct timeval tv;
+    gettimeofday(&tv,NULL);
+    unsigned long long currentTime = 1000000 * tv.tv_sec + tv.tv_usec;
+    serverLog(LL_NOTICE, "PerfStat (clientNum, throughput (kops/sec), timestamp): "
+            "%4lu %7.2f %12llu",
+            listLength(server.clients) - 1,
+            (double)(server.currentOpNum - server.last_client_connected_opNum)
+                * 1e3 / (currentTime - server.last_client_connected_usec),
+            currentTime);
+    server.last_client_connected_usec = currentTime;
+    server.last_client_connected_opNum = server.currentOpNum;
 }
 
 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {

src/object.c

@@ -631,6 +631,25 @@ int getLongLongFromObject(robj *o, long long *target) {
     return C_OK;
 }
 
+int getLongLongFromObjectInBase64(robj *o, long long *target) {
+    long long value;
+
+    if (o == NULL) {
+        value = 0;
+    } else {
+        serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
+        if (sdsEncodedObject(o)) {
+            if (base64int2ll(o->ptr,sdslen(o->ptr),&value) == 0) return C_ERR;
+        } else if (o->encoding == OBJ_ENCODING_INT) {
+            value = (long)o->ptr;
+        } else {
+            serverPanic("Unknown string encoding");
+        }
+    }
+    if (target) *target = value;
+    return C_OK;
+}
+
 int getLongLongFromObjectOrReply(client *c, robj *o, long long *target, const char *msg) {
     long long value;
     if (getLongLongFromObject(o, &value) != C_OK) {

src/server.c

@@ -144,7 +144,7 @@ struct redisCommand redisCommandTable[] = {
     {"decr",decrCommand,2,"wmF",0,NULL,1,1,1,0,0},
     {"mget",mgetCommand,-2,"r",0,NULL,1,-1,1,0,0},
     {"rpush",rpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
-    {"lpush",lpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
+    {"lpush",lpushCommand,-5,"wmFO",0,NULL,1,1,1,0,0},
     {"rpushx",rpushxCommand,3,"wmF",0,NULL,1,1,1,0,0},
     {"lpushx",lpushxCommand,3,"wmF",0,NULL,1,1,1,0,0},
     {"linsert",linsertCommand,5,"wm",0,NULL,1,1,1,0,0},
@@ -1296,6 +1296,8 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     run_with_period(1000) {
         /* If server haven't received client's replay request for a while,
          * assume completion of recover and start normal processing. */
+        // TODO(seojin): check recovered by witness. And don't wait if
+        // recovered by witness.
         if (server.serverState == SERVER_STATE_ACCEPTING_REPLAY &&
                 server.unixtime - server.last_client_replay > SECONDS_WAITING_REPLAY) {
             server.serverState = SERVER_STATE_NORMAL;
@@ -1482,6 +1484,8 @@ void initServerConfig(void) {
     server.currentOpNum = 0;
     server.port = CONFIG_DEFAULT_SERVER_PORT;
     server.portForRecovery = CONFIG_DEFAULT_RECOVERY_PORT;
+    server.last_client_connected_usec = 0;
+    server.last_client_connected_opNum = 0;
     server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
     server.bindaddr_count = 0;
     server.unixsocket = NULL;
@@ -1506,6 +1510,7 @@ void initServerConfig(void) {
     server.supervised_mode = SUPERVISED_NONE;
     server.aof_state = AOF_OFF;
     server.aof_fsync = CONFIG_DEFAULT_AOF_FSYNC;
+    server.must_aof_fsync = false;
     server.aof_no_fsync_on_rewrite = CONFIG_DEFAULT_AOF_NO_FSYNC_ON_REWRITE;
     server.aof_rewrite_perc = AOF_REWRITE_PERC;
     server.aof_rewrite_min_size = AOF_REWRITE_MIN_SIZE;
@@ -2296,8 +2301,13 @@ void call(client *c, int flags) {
 
     // 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);
+//        getLongLongFromObject(c->argv[c->argc-2], &c->clientId);
+//        getLongLongFromObject(c->argv[c->argc-1], &c->requestId);
+        getLongLongFromObjectInBase64(c->argv[c->argc-2], &c->clientId);
+        getLongLongFromObjectInBase64(c->argv[c->argc-1], &c->requestId);
+
+//        serverLog(LL_NOTICE,"clientId: %lld, requestId: %lld", c->clientId, c->requestId);
+
         if (!riflCheckClientIdOk(c)) {
             addReply(c, shared.riflClientIdCollision);
             return;
@@ -2307,13 +2317,13 @@ void call(client *c, int flags) {
             return;
         }
     }
-    if (c->cmd->proc != selectCommand) {
+    if (c->cmd->proc != selectCommand && c->cmd->flags & CMD_WRITE) {
         ++server.currentOpNum;
     }
     c->cmd->proc(c);
 
     // Track unsynced change.
-    if (c->cmd->flags & CMD_AT_MOST_ONCE) {
+    if (c->cmd->flags & CMD_AT_MOST_ONCE && server.numWitness > 0) {
         trackUnsyncedRpc(c);
     }
 

src/server.h

@@ -446,7 +446,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #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
+#define SECONDS_WAITING_REPLAY 2       /* server waits extra seconds after last
                                            client replay before swithing from
                         SERVER_STATE_ACCEPTING_REPLAY to SERVER_STATE_NORMAL. */
 
@@ -770,6 +770,9 @@ struct redisServer {
     char *addrToWitness[CONFIG_WITNESS_MAX];
     int fdToWitness[CONFIG_WITNESS_MAX];
     int numWitness;
+    /* For throughput benchmark */
+    unsigned long long last_client_connected_usec;
+    long long last_client_connected_opNum;
     /* RDB / AOF loading information */
     int loading;                /* We are loading data from disk if true */
     off_t loading_total_bytes;
@@ -823,6 +826,7 @@ struct redisServer {
     /* AOF persistence */
     int aof_state;                  /* AOF_(ON|OFF|WAIT_REWRITE) */
     int aof_fsync;                  /* Kind of fsync() policy */
+    bool must_aof_fsync;                  /* indicate this ae cycle need to fsync before responding to client. */
     char *aof_filename;             /* Name of the AOF file */
     int aof_no_fsync_on_rewrite;    /* Don't fsync if a rewrite is in prog. */
     int aof_rewrite_perc;           /* Rewrite AOF if % growth is > M and... */

src/t_hash.c

@@ -530,9 +530,10 @@ void hmsetCommand(client *c) {
         hashTypeSet(o,c->argv[i],c->argv[i+1]);
     }
 //    addReply(c, shared.ok);
-    sds s = sdsempty();
-    addReplySds(c, sdscatfmt(s, "%S %U %U\r\n", shared.unsyncedOk->ptr,
-            server.currentOpNum, server.aof_last_fsync_opNum));
+    addReplyOkCgar(c);
+//    sds s = sdsempty();
+//    addReplySds(c, sdscatfmt(s, "%S %U %U\r\n", shared.unsyncedOk->ptr,
+//            server.currentOpNum, server.aof_last_fsync_opNum));
 
     signalModifiedKey(c->db,c->argv[1]);
     notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id);

src/t_list.c

@@ -203,7 +203,7 @@ void pushGenericCommand(client *c, int where) {
         return;
     }
 
-    for (j = 2; j < c->argc; j++) {
+    for (j = 2; j < c->argc - 2; j++) {
         c->argv[j] = tryObjectEncoding(c->argv[j]);
         if (!lobj) {
             lobj = createQuicklistObject();

src/t_string.c

@@ -94,8 +94,18 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
     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));
+        addReplyOkCgar(c);
+//        char reply[28]; // "@OK " + two 64-bit ints encoded in 64-base.
+//        bzero(reply, 28);
+//        memcpy(reply, shared.unsyncedOk->ptr, 4);
+//        int offset = 4;
+//        offset += ulltoa64(reply + offset, 28 - offset, server.currentOpNum);
+//        reply[offset++] = ' ';
+//        ulltoa64(reply + offset, 28 - offset, server.aof_last_fsync_opNum);
+//        addReplyString(c, reply, offset);
+
+//        sds s = sdsempty();
+//        addReplySds(c, sdscatfmt(s, "%S %U %U\r\n", shared.unsyncedOk->ptr, server.currentOpNum, server.aof_last_fsync_opNum));
     }
 }
 
@@ -382,8 +392,9 @@ void incrDecrCommand(client *c, long long incr) {
     addReply(c,shared.colon);
     addReply(c,new);
 
-    sds s = sdsempty();
-    addReplySds(c, sdscatfmt(s, " %U %U\r\n", server.currentOpNum, server.aof_last_fsync_opNum));
+    addReplyOkCgar(c);
+//    sds s = sdsempty();
+//    addReplySds(c, sdscatfmt(s, " %U %U\r\n", server.currentOpNum, server.aof_last_fsync_opNum));
 //    addReply(c,shared.crlf);
 }
 

src/util.c

@@ -328,6 +328,78 @@ int ll2string(char* dst, size_t dstlen, long long svalue) {
     return length;
 }
 
+int ulltoa64(char* dst, size_t dstlen, long long svalue) {
+    char* ptr = dst, *ptr1 = dst, tmp_char;
+    uint64_t tmp_value;
+    size_t plen = 0;
+
+    do {
+        if (plen >= dstlen) {
+            return 0; // Error. Not enough space.
+        }
+        tmp_value = svalue & 63;
+        svalue >>= 6;
+        *ptr++ = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=" [tmp_value];
+        plen++;
+    } while (svalue);
+
+    *ptr-- = '\0';
+    while (ptr1 < ptr) {
+        tmp_char = *ptr;
+        *ptr-- = *ptr1;
+        *ptr1++ = tmp_char;
+    }
+    return (int)plen;
+}
+
+/* Convert a string of 64-base encoded integer into a long long.
+ * Returns 1 if the string could be parsed into a (non-overflowing) long long,
+ * 0 otherwise. The value will be set to the parsed value when appropriate. */
+int base64int2ll(const char *s, size_t slen, long long *value) {
+    long long v = 0;
+    const char *p = s;
+    uint plen = 0;
+//    int base64inv[128];
+//    const char* code = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=";
+//    size_t codeLen = strlen(code);
+//    for (int i = 0; i < 128; ++i) {
+//        base64inv[i] = -1;
+//    }
+//    for (uint i = 0; i < codeLen; ++i) {
+//        base64inv[(uint)code[i]] = i;
+//    }
+//
+//    printf("{");
+//    for (int i = 0; i < 128; ++i) {
+//        printf("%d, ", base64inv[i]);
+//    }
+//    printf("}");
+
+    // This table is generated by the code above.
+    int base64inv[] = {-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+                       -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+                       -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+                       -1, 62, -1, -1, -1, 63, 52, 53, 54, 55, 56, 57, 58, 59,
+                       60, 61, -1, -1, -1, 64, -1, -1, -1, 0, 1, 2, 3, 4, 5,
+                       6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,
+                       21, 22, 23, 24, 25, -1, -1, -1, -1, -1, -1, 26, 27, 28,
+                       29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42,
+                       43, 44, 45, 46, 47, 48, 49, 50, 51, -1, -1, -1, -1, -1};
+
+    while (plen < slen && p[0] != 0) {
+        int digit = base64inv[(uint)p[0]];
+        if (digit < 0) {
+            return 0; // error!
+        }
+        v <<= 6;
+        v += digit;
+        p++;
+        plen++;
+    }
+    *value = v;
+    return 1;
+}
+
 /* Convert a string into a long long. Returns 1 if the string could be parsed
  * into a (non-overflowing) long long, 0 otherwise. The value will be set to
  * the parsed value when appropriate. */