Add ACTIVE-EXPIRY and ACTIVE-DEFRAG for hash objects with volatile items

src/Makefile

@@ -626,7 +626,7 @@ gcov:
 	$(MAKE) SERVER_CFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" SERVER_LDFLAGS="-fprofile-arcs -ftest-coverage"
 
 noopt:
-	$(MAKE) OPTIMIZATION="-O0"
+	$(MAKE) OPTIMIZATION="-O0 -fno-lto"
 
 valgrind:
 	$(MAKE) OPTIMIZATION="-O0" MALLOC="libc"

src/db.c

@@ -483,6 +483,13 @@ int dbGenericDeleteWithDictIndex(serverDb *db, robj *key, int async, int flags,
             debugServerAssert(0 == kvstoreHashtableDelete(db->expires, dict_index, key->ptr));
         }
 
+        /* If deleting a hash object, untrack the object if it contains volatile items. */
+        if (val->type == OBJ_HASH && val->encoding == OBJ_ENCODING_HASHTABLE && hashTypeHasVolatileElements(val)) {
+            dbUntrackKeyWithVolaItems(db, val); // TODO xbasel, dbUntrackKeyWithVolaItems should accept optional dict_index (it's available here).
+        } else {
+            debugServerAssert(0 == dbUntrackKeyWithVolaItems(db, val));
+        }
+
         if (async) {
             freeObjAsync(key, val, db->id);
         } else {
@@ -501,6 +508,18 @@ int dbGenericDelete(serverDb *db, robj *key, int async, int flags) {
     return dbGenericDeleteWithDictIndex(db, key, async, flags, dict_index);
 }
 
+/* Add a key with volatile items to the tracking kvstore.  */
+int dbTrackKeyWithVolaItems(serverDb *db, robj *o) {
+    int dict_index = getKVStoreIndexForKey(objectGetKey(o));
+    return kvstoreHashtableAdd(db->keys_with_volatile_items, dict_index, o);
+}
+
+/* Delete a key from the keys with volatile entries tracking kvstore  */
+int dbUntrackKeyWithVolaItems(serverDb *db, robj *o) {
+    int dict_index = getKVStoreIndexForKey(objectGetKey(o));
+    return kvstoreHashtableDelete(db->keys_with_volatile_items, dict_index, objectGetKey(o));
+}
+
 /* Delete a key, value, and associated expiration entry if any, from the DB */
 int dbSyncDelete(serverDb *db, robj *key) {
     return dbGenericDelete(db, key, 0, DB_FLAG_KEY_DELETED);
@@ -582,6 +601,7 @@ long long emptyDbStructure(serverDb **dbarray, int dbnum, int async, void(callba
         } else {
             kvstoreEmpty(dbarray[j]->keys, callback);
             kvstoreEmpty(dbarray[j]->expires, callback);
+            kvstoreEmpty(dbarray[j]->keys_with_volatile_items, callback);
         }
         /* Because all keys of database are removed, reset average ttl. */
         dbarray[j]->avg_ttl = 0;
@@ -1550,7 +1570,7 @@ void copyCommand(client *c) {
     case OBJ_LIST: newobj = listTypeDup(o); break;
     case OBJ_SET: newobj = setTypeDup(o); break;
     case OBJ_ZSET: newobj = zsetDup(o); break;
-    case OBJ_HASH: newobj = hashTypeDup(o); break;
+    case OBJ_HASH: newobj = hashTypeDup(c->db, o); break;
     case OBJ_STREAM: newobj = streamDup(o); break;
     case OBJ_MODULE:
         newobj = moduleTypeDupOrReply(c, key, newkey, dst->id, o);

src/defrag.c

@@ -146,6 +146,11 @@ typedef struct {
 } defragPubSubCtx;
 static_assert(offsetof(defragPubSubCtx, kvstate) == 0, "defragStageKvstoreHelper requires this");
 
+/* Context for defragmenting hash objects: holds the DB and the hash key object. */
+typedef struct {
+    serverDb *db;
+    robj *o;
+} objectDbContext;
 
 /* When scanning a main kvstore, large elements are queued for later handling rather than
  * causing a large latency spike while processing a hash table bucket.  This list is only used
@@ -452,18 +457,21 @@ static void activeDefragEntry(void *privdata, void *element_ref) {
     if (new_entry) {
         /* In case the entry is tracked we need to update it in the volatile set */
         if (entryHasExpiry(new_entry)) {
-            robj *obj = (robj *)privdata;
-            serverAssert(obj);
-            hashTypeTrackUpdateEntry(obj, old_entry, new_entry, old_expiry, entryGetExpiry(new_entry));
+            objectDbContext *ctx = privdata;
+            serverAssert(ctx->o);
+            serverAssert(ctx->db);
+            hashTypeTrackUpdateEntry(ctx->db, ctx->o, old_entry, new_entry, old_expiry, entryGetExpiry(new_entry));
         }
         *entry_ref = new_entry;
     }
 }
 
-static void scanLaterHash(robj *ob, unsigned long *cursor) {
+static void scanLaterHash(robj *ob, unsigned long *cursor, int dbid) {
+    serverDb *db = server.db[dbid];
     serverAssert(ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_HASHTABLE);
     hashtable *ht = ob->ptr;
-    *cursor = hashtableScanDefrag(ht, *cursor, activeDefragEntry, ob, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
+    objectDbContext ctx = {db, ob};
+    *cursor = hashtableScanDefrag(ht, *cursor, activeDefragEntry, &ctx, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
 }
 
 static void defragQuicklist(robj *ob) {
@@ -500,15 +508,16 @@ static void defragZsetSkiplist(robj *ob) {
     }
 }
 
-static void defragHash(robj *ob) {
+static void defragHash(serverDb *db, robj *ob) {
     serverAssert(ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_HASHTABLE);
     hashtable *ht = ob->ptr;
     if (hashtableSize(ht) > server.active_defrag_max_scan_fields) {
         defragLater(ob);
     } else {
         unsigned long cursor = 0;
         do {
-            cursor = hashtableScanDefrag(ht, cursor, activeDefragEntry, ob, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
+            objectDbContext ctx = {db, ob};
+            cursor = hashtableScanDefrag(ht, cursor, activeDefragEntry, &ctx, activeDefragAlloc, HASHTABLE_SCAN_EMIT_REF);
         } while (cursor != 0);
     }
     /* defrag the hashtable struct and tables */
@@ -744,7 +753,7 @@ static void defragKey(defragKeysCtx *ctx, robj **elemref) {
         if (ob->encoding == OBJ_ENCODING_LISTPACK) {
             if ((newzl = activeDefragAlloc(ob->ptr))) ob->ptr = newzl;
         } else if (ob->encoding == OBJ_ENCODING_HASHTABLE) {
-            defragHash(ob);
+            defragHash(db, ob);
         } else {
             serverPanic("Unknown hash encoding");
         }
@@ -813,7 +822,7 @@ static int defragLaterItem(robj *ob, unsigned long *cursor, monotime endtime, in
         } else if (ob->type == OBJ_ZSET && ob->encoding == OBJ_ENCODING_SKIPLIST) {
             scanLaterZset(ob, cursor);
         } else if (ob->type == OBJ_HASH && ob->encoding == OBJ_ENCODING_HASHTABLE) {
-            scanLaterHash(ob, cursor);
+            scanLaterHash(ob, cursor, dbid);
         } else if (ob->type == OBJ_STREAM && ob->encoding == OBJ_ENCODING_STREAM) {
             return scanLaterStreamListpacks(ob, cursor, endtime);
         } else if (ob->type == OBJ_MODULE) {
@@ -973,6 +982,16 @@ static doneStatus defragStageExpiresKvstore(monotime endtime, void *target, void
                                     scanHashtableCallbackCountScanned, NULL, NULL);
 }
 
+// Target is a DBID
+static doneStatus defragStageKeysWithvolaItemsKvstore(monotime endtime, void *target, void *privdata) {
+    UNUSED(privdata);
+    int dbid = (uintptr_t)target;
+    serverDb *db = server.db[dbid];
+    return defragStageKvstoreHelper(endtime, db->keys_with_volatile_items,
+                                    scanHashtableCallbackCountScanned, NULL, NULL);
+}
+
+
 
 static doneStatus defragStagePubsubKvstore(monotime endtime, void *target, void *privdata) {
     // target is server.pubsub_channels or server.pubsubshard_channels
@@ -1240,6 +1259,7 @@ static void beginDefragCycle(void) {
         if (dbHasNoKeys(dbid)) continue;
         addDefragStage(defragStageDbKeys, (void *)(uintptr_t)dbid, NULL);
         addDefragStage(defragStageExpiresKvstore, (void *)(uintptr_t)dbid, NULL);
+        addDefragStage(defragStageKeysWithvolaItemsKvstore, (void *)(uintptr_t)dbid, NULL);
     }
 
     static getClientChannelsFnWrapper getClientPubSubChannelsFn = {getClientPubSubChannels};