add expirationPools to optimize expire efficiency

Signed-off-by: Ray Cao <zisong.cw@alibaba-inc.com>

Author: RayaCoo

src/expire.c

@@ -37,6 +37,68 @@
 
 #include "server.h"
 
+
+/* TODO: comment */
+
+#define EPPOOL_SIZE 20
+#define EPPOOL_CACHED_SDS_SIZE 255
+struct expirationPoolEntry {
+    long long expiretime; /* Absolute expiration time */
+    robj *entry;          /* The entry with expire time */
+};
+
+/* Create an array of expire pools, one for each database. */
+static struct expirationPoolEntry **ExpirationPools;
+
+void expirationPoolAlloc(void) {
+    int dbnum = server.dbnum;
+
+    ExpirationPools = zmalloc(sizeof(struct expirationPoolEntry *) * dbnum);
+
+    for (int i = 0; i < dbnum; i++) {
+        ExpirationPools[i] = zmalloc(sizeof(struct expirationPoolEntry) * EPPOOL_SIZE);
+        for (int j = 0; j < EPPOOL_SIZE; j++) {
+            ExpirationPools[i][j].expiretime = 0;
+            ExpirationPools[i][j].entry = NULL;
+        }
+    }
+}
+
+void expirationPoolPopulate(serverDb *db, robj *val, long long expiretime) {
+    struct expirationPoolEntry *pool = ExpirationPools[db->id];
+    int k;
+
+    k = 0;
+    while (k < EPPOOL_SIZE && pool[k].entry && pool[k].expiretime < expiretime) k++;
+
+    /* If we can't insert (all slots filled with sooner-to-expire entries), return */
+    if (k == 0 && pool[EPPOOL_SIZE - 1].entry != NULL) {
+        return;
+    } else if (k < EPPOOL_SIZE && pool[k].entry == NULL) {
+        /* Inserting into empty position. No setup needed before insert. */
+    } else {
+        /* Inserting in the middle. Now k points to the first element
+         * with expire time greater than the element to insert. */
+        if (pool[EPPOOL_SIZE - 1].entry == NULL) {
+            /* Free space on the right? Insert at k shifting
+             * all the elements from k to end to the right. */
+            memmove(pool + k + 1, pool + k, sizeof(pool[0]) * (EPPOOL_SIZE - k - 1));
+        } else {
+            /* No free space on right? Insert at k-1 */
+            k--;
+            /* Shift all elements on the left of k (included) to the
+             * left, so we discard the element with smallest expire time. */
+            if (pool[0].entry) decrRefCount(pool[0].entry);
+            memmove(pool, pool + 1, sizeof(pool[0]) * k);
+        }
+    }
+
+    /* Store the entry and increment its refcount */
+    pool[k].entry = val;
+    incrRefCount(val);
+    pool[k].expiretime = expiretime;
+}
+
 /*-----------------------------------------------------------------------------
  * Incremental collection of expired keys.
  *
@@ -135,7 +197,8 @@ typedef struct {
 void expireScanCallback(void *privdata, void *entry) {
     robj *val = entry;
     expireScanData *data = privdata;
-    long long ttl = objectGetExpire(val) - data->now;
+    long long expiretime = objectGetExpire(val);
+    long long ttl = expiretime - data->now;
     if (activeExpireCycleTryExpire(data->db, val, data->now)) {
         data->expired++;
         /* Propagate the DEL command */
@@ -145,6 +208,9 @@ void expireScanCallback(void *privdata, void *entry) {
         /* We want the average TTL of keys yet not expired. */
         data->ttl_sum += ttl;
         data->ttl_samples++;
+
+        /* Try to add this entry to the expiration pool for future expiration */
+        expirationPoolPopulate(data->db, val, expiretime);
     }
     data->sampled++;
 }
@@ -252,6 +318,39 @@ void activeExpireCycle(int type) {
 
         if (kvstoreSize(db->expires)) dbs_performed++;
 
+        /* First, try to expire keys from the expire pool */
+        struct expirationPoolEntry *pool = ExpirationPools[db->id];
+        long long now = mstime();
+        int expired_from_pool = 0;
+
+        /* Traverse from back to front, stop when we find a non-expired key */
+        for (int i = EPPOOL_SIZE - 1; i >= 0; i--) {
+            if (pool[i].entry == NULL) continue;
+            if (pool[i].expiretime <= now) {
+                /* Key is expired, try to expire it */
+                if (activeExpireCycleTryExpire(db, pool[i].entry, now)) {
+                    expired_from_pool++;
+                    /* Propagate the DEL command */
+                    postExecutionUnitOperations();
+                }
+
+                /* Clean up the pool entry */
+                decrRefCount(pool[i].entry);
+                pool[i].entry = NULL;
+                pool[i].expiretime = 0;
+            } else {
+                /* Since we're traversing from back to front and keys are sorted by expire time,
+                 * if we find a non-expired key, all keys before it are also not expired */
+                break;
+            }
+        }
+
+        /* If we found expired keys in the pool, update stats */
+        if (expired_from_pool > 0) {
+            total_expired += expired_from_pool;
+            total_sampled += expired_from_pool;
+        }
+
         /* Continue to expire if at the end of the cycle there are still
          * a big percentage of keys to expire, compared to the number of keys
          * we scanned. The percentage, stored in config_cycle_acceptable_stale

src/server.c

@@ -2844,6 +2844,7 @@ void initServer(void) {
         server.db[j].avg_ttl = 0;
     }
     evictionPoolAlloc(); /* Initialize the LRU keys pool. */
+    expirationPoolAlloc();   /* Initialize the expire pools. */
     /* Note that server.pubsub_channels was chosen to be a kvstore (with only one dict, which
      * seems odd) just to make the code cleaner by making it be the same type as server.pubsubshard_channels
      * (which has to be kvstore), see pubsubtype.serverPubSubChannels */

src/server.h

@@ -3555,6 +3555,8 @@ void handleBlockedClientsTimeout(void);
 int clientsCronHandleTimeout(client *c, mstime_t now_ms);
 
 /* expire.c -- Handling of expired keys */
+void expirationPoolAlloc(void);
+void expirationPoolPopulate(serverDb *db, robj *val, long long expiretime);
 void activeExpireCycle(int type);
 void expireReplicaKeys(void);
 void rememberReplicaKeyWithExpire(serverDb *db, robj *key);