Fix for skiplists backward link

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index 2e78d9c77e48ae86b5d1b24c8fff4e0dbd4ef732..202b0a6992bdce35a32a27ae5b210f45644376cf 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -311,12 +311,13 @@ typedef struct _redisSortOperation {
 
 typedef struct zskiplistNode {
     struct zskiplistNode **forward;
+    struct zskiplistNode *backward;
     double score;
     robj *obj;
 } zskiplistNode;
 
 typedef struct zskiplist {
-    struct zskiplistNode *header;
+    struct zskiplistNode *header, *tail;
     long length;
     int level;
 } zskiplist;
@@ -435,7 +436,9 @@ static void msetCommand(redisClient *c);
 static void msetnxCommand(redisClient *c);
 static void zaddCommand(redisClient *c);
 static void zrangeCommand(redisClient *c);
+static void zrevrangeCommand(redisClient *c);
 static void zlenCommand(redisClient *c);
+static void zremCommand(redisClient *c);
 
 /*================================= Globals ================================= */
 
@@ -475,7 +478,9 @@ static struct redisCommand cmdTable[] = {
     {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
     {"zadd",zaddCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
+    {"zrem",zremCommand,3,REDIS_CMD_BULK},
     {"zrange",zrangeCommand,4,REDIS_CMD_INLINE},
+    {"zrevrange",zrevrangeCommand,4,REDIS_CMD_INLINE},
     {"zlen",zlenCommand,2,REDIS_CMD_INLINE},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
@@ -2324,7 +2329,7 @@ static int rdbSave(char *filename) {
     /* Use RENAME to make sure the DB file is changed atomically only
      * if the generate DB file is ok. */
     if (rename(tmpfile,filename) == -1) {
-        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
+        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
         unlink(tmpfile);
         return REDIS_ERR;
     }
@@ -3719,6 +3724,8 @@ static zskiplist *zslCreate(void) {
     zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
     for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++)
         zsl->header->forward[j] = NULL;
+    zsl->header->backward = NULL;
+    zsl->tail = NULL;
     return zsl;
 }
 
@@ -3769,6 +3776,11 @@ static void zslInsert(zskiplist *zsl, double score, robj *obj) {
         x->forward[i] = update[i]->forward[i];
         update[i]->forward[i] = x;
     }
+    x->backward = (update[0] == zsl->header) ? NULL : update[0];
+    if (x->forward[0])
+        x->forward[0]->backward = x;
+    else
+        zsl->tail = x;
     zsl->length++;
 }
 
@@ -3791,9 +3803,16 @@ static int zslDelete(zskiplist *zsl, double score, robj *obj) {
                 if (update[i]->forward[i] != x) break;
                 update[i]->forward[i] = x->forward[i];
             }
+            if (x->forward[0]) {
+                x->forward[0]->backward = (x->backward == zsl->header) ?
+                                            NULL : x->backward;
+            } else {
+                zsl->tail = x->backward;
+            }
             zslFreeNode(x);
             while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
                 zsl->level--;
+            zsl->length--;
             return 1;
         } else {
             x = x->forward[0];
@@ -3852,7 +3871,42 @@ static void zaddCommand(redisClient *c) {
     }
 }
 
-static void zrangeCommand(redisClient *c) {
+static void zremCommand(redisClient *c) {
+    robj *zsetobj;
+    zset *zs;
+
+    zsetobj = lookupKeyWrite(c->db,c->argv[1]);
+    if (zsetobj == NULL) {
+        addReply(c,shared.czero);
+    } else {
+        dictEntry *de;
+        double *oldscore;
+        int deleted;
+
+        if (zsetobj->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+        zs = zsetobj->ptr;
+        de = dictFind(zs->dict,c->argv[2]);
+        if (de == NULL) {
+            addReply(c,shared.czero);
+            return;
+        }
+        /* Delete from the skiplist */
+        oldscore = dictGetEntryVal(de);
+        deleted = zslDelete(zs->zsl,*oldscore,c->argv[2]);
+        assert(deleted != 0);
+
+        /* Delete from the hash table */
+        dictDelete(zs->dict,c->argv[2]);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        server.dirty++;
+        addReply(c,shared.cone);
+    }
+}
+
+static void zrangeGenericCommand(redisClient *c, int reverse) {
     robj *o;
     int start = atoi(c->argv[2]->ptr);
     int end = atoi(c->argv[3]->ptr);
@@ -3888,9 +3942,15 @@ static void zrangeCommand(redisClient *c) {
             rangelen = (end-start)+1;
 
             /* Return the result in form of a multi-bulk reply */
-            ln = zsl->header->forward[0];
-            while (start--)
-                ln = ln->forward[0];
+            if (reverse) {
+                ln = zsl->tail;
+                while (start--)
+                    ln = ln->backward;
+            } else {
+                ln = zsl->header->forward[0];
+                while (start--)
+                    ln = ln->forward[0];
+            }
 
             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
             for (j = 0; j < rangelen; j++) {
@@ -3898,12 +3958,20 @@ static void zrangeCommand(redisClient *c) {
                 addReplyBulkLen(c,ele);
                 addReply(c,ele);
                 addReply(c,shared.crlf);
-                ln = ln->forward[0];
+                ln = reverse ? ln->backward : ln->forward[0];
             }
         }
     }
 }
 
+static void zrangeCommand(redisClient *c) {
+    zrangeGenericCommand(c,0);
+}
+
+static void zrevrangeCommand(redisClient *c) {
+    zrangeGenericCommand(c,1);
+}
+
 static void zlenCommand(redisClient *c) {
     robj *o;
     zset *zs;
@@ -4978,7 +5046,10 @@ static struct redisFunctionSym symsTable[] = {
 {"zslDelete",(unsigned long)zslDelete},
 {"createZsetObject",(unsigned long)createZsetObject},
 {"zaddCommand",(unsigned long)zaddCommand},
+{"zrangeGenericCommand",(unsigned long)zrangeGenericCommand},
 {"zrangeCommand",(unsigned long)zrangeCommand},
+{"zrevrangeCommand",(unsigned long)zrevrangeCommand},
+{"zremCommand",(unsigned long)zremCommand},
 {NULL,0}
 };