Merge branch '2.4' into rdio-2.4

src/redis.c

@@ -121,6 +121,8 @@ struct redisCommand readonlyCommandTable[] = {
     {"zadd",zaddCommand,-4,REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"zaddnx",zaddnxCommand,-4,REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"zaddcmp",zaddcmpCommand,-4,REDIS_CMD_DENYOOM,NULL,1,1,1},
+    {"zaddcap",zaddcapCommand,-5,REDIS_CMD_DENYOOM,NULL,1,1,1},
+    {"zaddcaprev",zaddcaprevCommand,-5,REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"zincrby",zincrbyCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"zrem",zremCommand,-3,0,NULL,1,1,1},
     {"zremrangebyscore",zremrangebyscoreCommand,4,0,NULL,1,1,1},

src/redis.h

@@ -1022,6 +1022,8 @@ void msetnxCommand(redisClient *c);
 void zaddCommand(redisClient *c);
 void zaddnxCommand(redisClient *c);
 void zaddcmpCommand(redisClient *c);
+void zaddcapCommand(redisClient *c);
+void zaddcaprevCommand(redisClient *c);
 void zincrbyCommand(redisClient *c);
 void zrangeCommand(redisClient *c);
 void zrangebyscoreCommand(redisClient *c);

src/t_zset.c

@@ -812,15 +812,16 @@ void zsetConvert(robj *zobj, int encoding) {
  *----------------------------------------------------------------------------*/
 
 /* This generic command implements both ZADD and ZINCRBY. */
-void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp) {
+void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp, int cap) {
     static char *nanerr = "resulting score is not a number (NaN)";
     robj *key = c->argv[1];
     robj *ele;
     robj *zobj;
     robj *curobj;
     double score = 0, *scores, curscore = 0.0;
-    int j, elements = (c->argc-2)/2;
+    int j, elements = (c->argc-(cap?3:2))/2;
     int added = 0;
+    long capLen = 0;
 
     if (cmp) {
         if (c->argc % 2) {
@@ -833,17 +834,22 @@ void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp) {
                 return;
             }
         }
-    } else if (c->argc % 2) {
-        addReply(c,shared.syntaxerr);
-        return;
+    } else {
+        if (cap && getLongFromObjectOrReply(c, c->argv[2], &capLen, NULL) != REDIS_OK)
+            return;
+
+        if ((c->argc + (cap?1:0)) % 2 || (cap && capLen < 1)) {
+            addReply(c,shared.syntaxerr);
+            return;
+        }
     }
 
     /* Start parsing all the scores, we need to emit any syntax error
      * before executing additions to the sorted set, as the command should
      * either execute fully or nothing at all. */
     scores = zmalloc(sizeof(double)*elements);
     for (j = 0; j < elements; j++) {
-        if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
+        if (getDoubleFromObjectOrReply(c,c->argv[(cap?3:2)+j*2],&scores[j],NULL)
             != REDIS_OK)
         {
             zfree(scores);
@@ -877,7 +883,7 @@ void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp) {
             unsigned char *eptr;
 
             /* Prefer non-encoded element when dealing with ziplists. */
-            ele = c->argv[3+j*2];
+            ele = c->argv[(cap?4:3)+j*2];
             if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
                 if (nx)
                     continue;
@@ -921,7 +927,7 @@ void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp) {
             zskiplistNode *znode;
             dictEntry *de;
 
-            ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
+            ele = c->argv[(cap?4:3)+j*2] = tryObjectEncoding(c->argv[(cap?4:3)+j*2]);
             de = dictFind(zs->dict,ele);
             if (de != NULL) {
                 curobj = dictGetEntryKey(de);
@@ -970,26 +976,59 @@ void zaddGenericCommand(redisClient *c, int incr, int nx, int cmp) {
         }
     }
     zfree(scores);
+
+    if (cap) {
+        unsigned int llen = zsetLength(zobj);
+        long capExtra = (long)llen-capLen;
+        if (capExtra > 0) {
+            unsigned long deleted=0;
+            /* Correct for 1-based rank. */
+            unsigned int start=(cap==1?1:llen-capExtra+1), end=(cap==1?capExtra:llen);
+            if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
+                zobj->ptr = zzlDeleteRangeByRank(zobj->ptr,start,end,&deleted);
+                if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key);
+            } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
+                zset *zs = zobj->ptr;
+                deleted = zslDeleteRangeByRank(zs->zsl,start,end,zs->dict);
+                if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+                if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
+            } else {
+                redisPanic("Unknown sorted set encoding");
+            }
+
+            if (deleted) signalModifiedKey(c->db,key);
+            server.dirty += deleted;
+        }
+    }
+
     if (incr) /* ZINCRBY */
         addReplyDouble(c,score);
     else /* ZADD */
         addReplyLongLong(c,added);
 }
 
 void zaddCommand(redisClient *c) {
-    zaddGenericCommand(c,0,0,0);
+    zaddGenericCommand(c,0,0,0,0);
 }
 
 void zaddnxCommand(redisClient *c) {
-    zaddGenericCommand(c,0,1,0);
+    zaddGenericCommand(c,0,1,0,0);
 }
 
 void zaddcmpCommand(redisClient *c) {
-    zaddGenericCommand(c,0,0,1);
+    zaddGenericCommand(c,0,0,1,0);
 }
 
 void zincrbyCommand(redisClient *c) {
-    zaddGenericCommand(c,1,0,0);
+    zaddGenericCommand(c,1,0,0,0);
+}
+
+void zaddcapCommand(redisClient *c) {
+    zaddGenericCommand(c,0,0,0,1);
+}
+
+void zaddcaprevCommand(redisClient *c) {
+    zaddGenericCommand(c,0,0,0,2);
 }
 
 void zremCommand(redisClient *c) {

tests/unit/type/zset.tcl

@@ -106,6 +106,68 @@ start_server {tags {"zset"}} {
             assert_equal 0 [r zcard zdoesntexist]
         }
 
+        test {ZSETCAP - Invalid negative cap} {
+            assert_error "*ERR*syntax*" {r zaddcap myzset -1 0 abc}
+        }
+
+        test {ZSETCAP - Invalid zero cap} {
+            assert_error "*ERR*syntax*" {r zaddcap myzset 0 0 abc}
+        }
+
+        test {ZSETCAPREV - Invalid negative cap} {
+            assert_error "*ERR*syntax*" {r zaddcaprev myzset -1 0 abc}
+        }
+
+        test {ZSETCAPREV - Invalid zero cap} {
+            assert_error "*ERR*syntax*" {r zaddcaprev myzset 0 0 abc}
+        }
+
+        test "ZSETCAP - Generic - $encoding" {
+            r del ztmp
+            r zaddcap ztmp 2 10 x
+            r zaddcap ztmp 2 20 y
+            r zaddcap ztmp 2 30 z
+            assert_equal {y z} [r zrange ztmp 0 -1]
+
+            r zaddcap ztmp 2 10 xx
+            assert_equal {y z} [r zrange ztmp 0 -1]
+
+            r zaddcap ztmp 2 15 z
+            assert_equal {z y} [r zrange ztmp 0 -1]
+            
+            r zaddcap ztmp 2 25 yy
+            assert_equal {y yy} [r zrange ztmp 0 -1]
+        }
+
+        test "ZSETCAPREV - Generic - $encoding" {
+            r del ztmp
+            r zaddcaprev ztmp 2 10 x
+            r zaddcaprev ztmp 2 20 y
+            r zaddcaprev ztmp 2 30 z
+            assert_equal {x y} [r zrange ztmp 0 -1]
+
+            r zaddcaprev ztmp 2 5 xx
+            assert_equal {xx x} [r zrange ztmp 0 -1]
+
+            r zaddcaprev ztmp 2 15 xx
+            assert_equal {x xx} [r zrange ztmp 0 -1]
+            
+            r zaddcaprev ztmp 2 5 yy
+            assert_equal {yy x} [r zrange ztmp 0 -1]
+        }
+
+        test "ZSETCAP - Variadic version - $encoding" {
+            r del ztmp
+            r zaddcap ztmp 2 10 x 20 y 30 z
+            assert_equal {y z} [r zrange ztmp 0 -1]
+        }
+
+        test "ZSETCAPREV - Variadic version - $encoding" {
+            r del ztmp
+            r zaddcaprev ztmp 2 30 z 20 y 10 x
+            assert_equal {x y} [r zrange ztmp 0 -1]
+        }
+
         test "ZREM removes key after last element is removed" {
             r del ztmp
             r zadd ztmp 10 x