chore: qlist improvements (#4194)

fix: qlist improvements + bug fix in Erase

1. Reduce code duplication.
2. Expose qlist node count via "debug object"
3. Add more tests to qlist_test
4. Fix a bug in QList::Erase

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

src/core/qlist.cc

@@ -60,6 +60,8 @@ namespace dfly {
 
 namespace {
 
+static_assert(sizeof(QList) == 32);
+
 enum IterDir : uint8_t { FWD = 1, REV = 0 };
 
 /* This is for test suite development purposes only, 0 means disabled. */
@@ -167,10 +169,15 @@ quicklistNode* CreateNode(int container, string_view value) {
   return new_node;
 }
 
-void NodeUpdateSz(quicklistNode* node) {
+inline void NodeUpdateSz(quicklistNode* node) {
   node->sz = lpBytes((node)->entry);
 }
 
+inline void NodeOnAddItem(quicklistNode* node) {
+  node->count++;
+  NodeUpdateSz(node);
+}
+
 /* 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. */
@@ -345,12 +352,7 @@ void QList::Push(string_view value, Where where) {
   if (tail_)
     DCHECK(tail_->encoding != QUICKLIST_NODE_ENCODING_LZF);
 
-  if (where == HEAD) {
+  PushSentinel(value, where);
-    PushHead(value);
-  } else {
-    DCHECK_EQ(TAIL, where);
-    PushTail(value);
-  }
 }
 
 string QList::Pop(Where where) {
@@ -459,50 +461,28 @@ void QList::Iterate(IterateFunc cb, long start, long end) const {
   }
 }
 
-bool QList::PushHead(string_view value) {
+bool QList::PushSentinel(string_view value, Where where) {
-  quicklistNode* orig = head_;
+  quicklistNode* orig = where == HEAD ? head_ : tail_;
+  InsertOpt opt = where == HEAD ? BEFORE : AFTER;
+
   size_t sz = value.size();
   if (ABSL_PREDICT_FALSE(IsLargeElement(sz, fill_))) {
-    InsertPlainNode(head_, value, BEFORE);
+    InsertPlainNode(orig, value, opt);
-    return true;
-  }
-
-  // TODO: we can deduplicate this code with PushTail once we complete the functionality
-  // of this class.
-  count_++;
-
-  if (ABSL_PREDICT_TRUE(NodeAllowInsert(head_, fill_, sz))) {
-    head_->entry = LP_Prepend(head_->entry, value);
-    NodeUpdateSz(head_);
-  } else {
-    quicklistNode* node = CreateNode(QUICKLIST_NODE_CONTAINER_PACKED, value);
-    InsertNode(head_, node, BEFORE);
-  }
-
-  head_->count++;
-  return (orig != head_);
-}
-
-// Returns false if used existing head, true if new head created.
-bool QList::PushTail(string_view value) {
-  quicklistNode* orig = tail_;
-  size_t sz = value.size();
-  if (ABSL_PREDICT_FALSE(IsLargeElement(sz, fill_))) {
-    InsertPlainNode(orig, value, AFTER);
     return true;
   }
 
   count_++;
+
   if (ABSL_PREDICT_TRUE(NodeAllowInsert(orig, fill_, sz))) {
-    orig->entry = LP_Append(orig->entry, value);
+    auto func = (where == HEAD) ? LP_Prepend : LP_Append;
+    orig->entry = func(orig->entry, value);
     NodeUpdateSz(orig);
     orig->count++;
     return false;
   }
 
   quicklistNode* node = CreateNode(QUICKLIST_NODE_CONTAINER_PACKED, value);
-  InsertNode(orig, node, AFTER);
+  InsertNode(orig, node, opt);
-
   return true;
 }
 
@@ -557,6 +537,9 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
 
   if (!node) {
     /* we have no reference node, so let's create only node in the list */
+    DCHECK_EQ(count_, 0u);
+    DCHECK_EQ(len_, 0u);
+
     if (ABSL_PREDICT_FALSE(IsLargeElement(sz, fill_))) {
       InsertPlainNode(tail_, elem, insert_opt);
       return;
@@ -605,14 +588,12 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
   if (!full && after) {
     DecompressNodeIfNeeded(true, node);
     node->entry = LP_Insert(node->entry, elem, it.zi_, LP_AFTER);
-    node->count++;
+    NodeOnAddItem(node);
-    NodeUpdateSz(node);
     RecompressOnly(node);
   } else if (!full && !after) {
     DecompressNodeIfNeeded(true, node);
     node->entry = LP_Insert(node->entry, elem, it.zi_, LP_BEFORE);
-    node->count++;
+    NodeOnAddItem(node);
-    NodeUpdateSz(node);
     RecompressOnly(node);
   } else if (full && at_tail && avail_next && after) {
     /* If we are: at tail, next has free space, and inserting after:
@@ -620,8 +601,7 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
     new_node = node->next;
     DecompressNodeIfNeeded(true, new_node);
     new_node->entry = LP_Prepend(new_node->entry, elem);
-    new_node->count++;
+    NodeOnAddItem(new_node);
-    NodeUpdateSz(new_node);
     RecompressOnly(new_node);
     RecompressOnly(node);
   } else if (full && at_head && avail_prev && !after) {
@@ -630,8 +610,7 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
     new_node = node->prev;
     DecompressNodeIfNeeded(true, new_node);
     new_node->entry = LP_Append(new_node->entry, elem);
-    new_node->count++;
+    NodeOnAddItem(new_node);
-    NodeUpdateSz(new_node);
     RecompressOnly(new_node);
     RecompressOnly(node);
   } else if (full && ((at_tail && !avail_next && after) || (at_head && !avail_prev && !after))) {
@@ -648,8 +627,7 @@ void QList::Insert(Iterator it, std::string_view elem, InsertOpt insert_opt) {
       new_node->entry = LP_Prepend(new_node->entry, elem);
     else
       new_node->entry = LP_Append(new_node->entry, elem);
-    new_node->count++;
+    NodeOnAddItem(new_node);
-    NodeUpdateSz(new_node);
     InsertNode(node, new_node, insert_opt);
     MergeNodes(node);
   }
@@ -1000,10 +978,10 @@ auto QList::Erase(Iterator it) -> Iterator {
 
   /* If current node is deleted, we must update iterator node and offset. */
   if (deleted_node) {
-    if (it.direction_ == AL_START_HEAD) {
+    if (it.direction_ == FWD) {
       it.current_ = next;
       it.offset_ = 0;
-    } else if (it.direction_ == AL_START_TAIL) {
+    } else if (it.direction_ == REV) {
       it.current_ = prev;
       it.offset_ = -1;
     }

src/core/qlist.h

@@ -146,8 +146,8 @@ class QList {
     return compress_ != 0;
   }
 
-  // Returns false if used existing head, true if new head created.
+  // Returns false if used existing sentinel, true if a new sentinel was created.
-  bool PushHead(std::string_view value);
+  bool PushSentinel(std::string_view value, Where where);
 
   // Returns false if used existing head, true if new head created.
   bool PushTail(std::string_view value);

src/core/qlist_test.cc

@@ -21,6 +21,7 @@ namespace dfly {
 
 using namespace std;
 using namespace testing;
+using absl::StrCat;
 
 static int _ql_verify_compress(const QList& ql) {
   int errors = 0;
@@ -162,6 +163,8 @@ TEST_F(QListTest, Basic) {
   EXPECT_EQ(0, ql_.Size());
   ql_.Push("abc", QList::HEAD);
   EXPECT_EQ(1, ql_.Size());
+  EXPECT_TRUE(ql_.Head()->prev == nullptr);
+  EXPECT_TRUE(ql_.Tail() == ql_.Head());
 
   auto it = ql_.GetIterator(QList::HEAD);
   ASSERT_TRUE(it.Next());  // Needed to initialize the iterator.
@@ -382,4 +385,132 @@ TEST_P(OptionsTest, DelRangeD) {
   ASSERT_EQ(30, ql_.Size());
 }
 
+TEST_P(OptionsTest, DelRangeNode) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(-2, compress);
+
+  for (int i = 0; i < 32; i++)
+    ql_.Push(StrCat("hello", i), QList::HEAD);
+
+  ql_verify(ql_, 1, 32, 32, 32);
+  ql_.Erase(0, 32);
+  ql_verify(ql_, 0, 0, 0, 0);
+}
+
+TEST_P(OptionsTest, DelRangeNodeOverflow) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(-2, compress);
+
+  for (int i = 0; i < 32; i++)
+    ql_.Push(StrCat("hello", i), QList::HEAD);
+  ql_verify(ql_, 1, 32, 32, 32);
+  ql_.Erase(0, 128);
+  ql_verify(ql_, 0, 0, 0, 0);
+}
+
+TEST_P(OptionsTest, DelRangeMiddle100of500) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+
+  for (int i = 0; i < 500; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+
+  ql_verify(ql_, 16, 500, 32, 20);
+  ql_.Erase(200, 100);
+  ql_verify(ql_, 14, 400, 32, 20);
+}
+
+TEST_P(OptionsTest, DelLessFillAcrossNodes) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+
+  for (int i = 0; i < 500; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+  ql_verify(ql_, 16, 500, 32, 20);
+  ql_.Erase(60, 10);
+  ql_verify(ql_, 16, 490, 32, 20);
+}
+
+TEST_P(OptionsTest, DelNegOne) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+  for (int i = 0; i < 500; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+  ql_verify(ql_, 16, 500, 32, 20);
+  ql_.Erase(-1, 1);
+  ql_verify(ql_, 16, 499, 32, 19);
+}
+
+TEST_P(OptionsTest, DelNegOneOverflow) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+  for (int i = 0; i < 500; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+
+  ql_verify(ql_, 16, 500, 32, 20);
+  ql_.Erase(-1, 128);
+
+  ql_verify(ql_, 16, 499, 32, 19);
+}
+
+TEST_P(OptionsTest, DelNeg100From500) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+  for (int i = 0; i < 500; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+  ql_.Erase(-100, 100);
+  ql_verify(ql_, 13, 400, 32, 16);
+}
+
+TEST_P(OptionsTest, DelMin10_5_from50) {
+  auto [_, compress] = GetParam();
+  ql_ = QList(32, compress);
+
+  for (int i = 0; i < 50; i++)
+    ql_.Push(StrCat("hello", i + 1), QList::TAIL);
+  ql_verify(ql_, 2, 50, 32, 18);
+  ql_.Erase(-10, 5);
+  ql_verify(ql_, 2, 45, 32, 13);
+}
+
+TEST_P(OptionsTest, DelElems) {
+  auto [fill, compress] = GetParam();
+  ql_ = QList(fill, compress);
+
+  const char* words[] = {"abc", "foo", "bar", "foobar", "foobared", "zap", "bar", "test", "foo"};
+  const char* result[] = {"abc", "foo", "foobar", "foobared", "zap", "test", "foo"};
+  const char* resultB[] = {"abc", "foo", "foobar", "foobared", "zap", "test"};
+
+  for (int i = 0; i < 9; i++)
+    ql_.Push(words[i], QList::TAIL);
+
+  /* lrem 0 bar */
+  auto iter = ql_.GetIterator(QList::HEAD);
+  while (iter.Next()) {
+    if (iter.Get() == "bar") {
+      iter = ql_.Erase(iter);
+    }
+  }
+  EXPECT_THAT(ToItems(), ElementsAreArray(result));
+
+  ql_.Push("foo", QList::TAIL);
+
+  /* lrem -2 foo */
+  iter = ql_.GetIterator(QList::TAIL);
+  int del = 2;
+  while (iter.Next()) {
+    if (iter.Get() == "foo") {
+      iter = ql_.Erase(iter);
+      del--;
+    }
+    if (del == 0)
+      break;
+  }
+
+  /* check result of lrem -2 foo */
+  /* (we're ignoring the '2' part and still deleting all foo
+   * because we only have two foo) */
+  EXPECT_THAT(ToItems(), ElementsAreArray(resultB));
+}
+
 }  // namespace dfly

src/server/debugcmd.cc

@@ -18,6 +18,7 @@ extern "C" {
 #include "base/flags.h"
 #include "base/logging.h"
 #include "core/compact_object.h"
+#include "core/qlist.h"
 #include "core/string_map.h"
 #include "server/blocking_controller.h"
 #include "server/container_utils.h"
@@ -62,6 +63,9 @@ struct ObjInfo {
   unsigned bucket_id = 0;
   unsigned slot_id = 0;
 
+  // for lists - how many nodes do they have.
+  unsigned num_nodes = 0;
+
   enum LockStatus { NONE, S, X } lock_status = NONE;
 
   int64_t ttl = INT64_MAX;
@@ -314,6 +318,12 @@ ObjInfo InspectOp(ConnectionContext* cntx, string_view key) {
     oinfo.encoding = pv.Encoding();
     oinfo.bucket_id = it.bucket_id();
     oinfo.slot_id = it.slot_id();
+
+    if (pv.ObjType() == OBJ_LIST && pv.Encoding() == kEncodingQL2) {
+      const QList* qlist = static_cast<const QList*>(pv.RObjPtr());
+      oinfo.num_nodes = qlist->node_count();
+    }
+
     if (pv.IsExternal()) {
       oinfo.external_len.emplace(pv.GetExternalSlice().second);
     }
@@ -406,7 +416,6 @@ const char* EncodingName(unsigned obj_type, unsigned encoding) {
       break;
     case OBJ_STREAM:
       return "stream";
-    default:;
   }
   return "unknown";
 }
@@ -876,6 +885,10 @@ void DebugCmd::Inspect(string_view key, CmdArgList args, facade::SinkReplyBuilde
       StrAppend(&resp, " spill_len:", *res.external_len);
     }
 
+    if (res.num_nodes) {
+      StrAppend(&resp, " ns:", res.num_nodes);
+    }
+
     if (res.lock_status != ObjInfo::NONE) {
       StrAppend(&resp, " lock:", res.lock_status == ObjInfo::X ? "x" : "s");
     }

src/server/list_family_test.cc

@@ -377,6 +377,11 @@ TEST_F(ListFamilyTest, LRem) {
 
   ASSERT_THAT(Run({"lrem", kKey2, "1", "12345678"}), IntArg(1));
   ASSERT_THAT(Run({"lrem", kKey2, "1", val}), IntArg(1));
+
+  ASSERT_THAT(Run({"lpush", kKey3, "bar", "bar", "foo"}), IntArg(3));
+  ASSERT_THAT(Run({"lrem", kKey3, "-2", "bar"}), IntArg(2));
+  resp = Run({"lrange", kKey3, "0", "-1"});
+  ASSERT_EQ(resp, "foo");
 }
 
 TEST_F(ListFamilyTest, DumpRestorePlain) {