Ensure correct compression for a node that was too small to compress but grows larger

src/quicklist.c

@@ -180,6 +180,7 @@ REDIS_STATIC quicklistNode *quicklistCreateNode(void) {
  node->container = QUICKLIST_NODE_CONTAINER_PACKED;
  node->recompress = 0;
  node->dont_compress = 0;
+ node->attempted_compress = 0;
  return node;
 }
 
@@ -212,10 +213,8 @@ void quicklistRelease(quicklist *quicklist) {
  * Returns 1 if listpack compressed successfully.
  * Returns 0 if compression failed or if listpack too small to compress. */
 REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
-#ifdef REDIS_TEST
- node->attempted_compress = 1;
-#endif
  if (node->dont_compress) return 0;
+ node->attempted_compress = 1;
 
  /* validate that the node is neither
  * tail nor head (it has prev and next)*/
@@ -240,6 +239,7 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
  zfree(node->entry);
  node->entry = (unsigned char *)lzf;
  node->encoding = QUICKLIST_NODE_ENCODING_LZF;
+ node->attempted_compress = 0;
  return 1;
 }
 
@@ -254,9 +254,7 @@ REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
 /* Uncompress the listpack in 'node' and update encoding details.
  * Returns 1 on successful decode, 0 on failure to decode. */
 REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
-#ifdef REDIS_TEST
  node->attempted_compress = 0;
-#endif
  node->recompress = 0;
 
  void *decompressed = zmalloc(node->sz);
@@ -275,17 +273,26 @@ REDIS_STATIC int __quicklistDecompressNode(quicklistNode *node) {
 /* Decompress only compressed nodes. */
 #define quicklistDecompressNode(_node) \
  do { \
- if ((_node) && (_node)->encoding == QUICKLIST_NODE_ENCODING_LZF) { \
- __quicklistDecompressNode((_node)); \
+ if ((_node)) { \
+ if ((_node)->encoding == QUICKLIST_NODE_ENCODING_LZF) { \
+ __quicklistDecompressNode((_node)); \
+ } \
+ (_node)->attempted_compress = 0; \
  } \
  } while (0)
 
-/* Force node to not be immediately re-compressible */
+/* Force node to be immediately re-compressible. If the node was too small to
+ * compress before, we also allow it to be immediately re-compressible. */
 #define quicklistDecompressNodeForUse(_node) \
  do { \
- if ((_node) && (_node)->encoding == QUICKLIST_NODE_ENCODING_LZF) { \
- __quicklistDecompressNode((_node)); \
- (_node)->recompress = 1; \
+ if ((_node)) { \
+ if ((_node)->encoding == QUICKLIST_NODE_ENCODING_LZF) { \
+ __quicklistDecompressNode((_node)); \
+ (_node)->recompress = 1; \
+ } else if ((_node)->attempted_compress) { \
+ (_node)->recompress = 1; \
+ (_node)->attempted_compress = 0; \
+ } \
  } \
  } while (0)
 
@@ -3259,6 +3266,34 @@ int quicklistTest(int argc, char *argv[], int flags) {
  quicklistRelease(ql);
  }
 
+ TEST("Test that a small node is compressed correctly when its size increases enough") {
+ quicklistIter *iter;
+ quicklistEntry entry;
+ quicklist *ql = quicklistNew(2, 1);
+ quicklistPushTail(ql, "0", 1);
+ quicklistPushTail(ql, "1", 1);
+ quicklistPushTail(ql, "3", 1);
+ quicklistPushTail(ql, "4", 1);
+
+ /* Create a compression failed node in the middle. */
+ iter = quicklistGetIteratorEntryAtIdx(ql, 1, &entry);
+ quicklistInsertAfter(iter, &entry, "2", 1);
+ quicklistReleaseIterator(iter);
+ assert(ql->head->next->attempted_compress == 1); /* too small to compress */
+
+ /* Add a large element to the node that failed to compress
+ * due to its small size, then ensure it can be compressed. */
+ size_t sz = 1024;
+ unsigned char *s = zmalloc(sz);
+ memset(s, 'a', sz);
+ iter = quicklistGetIteratorEntryAtIdx(ql, 2, &entry);
+ quicklistInsertAfter(iter, &entry, s, sz);
+ quicklistReleaseIterator(iter);
+ assert(ql->head->next->encoding == QUICKLIST_NODE_ENCODING_LZF);
+ zfree(s);
+ quicklistRelease(ql);
+ }
+
  if (flags & REDIS_TEST_LARGE_MEMORY) {
  TEST("compress and decompress quicklist listpack node") {
  quicklistNode *node = quicklistCreateNode();

tests/unit/type/list-3.tcl

@@ -119,6 +119,32 @@ start_server {
  r ping
  }
 
+ test {A node which is too small to compress, is correctly compressed when its size increases sufficiently} {
+ set orig_depth [config_get_set list-compress-depth 1]
+ set orig_fill [config_get_set list-max-listpack-size 2]
+
+ # Creates a node in the middle of the list that should
+ # be compressed, but is too small to be compressed.
+ r del key
+ r rpush key a b d e
+ r linsert key after b c
+ # Wait for the log message that confirms the node is too small to be compressed
+ set loglines [count_log_lines 0]
+ r debug quicklist key
+ wait_for_log_messages 0 {"*container : PACKED, encoding: RAW, size: 10, count: 1, recompress: 0, attempted_compress: 1*"} $loglines 1000 10
+
+ # Add a element to the failed node to make it big enough,
+ # then make sure it can be compressed correctly.
+ r linsert key after c [string repeat "x" 1000]
+ # Wait for the log message that confirms the node is now compressed correctly
+ set loglines [count_log_lines 0]
+ r debug quicklist key
+ wait_for_log_messages 0 {"*container : PACKED, encoding: LZF, size: 1014, count: 2, recompress: 0, attempted_compress: 0*"} $loglines 1000 10
+
+ config_set list-compress-depth $orig_depth
+ config_set list-max-listpack-size $orig_fill
+ } {} {needs:debug external:skip}
+
 foreach comp {2 1 0} {
  set cycles 1000
  if {$::accurate} { set cycles 10000 }