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.

20 files changed, 834 insertions, 280 deletions

diff --git a/src/aof.c b/src/aof.c
index f1bf9a1a6..2d1e1c441 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -1775,42 +1775,40 @@ int rioWriteBulkObject(rio *r, robj *obj) {
 int rewriteListObject(rio *r, robj *key, robj *o) {
     long long count = 0, items = listTypeLength(o);
 
-    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
-        quicklist *list = o->ptr;
-        quicklistIter *li = quicklistGetIterator(list, AL_START_HEAD);
-        quicklistEntry entry;
-
-        while (quicklistNext(li,&entry)) {
-            if (count == 0) {
-                int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
-                    AOF_REWRITE_ITEMS_PER_CMD : items;
-                if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
-                    !rioWriteBulkString(r,"RPUSH",5) ||
-                    !rioWriteBulkObject(r,key)) 
-                {
-                    quicklistReleaseIterator(li);
-                    return 0;
-                }
+    listTypeIterator *li = listTypeInitIterator(o,0,LIST_TAIL);
+    listTypeEntry entry;
+    while (listTypeNext(li,&entry)) {
+        if (count == 0) {
+            int cmd_items = (items > AOF_REWRITE_ITEMS_PER_CMD) ?
+                AOF_REWRITE_ITEMS_PER_CMD : items;
+            if (!rioWriteBulkCount(r,'*',2+cmd_items) ||
+                !rioWriteBulkString(r,"RPUSH",5) ||
+                !rioWriteBulkObject(r,key)) 
+            {
+                listTypeReleaseIterator(li);
+                return 0;
             }
+        }
 
-            if (entry.value) {
-                if (!rioWriteBulkString(r,(char*)entry.value,entry.sz)) {
-                    quicklistReleaseIterator(li);
-                    return 0;
-                }
-            } else {
-                if (!rioWriteBulkLongLong(r,entry.longval)) {
-                    quicklistReleaseIterator(li);
-                    return 0;
-                }
+        unsigned char *vstr;
+        size_t vlen;
+        long long lval;
+        vstr = listTypeGetValue(&entry,&vlen,&lval);
+        if (vstr) {
+            if (!rioWriteBulkString(r,(char*)vstr,vlen)) {
+                listTypeReleaseIterator(li);
+                return 0;
+            }
+        } else {
+            if (!rioWriteBulkLongLong(r,lval)) {
+                listTypeReleaseIterator(li);
+                return 0;
             }
-            if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
-            items--;
         }
-        quicklistReleaseIterator(li);
-    } else {
-        serverPanic("Unknown list encoding");
+        if (++count == AOF_REWRITE_ITEMS_PER_CMD) count = 0;
+        items--;
     }
+    listTypeReleaseIterator(li);
     return 1;
 }
 
diff --git a/src/defrag.c b/src/defrag.c
index e78c07929..dbdf2ab62 100644
--- a/src/defrag.c
+++ b/src/defrag.c
@@ -868,6 +868,9 @@ long defragKey(redisDb *db, dictEntry *de) {
     } else if (ob->type == OBJ_LIST) {
         if (ob->encoding == OBJ_ENCODING_QUICKLIST) {
             defragged += defragQuicklist(db, de);
+        } else if (ob->encoding == OBJ_ENCODING_LISTPACK) {
+            if ((newzl = activeDefragAlloc(ob->ptr)))
+                defragged++, ob->ptr = newzl;
         } else {
             serverPanic("Unknown list encoding");
         }
diff --git a/src/lazyfree.c b/src/lazyfree.c
index 4e336255e..a44ad2df4 100644
--- a/src/lazyfree.c
+++ b/src/lazyfree.c
@@ -102,7 +102,7 @@ void lazyfreeResetStats() {
  * For lists the function returns the number of elements in the quicklist
  * representing the list. */
 size_t lazyfreeGetFreeEffort(robj *key, robj *obj, int dbid) {
-    if (obj->type == OBJ_LIST) {
+    if (obj->type == OBJ_LIST && obj->encoding == OBJ_ENCODING_QUICKLIST) {
         quicklist *ql = obj->ptr;
         return ql->len;
     } else if (obj->type == OBJ_SET && obj->encoding == OBJ_ENCODING_HT) {
diff --git a/src/listpack.c b/src/listpack.c
index abe2142de..f36d9d64d 100644
--- a/src/listpack.c
+++ b/src/listpack.c
@@ -1209,6 +1209,13 @@ unsigned char *lpMerge(unsigned char **first, unsigned char **second) {
     return target;
 }
 
+unsigned char *lpDup(unsigned char *lp) {
+    size_t lpbytes = lpBytes(lp);
+    unsigned char *newlp = lp_malloc(lpbytes);
+    memcpy(newlp, lp, lpbytes);
+    return newlp;
+}
+
 /* Return the total number of bytes the listpack is composed of. */
 size_t lpBytes(unsigned char *lp) {
     return lpGetTotalBytes(lp);
diff --git a/src/listpack.h b/src/listpack.h
index ce27ea367..304ccd32c 100644
--- a/src/listpack.h
+++ b/src/listpack.h
@@ -72,6 +72,7 @@ unsigned char *lpDeleteRangeWithEntry(unsigned char *lp, unsigned char **p, unsi
 unsigned char *lpDeleteRange(unsigned char *lp, long index, unsigned long num);
 unsigned char *lpBatchDelete(unsigned char *lp, unsigned char **ps, unsigned long count);
 unsigned char *lpMerge(unsigned char **first, unsigned char **second);
+unsigned char *lpDup(unsigned char *lp);
 unsigned long lpLength(unsigned char *lp);
 unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf);
 unsigned char *lpGetValue(unsigned char *p, unsigned int *slen, long long *lval);
diff --git a/src/module.c b/src/module.c
index c42536449..e938b98d7 100644
--- a/src/module.c
+++ b/src/module.c
@@ -625,9 +625,7 @@ int moduleCreateEmptyKey(RedisModuleKey *key, int type) {
 
     switch(type) {
     case REDISMODULE_KEYTYPE_LIST:
-        obj = createQuicklistObject();
-        quicklistSetOptions(obj->ptr, server.list_max_listpack_size,
-                            server.list_compress_depth);
+        obj = createListListpackObject();
         break;
     case REDISMODULE_KEYTYPE_ZSET:
         obj = createZsetListpackObject();
@@ -660,6 +658,14 @@ static void moduleFreeKeyIterator(RedisModuleKey *key) {
     key->iter = NULL;
 }
 
+/* Callback for listTypeTryConversion().
+ * Frees list iterator and sets it to NULL. */
+static void moduleFreeListIterator(void *data) {
+    RedisModuleKey *key = (RedisModuleKey*)data;
+    serverAssert(key->value->type == OBJ_LIST);
+    if (key->iter) moduleFreeKeyIterator(key);
+}
+
 /* This function is called in low-level API implementation functions in order
  * to check if the value associated with the key remained empty after an
  * operation that removed elements from an aggregate data type.
@@ -4148,7 +4154,7 @@ int moduleListIteratorSeek(RedisModuleKey *key, long index, int mode) {
 
     /* Seek the iterator to the requested index. */
     unsigned char dir = key->u.list.index < index ? LIST_TAIL : LIST_HEAD;
-    listTypeSetIteratorDirection(key->iter, dir);
+    listTypeSetIteratorDirection(key->iter, &key->u.list.entry, dir);
     while (key->u.list.index != index) {
         serverAssert(listTypeNext(key->iter, &key->u.list.entry));
         key->u.list.index += dir == LIST_HEAD ? -1 : 1;
@@ -4183,6 +4189,7 @@ int RM_ListPush(RedisModuleKey *key, int where, RedisModuleString *ele) {
     if (key->value && key->value->type != OBJ_LIST) return REDISMODULE_ERR;
     if (key->iter) moduleFreeKeyIterator(key);
     if (key->value == NULL) moduleCreateEmptyKey(key,REDISMODULE_KEYTYPE_LIST);
+    listTypeTryConversionAppend(key->value, &ele, 0, 0, moduleFreeListIterator, key);
     listTypePush(key->value, ele,
         (where == REDISMODULE_LIST_HEAD) ? LIST_HEAD : LIST_TAIL);
     return REDISMODULE_OK;
@@ -4216,7 +4223,8 @@ RedisModuleString *RM_ListPop(RedisModuleKey *key, int where) {
         (where == REDISMODULE_LIST_HEAD) ? LIST_HEAD : LIST_TAIL);
     robj *decoded = getDecodedObject(ele);
     decrRefCount(ele);
-    moduleDelKeyIfEmpty(key);
+    if (!moduleDelKeyIfEmpty(key))
+        listTypeTryConversion(key->value, LIST_CONV_SHRINKING, moduleFreeListIterator, key);
     autoMemoryAdd(key->ctx,REDISMODULE_AM_STRING,decoded);
     return decoded;
 }
@@ -4270,6 +4278,11 @@ int RM_ListSet(RedisModuleKey *key, long index, RedisModuleString *value) {
         errno = EINVAL;
         return REDISMODULE_ERR;
     }
+    if (!key->value || key->value->type != OBJ_LIST) {
+        errno = ENOTSUP;
+        return REDISMODULE_ERR;
+    }
+    listTypeTryConversionAppend(key->value, &value, 0, 0, moduleFreeListIterator, key);
     if (moduleListIteratorSeek(key, index, REDISMODULE_WRITE)) {
         listTypeReplace(&key->u.list.entry, value);
         /* A note in quicklist.c forbids use of iterator after insert, so
@@ -4315,6 +4328,7 @@ int RM_ListInsert(RedisModuleKey *key, long index, RedisModuleString *value) {
         /* Insert before the first element => push head. */
         return RM_ListPush(key, REDISMODULE_LIST_HEAD, value);
     }
+    listTypeTryConversionAppend(key->value, &value, 0, 0, moduleFreeListIterator, key);
     if (moduleListIteratorSeek(key, index, REDISMODULE_WRITE)) {
         int where = index < 0 ? LIST_TAIL : LIST_HEAD;
         listTypeInsert(&key->u.list.entry, value, where);
@@ -4341,6 +4355,8 @@ int RM_ListDelete(RedisModuleKey *key, long index) {
     if (moduleListIteratorSeek(key, index, REDISMODULE_WRITE)) {
         listTypeDelete(key->iter, &key->u.list.entry);
         if (moduleDelKeyIfEmpty(key)) return REDISMODULE_OK;
+        listTypeTryConversion(key->value, LIST_CONV_SHRINKING, moduleFreeListIterator, key);
+        if (!key->iter) return REDISMODULE_OK; /* Return ASAP if iterator has been freed */
         if (listTypeNext(key->iter, &key->u.list.entry)) {
             /* After delete entry at position 'index', we need to update
              * 'key->u.list.index' according to the following cases:
diff --git a/src/object.c b/src/object.c
index d48af4128..7d71f59c4 100644
--- a/src/object.c
+++ b/src/object.c
@@ -233,6 +233,13 @@ robj *createQuicklistObject(void) {
     return o;
 }
 
+robj *createListListpackObject(void) {
+    unsigned char *lp = lpNew(0);
+    robj *o = createObject(OBJ_LIST,lp);
+    o->encoding = OBJ_ENCODING_LISTPACK;
+    return o;
+}
+
 robj *createSetObject(void) {
     dict *d = dictCreate(&setDictType);
     robj *o = createObject(OBJ_SET,d);
@@ -302,6 +309,8 @@ void freeStringObject(robj *o) {
 void freeListObject(robj *o) {
     if (o->encoding == OBJ_ENCODING_QUICKLIST) {
         quicklistRelease(o->ptr);
+    } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+        lpFree(o->ptr);
     } else {
         serverPanic("Unknown list encoding type");
     }
@@ -423,6 +432,8 @@ void dismissListObject(robj *o, size_t size_hint) {
                 node = node->next;
             }
         }
+    } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+        dismissMemory(o->ptr, lpBytes((unsigned char*)o->ptr));
     } else {
         serverPanic("Unknown list encoding type");
     }
@@ -1005,6 +1016,8 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
                 samples++;
             } while ((node = node->next) && samples < sample_size);
             asize += (double)elesize/samples*ql->len;
+        } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+            asize = sizeof(*o)+zmalloc_size(o->ptr);
         } else {
             serverPanic("Unknown list encoding");
         }
diff --git a/src/quicklist.c b/src/quicklist.c
index 954331365..301a2166e 100644
--- a/src/quicklist.c
+++ b/src/quicklist.c
@@ -30,6 +30,7 @@
 
 #include <stdio.h>
 #include <string.h> /* for memcpy */
+#include <limits.h>
 #include "quicklist.h"
 #include "zmalloc.h"
 #include "config.h"
@@ -447,25 +448,45 @@ REDIS_STATIC void _quicklistInsertNodeAfter(quicklist *quicklist,
     __quicklistInsertNode(quicklist, old_node, new_node, 1);
 }
 
-REDIS_STATIC int
-_quicklistNodeSizeMeetsOptimizationRequirement(const size_t sz,
-                                               const int fill) {
-    if (fill >= 0)
-        return 0;
+#define sizeMeetsSafetyLimit(sz) ((sz) <= SIZE_SAFETY_LIMIT)
 
-    size_t offset = (-fill) - 1;
-    if (offset < (sizeof(optimization_level) / sizeof(*optimization_level))) {
-        if (sz <= optimization_level[offset]) {
-            return 1;
-        } else {
-            return 0;
-        }
+/* Calculate the size limit or length limit of the quicklist node
+ * based on 'fill', and is also used to limit list listpack. */
+void quicklistNodeLimit(int fill, size_t *size, unsigned int *count) {
+    *size = SIZE_MAX;
+    *count = UINT_MAX;
+
+    if (fill >= 0) {
+        /* Ensure that one node have at least one entry */
+        *count = (fill == 0) ? 1 : fill;
     } else {
-        return 0;
+        size_t offset = (-fill) - 1;
+        size_t max_level = sizeof(optimization_level) / sizeof(*optimization_level);
+        if (offset >= max_level) offset = max_level - 1;
+        *size = optimization_level[offset];
     }
 }
 
-#define sizeMeetsSafetyLimit(sz) ((sz) <= SIZE_SAFETY_LIMIT)
+/* Check if the limit of the quicklist node has been reached to determine if
+ * insertions, merges or other operations that would increase the size of
+ * the node can be performed.
+ * Return 1 if exceeds the limit, otherwise 0. */
+int quicklistNodeExceedsLimit(int fill, size_t new_sz, unsigned int new_count) {
+    size_t sz_limit;
+    unsigned int count_limit;
+    quicklistNodeLimit(fill, &sz_limit, &count_limit);
+
+    if (likely(sz_limit != SIZE_MAX)) {
+        return new_sz > sz_limit;
+    } else if (count_limit != UINT_MAX) {
+        /* when we reach here we know that the limit is a size limit (which is
+         * safe, see comments next to optimization_level and SIZE_SAFETY_LIMIT) */
+        if (!sizeMeetsSafetyLimit(new_sz)) return 1;
+        return new_count > count_limit;
+    }
+
+    redis_unreachable();
+}
 
 REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
                                            const int fill, const size_t sz) {
@@ -481,16 +502,9 @@ REDIS_STATIC int _quicklistNodeAllowInsert(const quicklistNode *node,
      * Note: No need to check for overflow below since both `node->sz` and
      * `sz` are to be less than 1GB after the plain/large element check above. */
     size_t new_sz = node->sz + sz + SIZE_ESTIMATE_OVERHEAD;
-    if (likely(_quicklistNodeSizeMeetsOptimizationRequirement(new_sz, fill)))
-        return 1;
-    /* when we return 1 above we know that the limit is a size limit (which is
-     * safe, see comments next to optimization_level and SIZE_SAFETY_LIMIT) */
-    else if (!sizeMeetsSafetyLimit(new_sz))
-        return 0;
-    else if ((int)node->count < fill)
-        return 1;
-    else
+    if (unlikely(quicklistNodeExceedsLimit(fill, new_sz, node->count + 1)))
         return 0;
+    return 1;
 }
 
 REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a,
@@ -505,16 +519,9 @@ REDIS_STATIC int _quicklistNodeAllowMerge(const quicklistNode *a,
     /* approximate merged listpack size (- 7 to remove one listpack
      * header/trailer, see LP_HDR_SIZE and LP_EOF) */
     unsigned int merge_sz = a->sz + b->sz - 7;
-    if (likely(_quicklistNodeSizeMeetsOptimizationRequirement(merge_sz, fill)))
-        return 1;
-    /* when we return 1 above we know that the limit is a size limit (which is
-     * safe, see comments next to optimization_level and SIZE_SAFETY_LIMIT) */
-    else if (!sizeMeetsSafetyLimit(merge_sz))
-        return 0;
-    else if ((int)(a->count + b->count) <= fill)
-        return 1;
-    else
+    if (unlikely(quicklistNodeExceedsLimit(fill, merge_sz, a->count + b->count)))
         return 0;
+    return 1;
 }
 
 #define quicklistNodeUpdateSz(node)                                            \
diff --git a/src/quicklist.h b/src/quicklist.h
index 411231821..f17834b99 100644
--- a/src/quicklist.h
+++ b/src/quicklist.h
@@ -192,6 +192,8 @@ int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
 unsigned long quicklistCount(const quicklist *ql);
 int quicklistCompare(quicklistEntry *entry, unsigned char *p2, const size_t p2_len);
 size_t quicklistGetLzf(const quicklistNode *node, void **data);
+void quicklistNodeLimit(int fill, size_t *size, unsigned int *count);
+int quicklistNodeExceedsLimit(int fill, size_t new_sz, unsigned int new_count);
 void quicklistRepr(unsigned char *ql, int full);
 
 /* bookmarks */
diff --git a/src/rdb.c b/src/rdb.c
index bbb61aafb..5118c17b4 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -656,7 +656,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
     case OBJ_STRING:
         return rdbSaveType(rdb,RDB_TYPE_STRING);
     case OBJ_LIST:
-        if (o->encoding == OBJ_ENCODING_QUICKLIST)
+        if (o->encoding == OBJ_ENCODING_QUICKLIST || o->encoding == OBJ_ENCODING_LISTPACK)
             return rdbSaveType(rdb, RDB_TYPE_LIST_QUICKLIST_2);
         else
             serverPanic("Unknown list encoding");
@@ -828,6 +828,16 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
                 }
                 node = node->next;
             }
+        } else if (o->encoding == OBJ_ENCODING_LISTPACK) {
+            unsigned char *lp = o->ptr;
+
+            /* Save list listpack as a fake quicklist that only has a single node. */
+            if ((n = rdbSaveLen(rdb,1)) == -1) return -1;
+            nwritten += n;
+            if ((n = rdbSaveLen(rdb,QUICKLIST_NODE_CONTAINER_PACKED)) == -1) return -1;
+            nwritten += n;
+            if ((n = rdbSaveRawString(rdb,lp,lpBytes(lp))) == -1) return -1;
+            nwritten += n;
         } else {
             serverPanic("Unknown list encoding");
         }
@@ -1802,6 +1812,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
             decrRefCount(dec);
             decrRefCount(ele);
         }
+
+        listTypeTryConversion(o,LIST_CONV_AUTO,NULL,NULL);
     } else if (rdbtype == RDB_TYPE_SET) {
         /* Read Set value */
         if ((len = rdbLoadLen(rdb,NULL)) == RDB_LENERR) return NULL;
@@ -2173,6 +2185,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
             decrRefCount(o);
             goto emptykey;
         }
+
+        listTypeTryConversion(o,LIST_CONV_AUTO,NULL,NULL);
     } else if (rdbtype == RDB_TYPE_HASH_ZIPMAP  ||
                rdbtype == RDB_TYPE_LIST_ZIPLIST ||
                rdbtype == RDB_TYPE_SET_INTSET   ||
diff --git a/src/server.h b/src/server.h
index c96c1c743..3af135114 100644
--- a/src/server.h
+++ b/src/server.h
@@ -2318,12 +2318,15 @@ typedef struct {
     robj *subject;
     unsigned char encoding;
     unsigned char direction; /* Iteration direction */
-    quicklistIter *iter;
+
+    unsigned char *lpi; /* listpack iterator */
+    quicklistIter *iter; /* quicklist iterator */
 } listTypeIterator;
 
 /* Structure for an entry while iterating over a list. */
 typedef struct {
     listTypeIterator *li;
+    unsigned char *lpe; /* Entry in listpack */
     quicklistEntry entry; /* Entry in quicklist */
 } listTypeEntry;
 
@@ -2603,18 +2606,27 @@ robj *listTypePop(robj *subject, int where);
 unsigned long listTypeLength(const robj *subject);
 listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction);
 void listTypeReleaseIterator(listTypeIterator *li);
-void listTypeSetIteratorDirection(listTypeIterator *li, unsigned char direction);
+void listTypeSetIteratorDirection(listTypeIterator *li, listTypeEntry *entry, unsigned char direction);
 int listTypeNext(listTypeIterator *li, listTypeEntry *entry);
 robj *listTypeGet(listTypeEntry *entry);
+unsigned char *listTypeGetValue(listTypeEntry *entry, size_t *vlen, long long *lval);
 void listTypeInsert(listTypeEntry *entry, robj *value, int where);
 void listTypeReplace(listTypeEntry *entry, robj *value);
 int listTypeEqual(listTypeEntry *entry, robj *o);
 void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry);
 robj *listTypeDup(robj *o);
-int listTypeDelRange(robj *o, long start, long stop);
+void listTypeDelRange(robj *o, long start, long stop);
 void unblockClientWaitingData(client *c);
 void popGenericCommand(client *c, int where);
 void listElementsRemoved(client *c, robj *key, int where, robj *o, long count, int signal, int *deleted);
+typedef enum {
+    LIST_CONV_AUTO,
+    LIST_CONV_GROWING,
+    LIST_CONV_SHRINKING,
+} list_conv_type;
+typedef void (*beforeConvertCB)(void *data);
+void listTypeTryConversion(robj *o, list_conv_type lct, beforeConvertCB fn, void *data);
+void listTypeTryConversionAppend(robj *o, robj **argv, int start, int end, beforeConvertCB fn, void *data);
 
 /* MULTI/EXEC/WATCH... */
 void unwatchAllKeys(client *c);
@@ -2656,6 +2668,7 @@ robj *createStringObjectFromLongLong(long long value);
 robj *createStringObjectFromLongLongForValue(long long value);
 robj *createStringObjectFromLongDouble(long double value, int humanfriendly);
 robj *createQuicklistObject(void);
+robj *createListListpackObject(void);
 robj *createSetObject(void);
 robj *createIntsetObject(void);
 robj *createSetListpackObject(void);
diff --git a/src/sort.c b/src/sort.c
index 62e7ad701..3132d17e1 100644
--- a/src/sort.c
+++ b/src/sort.c
@@ -546,6 +546,8 @@ void sortCommandGeneric(client *c, int readonly) {
             }
         }
     } else {
+        /* We can't predict the size and encoding of the stored list, we
+         * assume it's a large list and then convert it at the end if needed. */
         robj *sobj = createQuicklistObject();
 
         /* STORE option specified, set the sorting result as a List object */
@@ -578,6 +580,7 @@ void sortCommandGeneric(client *c, int readonly) {
             }
         }
         if (outputlen) {
+            listTypeTryConversion(sobj,LIST_CONV_AUTO,NULL,NULL);
             setKey(c,c->db,storekey,sobj,0);
             notifyKeyspaceEvent(NOTIFY_LIST,"sortstore",storekey,
                                 c->db->id);
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",
diff --git a/tests/integration/rdb.tcl b/tests/integration/rdb.tcl
index 1dd6bb1d7..d4c6227fc 100644
--- a/tests/integration/rdb.tcl
+++ b/tests/integration/rdb.tcl
@@ -6,10 +6,11 @@ set server_path [tmpdir "server.rdb-encoding-test"]
 exec cp tests/assets/encodings.rdb $server_path
 exec cp tests/assets/list-quicklist.rdb $server_path
 
-start_server [list overrides [list "dir" $server_path "dbfilename" "list-quicklist.rdb"]] {
+start_server [list overrides [list "dir" $server_path "dbfilename" "list-quicklist.rdb" save ""]] {
     test "test old version rdb file" {
         r select 0
         assert_equal [r get x] 7
+        assert_encoding listpack list
         r lpop list
     } {7}
 }
diff --git a/tests/unit/aofrw.tcl b/tests/unit/aofrw.tcl
index 45db1939f..00fc9e3bd 100644
--- a/tests/unit/aofrw.tcl
+++ b/tests/unit/aofrw.tcl
@@ -78,10 +78,16 @@ start_server {tags {"aofrw external:skip"} overrides {aof-use-rdb-preamble no}}
     }
 
     foreach d {string int} {
-        foreach e {quicklist} {
+        foreach e {listpack quicklist} {
             test "AOF rewrite of list with $e encoding, $d data" {
                 r flushall
-                set len 1000
+                if {$e eq {listpack}} {
+                    r config set list-max-listpack-size -2
+                    set len 10
+                } else {
+                    r config set list-max-listpack-size 10
+                    set len 1000
+                }
                 for {set j 0} {$j < $len} {incr j} {
                     if {$d eq {string}} {
                         set data [randstring 0 16 alpha]
diff --git a/tests/unit/keyspace.tcl b/tests/unit/keyspace.tcl
index 97ffef6bd..b173e0efc 100644
--- a/tests/unit/keyspace.tcl
+++ b/tests/unit/keyspace.tcl
@@ -244,10 +244,15 @@ start_server {tags {"keyspace"}} {
         assert {[r get mynewkey{t}] eq "foobar"}
     }
 
-    test {COPY basic usage for list} {
+source "tests/unit/type/list-common.tcl"
+foreach {type large} [array get largevalue] {
+    set origin_config [config_get_set list-max-listpack-size -1]
+    test "COPY basic usage for list - $type" {
         r del mylist{t} mynewlist{t}
-        r lpush mylist{t} a b c d
+        r lpush mylist{t} a b $large c d
+        assert_encoding $type mylist{t}
         r copy mylist{t} mynewlist{t}
+        assert_encoding $type mynewlist{t}
         set digest [debug_digest_value mylist{t}]
         assert_equal $digest [debug_digest_value mynewlist{t}]
         assert_refcount 1 mylist{t}
@@ -255,6 +260,8 @@ start_server {tags {"keyspace"}} {
         r del mylist{t}
         assert_equal $digest [debug_digest_value mynewlist{t}]
     }
+    config_set list-max-listpack-size $origin_config
+}
 
     foreach type {intset listpack hashtable} {
         test {COPY basic usage for $type set} {
diff --git a/tests/unit/moduleapi/list.tcl b/tests/unit/moduleapi/list.tcl
index 0c440555d..11f3b7534 100644
--- a/tests/unit/moduleapi/list.tcl
+++ b/tests/unit/moduleapi/list.tcl
@@ -17,6 +17,7 @@ start_server {tags {"modules"}} {
 
     test {Module list set, get, insert, delete} {
         r del k
+        assert_error {WRONGTYPE Operation against a key holding the wrong kind of value*} {r list.set k 1 xyz}
         r rpush k x
         # insert, set, get
         r list.insert k 0 foo
@@ -76,6 +77,41 @@ start_server {tags {"modules"}} {
         r list.getall k
     } {bar y foo}
 
+    test {Module list - encoding conversion while inserting} {
+        r config set list-max-listpack-size 4
+        r del k
+        r rpush k a b c d
+        assert_encoding listpack k
+
+        # Converts to quicklist after inserting.
+        r list.edit k dii foo bar
+        assert_encoding quicklist k
+        assert_equal [r list.getall k] {foo bar b c d}
+
+        # Converts to listpack after deleting three entries.
+        r list.edit k ddd e
+        assert_encoding listpack k
+        assert_equal [r list.getall k] {c d}
+    }
+
+    test {Module list - encoding conversion while replacing} {
+        r config set list-max-listpack-size -1
+        r del k
+        r rpush k x y z
+        assert_encoding listpack k
+
+        # Converts to quicklist after replacing.
+        set big [string repeat "x" 4096]
+        r list.edit k r $big
+        assert_encoding quicklist k
+        assert_equal [r list.getall k] "$big y z"
+
+        # Converts to listpack after deleting the big entry.
+        r list.edit k d
+        assert_encoding listpack k
+        assert_equal [r list.getall k] {y z}
+    }
+
     test {Module list - list entry and index should be updated when deletion} {
         set original_config [config_get_set list-max-listpack-size 1]
 
diff --git a/tests/unit/sort.tcl b/tests/unit/sort.tcl
index a9264eadf..7946211eb 100644
--- a/tests/unit/sort.tcl
+++ b/tests/unit/sort.tcl
@@ -1,7 +1,7 @@
 start_server {
     tags {"sort"}
     overrides {
-        "list-max-ziplist-size" 32
+        "list-max-ziplist-size" 16
         "set-max-intset-entries" 32
     }
 } {
@@ -34,10 +34,22 @@ start_server {
         set _ $result
     }
 
+    proc check_sort_store_encoding {key} {
+        set listpack_max_size [lindex [r config get list-max-ziplist-size] 1]
+
+        # When the length or size of quicklist is less than the limit,
+        # it will be converted to listpack.
+        if {[r llen $key] <= $listpack_max_size} {
+            assert_encoding listpack $key
+        } else {
+            assert_encoding quicklist $key
+        }
+    }
+
     foreach {num cmd enc title} {
-        16 lpush quicklist "Old Ziplist"
-        1000 lpush quicklist "Old Linked list"
-        10000 lpush quicklist "Old Big Linked list"
+        16 lpush listpack "Listpack"
+        1000 lpush quicklist "Quicklist"
+        10000 lpush quicklist "Big Quicklist"
         16 sadd intset "Intset"
         1000 sadd hashtable "Hash table"
         10000 sadd hashtable "Big Hash table"
@@ -84,14 +96,14 @@ start_server {
         r sort tosort BY weight_* store sort-res
         assert_equal $result [r lrange sort-res 0 -1]
         assert_equal 16 [r llen sort-res]
-        assert_encoding quicklist sort-res
+        check_sort_store_encoding sort-res
     } {} {cluster:skip}
 
     test "SORT BY hash field STORE" {
         r sort tosort BY wobj_*->weight store sort-res
         assert_equal $result [r lrange sort-res 0 -1]
         assert_equal 16 [r llen sort-res]
-        assert_encoding quicklist sort-res
+        check_sort_store_encoding sort-res
     } {} {cluster:skip}
 
     test "SORT extracts STORE correctly" {
diff --git a/tests/unit/type/list-common.tcl b/tests/unit/type/list-common.tcl
index ab45f0b31..b393737c9 100644
--- a/tests/unit/type/list-common.tcl
+++ b/tests/unit/type/list-common.tcl
@@ -1,5 +1,4 @@
-# We need a value larger than list-max-ziplist-value to make sure
-# the list has the right encoding when it is swapped in again.
+# We need a value to make sure the list has the right encoding when it is inserted.
 array set largevalue {}
-set largevalue(ziplist) "hello"
-set largevalue(linkedlist) [string repeat "hello" 4]
+set largevalue(listpack) "hello"
+set largevalue(quicklist) [string repeat "x" 8192]
diff --git a/tests/unit/type/list.tcl b/tests/unit/type/list.tcl
index fb204d308..f3f8092f4 100644
--- a/tests/unit/type/list.tcl
+++ b/tests/unit/type/list.tcl
@@ -405,7 +405,7 @@ if {[lindex [r config get proto-max-bulk-len] 1] == 10000000000} {
 start_server {
     tags {"list"}
     overrides {
-        "list-max-ziplist-size" 5
+        "list-max-ziplist-size" -1
     }
 } {
     source "tests/unit/type/list-common.tcl"
@@ -432,9 +432,22 @@ start_server {
         }
     }
 
-    test {LPOS basic usage} {
+    proc create_listpack {key entries} {
+        r del $key
+        foreach entry $entries { r rpush $key $entry }
+        assert_encoding listpack $key
+    }
+
+    proc create_quicklist {key entries} {
+        r del $key
+        foreach entry $entries { r rpush $key $entry }
+        assert_encoding quicklist $key
+    }
+
+foreach {type large} [array get largevalue] {
+    test "LPOS basic usage - $type" {
         r DEL mylist
-        r RPUSH mylist a b c 1 2 3 c c
+        r RPUSH mylist a b c $large 2 3 c c
         assert {[r LPOS mylist a] == 0}
         assert {[r LPOS mylist c] == 2}
     }
@@ -483,59 +496,33 @@ start_server {
         assert {[r LPOS mylist b COUNT 10 RANK 5] eq {}}
     }
 
-    test {LPUSH, RPUSH, LLENGTH, LINDEX, LPOP - ziplist} {
-        # first lpush then rpush
-        assert_equal 1 [r lpush myziplist1 aa]
-        assert_equal 2 [r rpush myziplist1 bb]
-        assert_equal 3 [r rpush myziplist1 cc]
-        assert_equal 3 [r llen myziplist1]
-        assert_equal aa [r lindex myziplist1 0]
-        assert_equal bb [r lindex myziplist1 1]
-        assert_equal cc [r lindex myziplist1 2]
-        assert_equal {} [r lindex myziplist2 3]
-        assert_equal cc [r rpop myziplist1]
-        assert_equal aa [r lpop myziplist1]
-        assert_encoding quicklist myziplist1
-
-        # first rpush then lpush
-        assert_equal 1 [r rpush myziplist2 a]
-        assert_equal 2 [r lpush myziplist2 b]
-        assert_equal 3 [r lpush myziplist2 c]
-        assert_equal 3 [r llen myziplist2]
-        assert_equal c [r lindex myziplist2 0]
-        assert_equal b [r lindex myziplist2 1]
-        assert_equal a [r lindex myziplist2 2]
-        assert_equal {} [r lindex myziplist2 3]
-        assert_equal a [r rpop myziplist2]
-        assert_equal c [r lpop myziplist2]
-        assert_encoding quicklist myziplist2
-    }
-
-    test {LPUSH, RPUSH, LLENGTH, LINDEX, LPOP - regular list} {
+    test "LPUSH, RPUSH, LLENGTH, LINDEX, LPOP - $type" {
         # first lpush then rpush
-        assert_equal 1 [r lpush mylist1 $largevalue(linkedlist)]
-        assert_encoding quicklist mylist1
+        r del mylist1
+        assert_equal 1 [r lpush mylist1 $large]
+        assert_encoding $type mylist1
         assert_equal 2 [r rpush mylist1 b]
         assert_equal 3 [r rpush mylist1 c]
         assert_equal 3 [r llen mylist1]
-        assert_equal $largevalue(linkedlist) [r lindex mylist1 0]
+        assert_equal $large [r lindex mylist1 0]
         assert_equal b [r lindex mylist1 1]
         assert_equal c [r lindex mylist1 2]
         assert_equal {} [r lindex mylist1 3]
         assert_equal c [r rpop mylist1]
-        assert_equal $largevalue(linkedlist) [r lpop mylist1]
+        assert_equal $large [r lpop mylist1]
 
         # first rpush then lpush
-        assert_equal 1 [r rpush mylist2 $largevalue(linkedlist)]
-        assert_encoding quicklist mylist2
+        r del mylist2
+        assert_equal 1 [r rpush mylist2 $large]
         assert_equal 2 [r lpush mylist2 b]
         assert_equal 3 [r lpush mylist2 c]
+        assert_encoding $type mylist2
         assert_equal 3 [r llen mylist2]
         assert_equal c [r lindex mylist2 0]
         assert_equal b [r lindex mylist2 1]
-        assert_equal $largevalue(linkedlist) [r lindex mylist2 2]
+        assert_equal $large [r lindex mylist2 2]
         assert_equal {} [r lindex mylist2 3]
-        assert_equal $largevalue(linkedlist) [r rpop mylist2]
+        assert_equal $large [r rpop mylist2]
         assert_equal c [r lpop mylist2]
     }
 
@@ -544,15 +531,16 @@ start_server {
         assert_error {*wrong number of arguments for 'rpop' command} {r rpop key 2 2}
     }
 
-    test {RPOP/LPOP with the optional count argument} {
-        assert_equal 7 [r lpush listcount aa bb cc dd ee ff gg]
+    test "RPOP/LPOP with the optional count argument - $type" {
+        assert_equal 7 [r lpush listcount aa $large cc dd ee ff gg]
         assert_equal {gg} [r lpop listcount 1]
         assert_equal {ff ee} [r lpop listcount 2]
-        assert_equal {aa bb} [r rpop listcount 2]
+        assert_equal "aa $large" [r rpop listcount 2]
         assert_equal {cc} [r rpop listcount 1]
         assert_equal {dd} [r rpop listcount 123]
         assert_error "*ERR*range*" {r lpop forbarqaz -123}
     }
+}
 
     proc verify_resp_response {resp response resp2_response resp3_response} {
         if {$resp == 2} {
@@ -611,17 +599,11 @@ start_server {
         assert_equal 0 [r llen mylist2]
     }
 
-    proc create_list {key entries} {
-        r del $key
-        foreach entry $entries { r rpush $key $entry }
-        assert_encoding quicklist $key
-    }
-
     foreach {type large} [array get largevalue] {
     foreach {pop} {BLPOP BLMPOP_LEFT} {
         test "$pop: single existing list - $type" {
             set rd [redis_deferring_client]
-            create_list blist "a b $large c d"
+            create_$type blist "a b $large c d"
 
             bpop_command $rd $pop blist 1
             assert_equal {blist a} [$rd read]
@@ -647,8 +629,8 @@ start_server {
 
         test "$pop: multiple existing lists - $type" {
             set rd [redis_deferring_client]
-            create_list blist1{t} "a $large c"
-            create_list blist2{t} "d $large f"
+            create_$type blist1{t} "a $large c"
+            create_$type blist2{t} "d $large f"
 
             bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
             assert_equal {blist1{t} a} [$rd read]
@@ -677,7 +659,7 @@ start_server {
         test "$pop: second list has an entry - $type" {
             set rd [redis_deferring_client]
             r del blist1{t}
-            create_list blist2{t} "d $large f"
+            create_$type blist2{t} "d $large f"
 
             bpop_command_two_key $rd $pop blist1{t} blist2{t} 1
             assert_equal {blist2{t} d} [$rd read]
@@ -698,7 +680,7 @@ start_server {
             r rpush target{t} bar
 
             set rd [redis_deferring_client]
-            create_list blist{t} "a b $large c d"
+            create_$type blist{t} "a b $large c d"
 
             $rd brpoplpush blist{t} target{t} 1
             assert_equal d [$rd read]
@@ -715,7 +697,7 @@ start_server {
                     r rpush target{t} bar
 
                     set rd [redis_deferring_client]
-                    create_list blist{t} "a b $large c d"
+                    create_$type blist{t} "a b $large c d"
 
                     $rd blmove blist{t} target{t} $wherefrom $whereto 1
                     set poppedelement [$rd read]
@@ -1374,7 +1356,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
     foreach {type large} [array get largevalue] {
         test "LPUSHX, RPUSHX - $type" {
-            create_list xlist "$large c"
+            create_$type xlist "$large c"
             assert_equal 3 [r rpushx xlist d]
             assert_equal 4 [r lpushx xlist a]
             assert_equal 6 [r rpushx xlist 42 x]
@@ -1383,7 +1365,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         }
 
         test "LINSERT - $type" {
-            create_list xlist "a $large c d"
+            create_$type xlist "a $large c d"
             assert_equal 5 [r linsert xlist before c zz] "before c"
             assert_equal "a $large zz c d" [r lrange xlist 0 10] "lrangeA"
             assert_equal 6 [r linsert xlist after c yy] "after c"
@@ -1406,7 +1388,14 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         set e
     } {*ERR*syntax*error*}
 
-    foreach {type num} {quicklist 250 quicklist 500} {
+foreach type {listpack quicklist} {
+    foreach {num} {250 500} {
+        if {$type == "quicklist"} {
+            set origin_config [config_get_set list-max-listpack-size 5]
+        } else {
+            set origin_config [config_get_set list-max-listpack-size -1]
+        }
+
         proc check_numbered_list_consistency {key} {
             set len [r llen $key]
             for {set i 0} {$i < $len} {incr i} {
@@ -1444,7 +1433,10 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
             check_numbered_list_consistency mylist
             check_random_access_consistency mylist
         } {} {needs:debug}
+
+        config_set list-max-listpack-size $origin_config
     }
+}
 
     test {LLEN against non-list value error} {
         r del mylist
@@ -1475,12 +1467,14 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
     foreach {type large} [array get largevalue] {
         test "RPOPLPUSH base case - $type" {
             r del mylist1{t} mylist2{t}
-            create_list mylist1{t} "a $large c d"
+            create_$type mylist1{t} "a $large c d"
             assert_equal d [r rpoplpush mylist1{t} mylist2{t}]
             assert_equal c [r rpoplpush mylist1{t} mylist2{t}]
-            assert_equal "a $large" [r lrange mylist1{t} 0 -1]
-            assert_equal "c d" [r lrange mylist2{t} 0 -1]
-            assert_encoding quicklist mylist2{t}
+            assert_equal $large [r rpoplpush mylist1{t} mylist2{t}]
+            assert_equal "a" [r lrange mylist1{t} 0 -1]
+            assert_equal "$large c d" [r lrange mylist2{t} 0 -1]
+            assert_encoding listpack mylist1{t} ;# converted to listpack after shrinking
+            assert_encoding $type mylist2{t}
         }
 
         foreach wherefrom {left right} {
@@ -1489,9 +1483,9 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
                     r del mylist1{t} mylist2{t}
 
                     if {$wherefrom eq "right"} {
-                        create_list mylist1{t} "c d $large a"
+                        create_$type mylist1{t} "c d $large a"
                     } else {
-                        create_list mylist1{t} "a $large c d"
+                        create_$type mylist1{t} "a $large c d"
                     }
                     assert_equal a [r lmove mylist1{t} mylist2{t} $wherefrom $whereto]
                     assert_equal $large [r lmove mylist1{t} mylist2{t} $wherefrom $whereto]
@@ -1501,13 +1495,13 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
                     } else {
                         assert_equal "$large a" [r lrange mylist2{t} 0 -1]
                     }
-                    assert_encoding quicklist mylist2{t}
+                    assert_encoding $type mylist2{t}
                 }
             }
         }
 
         test "RPOPLPUSH with the same list as src and dst - $type" {
-            create_list mylist{t} "a $large c"
+            create_$type mylist{t} "a $large c"
             assert_equal "a $large c" [r lrange mylist{t} 0 -1]
             assert_equal c [r rpoplpush mylist{t} mylist{t}]
             assert_equal "c a $large" [r lrange mylist{t} 0 -1]
@@ -1517,10 +1511,10 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
             foreach whereto {left right} {
                 test "LMOVE $wherefrom $whereto with the same list as src and dst - $type" {
                     if {$wherefrom eq "right"} {
-                        create_list mylist{t} "a $large c"
+                        create_$type mylist{t} "a $large c"
                         assert_equal "a $large c" [r lrange mylist{t} 0 -1]
                     } else {
-                        create_list mylist{t} "c a $large"
+                        create_$type mylist{t} "c a $large"
                         assert_equal "c a $large" [r lrange mylist{t} 0 -1]
                     }
                     assert_equal c [r lmove mylist{t} mylist{t} $wherefrom $whereto]
@@ -1535,8 +1529,8 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
         foreach {othertype otherlarge} [array get largevalue] {
             test "RPOPLPUSH with $type source and existing target $othertype" {
-                create_list srclist{t} "a b c $large"
-                create_list dstlist{t} "$otherlarge"
+                create_$type srclist{t} "a b c $large"
+                create_$othertype dstlist{t} "$otherlarge"
                 assert_equal $large [r rpoplpush srclist{t} dstlist{t}]
                 assert_equal c [r rpoplpush srclist{t} dstlist{t}]
                 assert_equal "a b" [r lrange srclist{t} 0 -1]
@@ -1544,7 +1538,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
                 # When we rpoplpush'ed a large value, dstlist should be
                 # converted to the same encoding as srclist.
-                if {$type eq "linkedlist"} {
+                if {$type eq "quicklist"} {
                     assert_encoding quicklist dstlist{t}
                 }
             }
@@ -1552,12 +1546,12 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
             foreach wherefrom {left right} {
                 foreach whereto {left right} {
                     test "LMOVE $wherefrom $whereto with $type source and existing target $othertype" {
-                        create_list dstlist{t} "$otherlarge"
+                        create_$othertype dstlist{t} "$otherlarge"
 
                         if {$wherefrom eq "right"} {
-                            create_list srclist{t} "a b c $large"
+                            create_$type srclist{t} "a b c $large"
                         } else {
-                            create_list srclist{t} "$large c a b"
+                            create_$type srclist{t} "$large c a b"
                         }
                         assert_equal $large [r lmove srclist{t} dstlist{t} $wherefrom $whereto]
                         assert_equal c [r lmove srclist{t} dstlist{t} $wherefrom $whereto]
@@ -1571,7 +1565,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
                         # When we lmoved a large value, dstlist should be
                         # converted to the same encoding as srclist.
-                        if {$type eq "linkedlist"} {
+                        if {$type eq "quicklist"} {
                             assert_encoding quicklist dstlist{t}
                         }
                     }
@@ -1595,28 +1589,30 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         assert_equal 0 [r exists newlist{t}]
     }
 
-    test {RPOPLPUSH against non list dst key} {
-        create_list srclist{t} {a b c d}
+foreach {type large} [array get largevalue] {
+    test "RPOPLPUSH against non list dst key - $type" {
+        create_$type srclist{t} "a $large c d"
         r set dstlist{t} x
         assert_error WRONGTYPE* {r rpoplpush srclist{t} dstlist{t}}
         assert_type string dstlist{t}
-        assert_equal {a b c d} [r lrange srclist{t} 0 -1]
+        assert_equal "a $large c d" [r lrange srclist{t} 0 -1]
     }
+}
 
     test {RPOPLPUSH against non existing src key} {
         r del srclist{t} dstlist{t}
         assert_equal {} [r rpoplpush srclist{t} dstlist{t}]
     } {}
 
-    foreach {type large} [array get largevalue] {
+    foreach {type large} [array get largevalue] { 
         test "Basic LPOP/RPOP/LMPOP - $type" {
-            create_list mylist "$large 1 2"
+            create_$type mylist "$large 1 2"
             assert_equal $large [r lpop mylist]
             assert_equal 2 [r rpop mylist]
             assert_equal 1 [r lpop mylist]
             assert_equal 0 [r llen mylist]
 
-            create_list mylist "$large 1 2"
+            create_$type mylist "$large 1 2"
             assert_equal "mylist $large" [r lmpop 1 mylist left count 1]
             assert_equal {mylist {2 1}} [r lmpop 2 mylist mylist right count 2]
         }
@@ -1651,11 +1647,14 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         assert_error "WRONGTYPE*" {r blmpop 0 2 notalist{t} notalist2{t} left count 1}
     }
 
-    foreach {type num} {quicklist 250 quicklist 500} {
+    foreach {num} {250 500} {
         test "Mass RPOP/LPOP - $type" {
             r del mylist
             set sum1 0
             for {set i 0} {$i < $num} {incr i} {
+                if {$i == [expr $num/2]} {
+                    r lpush mylist $large
+                }
                 r lpush mylist $i
                 incr sum1 $i
             }
@@ -1691,56 +1690,58 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         assert_error "ERR count*" {r lmpop 2 mylist{t} mylist2{t} RIGHT COUNT -1}
     }
 
-    test {LMPOP single existing list} {
+foreach {type large} [array get largevalue] {
+    test "LMPOP single existing list - $type" {
         # Same key multiple times.
-        create_list mylist{t} "a b c d e f"
+        create_$type mylist{t} "a b $large d e f"
         assert_equal {mylist{t} {a b}} [r lmpop 2 mylist{t} mylist{t} left count 2]
         assert_equal {mylist{t} {f e}} [r lmpop 2 mylist{t} mylist{t} right count 2]
         assert_equal 2 [r llen mylist{t}]
 
         # First one exists, second one does not exist.
-        create_list mylist{t} "a b c d e"
+        create_$type mylist{t} "a b $large d e"
         r del mylist2{t}
         assert_equal {mylist{t} a} [r lmpop 2 mylist{t} mylist2{t} left count 1]
         assert_equal 4 [r llen mylist{t}]
-        assert_equal {mylist{t} {e d c b}} [r lmpop 2 mylist{t} mylist2{t} right count 10]
+        assert_equal "mylist{t} {e d $large b}" [r lmpop 2 mylist{t} mylist2{t} right count 10]
         assert_equal {} [r lmpop 2 mylist{t} mylist2{t} right count 1]
 
         # First one does not exist, second one exists.
         r del mylist{t}
-        create_list mylist2{t} "1 2 3 4 5"
+        create_$type mylist2{t} "1 2 $large 4 5"
         assert_equal {mylist2{t} 5} [r lmpop 2 mylist{t} mylist2{t} right count 1]
         assert_equal 4 [r llen mylist2{t}]
-        assert_equal {mylist2{t} {1 2 3 4}} [r lmpop 2 mylist{t} mylist2{t} left count 10]
+        assert_equal "mylist2{t} {1 2 $large 4}" [r lmpop 2 mylist{t} mylist2{t} left count 10]
 
         assert_equal 0 [r exists mylist{t} mylist2{t}]
     }
 
-    test {LMPOP multiple existing lists} {
-        create_list mylist{t} "a b c d e"
-        create_list mylist2{t} "1 2 3 4 5"
+    test "LMPOP multiple existing lists - $type" {
+        create_$type mylist{t} "a b $large d e"
+        create_$type mylist2{t} "1 2 $large 4 5"
 
         # Pop up from the first key.
         assert_equal {mylist{t} {a b}} [r lmpop 2 mylist{t} mylist2{t} left count 2]
         assert_equal 3 [r llen mylist{t}]
-        assert_equal {mylist{t} {e d c}} [r lmpop 2 mylist{t} mylist2{t} right count 3]
+        assert_equal "mylist{t} {e d $large}" [r lmpop 2 mylist{t} mylist2{t} right count 3]
         assert_equal 0 [r exists mylist{t}]
 
         # Pop up from the second key.
-        assert_equal {mylist2{t} {1 2 3}} [r lmpop 2 mylist{t} mylist2{t} left count 3]
+        assert_equal "mylist2{t} {1 2 $large}" [r lmpop 2 mylist{t} mylist2{t} left count 3]
         assert_equal 2 [r llen mylist2{t}]
         assert_equal {mylist2{t} {5 4}} [r lmpop 2 mylist{t} mylist2{t} right count 2]
         assert_equal 0 [r exists mylist{t}]
 
         # Pop up all elements.
-        create_list mylist{t} "a b c"
-        create_list mylist2{t} "1 2 3"
-        assert_equal {mylist{t} {a b c}} [r lmpop 2 mylist{t} mylist2{t} left count 10]
+        create_$type mylist{t} "a $large c"
+        create_$type mylist2{t} "1 $large 3"
+        assert_equal "mylist{t} {a $large c}" [r lmpop 2 mylist{t} mylist2{t} left count 10]
         assert_equal 0 [r llen mylist{t}]
-        assert_equal {mylist2{t} {3 2 1}} [r lmpop 2 mylist{t} mylist2{t} right count 10]
+        assert_equal "mylist2{t} {3 $large 1}" [r lmpop 2 mylist{t} mylist2{t} right count 10]
         assert_equal 0 [r llen mylist2{t}]
         assert_equal 0 [r exists mylist{t} mylist2{t}]
     }
+}
 
     test {LMPOP propagate as pop with count command to replica} {
         set repl [attach_to_replication_stream]
@@ -1788,24 +1789,24 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
     foreach {type large} [array get largevalue] {
         test "LRANGE basics - $type" {
-            create_list mylist "$large 1 2 3 4 5 6 7 8 9"
+            create_$type mylist "$large 1 2 3 4 5 6 7 8 9"
             assert_equal {1 2 3 4 5 6 7 8} [r lrange mylist 1 -2]
             assert_equal {7 8 9} [r lrange mylist -3 -1]
             assert_equal {4} [r lrange mylist 4 4]
         }
 
         test "LRANGE inverted indexes - $type" {
-            create_list mylist "$large 1 2 3 4 5 6 7 8 9"
+            create_$type mylist "$large 1 2 3 4 5 6 7 8 9"
             assert_equal {} [r lrange mylist 6 2]
         }
 
         test "LRANGE out of range indexes including the full list - $type" {
-            create_list mylist "$large 1 2 3"
+            create_$type mylist "$large 1 2 3"
             assert_equal "$large 1 2 3" [r lrange mylist -1000 1000]
         }
 
         test "LRANGE out of range negative end index - $type" {
-            create_list mylist "$large 1 2 3"
+            create_$type mylist "$large 1 2 3"
             assert_equal $large [r lrange mylist 0 -4]
             assert_equal {} [r lrange mylist 0 -5]
         }
@@ -1816,7 +1817,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
     }
 
     test {LRANGE with start > end yields an empty array for backward compatibility} {
-        create_list mylist "1 2 3"
+        create_$type mylist "1 $large 3"
         assert_equal {} [r lrange mylist 1 0]
         assert_equal {} [r lrange mylist -1 -2]
     }
@@ -1825,7 +1826,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         proc trim_list {type min max} {
             upvar 1 large large
             r del mylist
-            create_list mylist "1 2 3 4 $large"
+            create_$type mylist "1 2 3 4 $large"
             r ltrim mylist $min $max
             r lrange mylist 0 -1
         }
@@ -1850,11 +1851,8 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
             assert_equal {} [trim_list $type 0 -6]
         }
 
-    }
-
-    foreach {type large} [array get largevalue] {
         test "LSET - $type" {
-            create_list mylist "99 98 $large 96 95"
+            create_$type mylist "99 98 $large 96 95"
             r lset mylist 1 foo
             r lset mylist -1 bar
             assert_equal "99 foo $large 96 bar" [r lrange mylist 0 -1]
@@ -1876,7 +1874,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
 
     foreach {type e} [array get largevalue] {
         test "LREM remove all the occurrences - $type" {
-            create_list mylist "$e foo bar foobar foobared zap bar test foo"
+            create_$type mylist "$e foo bar foobar foobared zap bar test foo"
             assert_equal 2 [r lrem mylist 0 bar]
             assert_equal "$e foo foobar foobared zap test foo" [r lrange mylist 0 -1]
         }
@@ -1892,7 +1890,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         }
 
         test "LREM starting from tail with negative count - $type" {
-            create_list mylist "$e foo bar foobar foobared zap bar test foo foo"
+            create_$type mylist "$e foo bar foobar foobared zap bar test foo foo"
             assert_equal 1 [r lrem mylist -1 bar]
             assert_equal "$e foo bar foobar foobared zap test foo foo" [r lrange mylist 0 -1]
         }
@@ -1903,7 +1901,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
         }
 
         test "LREM deleting objects that may be int encoded - $type" {
-            create_list myotherlist "$e 1 2 3"
+            create_$type myotherlist "$e 1 2 3"
             assert_equal 1 [r lrem myotherlist 1 2]
             assert_equal 3 [r llen myotherlist]
         }
@@ -1988,7 +1986,124 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
     }
 }
 
-    test {List ziplist of various encodings} {
+    foreach {max_lp_size large} "3 $largevalue(listpack) -1 $largevalue(quicklist)" {
+        test "List listpack -> quicklist encoding conversion" {
+            set origin_conf [config_get_set list-max-listpack-size $max_lp_size]
+
+            # RPUSH
+            create_listpack lst "a b c"
+            r RPUSH lst $large
+            assert_encoding quicklist lst
+
+            # LINSERT
+            create_listpack lst "a b c"
+            r LINSERT lst after b $large
+            assert_encoding quicklist lst
+
+            # LSET
+            create_listpack lst "a b c"
+            r LSET lst 0 $large
+            assert_encoding quicklist lst
+
+            # LMOVE
+            create_quicklist lsrc{t} "a b c $large"
+            create_listpack ldes{t} "d e f"
+            r LMOVE lsrc{t} ldes{t} right right
+            assert_encoding quicklist ldes{t}
+
+            r config set list-max-listpack-size $origin_conf
+        }
+    }
+
+    test "List quicklist -> listpack encoding conversion" {
+        set origin_conf [config_get_set list-max-listpack-size 3]
+
+        # RPOP
+        create_quicklist lst "a b c d"
+        r RPOP lst 3
+        assert_encoding listpack lst
+
+        # LREM
+        create_quicklist lst "a a a d"
+        r LREM lst 3 a
+        assert_encoding listpack lst
+
+        # LTRIM
+        create_quicklist lst "a b c d"
+        r LTRIM lst 1 1
+        assert_encoding listpack lst
+
+        r config set list-max-listpack-size -1
+
+        # RPOP
+        create_quicklist lst "a b c $largevalue(quicklist)"
+        r RPOP lst 1
+        assert_encoding listpack lst
+
+        # LREM
+        create_quicklist lst "a $largevalue(quicklist)"
+        r LREM lst 1 $largevalue(quicklist)
+        assert_encoding listpack lst
+
+        # LTRIM
+        create_quicklist lst "a b $largevalue(quicklist)"
+        r LTRIM lst 0 1
+        assert_encoding listpack lst
+
+        # LSET
+        create_quicklist lst "$largevalue(quicklist) a b"
+        r RPOP lst 2
+        assert_encoding quicklist lst
+        r LSET lst -1 c
+        assert_encoding listpack lst
+
+        r config set list-max-listpack-size $origin_conf
+    }
+
+    test "List encoding conversion when RDB loading" {
+        set origin_conf [config_get_set list-max-listpack-size 3]
+        create_listpack lst "a b c"
+
+        # list is still a listpack after DEBUG RELOAD
+        r DEBUG RELOAD
+        assert_encoding listpack lst
+
+        # list is still a quicklist after DEBUG RELOAD
+        r RPUSH lst d
+        r DEBUG RELOAD
+        assert_encoding quicklist lst
+
+        # when a quicklist has only one packed node, it will be
+        # converted to listpack during rdb loading
+        r RPOP lst
+        assert_encoding quicklist lst 
+        r DEBUG RELOAD
+        assert_encoding listpack lst
+
+        r config set list-max-listpack-size $origin_conf
+    } {OK} {needs:debug}
+
+    test "List invalid list-max-listpack-size config" {
+        # ​When list-max-listpack-size is 0 we treat it as 1 and it'll
+        # still be listpack if there's a single element in the list.
+        r config set list-max-listpack-size 0
+        r DEL lst
+        r RPUSH lst a
+        assert_encoding listpack lst
+        r RPUSH lst b
+        assert_encoding quicklist lst
+
+        # When list-max-listpack-size < -5 we treat it as -5.
+        r config set list-max-listpack-size -6
+        r DEL lst
+        r RPUSH lst [string repeat "x" 60000]
+        assert_encoding listpack lst
+        # Converted to quicklist when the size of listpack exceed 65536
+        r RPUSH lst [string repeat "x" 5536]
+        assert_encoding quicklist lst
+    }
+
+    test "List of various encodings" {
         r del k
         r lpush k 127 ;# ZIP_INT_8B
         r lpush k 32767 ;# ZIP_INT_16B
@@ -1999,11 +2114,16 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
         r lpush k [string repeat x 31] ;# ZIP_STR_06B
         r lpush k [string repeat x 8191] ;# ZIP_STR_14B
         r lpush k [string repeat x 65535] ;# ZIP_STR_32B
+        assert_encoding quicklist k ;# exceeds the size limit of quicklist node
         set k [r lrange k 0 -1]
         set dump [r dump k]
 
+        # coverage for objectComputeSize
+        assert_morethan [memory_usage k] 0
+
         config_set sanitize-dump-payload no mayfail
-        r restore kk 0 $dump
+        r restore kk 0 $dump replace
+        assert_encoding quicklist kk
         set kk [r lrange kk 0 -1]
 
         # try some forward and backward searches to make sure all encodings
@@ -2021,9 +2141,10 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
         set _ $k
     } {12 0 9223372036854775808 2147483647 32767 127}
 
-    test {List ziplist of various encodings - sanitize dump} {
+    test "List of various encodings - sanitize dump" {
         config_set sanitize-dump-payload yes mayfail
         r restore kk 0 $dump replace
+        assert_encoding quicklist kk
         set k [r lrange k 0 -1]
         set kk [r lrange kk 0 -1]
 
@@ -2034,4 +2155,4 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
         assert_equal [lpop k] [string repeat x 31]
         set _ $k
     } {12 0 9223372036854775808 2147483647 32767 127}
-}
+} ;# stop servers