[feat & fix] add importing mode && fix func of swapThreadMain

1. Aimed at controling of some action during importing.
2. See details in Command IMPORT.
3. fixed of list adding node in swapThreadMain.

Author: warrick1016

.github/workflows/ci.yml

@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [ xredis_2_ror ]
+    branches: [ main,deprecated/xredis_2_ror_v1 ]
   pull_request:
-    branches: [ xredis_2_ror ]
+    branches: [ main,deprecated/xredis_2_ror_v1 ]
 
 jobs:
    unit:

src/ctrip_swap.c

@@ -299,15 +299,6 @@ void keyRequestSwapFinished(swapData *data, void *pd, int errcode) {
     ctx->finished(ctx->c,ctx);
 }
 
-/* Expired key should delete only if server is master, check expireIfNeeded
- * for more details. */
-int keyExpiredAndShouldDelete(redisDb *db, robj *key) {
-    if (!keyIsExpired(db,key)) return 0;
-    if (server.masterhost != NULL) return 0;
-    if (checkClientPauseTimeoutAndReturnIfPaused()) return 0;
-    return 1;
-}
-
 #define NOSWAP_REASON_KEYNOTEXISTS 1
 #define NOSWAP_REASON_NOTKEYLEVEL 2
 #define NOSWAP_REASON_KEYNOTSUPPORT 3

src/ctrip_swap_cmd.c

@@ -1007,6 +1007,9 @@ struct redisCommand redisCommandTable[SWAP_CMD_COUNT] = {
      "admin no-script ok-loading fast may-replicate",
      0,NULL,NULL,SWAP_NOP,0,0,0,0,0,0,0},
 
+    {"import",importCommand,-2,
+     "admin use-memory ok-loading fast @dangerous",
+     0,NULL,NULL,SWAP_NOP,0,0,0,0,0,0,0},
 };
 
 struct SwapDataTypeItem {

src/ctrip_swap_data.c

@@ -61,6 +61,8 @@ int swapDataMarkedPropagateExpire(swapData *data) {
 }
 
 static int swapDataExpiredAndShouldDelete(swapData *data) {
+    if (isImportingExpireDisabled()) return 0;
+
     if (!timestampIsExpired(data->expire)) return 0;
     if (server.masterhost != NULL) return 0;
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 0;

src/ctrip_swap_expire.c

@@ -61,11 +61,13 @@ int scanMetaExpireIfNeeded(redisDb *db, scanMeta *meta) {
     if (server.masterhost != NULL) return 1;
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 1;
 
-    /* Delete the key */
-    c = server.swap_expire_clients[db->id];
-    key = createStringObject(meta->key,sdslen(meta->key));
-    submitExpireClientRequest(c, key, 0);
-    decrRefCount(key);
+    if (!isImportingExpireDisabled()) {
+        /* Delete the key */
+        c = server.swap_expire_clients[db->id];
+        key = createStringObject(meta->key,sdslen(meta->key));
+        submitExpireClientRequest(c, key, 0);
+        decrRefCount(key);
+    }
 
     return 1;
 }

src/ctrip_swap_thread.c

@@ -47,8 +47,8 @@ void *swapThreadMain (void *arg) {
         listRewind(thread->pending_reqs, &li);
         processing_reqs = listCreate();
         while ((ln = listNext(&li))) {
-            swapRequest *req = listNodeValue(ln);
-            listAddNodeHead(processing_reqs, req);
+            swapRequestBatch *reqs = listNodeValue(ln);
+            listAddNodeTail(processing_reqs, reqs);
             listDelNode(thread->pending_reqs, ln);
         }
         pthread_mutex_unlock(&thread->lock);

src/db.c

@@ -1769,10 +1769,13 @@ int expireIfNeeded(redisDb *db, robj *key) {
      * have failed over and the new primary will send us the expire. */
     if (checkClientPauseTimeoutAndReturnIfPaused()) return 1;
 
-    /* Delete the key */
+    if (!isImportingExpireDisabled()) {
+        /* Delete the key */
 #ifndef ENABLE_SWAP
-    deleteExpiredKeyAndPropagate(db,key);
+        deleteExpiredKeyAndPropagate(db,key);
 #endif
+    }
+
     return 1;
 }
 

src/server.c

@@ -1129,6 +1129,10 @@ struct redisCommand redisCommandTable[] = {
     {"xslaveof",xslaveofCommand,3,
      "admin no-script ok-stale",
      0,NULL,0,0,0,0,0,0},
+
+    {"import",importCommand,-2,
+     "admin use-memory ok-loading fast @dangerous",
+     0,NULL,0,0,0,0,0,0},
 };
 #endif
 
@@ -1896,7 +1900,7 @@ void clientsCron(void) {
 void databasesCron(void) {
     /* Expire keys by random sampling. Not required for slaves
      * as master will synthesize DELs for us. */
-    if (server.active_expire_enabled) {
+    if (server.active_expire_enabled && !isImportingExpireDisabled()) {
         if (iAmMaster()) {
             activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
         } else {
@@ -2523,7 +2527,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     /* Run a fast expire cycle (the called function will return
      * ASAP if a fast cycle is not needed). */
-    if (server.active_expire_enabled && server.masterhost == NULL)
+    if (server.active_expire_enabled && !isImportingExpireDisabled() && server.masterhost == NULL)
         activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
 
     /* Unblock all the clients blocked for synchronous replication
@@ -3548,6 +3552,8 @@ void InitServerLast() {
     initThreadedIO();
     set_jemalloc_bg_thread(server.jemalloc_bg_thread);
     server.initial_memory_usage = zmalloc_used_memory();
+    server.importing_expire_enabled = 1;
+    server.importing_end_time = 0;
 }
 
 /* Parse the flags string description 'strflags' and set them to the
@@ -4823,6 +4829,115 @@ NULL
     }
 }
 
+/* The import command, only recommended to use in target server 
+ * during keys migration. It means time-based importing mode in server.
+ * As default, Some action will not be executed during this mode,
+ * which are:
+ * 1. expire,
+
+ * import <subcommand> [[arg] [value] [opt] ...]
+
+ * subcommand supported:
+ * import start [ttl]
+ * import end
+ * imoort status
+ * import set ((ttl <seconds>) | ( expire  < 1|0 > ) )
+ * import get (ttl | expire)
+ * 
+ *  */
+
+static inline int isImporting() {
+    return server.importing_end_time >= server.mstime;
+}
+
+void importCommand(client *c) {
+    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
+        const char *help[] = {
+            "start [ttl]",
+            "    importing mode start with seconds of ttl, default ttl as 60s,",
+            "    expire is default disabled.",
+            "end",
+            "    importing mode end.",
+            "status",
+            "    return 1 in importing mode, return 0 if not.",
+            "set ((ttl <seconds>) | ( expire < 1|0 > > ) )",
+            "    set some options.",
+            "get (ttl | expire)",
+            "    get some options.",
+            NULL};
+        addReplyHelp(c, help);
+    } else if ((c->argc == 2 || c->argc == 3) && !strcasecmp(c->argv[1]->ptr,"start")) {
+        if (!isImporting()) {
+            /* if importing mode already off, sub-status will not be inherited,
+             * which is reset to default value.
+             */
+            server.importing_expire_enabled = 0;
+        }
+
+        long long ttl;
+        if (c->argc == 2) {
+            ttl = 3600;
+        } else if (c->argc == 3) {
+            if (getLongLongFromObjectOrReply(c, c->argv[2], &ttl, NULL) != C_OK) {
+                return;
+            }
+        }
+        server.importing_end_time = mstime() + ttl * 1000;
+        addReply(c,shared.ok);
+    } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"end")) {
+        if (!isImporting()) {
+            addReplyError(c,"Importing mode already ended.");
+            return;
+        }
+        server.importing_end_time = -1;
+        addReply(c,shared.ok);
+    } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"status")) {
+        if (isImporting()) {
+            addReplyLongLong(c, 1);
+        } else {
+            addReplyLongLong(c, 0);
+        }
+    } else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr,"get")) {
+        if (!isImporting()) {
+            addReplyError(c,"IMPORT GET must be called in importing mode.");
+            return;
+        }
+
+        if (!strcasecmp(c->argv[2]->ptr,"ttl")) {
+            addReplyLongLong(c, (server.importing_end_time - mstime()) / 1000);
+        } else if (!strcasecmp(c->argv[2]->ptr,"expire")) {
+            addReplyLongLong(c, (long long)server.importing_expire_enabled);
+        } else {
+            addReplyError(c,"Invalid option.");
+        }
+    } else if (c->argc == 4 && !strcasecmp(c->argv[1]->ptr,"set")) {
+        if (!isImporting()) {
+            addReplyError(c,"IMPORT SET must be called in importing mode.");
+            return;
+        }
+
+        if (!strcasecmp(c->argv[2]->ptr,"ttl")) {
+            long long ttl;
+            if (getLongLongFromObjectOrReply(c, c->argv[3], &ttl, NULL) != C_OK) {
+                return;
+            }
+            server.importing_end_time = mstime() + ttl * 1000;
+            addReply(c,shared.ok);
+        } else if (!strcasecmp(c->argv[2]->ptr,"expire")) {
+            server.importing_expire_enabled = (atoi(c->argv[3]->ptr) != 0);
+            addReply(c,shared.ok);
+        } else {
+            addReplyError(c,"Invalid option.");
+        }
+    } else {
+        addReplyError(c,"Invalid subcommand.");
+    }
+}
+
+int isImportingExpireDisabled() {
+    return (isImporting() && (server.importing_expire_enabled == 0));
+}
+
 /* Convert an amount of bytes into a human readable string in the form
  * of 100B, 2G, 100M, 4K, and so forth. */
 void bytesToHuman(char *s, unsigned long long n) {
@@ -5288,6 +5403,8 @@ sds genRedisInfoString(const char *section) {
         atomicGet(server.stat_net_input_bytes, stat_net_input_bytes);
         atomicGet(server.stat_net_output_bytes, stat_net_output_bytes);
 
+        long long importing_ttl = isImporting()? (server.importing_end_time - mstime()):0;
+
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info,
             "# Stats\r\n"
@@ -5328,7 +5445,8 @@ sds genRedisInfoString(const char *section) {
             "total_reads_processed:%lld\r\n"
             "total_writes_processed:%lld\r\n"
             "io_threaded_reads_processed:%lld\r\n"
-            "io_threaded_writes_processed:%lld\r\n",
+            "io_threaded_writes_processed:%lld\r\n"
+            "importing:status=%d,ttl=%lld,expire=%d\r\n",
             server.stat_numconnections,
             server.stat_numcommands,
             getInstantaneousMetric(STATS_METRIC_COMMAND),
@@ -5366,7 +5484,8 @@ sds genRedisInfoString(const char *section) {
             stat_total_reads_processed,
             stat_total_writes_processed,
             server.stat_io_reads_processed,
-            server.stat_io_writes_processed);
+            server.stat_io_writes_processed,
+            isImporting(),importing_ttl,server.importing_expire_enabled);
     }
 
     /* Replication */

src/server.h

@@ -1747,6 +1747,10 @@ struct redisServer {
     long long gtid_ignored_cmd_count;
     long long gtid_executed_cmd_count;
     long long gtid_sync_stat[GTID_SYNC_TYPES];
+
+    /* importing mode */
+    mstime_t importing_end_time;  /* in milliseconds */
+    int importing_expire_enabled;
 };
 
 #define MAX_KEYS_BUFFER 256
@@ -2909,6 +2913,10 @@ void stralgoCommand(client *c);
 void resetCommand(client *c);
 void failoverCommand(client *c);
 
+/* importing mode */
+void importCommand(client *c);
+int isImportingExpireDisabled();
+
 #if defined(__GNUC__)
 void *calloc(size_t count, size_t size) __attribute__ ((deprecated));
 void free(void *ptr) __attribute__ ((deprecated));

tests/test_helper.tcl

@@ -148,6 +148,7 @@ set ::disk_tests {
 	swap/unit/slowlog
 	swap/unit/scripting
 	swap/unit/monitor
+    unit/import_mode
 }
 
 set ::all_tests [concat $::gtid_tests $::all_tests]