Add listpack encoding for list (#11303)

Improve memory efficiency of list keys

## Description of the feature
The new listpack encoding uses the old `list-max-listpack-size` config
to perform the conversion, which we can think it of as a node inside a
quicklist, but without 80 bytes overhead (internal fragmentation included)
of quicklist and quicklistNode structs.
For example, a list key with 5 items of 10 chars each, now takes 128 bytes
instead of 208 it used to take.

## Conversion rules
* Convert listpack to quicklist
  When the listpack length or size reaches the `list-max-listpack-size` limit,
  it will be converted to a quicklist.
* Convert quicklist to listpack
  When a quicklist has only one node, and its length or size is reduced to half
  of the `list-max-listpack-size` limit, it will be converted to a listpack.
  This is done to avoid frequent conversions when we add or remove at the bounding size or length.
    
## Interface changes
1. add list entry param to listTypeSetIteratorDirection
    When list encoding is listpack, `listTypeIterator->lpi` points to the next entry of current entry,
    so when changing the direction, we need to use the current node (listTypeEntry->p) to 
    update `listTypeIterator->lpi` to the next node in the reverse direction.

## Benchmark
### Listpack VS Quicklist with one node
* LPUSH - roughly 0.3% improvement
* LRANGE - roughly 13% improvement

### Both are quicklist
* LRANGE - roughly 3% improvement
* LRANGE without pipeline - roughly 3% improvement

From the benchmark, as we can see from the results
1. When list is quicklist encoding, LRANGE improves performance by <5%.
2. When list is listpack encoding, LRANGE improves performance by ~13%,
   the main enhancement is brought by `addListListpackRangeReply()`.

## Memory usage
1M lists(key:0~key:1000000) with 5 items of 10 chars ("hellohello") each.
shows memory usage down by 35.49%, from 214MB to 138MB.

## Note
1. Add conversion callback to support doing some work before conversion
    Since the quicklist iterator decompresses the current node when it is released, we can 
    no longer decompress the quicklist after we convert the list.

1 files changed, 373 insertions, 78 deletions

diff --git a/src/t_list.c b/src/t_list.c
index e0d324309..d7906d391 100644
--- a/src/t_list.c
+++ b/src/t_list.c
@@ -33,6 +33,131 @@
  * List API
  *----------------------------------------------------------------------------*/
 
+/* Check the length and size of a number of objects that will be added to list to see
+ * if we need to convert a listpack to a quicklist. Note that we only check string
+ * encoded objects as their string length can be queried in constant time.
+ *
+ * If callback is given the function is called in order for caller to do some work
+ * before the list conversion. */
+static void listTypeTryConvertListpack(robj *o, robj **argv, int start, int end,
+                                       beforeConvertCB fn, void *data)
+{
+    serverAssert(o->encoding == OBJ_ENCODING_LISTPACK);
+
+    size_t add_bytes = 0;
+    size_t add_length = 0;
+
+    if (argv) {
+        for (int i = start; i <= end; i++) {
+            if (!sdsEncodedObject(argv[i]))
+                continue;
+            add_bytes += sdslen(argv[i]->ptr);
+        }
+        add_length = end - start + 1;
+    }
+
+    if (quicklistNodeExceedsLimit(server.list_max_listpack_size,
+            lpBytes(o->ptr) + add_bytes, lpLength(o->ptr) + add_length))
+    {
+        /* Invoke callback before conversion. */
+        if (fn) fn(data);
+
+        quicklist *ql = quicklistCreate();
+        quicklistSetOptions(ql, server.list_max_listpack_size, server.list_compress_depth);
+
+        /* Append listpack to quicklist if it's not empty, otherwise release it. */
+        if (lpLength(o->ptr))
+            quicklistAppendListpack(ql, o->ptr);
+        else
+            lpFree(o->ptr);
+        o->ptr = ql;
+        o->encoding = OBJ_ENCODING_QUICKLIST;
+    }
+}
+
+/* Check the length and size of a quicklist to see if we need to convert it to listpack.
+ *
+ * 'shrinking' is 1 means that the conversion is due to a list shrinking, to avoid
+ * frequent conversions of quicklist and listpack due to frequent insertion and
+ * deletion, we don't convert quicklist to listpack until its length or size is
+ * below half of the limit.
+ *
+ * If callback is given the function is called in order for caller to do some work
+ * before the list conversion. */
+static void listTypeTryConvertQuicklist(robj *o, int shrinking, beforeConvertCB fn, void *data) {
+    serverAssert(o->encoding == OBJ_ENCODING_QUICKLIST);
+
+    size_t sz_limit;
+    unsigned int count_limit;
+    quicklist *ql = o->ptr;
+
+    /* A quicklist can be converted to listpack only if it has only one packed node. */
+    if (ql->len != 1 || ql->head->container != QUICKLIST_NODE_CONTAINER_PACKED)
+        return;
+
+    /* Check the length or size of the quicklist is below the limit. */
+    quicklistNodeLimit(server.list_max_listpack_size, &sz_limit, &count_limit);
+    if (shrinking) {
+        sz_limit /= 2;
+        count_limit /= 2;
+    }
+    if (ql->head->sz > sz_limit || ql->count > count_limit) return;
+
+    /* Invoke callback before conversion. */
+    if (fn) fn(data);
+
+    /* Extract the listpack from the unique quicklist node,
+     * then reset it and release the quicklist. */
+    o->ptr = ql->head->entry;
+    ql->head->entry = NULL;
+    quicklistRelease(ql);
+    o->encoding = OBJ_ENCODING_LISTPACK;
+}
+
+/* Check if the list needs to be converted to appropriate encoding due to
+ * growing, shrinking or other cases.
+ *
+ * 'lct' can be one of the following values:
+ * LIST_CONV_AUTO      - Used after we built a new list, and we want to let the
+ *                       function decide on the best encoding for that list.
+ * LIST_CONV_GROWING   - Used before or right after adding elements to the list,
+ *                       in which case we are likely to only consider converting
+ *                       from listpack to quicklist.
+ *                       'argv' is only used in this case to calculate the size
+ *                       of a number of objects that will be added to list.
+ * LIST_CONV_SHRINKING - Used after removing an element from the list, in which case we
+ *                       wanna consider converting from quicklist to listpack. When we
+ *                       know we're shrinking, we use a lower (more strict) threshold in
+ *                       order to avoid repeated conversions on every list change. */
+static void listTypeTryConversionRaw(robj *o, list_conv_type lct,
+                                     robj **argv, int start, int end,
+                                     beforeConvertCB fn, void *data)
+{
+    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
+        if (lct == LIST_CONV_GROWING) return; /* Growing has nothing to do with quicklist */
+        listTypeTryConvertQuicklist(o, lct == LIST_CONV_SHRINKING, fn, data);
+    } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+        if (lct == LIST_CONV_SHRINKING) return; /* Shrinking has nothing to do with listpack */
+        listTypeTryConvertListpack(o, argv, start, end, fn, data);
+    } else {
+        serverPanic("Unknown list encoding");
+    }
+}
+
+/* This is just a wrapper for listTypeTryConversionRaw() that is
+ * able to try conversion without passing 'argv'. */
+void listTypeTryConversion(robj *o, list_conv_type lct, beforeConvertCB fn, void *data) {
+    listTypeTryConversionRaw(o, lct, NULL, 0, 0, fn, data);
+}
+
+/* This is just a wrapper for listTypeTryConversionRaw() that is
+ * able to try conversion before adding elements to the list. */
+void listTypeTryConversionAppend(robj *o, robj **argv, int start, int end,
+                                 beforeConvertCB fn, void *data)
+{
+    listTypeTryConversionRaw(o, LIST_CONV_GROWING, argv, start, end, fn, data);
+}
+
 /* The function pushes an element to the specified list object 'subject',
  * at head or tail position as specified by 'where'.
  *
@@ -48,6 +173,16 @@ void listTypePush(robj *subject, robj *value, int where) {
         } else {
             quicklistPush(subject->ptr, value->ptr, sdslen(value->ptr), pos);
         }
+    } else if (subject->encoding == OBJ_ENCODING_LISTPACK) {
+        if (value->encoding == OBJ_ENCODING_INT) {
+            subject->ptr = (where == LIST_HEAD) ?
+                lpPrependInteger(subject->ptr, (long)value->ptr) :
+                lpAppendInteger(subject->ptr, (long)value->ptr);
+        } else {
+            subject->ptr = (where == LIST_HEAD) ?
+                lpPrepend(subject->ptr, value->ptr, sdslen(value->ptr)) :
+                lpAppend(subject->ptr, value->ptr, sdslen(value->ptr));
+        }
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -58,16 +193,28 @@ void *listPopSaver(unsigned char *data, size_t sz) {
 }
 
 robj *listTypePop(robj *subject, int where) {
-    long long vlong;
     robj *value = NULL;
 
-    int ql_where = where == LIST_HEAD ? QUICKLIST_HEAD : QUICKLIST_TAIL;
     if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
+        long long vlong;
+        int ql_where = where == LIST_HEAD ? QUICKLIST_HEAD : QUICKLIST_TAIL;
         if (quicklistPopCustom(subject->ptr, ql_where, (unsigned char **)&value,
                                NULL, &vlong, listPopSaver)) {
             if (!value)
                 value = createStringObjectFromLongLong(vlong);
         }
+    } else if (subject->encoding == OBJ_ENCODING_LISTPACK) {
+        unsigned char *p;
+        unsigned char *vstr;
+        int64_t vlen;
+        unsigned char intbuf[LP_INTBUF_SIZE];
+
+        p = (where == LIST_HEAD) ? lpFirst(subject->ptr) : lpLast(subject->ptr);
+        if (p) {
+            vstr = lpGet(p, &vlen, intbuf);
+            value = createStringObject((char*)vstr, vlen);
+            subject->ptr = lpDelete(subject->ptr, p, NULL);
+        }
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -77,6 +224,8 @@ robj *listTypePop(robj *subject, int where) {
 unsigned long listTypeLength(const robj *subject) {
     if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
         return quicklistCount(subject->ptr);
+    } else if (subject->encoding == OBJ_ENCODING_LISTPACK) {
+        return lpLength(subject->ptr);
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -92,11 +241,12 @@ listTypeIterator *listTypeInitIterator(robj *subject, long index,
     li->iter = NULL;
     /* LIST_HEAD means start at TAIL and move *towards* head.
      * LIST_TAIL means start at HEAD and move *towards tail. */
-    int iter_direction =
-        direction == LIST_HEAD ? AL_START_TAIL : AL_START_HEAD;
     if (li->encoding == OBJ_ENCODING_QUICKLIST) {
+        int iter_direction = direction == LIST_HEAD ? AL_START_TAIL : AL_START_HEAD;
         li->iter = quicklistGetIteratorAtIdx(li->subject->ptr,
                                              iter_direction, index);
+    } else if (li->encoding == OBJ_ENCODING_LISTPACK) {
+        li->lpi = lpSeek(subject->ptr, index);
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -104,15 +254,27 @@ listTypeIterator *listTypeInitIterator(robj *subject, long index,
 }
 
 /* Sets the direction of an iterator. */
-void listTypeSetIteratorDirection(listTypeIterator *li, unsigned char direction) {
+void listTypeSetIteratorDirection(listTypeIterator *li, listTypeEntry *entry, unsigned char direction) {
+    if (li->direction == direction) return;
+
     li->direction = direction;
-    int dir = direction == LIST_HEAD ? AL_START_TAIL : AL_START_HEAD;
-    quicklistSetDirection(li->iter, dir);
+    if (li->encoding == OBJ_ENCODING_QUICKLIST) {
+        int dir = direction == LIST_HEAD ? AL_START_TAIL : AL_START_HEAD;
+        quicklistSetDirection(li->iter, dir);
+    } else if (li->encoding == OBJ_ENCODING_LISTPACK) {
+        unsigned char *lp = li->subject->ptr;
+        /* Note that the iterator for listpack always points to the next of the current entry,
+         * so we need to update position of the iterator depending on the direction. */
+        li->lpi = (direction == LIST_TAIL) ? lpNext(lp, entry->lpe) : lpPrev(lp, entry->lpe);
+    } else {
+        serverPanic("Unknown list encoding");
+    }
 }
 
 /* Clean up the iterator. */
 void listTypeReleaseIterator(listTypeIterator *li) {
-    quicklistReleaseIterator(li->iter);
+    if (li->encoding == OBJ_ENCODING_QUICKLIST)
+        quicklistReleaseIterator(li->iter);
     zfree(li);
 }
 
@@ -126,62 +288,129 @@ int listTypeNext(listTypeIterator *li, listTypeEntry *entry) {
     entry->li = li;
     if (li->encoding == OBJ_ENCODING_QUICKLIST) {
         return quicklistNext(li->iter, &entry->entry);
+    } else if (li->encoding == OBJ_ENCODING_LISTPACK) {
+        entry->lpe = li->lpi;
+        if (entry->lpe != NULL) {
+            li->lpi = (li->direction == LIST_TAIL) ?
+                lpNext(li->subject->ptr,li->lpi) : lpPrev(li->subject->ptr,li->lpi);
+            return 1;
+        }
     } else {
         serverPanic("Unknown list encoding");
     }
     return 0;
 }
 
-/* Return entry or NULL at the current position of the iterator. */
-robj *listTypeGet(listTypeEntry *entry) {
-    robj *value = NULL;
+/* Get entry value at the current position of the iterator.
+ * When the function returns NULL, it populates the integer value by
+ * reference in 'lval'. Otherwise a pointer to the string is returned,
+ * and 'vlen' is set to the length of the string. */
+unsigned char *listTypeGetValue(listTypeEntry *entry, size_t *vlen, long long *lval) {
+    unsigned char *vstr = NULL;
     if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
         if (entry->entry.value) {
-            value = createStringObject((char *)entry->entry.value,
-                                       entry->entry.sz);
+            vstr = entry->entry.value;
+            *vlen = entry->entry.sz;
         } else {
-            value = createStringObjectFromLongLong(entry->entry.longval);
+            *lval = entry->entry.longval;
         }
+    } else if (entry->li->encoding == OBJ_ENCODING_LISTPACK) {
+        unsigned int slen;
+        vstr = lpGetValue(entry->lpe, &slen, lval);
+        *vlen = slen;
     } else {
         serverPanic("Unknown list encoding");
     }
-    return value;
+    return vstr;
+}
+
+/* Return entry or NULL at the current position of the iterator. */
+robj *listTypeGet(listTypeEntry *entry) {
+    unsigned char *vstr;
+    size_t vlen;
+    long long lval;
+
+    vstr = listTypeGetValue(entry, &vlen, &lval);
+    if (vstr) 
+        return createStringObject((char *)vstr, vlen);
+    else
+        return createStringObjectFromLongLong(lval);
 }
 
 void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
+    robj *subject = entry->li->subject;
+    value = getDecodedObject(value);
+    sds str = value->ptr;
+    size_t len = sdslen(str);
+
     if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
-        value = getDecodedObject(value);
-        sds str = value->ptr;
-        size_t len = sdslen(str);
         if (where == LIST_TAIL) {
             quicklistInsertAfter(entry->li->iter, &entry->entry, str, len);
         } else if (where == LIST_HEAD) {
             quicklistInsertBefore(entry->li->iter, &entry->entry, str, len);
         }
-        decrRefCount(value);
+    } else if (entry->li->encoding == OBJ_ENCODING_LISTPACK) {
+        int lpw = (where == LIST_TAIL) ? LP_AFTER : LP_BEFORE;
+        subject->ptr = lpInsertString(subject->ptr, (unsigned char *)str,
+                                      len, entry->lpe, lpw, &entry->lpe);
     } else {
         serverPanic("Unknown list encoding");
     }
+    decrRefCount(value);
 }
 
 /* Replaces entry at the current position of the iterator. */
 void listTypeReplace(listTypeEntry *entry, robj *value) {
+    robj *subject = entry->li->subject;
+    value = getDecodedObject(value);
+    sds str = value->ptr;
+    size_t len = sdslen(str);
+
     if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
-        value = getDecodedObject(value);
-        sds str = value->ptr;
-        size_t len = sdslen(str);
         quicklistReplaceEntry(entry->li->iter, &entry->entry, str, len);
-        decrRefCount(value);
+    } else if (entry->li->encoding == OBJ_ENCODING_LISTPACK) {
+        subject->ptr = lpReplace(subject->ptr, &entry->lpe, (unsigned char *)str, len);
+    } else {
+        serverPanic("Unknown list encoding");
+    }
+
+    decrRefCount(value);
+}
+
+/* Replace entry at offset 'index' by 'value'.
+ *
+ * Returns 1 if replace happened.
+ * Returns 0 if replace failed and no changes happened. */
+int listTypeReplaceAtIndex(robj *o, int index, robj *value) {
+    value = getDecodedObject(value);
+    sds vstr = value->ptr;
+    size_t vlen = sdslen(vstr);
+    int replaced = 0;
+
+    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
+        quicklist *ql = o->ptr;
+        replaced = quicklistReplaceAtIndex(ql, index, vstr, vlen);
+    } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+        unsigned char *p = lpSeek(o->ptr,index);
+        if (p) {
+            o->ptr = lpReplace(o->ptr, &p, (unsigned char *)vstr, vlen);
+            replaced = 1;
+        }
     } else {
         serverPanic("Unknown list encoding");
     }
+
+    decrRefCount(value);
+    return replaced;
 }
 
 /* Compare the given object with the entry at the current position. */
 int listTypeEqual(listTypeEntry *entry, robj *o) {
+    serverAssertWithInfo(NULL,o,sdsEncodedObject(o));
     if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
-        serverAssertWithInfo(NULL,o,sdsEncodedObject(o));
         return quicklistCompare(&entry->entry,o->ptr,sdslen(o->ptr));
+    } else if (entry->li->encoding == OBJ_ENCODING_LISTPACK) {
+        return lpCompare(entry->lpe,o->ptr,sdslen(o->ptr));
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -191,6 +420,22 @@ int listTypeEqual(listTypeEntry *entry, robj *o) {
 void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry) {
     if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
         quicklistDelEntry(iter->iter, &entry->entry);
+    } else if (entry->li->encoding == OBJ_ENCODING_LISTPACK) {
+        unsigned char *p = entry->lpe;
+        iter->subject->ptr = lpDelete(iter->subject->ptr,p,&p);
+
+        /* Update position of the iterator depending on the direction */
+        if (iter->direction == LIST_TAIL)
+            iter->lpi = p;
+        else {
+            if (p) {
+                iter->lpi = lpPrev(iter->subject->ptr,p);
+            } else {
+                /* We deleted the last element, so we need to set the
+                 * iterator to the last element. */
+                iter->lpi = lpLast(iter->subject->ptr);
+            }
+        }
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -207,21 +452,26 @@ robj *listTypeDup(robj *o) {
     serverAssert(o->type == OBJ_LIST);
 
     switch (o->encoding) {
+        case OBJ_ENCODING_LISTPACK:
+            lobj = createObject(OBJ_LIST, lpDup(o->ptr));
+            break;
         case OBJ_ENCODING_QUICKLIST:
             lobj = createObject(OBJ_LIST, quicklistDup(o->ptr));
-            lobj->encoding = o->encoding;
             break;
         default:
             serverPanic("Unknown list encoding");
             break;
     }
+    lobj->encoding = o->encoding;
     return lobj;
 }
 
 /* Delete a range of elements from the list. */
-int listTypeDelRange(robj *subject, long start, long count) {
+void listTypeDelRange(robj *subject, long start, long count) {
     if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
-        return quicklistDelRange(subject->ptr, start, count);
+        quicklistDelRange(subject->ptr, start, count);
+    } else if (subject->encoding == OBJ_ENCODING_LISTPACK) {
+        subject->ptr = lpDeleteRange(subject->ptr, start, count);
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -244,12 +494,11 @@ void pushGenericCommand(client *c, int where, int xx) {
             return;
         }
 
-        lobj = createQuicklistObject();
-        quicklistSetOptions(lobj->ptr, server.list_max_listpack_size,
-                            server.list_compress_depth);
+        lobj = createListListpackObject();
         dbAdd(c->db,c->argv[1],lobj);
     }
 
+    listTypeTryConversionAppend(lobj,c->argv,2,c->argc-1,NULL,NULL);
     for (j = 2; j < c->argc; j++) {
         listTypePush(lobj,c->argv[j],where);
         server.dirty++;
@@ -302,6 +551,13 @@ void linsertCommand(client *c) {
     if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,subject,OBJ_LIST)) return;
 
+    /* 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 listpack to a regular list if necessary. */
+    listTypeTryConversionAppend(subject,c->argv,4,4,NULL,NULL);
+
     /* Seek pivot from head to tail */
     iter = listTypeInitIterator(subject,0,LIST_TAIL);
     while (listTypeNext(iter,&entry)) {
@@ -343,22 +599,24 @@ void lindexCommand(client *c) {
     if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != C_OK))
         return;
 
-    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
-        quicklistEntry entry;
-        quicklistIter *iter = quicklistGetIteratorEntryAtIdx(o->ptr, index, &entry);
-        if (iter) {
-            if (entry.value) {
-                addReplyBulkCBuffer(c, entry.value, entry.sz);
-            } else {
-                addReplyBulkLongLong(c, entry.longval);
-            }
+    listTypeIterator *iter = listTypeInitIterator(o,index,LIST_TAIL);
+    listTypeEntry entry;
+    unsigned char *vstr;
+    size_t vlen;
+    long long lval;
+
+    if (listTypeNext(iter,&entry)) {
+        vstr = listTypeGetValue(&entry,&vlen,&lval);
+        if (vstr) {
+            addReplyBulkCBuffer(c, vstr, vlen);
         } else {
-            addReplyNull(c);
+            addReplyBulkLongLong(c, lval);
         }
-        quicklistReleaseIterator(iter);
     } else {
-        serverPanic("Unknown list encoding");
+        addReplyNull(c);
     }
+
+    listTypeReleaseIterator(iter);
 }
 
 /* LSET <key> <index> <element> */
@@ -371,20 +629,19 @@ void lsetCommand(client *c) {
     if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != C_OK))
         return;
 
-    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
-        quicklist *ql = o->ptr;
-        int replaced = quicklistReplaceAtIndex(ql, index,
-                                               value->ptr, sdslen(value->ptr));
-        if (!replaced) {
-            addReplyErrorObject(c,shared.outofrangeerr);
-        } else {
-            addReply(c,shared.ok);
-            signalModifiedKey(c,c->db,c->argv[1]);
-            notifyKeyspaceEvent(NOTIFY_LIST,"lset",c->argv[1],c->db->id);
-            server.dirty++;
-        }
+    listTypeTryConversionAppend(o,c->argv,3,3,NULL,NULL);
+    if (listTypeReplaceAtIndex(o,index,value)) {
+        addReply(c,shared.ok);
+        signalModifiedKey(c,c->db,c->argv[1]);
+        notifyKeyspaceEvent(NOTIFY_LIST,"lset",c->argv[1],c->db->id);
+        server.dirty++;
+
+        /* We might replace a big item with a small one or vice versa, but we've
+         * already handled the growing case in listTypeTryConversionAppend()
+         * above, so here we just need to try the conversion for shrinking. */
+        listTypeTryConversion(o,LIST_CONV_SHRINKING,NULL,NULL);
     } else {
-        serverPanic("Unknown list encoding");
+        addReplyErrorObject(c,shared.outofrangeerr);
     }
 }
 
@@ -417,6 +674,51 @@ void listPopRangeAndReplyWithKey(client *c, robj *o, robj *key, int where, long
     listElementsRemoved(c, key, where, o, rangelen, signal, deleted);
 }
 
+/* Extracted from `addListRangeReply()` to reply with a quicklist list.
+ * Note that the purpose is to make the methods small so that the
+ * code in the loop can be inlined better to improve performance. */
+void addListQuicklistRangeReply(client *c, robj *o, int from, int rangelen, int reverse) {
+    /* Return the result in form of a multi-bulk reply */
+    addReplyArrayLen(c,rangelen);
+
+    int direction = reverse ? AL_START_TAIL : AL_START_HEAD;
+    quicklistIter *iter = quicklistGetIteratorAtIdx(o->ptr, direction, from);
+    while(rangelen--) {
+        quicklistEntry qe;
+        serverAssert(quicklistNext(iter, &qe)); /* fail on corrupt data */
+        if (qe.value) {
+            addReplyBulkCBuffer(c,qe.value,qe.sz);
+        } else {
+            addReplyBulkLongLong(c,qe.longval);
+        }
+    }
+    quicklistReleaseIterator(iter);
+}
+
+/* Extracted from `addListRangeReply()` to reply with a listpack list.
+ * Note that the purpose is to make the methods small so that the
+ * code in the loop can be inlined better to improve performance. */
+void addListListpackRangeReply(client *c, robj *o, int from, int rangelen, int reverse) {
+    unsigned char *p = lpSeek(o->ptr, from);
+    unsigned char *vstr;
+    unsigned int vlen;
+    long long lval;
+
+    /* Return the result in form of a multi-bulk reply */
+    addReplyArrayLen(c,rangelen);
+
+    while(rangelen--) {
+        serverAssert(p); /* fail on corrupt data */
+        vstr = lpGetValue(p, &vlen, &lval);
+        if (vstr) {
+            addReplyBulkCBuffer(c,vstr,vlen);
+        } else {
+            addReplyBulkLongLong(c,lval);
+        }
+        p = reverse ? lpPrev(o->ptr,p) : lpNext(o->ptr,p);
+    }
+}
+
 /* A helper for replying with a list's range between the inclusive start and end
  * indexes as multi-bulk, with support for negative indexes. Note that start
  * must be less than end or an empty array is returned. When the reverse
@@ -439,27 +741,13 @@ void addListRangeReply(client *c, robj *o, long start, long end, int reverse) {
     if (end >= llen) end = llen-1;
     rangelen = (end-start)+1;
 
-    /* Return the result in form of a multi-bulk reply */
-    addReplyArrayLen(c,rangelen);
-    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
-        int from = reverse ? end : start;
-        int direction = reverse ? LIST_HEAD : LIST_TAIL;
-        listTypeIterator *iter = listTypeInitIterator(o,from,direction);
-
-        while(rangelen--) {
-            listTypeEntry entry;
-            serverAssert(listTypeNext(iter, &entry)); /* fail on corrupt data */
-            quicklistEntry *qe = &entry.entry;
-            if (qe->value) {
-                addReplyBulkCBuffer(c,qe->value,qe->sz);
-            } else {
-                addReplyBulkLongLong(c,qe->longval);
-            }
-        }
-        listTypeReleaseIterator(iter);
-    } else {
+    int from = reverse ? end : start;
+    if (o->encoding == OBJ_ENCODING_QUICKLIST)
+        addListQuicklistRangeReply(c, o, from, rangelen, reverse);
+    else if (o->encoding == OBJ_ENCODING_LISTPACK)
+        addListListpackRangeReply(c, o, from, rangelen, reverse);
+    else
         serverPanic("Unknown list encoding");
-    }
 }
 
 /* A housekeeping helper for list elements popping tasks.
@@ -478,6 +766,7 @@ void listElementsRemoved(client *c, robj *key, int where, robj *o, long count, i
         dbDelete(c->db, key);
         notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, c->db->id);
     } else {
+        listTypeTryConversion(o, LIST_CONV_SHRINKING, NULL, NULL);
         if (deleted) *deleted = 0;
     }
     if (signal) signalModifiedKey(c, c->db, key);
@@ -633,6 +922,9 @@ void ltrimCommand(client *c) {
     if (o->encoding == OBJ_ENCODING_QUICKLIST) {
         quicklistDelRange(o->ptr,0,ltrim);
         quicklistDelRange(o->ptr,-rtrim,rtrim);
+    } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+        o->ptr = lpDeleteRange(o->ptr,0,ltrim);
+        o->ptr = lpDeleteRange(o->ptr,-rtrim,rtrim);
     } else {
         serverPanic("Unknown list encoding");
     }
@@ -641,6 +933,8 @@ void ltrimCommand(client *c) {
     if (listTypeLength(o) == 0) {
         dbDelete(c->db,c->argv[1]);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"del",c->argv[1],c->db->id);
+    } else {
+        listTypeTryConversion(o,LIST_CONV_SHRINKING,NULL,NULL);
     }
     signalModifiedKey(c,c->db,c->argv[1]);
     server.dirty += (ltrim + rtrim);
@@ -799,6 +1093,8 @@ void lremCommand(client *c) {
     if (listTypeLength(subject) == 0) {
         dbDelete(c->db,c->argv[1]);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"del",c->argv[1],c->db->id);
+    } else if (removed) {
+        listTypeTryConversion(subject,LIST_CONV_SHRINKING,NULL,NULL);
     }
 
     addReplyLongLong(c,removed);
@@ -808,12 +1104,11 @@ void lmoveHandlePush(client *c, robj *dstkey, robj *dstobj, robj *value,
                      int where) {
     /* Create the list if the key does not exist */
     if (!dstobj) {
-        dstobj = createQuicklistObject();
-        quicklistSetOptions(dstobj->ptr, server.list_max_listpack_size,
-                            server.list_compress_depth);
+        dstobj = createListListpackObject();
         dbAdd(c->db,dstkey,dstobj);
     }
     signalModifiedKey(c,c->db,dstkey);
+    listTypeTryConversionAppend(dstobj,&value,0,0,NULL,NULL);
     listTypePush(dstobj,value,where);
     notifyKeyspaceEvent(NOTIFY_LIST,
                         where == LIST_HEAD ? "lpush" : "rpush",