feat: support lz4 compression method (#4610)

The feature is enabled via set_method call but since no code calls it
the lz4 compression is still disabled in dragonfly.

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

.clang-tidy

@@ -81,3 +81,4 @@ Checks: >
 # modernize-use-nullptr,
 # modernize-use-equals-default,
 # readability-qualified-auto,
+cppcoreguidelines-narrowing-conversions.WarnOnIntegerNarrowingConversion: 'false'

src/core/CMakeLists.txt

@@ -10,7 +10,7 @@ add_library(dfly_core allocation_tracker.cc bloom.cc compact_object.cc dense_set
     tx_queue.cc string_set.cc string_map.cc detail/bitpacking.cc)
 
 cxx_link(dfly_core base absl::flat_hash_map absl::str_format redis_lib TRDP::lua lua_modules
-    fibers2 ${SEARCH_LIB} jsonpath OpenSSL::Crypto TRDP::dconv)
+    fibers2 ${SEARCH_LIB} jsonpath OpenSSL::Crypto TRDP::dconv TRDP::lz4)
 
 add_executable(dash_bench dash_bench.cc)
 cxx_link(dash_bench dfly_core redis_test_lib)

src/core/qlist.cc

@@ -14,6 +14,7 @@ extern "C" {
 #include <absl/base/optimization.h>
 #include <absl/strings/escaping.h>
 #include <absl/strings/str_cat.h>
+#include <lz4frame.h>
 
 #include "base/logging.h"
 
@@ -33,12 +34,12 @@ using namespace std;
 #define SIZE_ESTIMATE_OVERHEAD 8
 
 /* Minimum listpack size in bytes for attempting compression. */
-#define MIN_COMPRESS_BYTES 48
+#define MIN_COMPRESS_BYTES 256
 
 /* Minimum size reduction in bytes to store compressed quicklistNode data.
  * This also prevents us from storing compression if the compression
  * resulted in a larger size than the original data. */
-#define MIN_COMPRESS_IMPROVE 8
+#define MIN_COMPRESS_IMPROVE 32
 
 /* This macro is used to compress a node.
  *
@@ -49,19 +50,22 @@ using namespace std;
  *
  * If the 'recompress' flag of the node is false, we check whether the node is
  * within the range of compress depth before compressing it. */
-#define quicklistCompress(_node) \
-  do {                           \
-    if ((_node)->recompress)     \
-      CompressNode((_node));     \
-    else                         \
-      this->Compress(_node);     \
+#define quicklistCompress(_node)                  \
+  do {                                            \
+    if ((_node)->recompress)                      \
+      CompressNode((_node), this->compr_method_); \
+    else                                          \
+      this->Compress(_node);                      \
   } while (0)
 
+#define QLIST_NODE_ENCODING_LZ4 3
+
 namespace dfly {
 
 namespace {
 
 static_assert(sizeof(QList) == 32);
+static_assert(sizeof(QList::Node) == 40);
 
 enum IterDir : uint8_t { FWD = 1, REV = 0 };
 
@@ -181,22 +185,13 @@ inline ssize_t NodeSetEntry(QList::Node* node, uint8_t* entry) {
   return diff;
 }
 
-/* Compress the listpack in 'node' and update encoding details.
- * Returns true if listpack compressed successfully.
- * Returns false if compression failed or if listpack too small to compress. */
-bool CompressNode(QList::Node* node) {
-  DCHECK(node->encoding == QUICKLIST_NODE_ENCODING_RAW);
-  DCHECK(!node->dont_compress);
-
-  /* validate that the node is neither
-   * tail nor head (it has prev and next)*/
-  DCHECK(node->prev && node->next);
-
-  node->recompress = 0;
-  /* Don't bother compressing small values */
-  if (node->sz < MIN_COMPRESS_BYTES)
-    return false;
+inline quicklistLZF* GetLzf(QList::Node* node) {
+  DCHECK(node->encoding == QUICKLIST_NODE_ENCODING_LZF ||
+         node->encoding == QLIST_NODE_ENCODING_LZ4);
+  return (quicklistLZF*)node->entry;
+}
 
+bool CompressLZF(QList::Node* node) {
   // We allocate LZF_STATE on heap, piggy-backing on the existing allocation.
   char* uptr = (char*)zmalloc(sizeof(quicklistLZF) + node->sz + sizeof(LZF_STATE));
   quicklistLZF* lzf = (quicklistLZF*)uptr;
@@ -208,10 +203,12 @@ bool CompressNode(QList::Node* node) {
     /* lzf_compress aborts/rejects compression if value not compressible. */
     DVLOG(2) << "Uncompressable " << node->sz << " vs " << lzf->sz;
     zfree(lzf);
-
+    QList::stats.bad_compression_attempts++;
     return false;
   }
   DVLOG(2) << "Compressed " << node->sz << " to " << lzf->sz;
+  QList::stats.compressed_bytes += lzf->sz;
+  QList::stats.raw_compressed_bytes += node->sz;
 
   lzf = (quicklistLZF*)zrealloc(lzf, sizeof(*lzf) + lzf->sz);
   zfree(node->entry);
@@ -220,26 +217,109 @@ bool CompressNode(QList::Node* node) {
   return true;
 }
 
-ssize_t CompressNodeIfNeeded(QList::Node* node) {
+bool CompressLZ4(QList::Node* node) {
+  LZ4F_cctx* cntx;
+  LZ4F_errorCode_t code = LZ4F_createCompressionContext(&cntx, LZ4F_VERSION);
+  CHECK(!LZ4F_isError(code));
+
+  LZ4F_preferences_t lz4_pref = LZ4F_INIT_PREFERENCES;
+  lz4_pref.compressionLevel = -1;
+  lz4_pref.frameInfo.contentSize = node->sz;
+  size_t buf_size = LZ4F_compressFrameBound(node->sz, &lz4_pref);
+
+  // We reuse quicklistLZF struct for LZ4 metadata.
+  quicklistLZF* dest = (quicklistLZF*)zmalloc(sizeof(quicklistLZF) + buf_size);
+  size_t compr_sz = LZ4F_compressFrame_usingCDict(cntx, dest->compressed, buf_size, node->entry,
+                                                  node->sz, nullptr /* dict */, &lz4_pref);
+  CHECK(!LZ4F_isError(compr_sz));
+
+  code = LZ4F_freeCompressionContext(cntx);
+  CHECK(!LZ4F_isError(code));
+
+  if (compr_sz + MIN_COMPRESS_IMPROVE >= node->sz) {
+    QList::stats.bad_compression_attempts++;
+    zfree(dest);
+    return false;
+  }
+
+  dest->sz = compr_sz;
+  dest = (quicklistLZF*)zrealloc(dest, sizeof(quicklistLZF) + compr_sz);
+  QList::stats.compressed_bytes += compr_sz;
+  QList::stats.raw_compressed_bytes += node->sz;
+
+  zfree(node->entry);
+  node->entry = (unsigned char*)dest;
+  node->encoding = QLIST_NODE_ENCODING_LZ4;
+  return true;
+}
+
+/* Compress the listpack in 'node' and update encoding details.
+ * Returns true if listpack compressed successfully.
+ * Returns false if compression failed or if listpack too small to compress. */
+bool CompressNode(QList::Node* node, unsigned method) {
+  DCHECK(node->encoding == QUICKLIST_NODE_ENCODING_RAW);
+  DCHECK(!node->dont_compress);
+
+  /* validate that the node is neither
+   * tail nor head (it has prev and next)*/
+  DCHECK(node->prev && node->next);
+
+  node->recompress = 0;
+  /* Don't bother compressing small values */
+  if (node->sz < MIN_COMPRESS_BYTES)
+    return false;
+
+  QList::stats.compression_attempts++;
+  if (method == static_cast<unsigned>(QList::LZF)) {
+    return CompressLZF(node);
+  }
+
+  return CompressLZ4(node);
+}
+
+ssize_t CompressNodeIfNeeded(QList::Node* node, unsigned method) {
   DCHECK(node);
-  if (node->encoding == QUICKLIST_NODE_ENCODING_RAW && !node->dont_compress) {
-    if (CompressNode(node))
-      return ((quicklistLZF*)node->entry)->sz - node->sz;
+  if (node->encoding == QUICKLIST_NODE_ENCODING_RAW) {
+    node->attempted_compress = 1;
+    if (!node->dont_compress) {
+      if (CompressNode(node, method))
+        return ssize_t(GetLzf(node)->sz) - node->sz;
+    }
   }
   return 0;
 }
 
 /* Uncompress the listpack in 'node' and update encoding details.
  * Returns 1 on successful decode, 0 on failure to decode. */
 bool DecompressNode(bool recompress, QList::Node* node) {
+  DCHECK(node->encoding == QUICKLIST_NODE_ENCODING_LZF ||
+         node->encoding == QLIST_NODE_ENCODING_LZ4);
+
   node->recompress = int(recompress);
 
   void* decompressed = zmalloc(node->sz);
-  quicklistLZF* lzf = (quicklistLZF*)node->entry;
-  if (lzf_decompress(lzf->compressed, lzf->sz, decompressed, node->sz) == 0) {
-    /* Someone requested decompress, but we can't decompress.  Not good. */
-    zfree(decompressed);
-    return false;
+  quicklistLZF* lzf = GetLzf(node);
+  QList::stats.decompression_calls++;
+  QList::stats.compressed_bytes -= lzf->sz;
+  QList::stats.raw_compressed_bytes -= node->sz;
+
+  if (node->encoding == QLIST_NODE_ENCODING_LZ4) {
+    LZ4F_dctx* dctx = nullptr;
+    LZ4F_errorCode_t code = LZ4F_createDecompressionContext(&dctx, LZ4F_VERSION);
+    CHECK(!LZ4F_isError(code));
+    size_t decompressed_sz = node->sz;
+    size_t left =
+        LZ4F_decompress(dctx, decompressed, &decompressed_sz, lzf->compressed, &lzf->sz, nullptr);
+    CHECK_EQ(left, 0u);
+    CHECK_EQ(decompressed_sz, node->sz);
+    LZ4F_freeDecompressionContext(dctx);
+  } else {
+    if (lzf_decompress(lzf->compressed, lzf->sz, decompressed, node->sz) == 0) {
+      LOG(DFATAL) << "Invalid LZF compressed data";
+      /* Someone requested decompress, but we can't decompress.  Not good. */
+      zfree(decompressed);
+      return false;
+    }
   }
   zfree(lzf);
   node->entry = (uint8_t*)decompressed;
@@ -252,19 +332,19 @@ bool DecompressNode(bool recompress, QList::Node* node) {
    returns by how much the size of the node has increased.
 */
 ssize_t DecompressNodeIfNeeded(bool recompress, QList::Node* node) {
-  if ((node) && (node)->encoding == QUICKLIST_NODE_ENCODING_LZF) {
-    size_t compressed_sz = ((quicklistLZF*)node->entry)->sz;
+  if (node && node->encoding != QUICKLIST_NODE_ENCODING_RAW) {
+    size_t compressed_sz = GetLzf(node)->sz;
     if (DecompressNode(recompress, node)) {
       return node->sz - compressed_sz;
     }
   }
   return 0;
 }
 
-ssize_t RecompressOnly(QList::Node* node) {
+ssize_t RecompressOnly(QList::Node* node, unsigned method) {
   if (node->recompress && !node->dont_compress) {
-    if (CompressNode(node))
-      return ((quicklistLZF*)node->entry)->sz - node->sz;
+    if (CompressNode(node, method))
+      return (GetLzf(node))->sz - node->sz;
   }
   return 0;
 }
@@ -299,11 +379,14 @@ QList::Node* SplitNode(QList::Node* node, int offset, bool after, ssize_t* diff)
 
 }  // namespace
 
+__thread QList::Stats QList::stats;
+
 void QList::SetPackedThreshold(unsigned threshold) {
   packed_threshold = threshold;
 }
 
 QList::QList(int fill, int compress) : fill_(fill), compress_(compress), bookmark_count_(0) {
+  compr_method_ = 0;
 }
 
 QList::QList(QList&& other)
@@ -342,7 +425,11 @@ void QList::Clear() {
 
   while (len_) {
     Node* next = current->next;
-
+    if (current->encoding != QUICKLIST_NODE_ENCODING_RAW) {
+      quicklistLZF* lzf = (quicklistLZF*)current->entry;
+      QList::stats.compressed_bytes -= lzf->sz;
+      QList::stats.raw_compressed_bytes -= current->sz;
+    }
     zfree(current->entry);
     zfree(current);
 
@@ -587,7 +674,7 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
     uint8_t* new_entry = LP_Insert(node->entry, elem, it.zi_, after ? LP_AFTER : LP_BEFORE);
     malloc_size_ += NodeSetEntry(node, new_entry);
     node->count++;
-    malloc_size_ += RecompressOnly(node);
+    malloc_size_ += RecompressOnly(node, compr_method_);
   } else {
     bool insert_tail = at_tail && after;
     bool insert_head = at_head && !after;
@@ -598,17 +685,17 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
       malloc_size_ += DecompressNodeIfNeeded(true, new_node);
       malloc_size_ += NodeSetEntry(new_node, LP_Prepend(new_node->entry, elem));
       new_node->count++;
-      malloc_size_ += RecompressOnly(new_node);
-      malloc_size_ += RecompressOnly(node);
+      malloc_size_ += RecompressOnly(new_node, compr_method_);
+      malloc_size_ += RecompressOnly(node, compr_method_);
     } else if (insert_head && avail_prev) {
       /* If we are: at head, previous has free space, and inserting before:
        *   - insert entry at tail of previous node. */
       auto* new_node = node->prev;
       malloc_size_ += DecompressNodeIfNeeded(true, new_node);
       malloc_size_ += NodeSetEntry(new_node, LP_Append(new_node->entry, elem));
       new_node->count++;
-      malloc_size_ += RecompressOnly(new_node);
-      malloc_size_ += RecompressOnly(node);
+      malloc_size_ += RecompressOnly(new_node, compr_method_);
+      malloc_size_ += RecompressOnly(node, compr_method_);
     } else if (insert_tail || insert_head) {
       /* If we are: full, and our prev/next has no available space, then:
        *   - create new node and attach to qlist */
@@ -732,12 +819,12 @@ void QList::Compress(Node* node) {
     reverse = reverse->prev;
   }
 
-  if (!in_depth && node)
-    malloc_size_ += CompressNodeIfNeeded(node);
-
+  if (!in_depth && node) {
+    malloc_size_ += CompressNodeIfNeeded(node, this->compr_method_);
+  }
   /* At this point, forward and reverse are one node beyond depth */
-  malloc_size_ += CompressNodeIfNeeded(forward);
-  malloc_size_ += CompressNodeIfNeeded(reverse);
+  malloc_size_ += CompressNodeIfNeeded(forward, this->compr_method_);
+  malloc_size_ += CompressNodeIfNeeded(reverse, this->compr_method_);
 }
 
 /* Attempt to merge listpacks within two nodes on either side of 'center'.
@@ -1063,7 +1150,7 @@ bool QList::Erase(const long start, unsigned count) {
       if (node->count == 0) {
         DelNode(node);
       } else {
-        malloc_size_ += RecompressOnly(node);
+        malloc_size_ += RecompressOnly(node, compr_method_);
       }
     }