Merge branch 'lists' of git://github.com/pietern/redis

adlist.c

@@ -123,6 +123,35 @@ list *listAddNodeTail(list *list, void *value)
     return list;
 }
 
+list *listInsertNode(list *list, listNode *old_node, void *value, int after) {
+    listNode *node;
+
+    if ((node = zmalloc(sizeof(*node))) == NULL)
+        return NULL;
+    node->value = value;
+    if (after) {
+        node->prev = old_node;
+        node->next = old_node->next;
+        if (list->tail == old_node) {
+            list->tail = node;
+        }
+    } else {
+        node->next = old_node;
+        node->prev = old_node->prev;
+        if (list->head == old_node) {
+            list->head = node;
+        }
+    }
+    if (node->prev != NULL) {
+        node->prev->next = node;
+    }
+    if (node->next != NULL) {
+        node->next->prev = node;
+    }
+    list->len++;
+    return list;
+}
+
 /* Remove the specified node from the specified list.
  * It's up to the caller to free the private value of the node.
  *
@@ -183,9 +212,9 @@ void listRewindTail(list *list, listIter *li) {
  * or NULL if there are no more elements, so the classical usage patter
  * is:
  *
- * iter = listGetItarotr(list,<direction>);
+ * iter = listGetIterator(list,<direction>);
- * while ((node = listNextIterator(iter)) != NULL) {
+ * while ((node = listNext(iter)) != NULL) {
- *     DoSomethingWith(listNodeValue(node));
+ *     doSomethingWith(listNodeValue(node));
  * }
  *
  * */

adlist.h

@@ -74,6 +74,7 @@ list *listCreate(void);
 void listRelease(list *list);
 list *listAddNodeHead(list *list, void *value);
 list *listAddNodeTail(list *list, void *value);
+list *listInsertNode(list *list, listNode *old_node, void *value, int after);
 void listDelNode(list *list, listNode *node);
 listIter *listGetIterator(list *list, int direction);
 listNode *listNext(listIter *iter);

redis.c

@@ -261,7 +261,7 @@ typedef struct redisObject {
     unsigned lru:22;        /* lru time (relative to server.lruclock) */
     int refcount;
     void *ptr;
-    /* VM fields, this are only allocated if VM is active, otherwise the
+    /* VM fields are only allocated if VM is active, otherwise the
      * object allocation function will just allocate
      * sizeof(redisObjct) minus sizeof(redisObjectVM), so using
      * Redis without VM active will not have any overhead. */
@@ -538,7 +538,7 @@ typedef struct zset {
 
 #define REDIS_SHARED_INTEGERS 10000
 struct sharedObjectsStruct {
-    robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
+    robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space,
     *colon, *nullbulk, *nullmultibulk, *queued,
     *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
     *outofrangeerr, *plus,
@@ -692,6 +692,9 @@ static void renameCommand(redisClient *c);
 static void renamenxCommand(redisClient *c);
 static void lpushCommand(redisClient *c);
 static void rpushCommand(redisClient *c);
+static void lpushxCommand(redisClient *c);
+static void rpushxCommand(redisClient *c);
+static void linsertCommand(redisClient *c);
 static void lpopCommand(redisClient *c);
 static void rpopCommand(redisClient *c);
 static void llenCommand(redisClient *c);
@@ -792,6 +795,9 @@ static struct redisCommand readonlyCommandTable[] = {
     {"mget",mgetCommand,-2,REDIS_CMD_INLINE,NULL,1,-1,1},
     {"rpush",rpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
+    {"rpushx",rpushxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
+    {"lpushx",lpushxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
+    {"linsert",linsertCommand,5,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
     {"rpop",rpopCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
     {"lpop",lpopCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
     {"brpop",brpopCommand,-3,REDIS_CMD_INLINE,NULL,1,1,1},
@@ -1671,6 +1677,7 @@ static void createSharedObjects(void) {
     shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
     shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
     shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
+    shared.cnegone = createObject(REDIS_STRING,sdsnew(":-1\r\n"));
     shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
     shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
     shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
@@ -4920,7 +4927,7 @@ static void listTypePush(robj *subject, robj *value, int where) {
     /* Check if we need to convert the ziplist */
     listTypeTryConversion(subject,value);
     if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
-        ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
+        ziplistLen(subject->ptr) >= server.list_max_ziplist_entries)
             listTypeConvert(subject,REDIS_ENCODING_LIST);
 
     if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
@@ -5082,6 +5089,36 @@ static robj *listTypeGet(listTypeEntry *entry) {
     return value;
 }
 
+static void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
+    robj *subject = entry->li->subject;
+    if (entry->li->encoding == REDIS_ENCODING_ZIPLIST) {
+        value = getDecodedObject(value);
+        if (where == REDIS_TAIL) {
+            unsigned char *next = ziplistNext(subject->ptr,entry->zi);
+
+            /* When we insert after the current element, but the current element
+             * is the tail of the list, we need to do a push. */
+            if (next == NULL) {
+                subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),REDIS_TAIL);
+            } else {
+                subject->ptr = ziplistInsert(subject->ptr,next,value->ptr,sdslen(value->ptr));
+            }
+        } else {
+            subject->ptr = ziplistInsert(subject->ptr,entry->zi,value->ptr,sdslen(value->ptr));
+        }
+        decrRefCount(value);
+    } else if (entry->li->encoding == REDIS_ENCODING_LIST) {
+        if (where == REDIS_TAIL) {
+            listInsertNode(subject->ptr,entry->ln,value,AL_START_TAIL);
+        } else {
+            listInsertNode(subject->ptr,entry->ln,value,AL_START_HEAD);
+        }
+        incrRefCount(value);
+    } else {
+        redisPanic("Unknown list encoding");
+    }
+}
+
 /* Compare the given object with the entry at the current position. */
 static int listTypeEqual(listTypeEntry *entry, robj *o) {
     listTypeIterator *li = entry->li;
@@ -5174,6 +5211,75 @@ static void rpushCommand(redisClient *c) {
     pushGenericCommand(c,REDIS_TAIL);
 }
 
+static void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
+    robj *subject;
+    listTypeIterator *iter;
+    listTypeEntry entry;
+    int inserted = 0;
+
+    if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
+        checkType(c,subject,REDIS_LIST)) return;
+
+    if (refval != NULL) {
+        /* Note: we expect refval to be string-encoded because it is *not* the
+         * last argument of the multi-bulk LINSERT. */
+        redisAssert(refval->encoding == REDIS_ENCODING_RAW);
+
+        /* We're not sure if this value can be inserted yet, but we cannot
+         * convert the list inside the iterator. We don't want to loop over
+         * the list twice (once to see if the value can be inserted and once
+         * to do the actual insert), so we assume this value can be inserted
+         * and convert the ziplist to a regular list if necessary. */
+        listTypeTryConversion(subject,val);
+
+        /* Seek refval from head to tail */
+        iter = listTypeInitIterator(subject,0,REDIS_TAIL);
+        while (listTypeNext(iter,&entry)) {
+            if (listTypeEqual(&entry,refval)) {
+                listTypeInsert(&entry,val,where);
+                inserted = 1;
+                break;
+            }
+        }
+        listTypeReleaseIterator(iter);
+
+        if (inserted) {
+            /* Check if the length exceeds the ziplist length threshold. */
+            if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
+                ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
+                    listTypeConvert(subject,REDIS_ENCODING_LIST);
+            server.dirty++;
+        } else {
+            /* Notify client of a failed insert */
+            addReply(c,shared.cnegone);
+            return;
+        }
+    } else {
+        listTypePush(subject,val,where);
+        server.dirty++;
+    }
+
+    addReplyUlong(c,listTypeLength(subject));
+}
+
+static void lpushxCommand(redisClient *c) {
+    pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD);
+}
+
+static void rpushxCommand(redisClient *c) {
+    pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL);
+}
+
+static void linsertCommand(redisClient *c) {
+    if (strcasecmp(c->argv[2]->ptr,"after") == 0) {
+        pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL);
+    } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) {
+        pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD);
+    } else {
+        addReply(c,shared.syntaxerr);
+    }
+}
+
 static void llenCommand(redisClient *c) {
     robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
     if (o == NULL || checkType(c,o,REDIS_LIST)) return;
@@ -9585,8 +9691,10 @@ static double computeObjectSwappability(robj *o) {
     /* actual age can be >= minage, but not < minage. As we use wrapping
      * 21 bit clocks with minutes resolution for the LRU. */
     time_t minage = abs(server.lruclock - o->lru);
-    long asize = 0;
+    long asize = 0, elesize;
+    robj *ele;
     list *l;
+    listNode *ln;
     dict *d;
     struct dictEntry *de;
     int z;
@@ -9601,18 +9709,19 @@ static double computeObjectSwappability(robj *o) {
         }
         break;
     case REDIS_LIST:
+        if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+            asize = sizeof(*o)+ziplistSize(o->ptr);
+        } else {
             l = o->ptr;
-        listNode *ln = listFirst(l);
+            ln = listFirst(l);
-
             asize = sizeof(list);
             if (ln) {
-            robj *ele = ln->value;
+                ele = ln->value;
-            long elesize;
-
                 elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
                                 (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o);
                 asize += (sizeof(listNode)+elesize)*listLength(l);
             }
+        }
         break;
     case REDIS_SET:
     case REDIS_ZSET:
@@ -9622,9 +9731,6 @@ static double computeObjectSwappability(robj *o) {
         asize = sizeof(dict)+(sizeof(struct dictEntry*)*dictSlots(d));
         if (z) asize += sizeof(zset)-sizeof(dict);
         if (dictSize(d)) {
-            long elesize;
-            robj *ele;
-
             de = dictGetRandomKey(d);
             ele = dictGetEntryKey(de);
             elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
@@ -9649,9 +9755,6 @@ static double computeObjectSwappability(robj *o) {
             d = o->ptr;
             asize = sizeof(dict)+(sizeof(struct dictEntry*)*dictSlots(d));
             if (dictSize(d)) {
-                long elesize;
-                robj *ele;
-
                 de = dictGetRandomKey(d);
                 ele = dictGetEntryKey(de);
                 elesize = (ele->encoding == REDIS_ENCODING_RAW) ?

tests/support/redis.tcl

@@ -40,7 +40,7 @@ array set ::redis::multibulkarg {}
 
 # Flag commands requiring last argument as a bulk write operation
 foreach redis_bulk_cmd {
-    set setnx rpush lpush lset lrem sadd srem sismember echo getset smove zadd zrem zscore zincrby append zrank zrevrank hget hdel hexists setex
+    set setnx rpush lpush rpushx lpushx linsert lset lrem sadd srem sismember echo getset smove zadd zrem zscore zincrby append zrank zrevrank hget hdel hexists setex
 } {
     set ::redis::bulkarg($redis_bulk_cmd) {}
 }

tests/unit/type/list.tcl

@@ -78,6 +78,89 @@ start_server {
         assert_encoding list $key
     }
 
+    test {LPUSHX, RPUSHX - generic} {
+        r del xlist
+        assert_equal 0 [r lpushx xlist a]
+        assert_equal 0 [r llen xlist]
+        assert_equal 0 [r rpushx xlist a]
+        assert_equal 0 [r llen xlist]
+    }
+
+    foreach type {ziplist list} {
+        test "LPUSHX, RPUSHX - $type" {
+            create_$type xlist {b c}
+            assert_equal 3 [r rpushx xlist d]
+            assert_equal 4 [r lpushx xlist a]
+            assert_equal {a b c d} [r lrange xlist 0 -1]
+        }
+
+        test "LINSERT - $type" {
+            create_$type xlist {a b c d}
+            assert_equal 5 [r linsert xlist before c zz]
+            assert_equal {a b zz c d} [r lrange xlist 0 10]
+            assert_equal 6 [r linsert xlist after c yy]
+            assert_equal {a b zz c yy d} [r lrange xlist 0 10]
+            assert_equal 7 [r linsert xlist after d dd]
+            assert_equal -1 [r linsert xlist after bad ddd]
+            assert_equal {a b zz c yy d dd} [r lrange xlist 0 10]
+            assert_equal 8 [r linsert xlist before a aa]
+            assert_equal -1 [r linsert xlist before bad aaa]
+            assert_equal {aa a b zz c yy d dd} [r lrange xlist 0 10]
+
+            # check inserting integer encoded value
+            assert_equal 9 [r linsert xlist before aa 42]
+            assert_equal 42 [r lrange xlist 0 0]
+        }
+    }
+
+    test {LPUSHX, RPUSHX convert from ziplist to list} {
+        set large_value "aaaaaaaaaaaaaaaaa"
+
+        # convert when a large value is pushed
+        create_ziplist xlist a
+        assert_equal 2 [r rpushx xlist $large_value]
+        assert_encoding list xlist
+        create_ziplist xlist a
+        assert_equal 2 [r lpushx xlist $large_value]
+        assert_encoding list xlist
+
+        # convert when the length threshold is exceeded
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal 257 [r rpushx xlist b]
+        assert_encoding list xlist
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal 257 [r lpushx xlist b]
+        assert_encoding list xlist
+    }
+
+    test {LINSERT convert from ziplist to list} {
+        set large_value "aaaaaaaaaaaaaaaaa"
+
+        # convert when a large value is inserted
+        create_ziplist xlist a
+        assert_equal 2 [r linsert xlist before a $large_value]
+        assert_encoding list xlist
+        create_ziplist xlist a
+        assert_equal 2 [r linsert xlist after a $large_value]
+        assert_encoding list xlist
+
+        # convert when the length threshold is exceeded
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal 257 [r linsert xlist before a a]
+        assert_encoding list xlist
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal 257 [r linsert xlist after a a]
+        assert_encoding list xlist
+
+        # don't convert when the value could not be inserted
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal -1 [r linsert xlist before foo a]
+        assert_encoding ziplist xlist
+        create_ziplist xlist [lrepeat 256 a]
+        assert_equal -1 [r linsert xlist after foo a]
+        assert_encoding ziplist xlist
+    }
+
     foreach {type num} {ziplist 250 list 500} {
         proc check_numbered_list_consistency {key} {
             set len [r llen $key]

ziplist.c

@@ -21,7 +21,6 @@
 #include <assert.h>
 #include <limits.h>
 #include "zmalloc.h"
-#include "sds.h"
 #include "ziplist.h"
 
 /* Important note: the ZIP_END value is used to depict the end of the
@@ -382,30 +381,6 @@ unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int sle
     return __ziplistInsert(zl,p,s,slen);
 }
 
-unsigned char *ziplistPop(unsigned char *zl, sds *target, int where) {
-    zlentry entry;
-    unsigned char *p;
-    long long value;
-    if (target) *target = NULL;
-
-    /* Get pointer to element to remove */
-    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_TAIL(zl);
-    if (*p == ZIP_END) return zl;
-
-    entry = zipEntry(p);
-    if (target) {
-        if (entry.encoding == ZIP_ENC_RAW) {
-            *target = sdsnewlen(p+entry.headersize,entry.len);
-        } else {
-            value = zipLoadInteger(p+entry.headersize,entry.encoding);
-            *target = sdscatprintf(sdsempty(), "%lld", value);
-        }
-    }
-
-    zl = __ziplistDelete(zl,p,1);
-    return zl;
-}
-
 /* Returns an offset to use for iterating with ziplistNext. When the given
  * index is negative, the list is traversed back to front. When the list
  * doesn't contain an element at the provided index, NULL is returned. */
@@ -644,12 +619,36 @@ void stress(int pos, int num, int maxsize, int dnum) {
     }
 }
 
+void pop(unsigned char *zl, int where) {
+    unsigned char *p, *vstr;
+    unsigned int vlen;
+    long long vlong;
+
+    p = ziplistIndex(zl,where == ZIPLIST_HEAD ? 0 : -1);
+    if (ziplistGet(p,&vstr,&vlen,&vlong)) {
+        if (where == ZIPLIST_HEAD)
+            printf("Pop head: ");
+        else
+            printf("Pop tail: ");
+
+        if (vstr)
+            fwrite(vstr,vlen,1,stdout);
+        else
+            printf("%lld", vlong);
+
+        printf("\n");
+        ziplistDeleteRange(zl,-1,1);
+    } else {
+        printf("ERROR: Could not pop\n");
+        exit(1);
+    }
+}
+
 int main(int argc, char **argv) {
     unsigned char *zl, *p;
     unsigned char *entry;
     unsigned int elen;
     long long value;
-    sds s;
 
     zl = createIntList();
     ziplistRepr(zl);
@@ -657,20 +656,16 @@ int main(int argc, char **argv) {
     zl = createList();
     ziplistRepr(zl);
 
-    zl = ziplistPop(zl, &s, ZIPLIST_TAIL);
+    pop(zl,ZIPLIST_TAIL);
-    printf("Pop tail: %s (length %ld)\n", s, sdslen(s));
     ziplistRepr(zl);
 
-    zl = ziplistPop(zl, &s, ZIPLIST_HEAD);
+    pop(zl,ZIPLIST_HEAD);
-    printf("Pop head: %s (length %ld)\n", s, sdslen(s));
     ziplistRepr(zl);
 
-    zl = ziplistPop(zl, &s, ZIPLIST_TAIL);
+    pop(zl,ZIPLIST_TAIL);
-    printf("Pop tail: %s (length %ld)\n", s, sdslen(s));
     ziplistRepr(zl);
 
-    zl = ziplistPop(zl, &s, ZIPLIST_TAIL);
+    pop(zl,ZIPLIST_TAIL);
-    printf("Pop tail: %s (length %ld)\n", s, sdslen(s));
     ziplistRepr(zl);
 
     printf("Get element at index 3:\n");

ziplist.h

@@ -3,7 +3,6 @@
 
 unsigned char *ziplistNew(void);
 unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where);
-unsigned char *ziplistPop(unsigned char *zl, sds *target, int where);
 unsigned char *ziplistIndex(unsigned char *zl, int index);
 unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);
 unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);