Convert pubsub dicts to hashtables (#2007)

Two dicts are converted to hashtables:

1. On each client, the set of channels/patterns/shard-channels the
   client is subscribed to
2. On each channel or pattern, the set of clients subscribed to it.

---------

Signed-off-by: Rain Valentine <rsg000@gmail.com>
Signed-off-by: Viktor Söderqvist <viktor.soderqvist@est.tech>
Co-authored-by: Rain Valentine <rsg000@gmail.com>

Author: zuiderkwast

src/acl.c

@@ -1899,41 +1899,44 @@ static list *getUpcomingChannelList(user *new, user *original) {
 /* Check if the client should be killed because it is subscribed to channels that were
  * permitted in the past, are not in the `upcoming` channel list. */
 static int ACLShouldKillPubsubClient(client *c, list *upcoming) {
-    robj *o;
-    int kill = 0;
-
     if (getClientType(c) == CLIENT_TYPE_PUBSUB) {
+        int kill = 0;
+
         /* Check for pattern violations. */
-        dictIterator *di = dictGetIterator(c->pubsub_data->pubsub_patterns);
-        dictEntry *de;
-        while (!kill && ((de = dictNext(di)) != NULL)) {
-            o = dictGetKey(de);
+        hashtableIterator iter;
+        hashtableInitIterator(&iter, c->pubsub_data->pubsub_patterns, 0);
+        void *next;
+        while (!kill && hashtableNext(&iter, &next)) {
+            robj *o = next;
             int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 1);
             kill = (res == ACL_DENIED_CHANNEL);
         }
-        dictReleaseIterator(di);
+        hashtableResetIterator(&iter);
 
         /* Check for channel violations. */
         if (!kill) {
             /* Check for global channels violation. */
-            di = dictGetIterator(c->pubsub_data->pubsub_channels);
-
-            while (!kill && ((de = dictNext(di)) != NULL)) {
-                o = dictGetKey(de);
+            hashtableIterator iter;
+            hashtableInitIterator(&iter, c->pubsub_data->pubsub_channels, 0);
+            void *next;
+            while (!kill && hashtableNext(&iter, &next)) {
+                robj *o = next;
                 int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 0);
                 kill = (res == ACL_DENIED_CHANNEL);
             }
-            dictReleaseIterator(di);
+            hashtableResetIterator(&iter);
         }
         if (!kill) {
             /* Check for shard channels violation. */
-            di = dictGetIterator(c->pubsub_data->pubsubshard_channels);
-            while (!kill && ((de = dictNext(di)) != NULL)) {
-                o = dictGetKey(de);
+            hashtableIterator iter;
+            hashtableInitIterator(&iter, c->pubsub_data->pubsubshard_channels, 0);
+            void *next;
+            while (!kill && hashtableNext(&iter, &next)) {
+                robj *o = next;
                 int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 0);
                 kill = (res == ACL_DENIED_CHANNEL);
             }
-            dictReleaseIterator(di);
+            hashtableResetIterator(&iter);
         }
 
         if (kill) {

src/defrag.c

@@ -51,7 +51,6 @@
 typedef enum { DEFRAG_NOT_DONE = 0,
                DEFRAG_DONE = 1 } doneStatus;
 
-
 /*
  * Defragmentation is performed in stages.  Each stage is serviced by a stage function
  * (defragStageFn).  The stage function is passed a target (void*) to defrag.  The contents of that
@@ -135,7 +134,7 @@ typedef struct {
 static_assert(offsetof(defragKeysCtx, kvstate) == 0, "defragStageKvstoreHelper requires this");
 
 // Private data for pubsub kvstores
-typedef dict *(*getClientChannelsFn)(client *);
+typedef hashtable *(*getClientChannelsFn)(client *);
 typedef struct {
     getClientChannelsFn fn;
 } getClientChannelsFnWrapper;
@@ -243,30 +242,6 @@ robj *activeDefragStringOb(robj *ob) {
     return new_robj;
 }
 
-/* Defrag helper for dict main allocations (dict struct, and hash tables).
- * Receives a pointer to the dict* and return a new dict* when the dict
- * struct itself was moved.
- *
- * Returns NULL in case the allocation wasn't moved.
- * When it returns a non-null value, the old pointer was already released
- * and should NOT be accessed. */
-static dict *dictDefragTables(dict *d) {
-    dict *ret = NULL;
-    dictEntry **newtable;
-    /* handle the dict struct */
-    if ((ret = activeDefragAlloc(d))) d = ret;
-    /* handle the first hash table */
-    if (!d->ht_table[0]) return ret; /* created but unused */
-    newtable = activeDefragAlloc(d->ht_table[0]);
-    if (newtable) d->ht_table[0] = newtable;
-    /* handle the second hash table */
-    if (d->ht_table[1]) {
-        newtable = activeDefragAlloc(d->ht_table[1]);
-        if (newtable) d->ht_table[1] = newtable;
-    }
-    return ret;
-}
-
 /* Internal function used by zslDefrag */
 static void zslUpdateNode(zskiplist *zsl, zskiplistNode *oldnode, zskiplistNode *newnode, zskiplistNode **update) {
     int i;
@@ -786,37 +761,33 @@ static void dbKeysScanCallback(void *privdata, void *elemref) {
 /* Defrag scan callback for a pubsub channels hashtable. */
 static void defragPubsubScanCallback(void *privdata, void *elemref) {
     defragPubSubCtx *ctx = privdata;
-    void **channel_dict_ref = (void **)elemref;
-    dict *newclients, *clients = *channel_dict_ref;
-    robj *newchannel, *channel = *(robj **)dictMetadata(clients);
-    size_t allocation_size;
+    void **clients_ref = (void **)elemref;
+    hashtable *newclients, *clients = *clients_ref;
+    robj *newchannel, *channel = *(robj **)hashtableMetadata(clients);
 
     /* Try to defrag the channel name. */
-    serverAssert(channel->refcount == (int)dictSize(clients) + 1);
-    newchannel = activeDefragStringObWithoutFree(channel, &allocation_size);
+    serverAssert(channel->refcount == (int)hashtableSize(clients) + 1);
+    newchannel = activeDefragStringOb(channel);
     if (newchannel) {
-        *(robj **)dictMetadata(clients) = newchannel;
+        *(robj **)hashtableMetadata(clients) = newchannel;
 
         /* The channel name is shared by the client's pubsub(shard) and server's
          * pubsub(shard), after defraging the channel name, we need to update
          * the reference in the clients' dictionary. */
-        dictIterator *di = dictGetIterator(clients);
-        dictEntry *clientde;
-        while ((clientde = dictNext(di)) != NULL) {
-            client *c = dictGetKey(clientde);
-            dict *client_channels = ctx->getPubSubChannels(c);
-            dictEntry *pubsub_channel = dictFind(client_channels, newchannel);
-            serverAssert(pubsub_channel);
-            dictSetKey(ctx->getPubSubChannels(c), pubsub_channel, newchannel);
+        hashtableIterator iter;
+        hashtableInitIterator(&iter, clients, 0);
+        void *c;
+        while (hashtableNext(&iter, &c)) {
+            hashtable *client_channels = ctx->getPubSubChannels(c);
+            int replaced = hashtableReplaceReallocatedEntry(client_channels, channel, newchannel);
+            serverAssert(replaced);
         }
-        dictReleaseIterator(di);
-        // Now that we're done correcting the references, we can safely free the old channel robj
-        allocatorDefragFree(channel, allocation_size);
+        hashtableResetIterator(&iter);
     }
 
     /* Try to defrag the dictionary of clients that is stored as the value part. */
-    if ((newclients = dictDefragTables(clients)))
-        *channel_dict_ref = newclients;
+    if ((newclients = hashtableDefragTables(clients, activeDefragAlloc)))
+        *clients_ref = newclients;
 
     server.stat_active_defrag_scanned++;
 }

src/networking.c

@@ -3504,9 +3504,9 @@ sds catClientInfoString(sds s, client *client, int hide_user_data) {
             " flags=%s", flags,
             " capa=%s", capa,
             " db=%i", client->db->id,
-            " sub=%i", client->pubsub_data ? (int)dictSize(client->pubsub_data->pubsub_channels) : 0,
-            " psub=%i", client->pubsub_data ? (int)dictSize(client->pubsub_data->pubsub_patterns) : 0,
-            " ssub=%i", client->pubsub_data ? (int)dictSize(client->pubsub_data->pubsubshard_channels) : 0,
+            " sub=%i", client->pubsub_data ? (int)hashtableSize(client->pubsub_data->pubsub_channels) : 0,
+            " psub=%i", client->pubsub_data ? (int)hashtableSize(client->pubsub_data->pubsub_patterns) : 0,
+            " ssub=%i", client->pubsub_data ? (int)hashtableSize(client->pubsub_data->pubsubshard_channels) : 0,
             " multi=%i", client->mstate ? client->mstate->count : -1,
             " watch=%i", client->mstate ? (int)listLength(&client->mstate->watched_keys) : 0,
             " qbuf=%U", client->querybuf ? (unsigned long long)sdslen(client->querybuf) : 0,