Revert " partial work. introduce set expirations"

This reverts commit 04f2006.

Author: ranshid

cmake/Modules/SourceFiles.cmake

@@ -118,8 +118,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/volatile_hashtable.c)
+    ${CMAKE_SOURCE_DIR}/src/volatile_set.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 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 volatile_hashtable.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 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+=$(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/aof.c

@@ -1955,7 +1955,7 @@ static int rioWriteHashIteratorCursor(rio *r, hashTypeIterator *hi, int what) {
  * The function returns 0 on error, 1 on success. */
 int rewriteHashObject(rio *r, robj *key, robj *o) {
     hashTypeIterator hi;
-    long long count = 0, volatile_items = objectNumVolatileElements(o), items = hashTypeLength(o) - volatile_items;
+    long long count = 0, volatile_items = hashTypeNumVolatileElements(o), items = hashTypeLength(o) - volatile_items;
 
     hashTypeInitIterator(o, &hi);
     while (hashTypeNext(&hi) != C_ERR) {
@@ -1983,7 +1983,7 @@ int rewriteHashObject(rio *r, robj *key, robj *o) {
     hashTypeResetIterator(&hi);
 
     /* Now serialize volatile items if exist */
-    if (objectHasVolatileElements(o)) {
+    if (hashTypeHasVolatileElements(o)) {
         hashTypeInitVolatileIterator(o, &hi);
         while (hashTypeNext(&hi) != C_ERR) {
             long long expiry = entryGetExpiry(hi.next);

src/object.c

@@ -526,7 +526,7 @@ void freeZsetObject(robj *o) {
 void freeHashObject(robj *o) {
     switch (o->encoding) {
     case OBJ_ENCODING_HASHTABLE:
-        objectFreeVolatileSet(o);
+        hashTypeFreeVolatileSet(o);
         hashtableRelease((hashtable *)o->ptr);
         break;
     case OBJ_ENCODING_LISTPACK: lpFree(o->ptr); break;

src/rdb.c

@@ -718,7 +718,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
         if (o->encoding == OBJ_ENCODING_LISTPACK)
             return rdbSaveType(rdb, RDB_TYPE_HASH_LISTPACK);
         else if (o->encoding == OBJ_ENCODING_HASHTABLE)
-            if (objectHasVolatileElements(o))
+            if (hashTypeHasVolatileElements(o))
                 return rdbSaveType(rdb, RDB_TYPE_HASH_2);
             else
                 return rdbSaveType(rdb, RDB_TYPE_HASH);
@@ -967,7 +967,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
             }
             nwritten += n;
             /* check if need to add expired time for the hash elements */
-            int add_expiry = objectHasVolatileElements(o);
+            int add_expiry = hashTypeHasVolatileElements(o);
             hashtableIterator iter;
             hashtableInitIterator(&iter, ht, HASHTABLE_ITER_AVOID_ACCESS);
             void *next;
@@ -2211,7 +2211,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
             }
 
             if (rdbtype == RDB_TYPE_HASH_2 && itemexpiry > 0) {
-                objectTrackEntry(o, entry);
+                hashTypeTrackEntry(o, entry);
             }
         }
 

src/server.c

@@ -544,20 +544,6 @@ const void *hashtableSubcommandGetKey(const void *element) {
     return command->declared_name;
 }
 
-const void *hashHashtableTypeGetKey(const void *entry) {
-    return (const void *)entryGetField(entry);
-}
-
-void hashHashtableTypeDestructor(void *entry) {
-    entryFree(entry);
-}
-
-size_t hashHashtableTypeMetadataSize(void) {
-    return sizeof(void *);
-}
-
-extern hashtableElementAccessState hashHashtableTypeAccess(hashtable *ht, void *entry);
-
 /* Generic hash table type where keys are Objects, Values
  * dummy pointers. */
 dictType objectKeyPointerValueDictType = {
@@ -591,17 +577,7 @@ hashtableType objectHashtableType = {
 hashtableType setHashtableType = {
     .hashFunction = dictSdsHash,
     .keyCompare = hashtableSdsKeyCompare,
-    .entryDestructor = hashHashtableTypeDestructor,
-    .getMetadataSize = hashHashtableTypeMetadataSize,
-};
-
-hashtableType setWithVolatileItemsHashtableType = {
-    .hashFunction = dictSdsHash,
-    .keyCompare = hashtableSdsKeyCompare,
-    .entryDestructor = hashHashtableTypeDestructor,
-    .getMetadataSize = hashHashtableTypeMetadataSize,
-    .accessElement = hashHashtableTypeAccess,
-};
+    .entryDestructor = dictSdsDestructor};
 
 const void *zsetHashtableGetKey(const void *element) {
     const zskiplistNode *node = element;
@@ -689,6 +665,19 @@ hashtableType subcommandSetType = {.entryGetKey = hashtableSubcommandGetKey,
                                    .instant_rehashing = 1};
 
 /* Hash type hash table (note that small hashes are represented with listpacks) */
+const void *hashHashtableTypeGetKey(const void *entry) {
+    return (const void *)entryGetField(entry);
+}
+
+void hashHashtableTypeDestructor(void *entry) {
+    entryFree(entry);
+}
+
+size_t hashHashtableTypeMetadataSize(void) {
+    return sizeof(void *);
+}
+
+extern hashtableElementAccessState hashHashtableTypeAccess(hashtable *ht, void *entry);
 
 hashtableType hashHashtableType = {
     .hashFunction = dictSdsHash,

src/server.h

@@ -2647,7 +2647,6 @@ extern dictType objectKeyPointerValueDictType;
 extern hashtableType objectHashtableType;
 extern dictType objectKeyHeapPointerValueDictType;
 extern hashtableType setHashtableType;
-extern hashtableType setWithVolatileItemsHashtableType;
 extern dictType BenchmarkDictType;
 extern hashtableType zsetHashtableType;
 extern hashtableType kvstoreKeysHashtableType;
@@ -3340,7 +3339,7 @@ unsigned long long dbScan(serverDb *db, unsigned long long cursor, hashtableScan
 /* Set data type */
 robj *setTypeCreate(sds value, size_t size_hint);
 int setTypeAdd(robj *subject, sds value);
-int setTypeAddAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds, long long expiry);
+int setTypeAddAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds);
 int setTypeRemove(robj *subject, sds value);
 int setTypeRemoveAux(robj *set, char *str, size_t len, int64_t llval, int str_is_sds);
 int setTypeIsMember(robj *subject, sds value);
@@ -3361,16 +3360,10 @@ robj *setTypeDup(robj *o);
 #define HASH_SET_KEEP_EXPIRY (1 << 2)
 #define HASH_SET_COPY 0
 
-/* Objects with volatile items API */
-volatile_set *objectGetVolatileSet(robj *o);
-volatile_set *objectGetOrcreateVolatileSet(robj *o);
-void objectFreeVolatileSet(robj *o);
-void objectTrackEntry(robj *o, void *entry);
-void objectUntrackEntry(robj *o, void *entry);
-void objectTrackUpdateEntry(robj *o, void *old_entry, void *new_entry, long long old_expiry, long long new_expiry);
-int objectHasVolatileElements(robj *o);
-size_t objectNumVolatileElements(robj *o);
-void objectIgnoreTTL(robj *o, int ignore);
+
+void hashTypeFreeVolatileSet(robj *o);
+void hashTypeTrackEntry(robj *o, void *entry);
+void hashTypeUntrackEntry(robj *o, void *entry);
 
 void hashTypeConvert(robj *o, int enc);
 void hashTypeTryConversion(robj *subject, robj **argv, int start, int end);
@@ -3392,6 +3385,8 @@ robj *hashTypeLookupWriteOrCreate(client *c, robj *key);
 robj *hashTypeGetValueObject(robj *o, sds field);
 int hashTypeSet(robj *o, sds field, sds value, long long expiry, int flags);
 robj *hashTypeDup(robj *o);
+int hashTypeHasVolatileElements(robj *o);
+size_t hashTypeNumVolatileElements(robj *o);
 
 /* Pub / Sub */
 int pubsubUnsubscribeAllChannels(client *c, int notify);

src/t_hash.c

@@ -43,12 +43,124 @@
 #include "entry.h"
 
 
+int hashTypeExpireEntry(entry *entry);
+
+volatileEntryType hashvolatileEntryType = {
+    .entryGetKey = (sds(*)(const void *entry))entryGetField,
+    .getExpiry = (long long (*)(const void *entry))entryGetExpiry,
+    .expire = hashTypeExpireEntry,
+};
+
+/*-----------------------------------------------------------------------------
+ * Hash type Expiry API
+ *----------------------------------------------------------------------------*/
+
+static volatile_set *hashTypeGetVolatileSet(robj *o) {
+    serverAssert(o->encoding == OBJ_ENCODING_HASHTABLE);
+    return *(volatile_set **)hashtableMetadata(o->ptr);
+}
+
+void hashTypeFreeVolatileSet(robj *o) {
+    volatile_set *set = hashTypeGetVolatileSet(o);
+    if (set)
+        freeVolatileSet(set);
+}
+
+int hashTypeHasVolatileElements(robj *o) {
+    return ((o->encoding == OBJ_ENCODING_HASHTABLE) && (hashTypeGetVolatileSet(o) != NULL));
+}
+
+size_t hashTypeNumVolatileElements(robj *o) {
+    if (hashTypeHasVolatileElements(o)) {
+        return volatileSetNumEntries(hashTypeGetVolatileSet(o));
+    }
+    return 0;
+}
+
+void hashTypeIgnoreTTL(robj *o, int ignore) {
+    if (o->encoding == OBJ_ENCODING_HASHTABLE) {
+        /* prevent placing access function if not needed */
+        if (!ignore && !hashTypeHasVolatileElements(o)) {
+            ignore = 0;
+        }
+        hashtableSetType(o->ptr, ignore ? &hashHashtableType : &hashWithVolatileItemsHashtableType);
+    }
+}
+
+static volatile_set *
+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);
+        /* serves mainly for optimization. Use type which supports access function only when needed. */
+        hashTypeIgnoreTTL(o, 0);
+    }
+    return *volatile_set_ref;
+}
+
+static void hashTypeDeleteVolatileSet(robj *o) {
+    volatile_set **volatile_set_ref = hashtableMetadata(o->ptr);
+    freeVolatileSet(*volatile_set_ref);
+    *volatile_set_ref = NULL;
+    /* serves mainly for optimization. by changing the hashtable type we can avoid extra function call in hashtable access */
+    hashTypeIgnoreTTL(o, 1);
+}
+
+void hashTypeTrackEntry(robj *o, void *entry) {
+    volatile_set *set = hashTypeGetOrcreateVolatileSet(o);
+    serverAssert(volatileSetAddEntry(set, entry, entryGetExpiry(entry)));
+}
+
+void hashTypeUntrackEntry(robj *o, void *entry) {
+    if (!entryHasExpiry(entry)) return;
+    volatile_set *set = hashTypeGetVolatileSet(o);
+    debugServerAssert(set);
+    serverAssert(volatileSetRemoveEntry(set, entry, entryGetExpiry(entry)));
+    if (volatileSetNumEntries(set) == 0) {
+        hashTypeDeleteVolatileSet(o);
+    }
+}
+
+static void hashTypeTrackUpdateEntry(robj *o, void *old_entry, void *new_entry, long long old_expiry, long long new_expiry) {
+    int old_tracked = (old_entry && old_expiry != EXPIRY_NONE);
+    int new_tracked = (new_entry && new_expiry != EXPIRY_NONE);
+    /* If entry was not tracked before and not going to be tracked now, we can simply return */
+    if (!old_tracked && !new_tracked)
+        return;
+
+    volatile_set *set = hashTypeGetOrcreateVolatileSet(o);
+    debugServerAssert(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) {
+        hashTypeDeleteVolatileSet(o);
+    }
+}
+
 int hashTypeExpireEntry(void *entry) {
     // TBD
     UNUSED(entry);
     return 1;
 }
 
+hashtableElementAccessState hashHashtableTypeAccess(hashtable *ht, void *entry) {
+    UNUSED(ht);
+
+    if (!canExpireWithFlags(0, NULL)) return ELEMENT_VALID;
+
+    if (!entryIsExpired(entry)) return ELEMENT_VALID;
+
+    return ELEMENT_INVALID;
+}
 
 /*-----------------------------------------------------------------------------
  * Hash type API
@@ -281,7 +393,7 @@ int hashTypeSet(robj *o, sds field, sds value, long long expiry, int flags) {
 
         /* We have to ignore the TTL when setting an element. this is mainly in order to be able to update an existing expired
          * entry and not have it remain in the hashtable with the same field/value. */
-        objectIgnoreTTL(o, 1);
+        hashTypeIgnoreTTL(o, 1);
         hashtablePosition position;
         void *existing;
         if (hashtableFindPositionForInsert(ht, field, &position, &existing)) {
@@ -290,7 +402,7 @@ int hashTypeSet(robj *o, sds field, sds value, long long expiry, int flags) {
             hashtableInsertAtPosition(ht, entry, &position);
             /* In case an expiry is set on the new entry, we need to track it */
             if (expiry != EXPIRY_NONE) {
-                objectTrackEntry(o, entry);
+                hashTypeTrackEntry(o, entry);
             }
         } else {
             /* exists: replace value */
@@ -307,11 +419,11 @@ int hashTypeSet(robj *o, sds field, sds value, long long expiry, int flags) {
                 int replaced = hashtableReplaceReallocatedEntry(ht, existing, new_entry);
                 serverAssert(replaced);
             }
-            objectTrackUpdateEntry(o, existing, new_entry, entry_expiry, expiry);
+            hashTypeTrackUpdateEntry(o, existing, new_entry, entry_expiry, expiry);
 
             update = is_expired ? 0 : 1;
         }
-        objectIgnoreTTL(o, 0);
+        hashTypeIgnoreTTL(o, 0);
     } else {
         serverPanic("Unknown hash encoding");
     }
@@ -383,7 +495,7 @@ int hashTypeSetExpire(robj *o, sds field, long long expiry, int flag) {
             }
         }
         *entry_ref = entrySetExpiry(current_entry, expiry);
-        objectTrackUpdateEntry(o, current_entry, *entry_ref, current_expire, expiry);
+        hashTypeTrackUpdateEntry(o, current_entry, *entry_ref, current_expire, expiry);
         return 1;
     }
     return -2; // we did not find anything to do. return -2
@@ -408,7 +520,7 @@ int hashTypePersist(robj *o, sds field) {
         entry *current_entry = *entry_ref;
         long long current_expire = entryGetExpiry(current_entry);
         if (current_expire != EXPIRY_NONE) {
-            objectUntrackEntry(o, current_entry);
+            hashTypeUntrackEntry(o, current_entry);
             *entry_ref = entryUpdate(current_entry, NULL, EXPIRY_NONE);
             return 1;
         }
@@ -441,7 +553,7 @@ int hashTypeDelete(robj *o, sds field) {
         void *entry = NULL;
         deleted = hashtablePop(ht, field, &entry);
         if (deleted) {
-            objectUntrackEntry(o, entry);
+            hashTypeUntrackEntry(o, entry);
             entryFree(entry);
         }
     } else {
@@ -486,7 +598,7 @@ void hashTypeInitVolatileIterator(robj *subject, hashTypeIterator *hi) {
     if (hi->encoding == OBJ_ENCODING_LISTPACK) {
         return;
     } else if (hi->encoding == OBJ_ENCODING_HASHTABLE) {
-        volatileSetStart(objectGetVolatileSet(subject), &hi->viter);
+        volatileSetStart(hashTypeGetVolatileSet(subject), &hi->viter);
     } else {
         serverPanic("Unknown hash encoding");
     }
@@ -701,7 +813,7 @@ robj *hashTypeDup(robj *o) {
             entry *entry = entryCreate(field, sdsdup(value), expiry);
             hashtableAdd(ht, entry);
             if (expiry != EXPIRY_NONE)
-                objectTrackEntry(hobj, entry);
+                hashTypeTrackEntry(hobj, entry);
         }
         hashTypeResetIterator(&hi);
     } else {