Merge branch 'main' into fix-test-hypothesis

.github/actions/regression-tests/action.yml

@@ -46,7 +46,7 @@ runs:
         export DRAGONFLY_PATH="${GITHUB_WORKSPACE}/${{inputs.build-folder-name}}/${{inputs.dfly-executable}}"
         export UBSAN_OPTIONS=print_stacktrace=1:halt_on_error=1 # to crash on errors
 
-        timeout 50m pytest -m "${{inputs.filter}}" --durations=10 --color=yes --json-report --json-report-file=report.json dragonfly --log-cli-level=INFO || code=$?
+        timeout 50m pytest -m "${{inputs.filter}}" --durations=10 --timeout=300 --color=yes --json-report --json-report-file=report.json dragonfly --log-cli-level=INFO || code=$?
 
         # timeout returns 124 if we exceeded the timeout duration
         if [[ $code -eq 124 ]]; then

.github/workflows/ci.yml

@@ -159,8 +159,6 @@ jobs:
           FLAGS_cluster_mode=emulated FLAGS_lock_on_hashtags=true timeout 20m ctest -V -L DFLY
 
           timeout 5m ./dragonfly_test
-          timeout 5m ./multi_test --multi_exec_mode=1
-          timeout 5m ./multi_test --multi_exec_mode=3
           timeout 5m ./json_family_test --jsonpathv2=false
           timeout 5m ./tiered_storage_test --vmodule=db_slice=2 --logtostderr
       - name: Upload unit logs on failure

src/core/compact_object.cc

@@ -791,8 +791,8 @@ uint64_t CompactObj::HashCode() const {
   }
 
   if (encoded) {
-    GetString(&tl.tmp_str);
-    return XXH3_64bits_withSeed(tl.tmp_str.data(), tl.tmp_str.size(), kHashSeed);
+    string_view sv = GetSlice(&tl.tmp_str);
+    return XXH3_64bits_withSeed(sv.data(), sv.size(), kHashSeed);
   }
 
   switch (taglen_) {

src/core/compact_object_test.cc

@@ -76,29 +76,38 @@ void DeallocateAtRandom(size_t steps, std::vector<void*>* ptrs) {
   }
 }
 
+static void InitThreadStructs() {
+  auto* tlh = mi_heap_get_backing();
+  init_zmalloc_threadlocal(tlh);
+  SmallString::InitThreadLocal(tlh);
+  thread_local MiMemoryResource mi_resource(tlh);
+  CompactObj::InitThreadLocal(&mi_resource);
+};
+
+static void CheckEverythingDeallocated() {
+  mi_heap_collect(mi_heap_get_backing(), true);
+
+  auto cb_visit = [](const mi_heap_t* heap, const mi_heap_area_t* area, void* block,
+                     size_t block_size, void* arg) {
+    LOG(ERROR) << "Unfreed allocations: block_size " << block_size
+               << ", allocated: " << area->used * block_size;
+    return true;
+  };
+
+  mi_heap_visit_blocks(mi_heap_get_backing(), false /* do not visit all blocks*/, cb_visit,
+                       nullptr);
+}
+
 class CompactObjectTest : public ::testing::Test {
  protected:
   static void SetUpTestSuite() {
     InitRedisTables();  // to initialize server struct.
 
-    auto* tlh = mi_heap_get_backing();
-    init_zmalloc_threadlocal(tlh);
-    SmallString::InitThreadLocal(tlh);
-    CompactObj::InitThreadLocal(PMR_NS::get_default_resource());
+    InitThreadStructs();
   }
 
   static void TearDownTestSuite() {
-    mi_heap_collect(mi_heap_get_backing(), true);
-
-    auto cb_visit = [](const mi_heap_t* heap, const mi_heap_area_t* area, void* block,
-                       size_t block_size, void* arg) {
-      LOG(ERROR) << "Unfreed allocations: block_size " << block_size
-                 << ", allocated: " << area->used * block_size;
-      return true;
-    };
-
-    mi_heap_visit_blocks(mi_heap_get_backing(), false /* do not visit all blocks*/, cb_visit,
-                         nullptr);
+    CheckEverythingDeallocated();
   }
 
   CompactObj cobj_;

src/core/dense_set.cc

@@ -168,14 +168,14 @@ auto DenseSet::PopPtrFront(DenseSet::ChainVectorIterator it) -> DensePtr {
   return front;
 }
 
-uint32_t DenseSet::ClearInternal(uint32_t start, uint32_t count) {
+uint32_t DenseSet::ClearStep(uint32_t start, uint32_t count) {
   constexpr unsigned kArrLen = 32;
   ClearItem arr[kArrLen];
   unsigned len = 0;
 
   size_t end = min<size_t>(entries_.size(), start + count);
   for (size_t i = start; i < end; ++i) {
-    DensePtr ptr = entries_[i];
+    DensePtr& ptr = entries_[i];
     if (ptr.IsEmpty())
       continue;
 
@@ -190,6 +190,7 @@ uint32_t DenseSet::ClearInternal(uint32_t start, uint32_t count) {
       dest.ptr = ptr;
       dest.obj = nullptr;
     }
+    ptr.Reset();
     if (len == kArrLen) {
       ClearBatch(kArrLen, arr);
       len = 0;

src/core/dense_set.h

@@ -215,9 +215,13 @@ class DenseSet {
   virtual ~DenseSet();
 
   void Clear() {
-    ClearInternal(0, entries_.size());
+    ClearStep(0, entries_.size());
   }
 
+  // Returns the next bucket index that should be cleared.
+  // Returns BucketCount when all objects are erased.
+  uint32_t ClearStep(uint32_t start, uint32_t count);
+
   // Returns the number of elements in the map. Note that it might be that some of these elements
   // have expired and can't be accessed.
   size_t UpperBoundSize() const {
@@ -303,11 +307,6 @@ class DenseSet {
 
   void* PopInternal();
 
-  // Note this does not free any dynamic allocations done by derived classes, that a DensePtr
-  // in the set may point to. This function only frees the allocated DenseLinkKeys created by
-  // DenseSet. All data allocated by a derived class should be freed before calling this
-  uint32_t ClearInternal(uint32_t start, uint32_t count);
-
   void IncreaseMallocUsed(size_t delta) {
     obj_malloc_used_ += delta;
   }

src/core/extent_tree.cc

@@ -15,47 +15,39 @@ void ExtentTree::Add(size_t start, size_t len) {
   DCHECK_GT(len, 0u);
   DCHECK_EQ(len_extents_.size(), extents_.size());
 
-  size_t end = start + len;
-
-  if (extents_.empty()) {
-    extents_.emplace(start, end);
-    len_extents_.emplace(len, start);
-
-    return;
-  }
-
   auto it = extents_.lower_bound(start);
-  bool merged = false;
+  optional<absl::btree_map<size_t, size_t>::iterator> prev_extent;
 
   if (it != extents_.begin()) {
     auto prev = it;
     --prev;
 
     DCHECK_LE(prev->second, start);
-    if (prev->second == start) {  // [first, second = start, end)
-      merged = true;
-      len_extents_.erase(pair{prev->second - prev->first, prev->first});
-
-      // check if we join: prev, [start, end), [it->first, it->second]
-      if (it != extents_.end() && end == it->first) {  // [first, end = it->first, it->second)
-        prev->second = it->second;
-        len_extents_.erase(pair{it->second - it->first, it->first});
-        extents_.erase(it);
-      } else {
-        prev->second = end;  // just extend prev
-      }
-      len_extents_.emplace(prev->second - prev->first, prev->first);
+    if (prev->second == start) {  // combine with the previous extent
+      size_t prev_len = prev->second - prev->first;
+      CHECK_EQ(1u, len_extents_.erase(pair{prev_len, prev->first}));
+      prev->second += len;
+      start = prev->first;
+      len += prev_len;
+      prev_extent = prev;
     }
   }
 
-  if (!merged) {
-    if (end == it->first) {  // [start, end), [it->first, it->second]
-      len_extents_.erase(pair{it->second - it->first, it->first});
-      end = it->second;
+  if (it != extents_.end()) {
+    DCHECK_GE(it->first, start + len);
+    if (start + len == it->first) {  // merge with the next extent
+      size_t it_len = it->second - it->first;
+      CHECK_EQ(1u, len_extents_.erase(pair{it_len, it->first}));
       extents_.erase(it);
+      len += it_len;
     }
-    extents_.emplace(start, end);
-    len_extents_.emplace(end - start, start);
+  }
+
+  len_extents_.emplace(len, start);
+  if (prev_extent) {
+    (*prev_extent)->second = start + len;
+  } else {
+    extents_.emplace(start, start + len);
   }
 }
 

src/core/extent_tree_test.cc

@@ -28,28 +28,28 @@ TEST_F(ExtentTreeTest, Basic) {
   tree_.Add(0, 256);
   auto op = tree_.GetRange(64, 16);
   EXPECT_TRUE(op);
-  EXPECT_THAT(*op, testing::Pair(0, 64));
+  EXPECT_THAT(*op, testing::Pair(0, 64));  // [64, 256)
 
   tree_.Add(56, 8);
   op = tree_.GetRange(64, 16);
   EXPECT_TRUE(op);
-  EXPECT_THAT(*op, testing::Pair(64, 128));
+  EXPECT_THAT(*op, testing::Pair(64, 128));  // {[56, 64), [128, 256)}
 
   op = tree_.GetRange(18, 2);
   EXPECT_TRUE(op);
-  EXPECT_THAT(*op, testing::Pair(128, 146));
+  EXPECT_THAT(*op, testing::Pair(128, 146));  // {[56, 64), [146, 256)}
 
   op = tree_.GetRange(80, 16);
   EXPECT_TRUE(op);
-  EXPECT_THAT(*op, testing::Pair(160, 240));
+  EXPECT_THAT(*op, testing::Pair(160, 240));  // {[56, 64), [146, 160), [240, 256)}
 
   op = tree_.GetRange(4, 1);
   EXPECT_TRUE(op);
-  EXPECT_THAT(*op, testing::Pair(56, 60));
+  EXPECT_THAT(*op, testing::Pair(56, 60));  // {[60, 64), [146, 160), [240, 256)}
 
   op = tree_.GetRange(32, 1);
   EXPECT_FALSE(op);
-  tree_.Add(64, 240 - 64);
+  tree_.Add(64, 146 - 64);
   op = tree_.GetRange(32, 4);
   EXPECT_TRUE(op);
   EXPECT_THAT(*op, testing::Pair(60, 92));

src/facade/reply_builder.cc

@@ -283,14 +283,18 @@ void MCReplyBuilder::SendMiss() {
     SendSimpleString("EN");
 }
 
+void MCReplyBuilder::SendDeleted() {
+  SendSimpleString(flag_.meta ? "HD" : "DELETED");
+}
+
 void MCReplyBuilder::SendRaw(std::string_view str) {
   ReplyScope scope(this);
   WriteRef(str);
 }
 
 void RedisReplyBuilderBase::SendNull() {
   ReplyScope scope(this);
-  resp3_ ? WritePieces(kNullStringR3) : WritePieces(kNullStringR2);
+  IsResp3() ? WritePieces(kNullStringR3) : WritePieces(kNullStringR2);
 }
 
 void RedisReplyBuilderBase::SendSimpleString(std::string_view str) {
@@ -323,7 +327,7 @@ void RedisReplyBuilderBase::SendDouble(double val) {
   static_assert(ABSL_ARRAYSIZE(buf) < kMaxInlineSize, "Write temporary string from buf inline");
   string_view val_str = FormatDouble(val, buf, ABSL_ARRAYSIZE(buf));
 
-  if (!resp3_)
+  if (!IsResp3())
     return SendBulkString(val_str);
 
   ReplyScope scope(this);
@@ -408,8 +412,7 @@ void RedisReplyBuilder::SendBulkStrArr(const facade::ArgRange& strs, CollectionT
     SendBulkString(str);
 }
 
-void RedisReplyBuilder::SendScoredArray(absl::Span<const std::pair<std::string, double>> arr,
-                                        bool with_scores) {
+void RedisReplyBuilder::SendScoredArray(ScoredArray arr, bool with_scores) {
   ReplyScope scope(this);
   StartArray((with_scores && !IsResp3()) ? arr.size() * 2 : arr.size());
   for (const auto& [str, score] : arr) {
@@ -421,6 +424,20 @@ void RedisReplyBuilder::SendScoredArray(absl::Span<const std::pair<std::string,
   }
 }
 
+void RedisReplyBuilder::SendLabeledScoredArray(std::string_view arr_label, ScoredArray arr) {
+  ReplyScope scope(this);
+
+  StartArray(2);
+
+  SendBulkString(arr_label);
+  StartArray(arr.size());
+  for (const auto& [str, score] : arr) {
+    StartArray(2);
+    SendBulkString(str);
+    SendDouble(score);
+  }
+}
+
 void RedisReplyBuilder::SendStored() {
   SendSimpleString("OK");
 }

src/facade/reply_builder.h

@@ -22,6 +22,8 @@ enum class ReplyMode {
   FULL       // All replies are recorded
 };
 
+enum class RespVersion { kResp2, kResp3 };
+
 // Base class for all reply builders. Offer a simple high level interface for controlling output
 // modes and sending basic response types.
 class SinkReplyBuilder {
@@ -172,6 +174,7 @@ class MCReplyBuilder : public SinkReplyBuilder {
   void SendClientError(std::string_view str);
   void SendNotFound();
   void SendMiss();
+  void SendDeleted();
   void SendGetEnd();
 
   void SendValue(std::string_view key, std::string_view value, uint64_t mc_ver, uint32_t mc_flag);
@@ -258,21 +261,26 @@ class RedisReplyBuilderBase : public SinkReplyBuilder {
   static std::string SerializeCommand(std::string_view command);
 
   bool IsResp3() const {
-    return resp3_;
+    return resp_ == RespVersion::kResp3;
+  }
+
+  void SetRespVersion(RespVersion resp_version) {
+    resp_ = resp_version;
   }
 
-  void SetResp3(bool resp3) {
-    resp3_ = resp3;
+  RespVersion GetRespVersion() {
+    return resp_;
   }
 
  private:
-  bool resp3_ = false;
+  RespVersion resp_ = RespVersion::kResp2;
 };
 
 // Non essential redis reply builder functions implemented on top of the base resp protocol
 class RedisReplyBuilder : public RedisReplyBuilderBase {
  public:
   using RedisReplyBuilderBase::CollectionType;
+  using ScoredArray = absl::Span<const std::pair<std::string, double>>;
 
   RedisReplyBuilder(io::Sink* sink) : RedisReplyBuilderBase(sink) {
   }
@@ -281,8 +289,8 @@ class RedisReplyBuilder : public RedisReplyBuilderBase {
 
   void SendSimpleStrArr(const facade::ArgRange& strs);
   void SendBulkStrArr(const facade::ArgRange& strs, CollectionType ct = ARRAY);
-  void SendScoredArray(absl::Span<const std::pair<std::string, double>> arr, bool with_scores);
-
+  void SendScoredArray(ScoredArray arr, bool with_scores);
+  void SendLabeledScoredArray(std::string_view arr_label, ScoredArray arr);
   void SendStored() final;
   void SendSetSkipped() final;