Skip to content

Commit 02002f1

Browse files
BorysTheDevBorysTheDev
committed
feat: Integrate IntrusiveStringSet (#5131)
1 parent f91e374 commit 02002f1

File tree

9 files changed

+390
-86
lines changed

9 files changed

+390
-86
lines changed

‎src/core/compact_object.cc‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ extern "C" {
2626
#include "core/bloom.h"
2727
#include "core/detail/bitpacking.h"
2828
#include "core/huff_coder.h"
29+
#include "core/intrusive_string_set.h"
2930
#include "core/qlist.h"
3031
#include "core/sorted_map.h"
3132
#include "core/string_map.h"
@@ -66,11 +67,12 @@ size_t UpdateSize(size_t size, int64_t update) {
6667

6768
inline void FreeObjSet(unsigned encoding, void* ptr, MemoryResource* mr) {
6869
switch (encoding) {
69-
case kEncodingStrMap2: {
70+
case kEncodingStrMap2:
7071
CompactObj::DeleteMR<StringSet>(ptr);
7172
break;
72-
}
73-
73+
case kEncodingIntrusiveSet:
74+
CompactObj::DeleteMR<IntrusiveStringSet>(ptr);
75+
break;
7476
case kEncodingIntSet:
7577
zfree((void*)ptr);
7678
break;
@@ -98,6 +100,10 @@ size_t MallocUsedSet(unsigned encoding, void* ptr) {
98100
StringSet* ss = (StringSet*)ptr;
99101
return ss->ObjMallocUsed() + ss->SetMallocUsed() + zmalloc_usable_size(ptr);
100102
}
103+
case kEncodingIntrusiveSet: {
104+
IntrusiveStringSet* ss = (IntrusiveStringSet*)ptr;
105+
return ss->ObjMallocUsed() + ss->SetMallocUsed() + zmalloc_usable_size(ptr);
106+
}
101107
case kEncodingIntSet:
102108
return intsetBlobLen((intset*)ptr);
103109
}
@@ -324,6 +330,10 @@ pair<void*, bool> DefragSet(unsigned encoding, void* ptr, float ratio) {
324330
case kEncodingStrMap2: {
325331
return DefragStrSet((StringSet*)ptr, ratio);
326332
}
333+
case kEncodingIntrusiveSet:
334+
// TODO add defrag logic
335+
LOG(WARNING) << "DefragSet for IntrusiveSet isn't implemented";
336+
return {ptr, false};
327337

328338
default:
329339
ABSL_UNREACHABLE();
@@ -455,6 +465,10 @@ size_t RobjWrapper::Size() const {
455465
StringSet* ss = (StringSet*)inner_obj_;
456466
return ss->UpperBoundSize();
457467
}
468+
case kEncodingIntrusiveSet: {
469+
auto* ss = (IntrusiveStringSet*)inner_obj_;
470+
return ss->UpperBoundSize();
471+
}
458472
default:
459473
LOG(FATAL) << "Unexpected encoding " << encoding_;
460474
};

‎src/core/compact_object.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
namespace dfly {
2020

2121
constexpr unsigned kEncodingIntSet = 0;
22+
constexpr unsigned kEncodingIntrusiveSet = 1;
2223
constexpr unsigned kEncodingStrMap2 = 2; // for set/map encodings of strings using DenseSet
2324
constexpr unsigned kEncodingQL2 = 1;
2425
constexpr unsigned kEncodingListPack = 3;

‎src/core/intrusive_string_set.h‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,9 @@
1515

1616
namespace dfly {
1717

18+
// NOTE: do not change the ABI of ISLEntry struct as long as we support
19+
// --stringset_experimental=false
20+
1821
class IntrusiveStringSet {
1922
using Buckets =
2023
std::vector<IntrusiveStringList, PMR_NS::polymorphic_allocator<IntrusiveStringList>>;
@@ -106,6 +109,8 @@ class IntrusiveStringSet {
106109
: entries_(mr) {
107110
}
108111

112+
static constexpr uint32_t kMaxBatchLen = 32;
113+
109114
ISLEntry Add(std::string_view str, uint32_t ttl_sec = UINT32_MAX) {
110115
if (entries_.empty() || size_ >= entries_.size()) {
111116
Reserve(Capacity() * 2);
@@ -262,6 +267,10 @@ class IntrusiveStringSet {
262267
return size_ == 0;
263268
}
264269

270+
std::uint32_t BucketCount() const {
271+
return entries_.size(); // the same as Capacity()
272+
}
273+
265274
std::uint32_t Capacity() const {
266275
return 1 << capacity_log_;
267276
}

‎src/server/container_utils.cc‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
#include "base/flags.h"
77
#include "base/logging.h"
8+
#include "core/intrusive_string_set.h"
89
#include "core/qlist.h"
910
#include "core/sorted_map.h"
1011
#include "core/string_map.h"
@@ -199,13 +200,20 @@ bool IterateSet(const PrimeValue& pv, const IterateFunc& func) {
199200
while (success && intsetGet(is, ii++, &ival)) {
200201
success = func(ContainerEntry{ival});
201202
}
202-
} else {
203+
} else if (pv.Encoding() == kEncodingStrMap2) {
203204
for (sds ptr : *static_cast<StringSet*>(pv.RObjPtr())) {
204205
if (!func(ContainerEntry{ptr, sdslen(ptr)})) {
205206
success = false;
206207
break;
207208
}
208209
}
210+
} else if (pv.Encoding() == kEncodingIntrusiveSet) {
211+
for (auto ptr : *static_cast<IntrusiveStringSet*>(pv.RObjPtr())) {
212+
if (!func(ContainerEntry{ptr.Key().data(), ptr.Key().size()})) {
213+
success = false;
214+
break;
215+
}
216+
}
209217
}
210218

211219
return success;

‎src/server/db_slice.cc‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ extern "C" {
1212

1313
#include "base/flags.h"
1414
#include "base/logging.h"
15+
#include "core/intrusive_string_set.h"
1516
#include "core/top_keys.h"
1617
#include "search/doc_index.h"
1718
#include "server/channel_store.h"
@@ -1671,19 +1672,27 @@ void DbSlice::PerformDeletionAtomic(Iterator del_it, ExpIterator exp_it, DbTable
16711672
table); // Key
16721673
AccountObjectMemory(del_it.key(), pv.ObjType(), -value_heap_size, table); // Value
16731674

1674-
if (del_it->first.IsAsyncDelete() && pv.ObjType() == OBJ_SET &&
1675-
pv.Encoding() == kEncodingStrMap2) {
1676-
DenseSet* ds = (DenseSet*)pv.RObjPtr();
1677-
pv.SetRObjPtr(nullptr);
1678-
const size_t kClearStepSize = 512;
1675+
if (del_it->first.IsAsyncDelete() && pv.ObjType() == OBJ_SET) {
1676+
if (pv.Encoding() == kEncodingStrMap2) {
1677+
DenseSet* ds = (DenseSet*)pv.RObjPtr();
1678+
pv.SetRObjPtr(nullptr);
1679+
const size_t kClearStepSize = 512;
1680+
1681+
uint32_t next = ds->ClearStep(0, kClearStepSize);
1682+
if (next < ds->BucketCount()) {
1683+
AsyncDeleter::EnqueDeletion(next, ds);
1684+
} else {
1685+
CompactObj::DeleteMR<DenseSet>(ds);
1686+
}
1687+
} // del_it->first.IsAsyncDelete()
1688+
else if (pv.Encoding() == kEncodingIntrusiveSet) {
1689+
IntrusiveStringSet* ds = (IntrusiveStringSet*)pv.RObjPtr();
1690+
pv.SetRObjPtr(nullptr);
16791691

1680-
uint32_t next = ds->ClearStep(0, kClearStepSize);
1681-
if (next < ds->BucketCount()) {
1682-
AsyncDeleter::EnqueDeletion(next, ds);
1683-
} else {
16841692
CompactObj::DeleteMR<DenseSet>(ds);
1693+
LOG(WARNING) << "Implement async deletion for IntrusiveStringSet";
16851694
}
1686-
} // del_it->first.IsAsyncDelete()
1695+
}
16871696

16881697
if (table->slots_stats) {
16891698
SlotId sid = KeySlot(del_it.key());

‎src/server/debugcmd.cc‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -465,6 +465,8 @@ const char* EncodingName(unsigned obj_type, unsigned encoding) {
465465
switch (encoding) {
466466
case kEncodingIntSet:
467467
return "intset";
468+
case kEncodingIntrusiveSet:
469+
return "intrusive_set";
468470
case kEncodingStrMap2:
469471
return "dense_set";
470472
case OBJ_ENCODING_SKIPLIST: // we kept the old enum for zset

‎src/server/rdb_load.cc‎

Lines changed: 54 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ extern "C" {
2929
#include "base/flags.h"
3030
#include "base/logging.h"
3131
#include "core/bloom.h"
32+
#include "core/intrusive_string_set.h"
3233
#include "core/json/json_object.h"
3334
#include "core/qlist.h"
3435
#include "core/sorted_map.h"
@@ -57,6 +58,7 @@ ABSL_DECLARE_FLAG(int32_t, list_max_listpack_size);
5758
ABSL_DECLARE_FLAG(int32_t, list_compress_depth);
5859
ABSL_DECLARE_FLAG(uint32_t, dbnum);
5960
ABSL_DECLARE_FLAG(bool, list_experimental_v2);
61+
ABSL_DECLARE_FLAG(bool, stringset_experimental);
6062
ABSL_FLAG(bool, rdb_load_dry_run, false, "Dry run RDB load without applying changes");
6163
ABSL_FLAG(bool, rdb_ignore_expiry, false, "Ignore Key Expiry when loding from RDB snapshot");
6264

@@ -410,7 +412,9 @@ void RdbLoaderBase::OpaqueObjLoader::CreateSet(const LoadTrace* ltrace) {
410412
return;
411413

412414
if (!config_.append) {
413-
pv_->InitRobj(OBJ_SET, is_intset ? kEncodingIntSet : kEncodingStrMap2, inner_obj);
415+
int str_set_encoding =
416+
absl::GetFlag(FLAGS_stringset_experimental) ? kEncodingIntrusiveSet : kEncodingStrMap2;
417+
pv_->InitRobj(OBJ_SET, is_intset ? kEncodingIntSet : str_set_encoding, inner_obj);
414418
}
415419
std::move(cleanup).Cancel();
416420
}
@@ -877,14 +881,26 @@ void RdbLoaderBase::OpaqueObjLoader::HandleBlob(string_view blob) {
877881
unsigned len = intsetLen(is);
878882

879883
if (len > SetFamily::MaxIntsetEntries()) {
880-
StringSet* set = SetFamily::ConvertToStrSet(is, len);
884+
if (!absl::GetFlag(FLAGS_stringset_experimental)) {
885+
StringSet* set = SetFamily::ConvertToStrSet(is, len);
881886

882-
if (!set) {
883-
LOG(ERROR) << "OOM in ConvertToStrSet " << len;
884-
ec_ = RdbError(errc::out_of_memory);
885-
return;
887+
if (!set) {
888+
LOG(ERROR) << "OOM in ConvertToStrSet " << len;
889+
ec_ = RdbError(errc::out_of_memory);
890+
return;
891+
}
892+
pv_->InitRobj(OBJ_SET, kEncodingStrMap2, set);
893+
} else {
894+
IntrusiveStringSet* set = SetFamily::ConvertToIntrStrSet(is, len);
895+
896+
if (!set) {
897+
LOG(ERROR) << "OOM in ConvertToStrSet " << len;
898+
ec_ = RdbError(errc::out_of_memory);
899+
return;
900+
}
901+
pv_->InitRobj(OBJ_SET, kEncodingIntrusiveSet, set);
886902
}
887-
pv_->InitRobj(OBJ_SET, kEncodingStrMap2, set);
903+
888904
} else {
889905
intset* mine = (intset*)zmalloc(blob.size());
890906
::memcpy(mine, blob.data(), blob.size());
@@ -898,21 +914,39 @@ void RdbLoaderBase::OpaqueObjLoader::HandleBlob(string_view blob) {
898914
}
899915

900916
unsigned char* lp = (unsigned char*)blob.data();
901-
StringSet* set = CompactObj::AllocateMR<StringSet>();
902-
for (unsigned char* cur = lpFirst(lp); cur != nullptr; cur = lpNext(lp, cur)) {
903-
unsigned char field_buf[LP_INTBUF_SIZE];
904-
string_view elem = container_utils::LpGetView(cur, field_buf);
905-
if (!set->Add(elem)) {
906-
LOG(ERROR) << "Duplicate member " << elem;
907-
ec_ = RdbError(errc::duplicate_key);
908-
break;
917+
if (!absl::GetFlag(FLAGS_stringset_experimental)) {
918+
StringSet* set = CompactObj::AllocateMR<StringSet>();
919+
for (unsigned char* cur = lpFirst(lp); cur != nullptr; cur = lpNext(lp, cur)) {
920+
unsigned char field_buf[LP_INTBUF_SIZE];
921+
string_view elem = container_utils::LpGetView(cur, field_buf);
922+
if (!set->Add(elem)) {
923+
LOG(ERROR) << "Duplicate member " << elem;
924+
ec_ = RdbError(errc::duplicate_key);
925+
break;
926+
}
909927
}
928+
if (ec_) {
929+
CompactObj::DeleteMR<StringSet>(set);
930+
return;
931+
}
932+
pv_->InitRobj(OBJ_SET, kEncodingStrMap2, set);
933+
} else {
934+
IntrusiveStringSet* set = CompactObj::AllocateMR<IntrusiveStringSet>();
935+
for (unsigned char* cur = lpFirst(lp); cur != nullptr; cur = lpNext(lp, cur)) {
936+
unsigned char field_buf[LP_INTBUF_SIZE];
937+
string_view elem = container_utils::LpGetView(cur, field_buf);
938+
if (!set->Add(elem)) {
939+
LOG(ERROR) << "Duplicate member " << elem;
940+
ec_ = RdbError(errc::duplicate_key);
941+
break;
942+
}
943+
}
944+
if (ec_) {
945+
CompactObj::DeleteMR<IntrusiveStringSet>(set);
946+
return;
947+
}
948+
pv_->InitRobj(OBJ_SET, kEncodingIntrusiveSet, set);
910949
}
911-
if (ec_) {
912-
CompactObj::DeleteMR<StringSet>(set);
913-
return;
914-
}
915-
pv_->InitRobj(OBJ_SET, kEncodingStrMap2, set);
916950
} else if (rdb_type_ == RDB_TYPE_HASH_ZIPLIST || rdb_type_ == RDB_TYPE_HASH_LISTPACK) {
917951
unsigned char* lp = lpNew(blob.size());
918952
switch (rdb_type_) {

0 commit comments

Comments
 (0)