Allow compression of interior quicklist nodes

Let user set how many nodes to *not* compress.

We can specify a compression "depth" of how many nodes
to leave uncompressed on each end of the quicklist.

Depth 0 = disable compression.
Depth 1 = only leave head/tail uncompressed.
  - (read as: "skip 1 node on each end of the list before compressing")
Depth 2 = leave head, head->next, tail->prev, tail uncompressed.
  - ("skip 2 nodes on each end of the list before compressing")
Depth 3 = Depth 2 + head->next->next + tail->prev->prev
  - ("skip 3 nodes...")
etc.

This also:
  - updates RDB storage to use native quicklist compression (if node is
    already compressed) instead of uncompressing, generating the RDB string,
    then re-compressing the quicklist node.
  - internalizes the "fill" parameter for the quicklist so we don't
    need to pass it to _every_ function.  Now it's just a property of
    the list.
  - allows a runtime-configurable compression option, so we can
    expose a compresion parameter in the configuration file if people
    want to trade slight request-per-second performance for up to 90%+
    memory savings in some situations.
  - updates the quicklist tests to do multiple passes: 200k+ tests now.

1 files changed, 66 insertions, 17 deletions

diff --git a/src/quicklist.h b/src/quicklist.h
index 6f21f789d..5c9530ccd 100644
--- a/src/quicklist.h
+++ b/src/quicklist.h
@@ -33,19 +33,50 @@
 
 /* Node, quicklist, and Iterator are the only data structures used currently. */
 
+/* quicklistNode is a 32 byte struct describing a ziplist for a quicklist.
+ * We use bit fields keep the quicklistNode at 32 bytes.
+ * count: 16 bits, max 65536 (max zl bytes is 65k, so max count actually < 32k).
+ * encoding: 2 bits, RAW=1, LZF=2.
+ * container: 2 bits, NONE=1, ZIPLIST=2.
+ * recompress: 1 bit, bool, true if node is temporarry decompressed for usage.
+ * attempted_compress: 1 bit, boolean, used for verifying during testing.
+ * extra: 12 bits, free for future use; pads out the remainder of 32 bits */
 typedef struct quicklistNode {
     struct quicklistNode *prev;
     struct quicklistNode *next;
     unsigned char *zl;
-    unsigned int count; /* cached count of items in ziplist */
-    unsigned int sz;    /* ziplist size in bytes */
+    unsigned int sz;             /* ziplist size in bytes */
+    unsigned int count : 16;     /* count of items in ziplist */
+    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
+    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
+    unsigned int recompress : 1; /* was this node previous compressed? */
+    unsigned int attempted_compress : 1; /* node can't compress; too small */
+    unsigned int extra : 10; /* more bits to steal for future usage */
 } quicklistNode;
 
+/* quicklistLZF is a 4+N byte struct holding 'sz' followed by 'compressed'.
+ * 'sz' is byte length of 'compressed' field.
+ * 'compressed' is LZF data with total (compressed) length 'sz'
+ * NOTE: uncompressed length is stored in quicklistNode->sz.
+ * When quicklistNode->zl is compressed, node->zl points to a quicklistLZF */
+typedef struct quicklistLZF {
+    unsigned int sz; /* LZF size in bytes*/
+    char compressed[];
+} quicklistLZF;
+
+/* quicklist is a 32 byte struct (on 64-bit systems) describing a quicklist.
+ * 'count' is the number of total entries.
+ * 'len' is the number of quicklist nodes.
+ * 'compress' is: -1 if compression disabled, otherwise it's the number
+ *                of quicklistNodes to leave uncompressed at ends of quicklist.
+ * 'fill' is the user-requested (or default) fill factor. */
 typedef struct quicklist {
     quicklistNode *head;
     quicklistNode *tail;
-    unsigned long len;   /* number of quicklistNodes */
-    unsigned long count; /* total count of all entries in all ziplists */
+    unsigned long count;        /* total count of all entries in all ziplists */
+    unsigned int len;           /* number of quicklistNodes */
+    int fill : 16;              /* fill factor for individual nodes */
+    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
 } quicklist;
 
 typedef struct quicklistIter {
@@ -69,23 +100,40 @@ typedef struct quicklistEntry {
 #define QUICKLIST_HEAD 0
 #define QUICKLIST_TAIL -1
 
+/* quicklist node encodings */
+#define QUICKLIST_NODE_ENCODING_RAW 1
+#define QUICKLIST_NODE_ENCODING_LZF 2
+
+/* quicklist compression disable */
+#define QUICKLIST_NOCOMPRESS 0
+
+/* quicklist container formats */
+#define QUICKLIST_NODE_CONTAINER_NONE 1
+#define QUICKLIST_NODE_CONTAINER_ZIPLIST 2
+
+#define quicklistNodeIsCompressed(node)                                        \
+    ((node)->encoding == QUICKLIST_NODE_ENCODING_LZF)
+
 /* Prototypes */
 quicklist *quicklistCreate(void);
+quicklist *quicklistNew(int fill, int compress);
+void quicklistSetCompressDepth(quicklist *quicklist, int depth);
+void quicklistSetFill(quicklist *quicklist, int fill);
+void quicklistSetOptions(quicklist *quicklist, int fill, int depth);
 void quicklistRelease(quicklist *quicklist);
-quicklist *quicklistPushHead(quicklist *quicklist, const int fill, void *value,
-                             const size_t sz);
-quicklist *quicklistPushTail(quicklist *quicklist, const int fill, void *value,
-                             const size_t sz);
-void quicklistPush(quicklist *quicklist, const int fill, void *value,
-                   const size_t sz, int where);
+int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz);
+int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);
+void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
+                   int where);
 void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);
 quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist,
-                                            const int fill, unsigned char *zl);
-quicklist *quicklistCreateFromZiplist(int fill, unsigned char *zl);
-void quicklistInsertAfter(quicklist *quicklist, const int fill,
-                          quicklistEntry *node, void *value, const size_t sz);
-void quicklistInsertBefore(quicklist *quicklist, const int fill,
-                           quicklistEntry *node, void *value, const size_t sz);
+                                            unsigned char *zl);
+quicklist *quicklistCreateFromZiplist(int fill, int compress,
+                                      unsigned char *zl);
+void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node,
+                          void *value, const size_t sz);
+void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node,
+                           void *value, const size_t sz);
 void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);
 int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                             int sz);
@@ -100,7 +148,7 @@ int quicklistIndex(const quicklist *quicklist, const long long index,
                    quicklistEntry *entry);
 void quicklistRewind(quicklist *quicklist, quicklistIter *li);
 void quicklistRewindTail(quicklist *quicklist, quicklistIter *li);
-void quicklistRotate(quicklist *quicklist, const int fill);
+void quicklistRotate(quicklist *quicklist);
 int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                        unsigned int *sz, long long *sval,
                        void *(*saver)(unsigned char *data, unsigned int sz));
@@ -108,6 +156,7 @@ int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                  unsigned int *sz, long long *slong);
 unsigned int quicklistCount(quicklist *ql);
 int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);
+size_t quicklistGetLzf(const quicklistNode *node, void **data);
 
 #ifdef REDIS_TEST
 int quicklistTest(int argc, char *argv[]);