Client-driven replay support for INCR, HMSET

Author: seojinpark

src/server.c

@@ -139,7 +139,7 @@ struct redisCommand redisCommandTable[] = {
     {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0},
     {"getrange",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
     {"substr",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
-    {"incr",incrCommand,2,"wmF",0,NULL,1,1,1,0,0},
+    {"incr",incrCommand,4,"wmFO",0,NULL,1,1,1,0,0},
     {"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},
@@ -198,7 +198,7 @@ struct redisCommand redisCommandTable[] = {
     {"hset",hsetCommand,4,"wmF",0,NULL,1,1,1,0,0},
     {"hsetnx",hsetnxCommand,4,"wmF",0,NULL,1,1,1,0,0},
     {"hget",hgetCommand,3,"rF",0,NULL,1,1,1,0,0},
-    {"hmset",hmsetCommand,-4,"wm",0,NULL,1,1,1,0,0},
+    {"hmset",hmsetCommand,-6,"wmO",0,NULL,1,1,1,0,0},
     {"hmget",hmgetCommand,-3,"r",0,NULL,1,1,1,0,0},
     {"hincrby",hincrbyCommand,4,"wmF",0,NULL,1,1,1,0,0},
     {"hincrbyfloat",hincrbyfloatCommand,4,"wmF",0,NULL,1,1,1,0,0},

src/t_hash.c

@@ -524,12 +524,16 @@ void hmsetCommand(client *c) {
     }
 
     if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
-    hashTypeTryConversion(o,c->argv,2,c->argc-1);
-    for (i = 2; i < c->argc; i += 2) {
+    hashTypeTryConversion(o,c->argv,2,c->argc-1 - 2 /* exclude RIFL args */);
+    for (i = 2; i < c->argc - 2;/* exclude RIFL args */ i += 2) {
         hashTypeTryObjectEncoding(o,&c->argv[i], &c->argv[i+1]);
         hashTypeSet(o,c->argv[i],c->argv[i+1]);
     }
-    addReply(c, shared.ok);
+//    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));
+
     signalModifiedKey(c->db,c->argv[1]);
     notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id);
     server.dirty++;

src/t_string.c

@@ -381,7 +381,10 @@ void incrDecrCommand(client *c, long long incr) {
     server.dirty++;
     addReply(c,shared.colon);
     addReply(c,new);
-    addReply(c,shared.crlf);
+
+    sds s = sdsempty();
+    addReplySds(c, sdscatfmt(s, " %U %U\r\n", server.currentOpNum, server.aof_last_fsync_opNum));
+//    addReply(c,shared.crlf);
 }
 
 void incrCommand(client *c) {