Introduce volatile_set implementation

cmake/Modules/SourceFiles.cmake

@@ -119,7 +119,7 @@ set(VALKEY_SERVER_SRCS
     ${CMAKE_SOURCE_DIR}/src/server.c
     ${CMAKE_SOURCE_DIR}/src/logreqres.c
     ${CMAKE_SOURCE_DIR}/src/entry.c
-    ${CMAKE_SOURCE_DIR}/src/volatile_set.c)
+    ${CMAKE_SOURCE_DIR}/src/vset.c)
 
 
 # valkey-cli

src/Makefile

@@ -423,7 +423,7 @@ ENGINE_NAME=valkey
 SERVER_NAME=$(ENGINE_NAME)-server$(PROG_SUFFIX)
 ENGINE_SENTINEL_NAME=$(ENGINE_NAME)-sentinel$(PROG_SUFFIX)
 ENGINE_TRACE_OBJ=trace/trace.o trace/trace_commands.o trace/trace_db.o trace/trace_bgsave.o trace/trace_cluster.o trace/trace_server.o trace/trace_aof.o
-ENGINE_SERVER_OBJ=threads_mngr.o adlist.o vector.o quicklist.o ae.o anet.o dict.o hashtable.o kvstore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o memory_prefetch.o io_threads.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o cluster_slot_stats.o crc16.o endianconv.o commandlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o valkey-check-rdb.o valkey-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o allocator_defrag.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script.o functions.o commands.o strl.o connection.o unix.o logreqres.o rdma.o scripting_engine.o entry.o volatile_set.o lua/script_lua.o lua/function_lua.o lua/engine_lua.o lua/debug_lua.o
+ENGINE_SERVER_OBJ=threads_mngr.o adlist.o vector.o quicklist.o ae.o anet.o dict.o hashtable.o kvstore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o memory_prefetch.o io_threads.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o cluster_slot_stats.o crc16.o endianconv.o commandlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crccombine.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o valkey-check-rdb.o valkey-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o allocator_defrag.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script.o functions.o commands.o strl.o connection.o unix.o logreqres.o rdma.o scripting_engine.o entry.o vset.o lua/script_lua.o lua/function_lua.o lua/engine_lua.o lua/debug_lua.o
 ENGINE_SERVER_OBJ+=$(ENGINE_TRACE_OBJ)
 ENGINE_CLI_NAME=$(ENGINE_NAME)-cli$(PROG_SUFFIX)
 ENGINE_CLI_OBJ=anet.o adlist.o dict.o valkey-cli.o zmalloc.o release.o ae.o serverassert.o crcspeed.o crccombine.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o cli_commands.o sds.o util.o sha256.o

src/object.c

@@ -1129,9 +1129,9 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
 
     if (o->type == OBJ_STRING) {
         if (o->encoding == OBJ_ENCODING_INT) {
-            asize = sizeof(*o);
+            asize = zmalloc_size((void *)o);
         } else if (o->encoding == OBJ_ENCODING_RAW) {
-            asize = sdsAllocSize(o->ptr) + sizeof(*o);
+            asize = sdsAllocSize(o->ptr) + zmalloc_size((void *)o);
         } else if (o->encoding == OBJ_ENCODING_EMBSTR) {
             asize = zmalloc_size((void *)o);
         } else {

src/server.h

@@ -80,7 +80,7 @@
 #include "rax.h"        /* Radix tree */
 #include "connection.h" /* Connection abstraction */
 #include "memory_prefetch.h"
-#include "volatile_set.h"
+#include "vset.h"
 #include "trace/trace.h"
 #include "entry.h"
 
@@ -2597,7 +2597,7 @@ typedef struct {
     unsigned char *fptr, *vptr;
 
     hashtableIterator iter;
-    volatileSetIterator viter;
+    vsetIterator viter;
     void *next;
 
 } hashTypeIterator;

src/t_hash.c

@@ -35,32 +35,26 @@
 #include "hashtable.h"
 #include "rax.h"
 #include "sds.h"
-#include "volatile_set.h"
+#include "vset.h"
 #include "server.h"
 #include "zmalloc.h"
 #include <math.h>
 #include <string.h>
 #include "entry.h"
 
-
-volatileEntryType hashVolatileEntryType = {
-    .entryGetKey = (sds(*)(const void *entry))entryGetField,
-    .getExpiry = (long long (*)(const void *entry))entryGetExpiry,
-};
-
 /*-----------------------------------------------------------------------------
  * Hash type Expiry API
  *----------------------------------------------------------------------------*/
 
-static volatile_set *hashTypeGetVolatileSet(robj *o) {
+static vset *hashTypeGetVolatileSet(robj *o) {
     serverAssert(o->encoding == OBJ_ENCODING_HASHTABLE);
-    return *(volatile_set **)hashtableMetadata(o->ptr);
+    return (vset *)hashtableMetadata(o->ptr);
 }
 
 void hashTypeFreeVolatileSet(robj *o) {
-    volatile_set *set = hashTypeGetVolatileSet(o);
+    vset *set = hashTypeGetVolatileSet(o);
     if (set)
-        freeVolatileSet(set);
+        vsetClear(set);
 }
 
 bool hashTypeHasVolatileElements(robj *o) {
@@ -79,36 +73,35 @@ void hashTypeIgnoreTTL(robj *o, bool ignore) {
     }
 }
 
-static volatile_set *hashTypeGetOrcreateVolatileSet(robj *o) {
+static vset *hashTypeGetOrcreateVolatileSet(robj *o) {
     serverAssert(o->encoding == OBJ_ENCODING_HASHTABLE);
-    volatile_set **volatile_set_ref = hashtableMetadata(o->ptr);
-    if (*volatile_set_ref == NULL) {
-        *volatile_set_ref = createVolatileSet(&hashVolatileEntryType);
+    vset *vset = hashtableMetadata(o->ptr);
+    if (*vset == NULL) {
+        vsetInit(vset);
         /* serves mainly for optimization. Use type which supports access function only when needed. */
         hashTypeIgnoreTTL(o, false);
     }
-    return *volatile_set_ref;
+    return vset;
 }
 
 static void hashTypeDeleteVolatileSet(robj *o) {
-    volatile_set **volatile_set_ref = hashtableMetadata(o->ptr);
-    freeVolatileSet(*volatile_set_ref);
-    *volatile_set_ref = NULL;
+    vset *vset = hashtableMetadata(o->ptr);
+    vsetClear(vset);
     /* serves mainly for optimization. by changing the hashtable type we can avoid extra function call in hashtable access */
     hashTypeIgnoreTTL(o, true);
 }
 
 void hashTypeTrackEntry(robj *o, void *entry) {
-    volatile_set *set = hashTypeGetOrcreateVolatileSet(o);
-    serverAssert(volatileSetAddEntry(set, entry, entryGetExpiry(entry)));
+    vset *set = hashTypeGetOrcreateVolatileSet(o);
+    serverAssert(vsetAddEntry(set, entryGetExpiry, entry));
 }
 
 void hashTypeUntrackEntry(robj *o, void *entry) {
     if (!entryHasExpiry(entry)) return;
-    volatile_set *set = hashTypeGetVolatileSet(o);
+    vset *set = hashTypeGetVolatileSet(o);
     debugServerAssert(set);
-    serverAssert(volatileSetRemoveEntry(set, entry, entryGetExpiry(entry)));
-    if (volatileSetNumEntries(set) == 0) {
+    serverAssert(vsetRemoveEntry(set, entryGetExpiry, entry));
+    if (vsetIsEmpty(set)) {
         hashTypeDeleteVolatileSet(o);
     }
 }
@@ -120,19 +113,12 @@ static void hashTypeTrackUpdateEntry(robj *o, void *old_entry, void *new_entry,
     if (!old_tracked && !new_tracked)
         return;
 
-    volatile_set *set = hashTypeGetOrcreateVolatileSet(o);
-    debugServerAssert(set);
+    vset *set = hashTypeGetOrcreateVolatileSet(o);
+    debugServerAssert(!old_tracked || !vsetIsEmpty(set));
 
-    if (old_tracked && !new_tracked)
-        serverAssert(volatileSetRemoveEntry(set, old_entry, old_expiry));
-    else if (new_tracked && !old_tracked)
-        serverAssert(volatileSetAddEntry(set, new_entry, new_expiry));
-    else {
-        volatile_set *set = hashTypeGetVolatileSet(o);
-        debugServerAssert(set);
-        serverAssert(volatileSetUpdateEntry(set, old_entry, new_entry, old_expiry, new_expiry) == 1);
-    }
-    if (volatileSetNumEntries(set) == 0) {
+    serverAssert(vsetUpdateEntry(set, entryGetExpiry, old_entry, new_entry, old_expiry, new_expiry) == 1);
+
+    if (vsetIsEmpty(set)) {
         hashTypeDeleteVolatileSet(o);
     }
 }
@@ -584,7 +570,7 @@ void hashTypeInitVolatileIterator(robj *subject, hashTypeIterator *hi) {
     if (hi->encoding == OBJ_ENCODING_LISTPACK) {
         return;
     } else if (hi->encoding == OBJ_ENCODING_HASHTABLE) {
-        volatileSetStart(hashTypeGetVolatileSet(subject), &hi->viter);
+        vsetStart(hashTypeGetVolatileSet(subject), &hi->viter);
     } else {
         serverPanic("Unknown hash encoding");
     }
@@ -595,7 +581,7 @@ void hashTypeResetIterator(hashTypeIterator *hi) {
         if (!hi->volatile_items_iter)
             hashtableResetIterator(&hi->iter);
         else
-            volatileSetReset(&hi->viter);
+            vsetStop(&hi->viter);
     }
 }
 
@@ -635,7 +621,7 @@ int hashTypeNext(hashTypeIterator *hi) {
         if (!hi->volatile_items_iter) {
             if (!hashtableNext(&hi->iter, &hi->next)) return C_ERR;
         } else {
-            if (!volatileSetNext(&hi->viter, &hi->next)) return C_ERR;
+            if (!vsetNext(&hi->viter, &hi->next)) return C_ERR;
         }
     } else {
         serverPanic("Unknown hash encoding");

src/unit/test_files.h

@@ -201,6 +201,11 @@ int test_version2num(int argc, char **argv, int flags);
 int test_reclaimFilePageCache(int argc, char **argv, int flags);
 int test_valkey_strtod(int argc, char **argv, int flags);
 int test_vector(int argc, char **argv, int flags);
+int test_vset_add_and_iterate(int argc, char **argv, int flags);
+int test_vset_large_batch_same_expiry(int argc, char **argv, int flags);
+int test_vset_iterate_multiple_expiries(int argc, char **argv, int flags);
+int test_vset_add_and_remove_all(int argc, char **argv, int flags);
+int test_vset_fuzzer(int argc, char **argv, int flags);
 int test_ziplistCreateIntList(int argc, char **argv, int flags);
 int test_ziplistPop(int argc, char **argv, int flags);
 int test_ziplistGetElementAtIndex3(int argc, char **argv, int flags);
@@ -262,6 +267,7 @@ unitTest __test_sha1_c[] = {{"test_sha1", test_sha1}, {NULL, NULL}};
 unitTest __test_util_c[] = {{"test_string2ll", test_string2ll}, {"test_string2l", test_string2l}, {"test_ll2string", test_ll2string}, {"test_ld2string", test_ld2string}, {"test_fixedpoint_d2string", test_fixedpoint_d2string}, {"test_version2num", test_version2num}, {"test_reclaimFilePageCache", test_reclaimFilePageCache}, {NULL, NULL}};
 unitTest __test_valkey_strtod_c[] = {{"test_valkey_strtod", test_valkey_strtod}, {NULL, NULL}};
 unitTest __test_vector_c[] = {{"test_vector", test_vector}, {NULL, NULL}};
+unitTest __test_vset_c[] = {{"test_vset_add_and_iterate", test_vset_add_and_iterate}, {"test_vset_large_batch_same_expiry", test_vset_large_batch_same_expiry}, {"test_vset_iterate_multiple_expiries", test_vset_iterate_multiple_expiries}, {"test_vset_add_and_remove_all", test_vset_add_and_remove_all}, {"test_vset_fuzzer", test_vset_fuzzer}, {NULL, NULL}};
 unitTest __test_ziplist_c[] = {{"test_ziplistCreateIntList", test_ziplistCreateIntList}, {"test_ziplistPop", test_ziplistPop}, {"test_ziplistGetElementAtIndex3", test_ziplistGetElementAtIndex3}, {"test_ziplistGetElementOutOfRange", test_ziplistGetElementOutOfRange}, {"test_ziplistGetLastElement", test_ziplistGetLastElement}, {"test_ziplistGetFirstElement", test_ziplistGetFirstElement}, {"test_ziplistGetElementOutOfRangeReverse", test_ziplistGetElementOutOfRangeReverse}, {"test_ziplistIterateThroughFullList", test_ziplistIterateThroughFullList}, {"test_ziplistIterateThroughListFrom1ToEnd", test_ziplistIterateThroughListFrom1ToEnd}, {"test_ziplistIterateThroughListFrom2ToEnd", test_ziplistIterateThroughListFrom2ToEnd}, {"test_ziplistIterateThroughStartOutOfRange", test_ziplistIterateThroughStartOutOfRange}, {"test_ziplistIterateBackToFront", test_ziplistIterateBackToFront}, {"test_ziplistIterateBackToFrontDeletingAllItems", test_ziplistIterateBackToFrontDeletingAllItems}, {"test_ziplistDeleteInclusiveRange0To0", test_ziplistDeleteInclusiveRange0To0}, {"test_ziplistDeleteInclusiveRange0To1", test_ziplistDeleteInclusiveRange0To1}, {"test_ziplistDeleteInclusiveRange1To2", test_ziplistDeleteInclusiveRange1To2}, {"test_ziplistDeleteWithStartIndexOutOfRange", test_ziplistDeleteWithStartIndexOutOfRange}, {"test_ziplistDeleteWithNumOverflow", test_ziplistDeleteWithNumOverflow}, {"test_ziplistDeleteFooWhileIterating", test_ziplistDeleteFooWhileIterating}, {"test_ziplistReplaceWithSameSize", test_ziplistReplaceWithSameSize}, {"test_ziplistReplaceWithDifferentSize", test_ziplistReplaceWithDifferentSize}, {"test_ziplistRegressionTestForOver255ByteStrings", test_ziplistRegressionTestForOver255ByteStrings}, {"test_ziplistRegressionTestDeleteNextToLastEntries", test_ziplistRegressionTestDeleteNextToLastEntries}, {"test_ziplistCreateLongListAndCheckIndices", test_ziplistCreateLongListAndCheckIndices}, {"test_ziplistCompareStringWithZiplistEntries", test_ziplistCompareStringWithZiplistEntries}, {"test_ziplistMergeTest", test_ziplistMergeTest}, {"test_ziplistStressWithRandomPayloadsOfDifferentEncoding", test_ziplistStressWithRandomPayloadsOfDifferentEncoding}, {"test_ziplistCascadeUpdateEdgeCases", test_ziplistCascadeUpdateEdgeCases}, {"test_ziplistInsertEdgeCase", test_ziplistInsertEdgeCase}, {"test_ziplistStressWithVariableSize", test_ziplistStressWithVariableSize}, {"test_BenchmarkziplistFind", test_BenchmarkziplistFind}, {"test_BenchmarkziplistIndex", test_BenchmarkziplistIndex}, {"test_BenchmarkziplistValidateIntegrity", test_BenchmarkziplistValidateIntegrity}, {"test_BenchmarkziplistCompareWithString", test_BenchmarkziplistCompareWithString}, {"test_BenchmarkziplistCompareWithNumber", test_BenchmarkziplistCompareWithNumber}, {"test_ziplistStress__ziplistCascadeUpdate", test_ziplistStress__ziplistCascadeUpdate}, {NULL, NULL}};
 unitTest __test_zipmap_c[] = {{"test_zipmapIterateWithLargeKey", test_zipmapIterateWithLargeKey}, {"test_zipmapIterateThroughElements", test_zipmapIterateThroughElements}, {NULL, NULL}};
 unitTest __test_zmalloc_c[] = {{"test_zmallocInitialUsedMemory", test_zmallocInitialUsedMemory}, {"test_zmallocAllocReallocCallocAndFree", test_zmallocAllocReallocCallocAndFree}, {"test_zmallocAllocZeroByteAndFree", test_zmallocAllocZeroByteAndFree}, {NULL, NULL}};
@@ -289,6 +295,7 @@ struct unitTestSuite {
     {"test_util.c", __test_util_c},
     {"test_valkey_strtod.c", __test_valkey_strtod_c},
     {"test_vector.c", __test_vector_c},
+    {"test_vset.c", __test_vset_c},
     {"test_ziplist.c", __test_ziplist_c},
     {"test_zipmap.c", __test_zipmap_c},
     {"test_zmalloc.c", __test_zmalloc_c},