Introduce atomic slot migration

cmake/Modules/SourceFiles.cmake

@@ -44,6 +44,7 @@ set(VALKEY_SERVER_SRCS
     ${CMAKE_SOURCE_DIR}/src/intset.c
     ${CMAKE_SOURCE_DIR}/src/syncio.c
     ${CMAKE_SOURCE_DIR}/src/cluster.c
+    ${CMAKE_SOURCE_DIR}/src/cluster_migrateslots.c
     ${CMAKE_SOURCE_DIR}/src/cluster_legacy.c
     ${CMAKE_SOURCE_DIR}/src/cluster_slot_stats.c
     ${CMAKE_SOURCE_DIR}/src/crc16.c

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_cluster.o trace/trace_server.o trace/trace_rdb.o trace/trace_aof.o
-ENGINE_SERVER_OBJ=threads_mngr.o adlist.o vector.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 vset.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 vector.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 cluster_migrateslots.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 vset.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

@@ -2218,6 +2218,103 @@ static int rewriteFunctions(rio *aof) {
     return 0;
 }
 
+int rewriteSelectDbRio(rio *aof, int db_num) {
+    char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
+    if (rioWrite(aof, selectcmd, sizeof(selectcmd) - 1) == 0) return C_ERR;
+    if (rioWriteBulkLongLong(aof, db_num) == 0) return C_ERR;
+    return C_OK;
+}
+
+int rewriteObjectRio(rio *aof, robj *o, int db_num) {
+    size_t aof_bytes_before_key = aof->processed_bytes;
+    sds keystr;
+    robj key;
+    long long expiretime;
+
+    keystr = objectGetKey(o);
+    initStaticStringObject(key, keystr);
+
+    expiretime = objectGetExpire(o);
+
+    /* Save the key and associated value */
+    if (o->type == OBJ_STRING) {
+        /* Emit a SET command */
+        char cmd[] = "*3\r\n$3\r\nSET\r\n";
+        if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) return C_ERR;
+        /* Key and value */
+        if (rioWriteBulkObject(aof, &key) == 0) return C_ERR;
+        if (rioWriteBulkObject(aof, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_LIST) {
+        if (rewriteListObject(aof, &key, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_SET) {
+        if (rewriteSetObject(aof, &key, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_ZSET) {
+        if (rewriteSortedSetObject(aof, &key, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_HASH) {
+        if (rewriteHashObject(aof, &key, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_STREAM) {
+        if (rewriteStreamObject(aof, &key, o) == 0) return C_ERR;
+    } else if (o->type == OBJ_MODULE) {
+        if (rewriteModuleObject(aof, &key, o, db_num) == 0) return C_ERR;
+    } else {
+        serverPanic("Unknown object type");
+    }
+
+    /* In fork child process, we can try to release memory back to the
+     * OS and possibly avoid or decrease COW. We give the dismiss
+     * mechanism a hint about an estimated size of the object we stored. */
+    size_t dump_size = aof->processed_bytes - aof_bytes_before_key;
+    if (server.in_fork_child) dismissObject(o, dump_size);
+
+    /* Save the expire time */
+    if (expiretime != -1) {
+        char cmd[] = "*3\r\n$9\r\nPEXPIREAT\r\n";
+        if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) return C_ERR;
+        if (rioWriteBulkObject(aof, &key) == 0) return C_ERR;
+        if (rioWriteBulkLongLong(aof, expiretime) == 0) return C_ERR;
+    }
+
+    /* Delay before next key if required (for testing) */
+    if (server.rdb_key_save_delay) debugDelay(server.rdb_key_save_delay);
+
+    return C_OK;
+}
+
+int rewriteSlotToAppendOnlyFileRio(rio *aof, int db_num, int hashslot, size_t *key_count) {
+    long long updated_time = 0;
+
+    if (rewriteFunctions(aof) == 0) return C_ERR;
+    if (dbHasNoKeys(db_num)) return C_OK;
+
+    serverDb *db = server.db[db_num];
+    if (kvstoreHashtableSize(db->keys, hashslot) == 0) return C_OK;
+
+    /* SELECT the DB */
+    if (rewriteSelectDbRio(aof, db_num) == C_ERR) return C_ERR;
+
+    kvstoreHashtableIterator *iter = kvstoreGetHashtableIterator(db->keys, hashslot, HASHTABLE_ITER_SAFE | HASHTABLE_ITER_PREFETCH_VALUES);
+    void *next;
+    while (kvstoreHashtableIteratorNext(iter, &next)) {
+        robj *o = next;
+
+        /* Update info every 1 second (approximately).
+         * in order to avoid calling mstime() on each iteration, we will
+         * check the diff every 1024 keys */
+        if (key_count && ((*key_count)++ & 1023) == 0) {
+            long long now = mstime();
+            if (now - updated_time >= 1000) {
+                sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, *key_count, "AOF rewrite");
+                updated_time = now;
+            }
+        }
+
+        if (rewriteObjectRio(aof, o, db_num) == C_ERR) return C_ERR;
+    }
+
+    kvstoreReleaseHashtableIterator(iter);
+    return C_OK;
+}
+
 int rewriteAppendOnlyFileRio(rio *aof) {
     int j;
     long key_count = 0;
@@ -2234,69 +2331,20 @@ int rewriteAppendOnlyFileRio(rio *aof) {
         sdsfree(ts);
     }
 
-    if (rewriteFunctions(aof) == 0) goto werr;
+    if (rewriteFunctions(aof) == C_ERR) goto werr;
 
     for (j = 0; j < server.dbnum; j++) {
-        char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
         if (dbHasNoKeys(j)) continue;
         serverDb *db = server.db[j];
 
         /* SELECT the new DB */
-        if (rioWrite(aof, selectcmd, sizeof(selectcmd) - 1) == 0) goto werr;
-        if (rioWriteBulkLongLong(aof, j) == 0) goto werr;
+        if (rewriteSelectDbRio(aof, j) == C_ERR) goto werr;
 
         kvs_it = kvstoreIteratorInit(db->keys, HASHTABLE_ITER_SAFE | HASHTABLE_ITER_PREFETCH_VALUES);
         /* Iterate this DB writing every entry */
         void *next;
         while (kvstoreIteratorNext(kvs_it, &next)) {
             robj *o = next;
-            sds keystr;
-            robj key;
-            long long expiretime;
-            size_t aof_bytes_before_key = aof->processed_bytes;
-
-            keystr = objectGetKey(o);
-            initStaticStringObject(key, keystr);
-
-            expiretime = objectGetExpire(o);
-
-            /* Save the key and associated value */
-            if (o->type == OBJ_STRING) {
-                /* Emit a SET command */
-                char cmd[] = "*3\r\n$3\r\nSET\r\n";
-                if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
-                /* Key and value */
-                if (rioWriteBulkObject(aof, &key) == 0) goto werr;
-                if (rioWriteBulkObject(aof, o) == 0) goto werr;
-            } else if (o->type == OBJ_LIST) {
-                if (rewriteListObject(aof, &key, o) == 0) goto werr;
-            } else if (o->type == OBJ_SET) {
-                if (rewriteSetObject(aof, &key, o) == 0) goto werr;
-            } else if (o->type == OBJ_ZSET) {
-                if (rewriteSortedSetObject(aof, &key, o) == 0) goto werr;
-            } else if (o->type == OBJ_HASH) {
-                if (rewriteHashObject(aof, &key, o) == 0) goto werr;
-            } else if (o->type == OBJ_STREAM) {
-                if (rewriteStreamObject(aof, &key, o) == 0) goto werr;
-            } else if (o->type == OBJ_MODULE) {
-                if (rewriteModuleObject(aof, &key, o, j) == 0) goto werr;
-            } else {
-                serverPanic("Unknown object type");
-            }
-
-            /* In fork child process, we can try to release memory back to the
-             * OS and possibly avoid or decrease COW. We give the dismiss
-             * mechanism a hint about an estimated size of the object we stored. */
-            size_t dump_size = aof->processed_bytes - aof_bytes_before_key;
-            if (server.in_fork_child) dismissObject(o, dump_size);
-
-            /* Save the expire time */
-            if (expiretime != -1) {
-                char cmd[] = "*3\r\n$9\r\nPEXPIREAT\r\n";
-                if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
-                if (rioWriteBulkObject(aof, &key) == 0) goto werr;
-                if (rioWriteBulkLongLong(aof, expiretime) == 0) goto werr;
-            }
 
             /* Update info every 1 second (approximately).
              * in order to avoid calling mstime() on each iteration, we will
@@ -2309,8 +2357,7 @@ int rewriteAppendOnlyFileRio(rio *aof) {
                 }
             }
 
-            /* Delay before next key if required (for testing) */
-            if (server.rdb_key_save_delay) debugDelay(server.rdb_key_save_delay);
+            if (rewriteObjectRio(aof, o, j) == C_ERR) goto werr;
         }
         kvstoreIteratorRelease(kvs_it);
     }

src/blocked.c

@@ -104,8 +104,8 @@ void freeClientBlockingState(client *c) {
  * flag is set client query buffer is not longer processed, but accumulated,
  * and will be processed when the client is unblocked. */
 void blockClient(client *c, int btype) {
-    /* Primary client should never be blocked unless pause or module */
-    serverAssert(!(c->flag.primary && btype != BLOCKED_MODULE && btype != BLOCKED_POSTPONE));
+    /* Replicated clients should never be blocked unless pause or module */
+    serverAssert(!(isReplicatedClient(c) && btype != BLOCKED_MODULE && btype != BLOCKED_POSTPONE));
 
     initClientBlockingState(c);
 

src/cluster.c

@@ -1596,7 +1596,6 @@ void resetClusterStats(void) {
     clusterSlotStatResetAll();
 }
 
-
 void clusterCommandFlushslot(client *c) {
     int slot;
     int lazy = server.lazyfree_lazy_user_flush;

src/cluster.h

@@ -28,6 +28,9 @@
 #define CLUSTER_REDIR_DOWN_UNBOUND 6  /* -CLUSTERDOWN, unbound slot. */
 #define CLUSTER_REDIR_DOWN_RO_STATE 7 /* -CLUSTERDOWN, allow reads. */
 
+/* Fixed timeout value for cluster operations (milliseconds) */
+#define CLUSTER_OPERATION_TIMEOUT 2000
+
 typedef struct _clusterNode clusterNode;
 struct clusterState;
 
@@ -38,6 +41,10 @@ struct clusterState;
 #define CLUSTER_MODULE_FLAG_NO_FAILOVER (1 << 1)
 #define CLUSTER_MODULE_FLAG_NO_REDIRECTION (1 << 2)
 
+/* For clusterBroadcastPong */
+#define CLUSTER_BROADCAST_ALL 0            /* All known instances. */
+#define CLUSTER_BROADCAST_LOCAL_REPLICAS 1 /* All replicas in my primary-replicas ring. */
+
 /* ---------------------- API exported outside cluster.c -------------------- */
 /* functions requiring mechanism specific implementations */
 void clusterInit(void);
@@ -62,6 +69,7 @@ void clusterUpdateMyselfAnnouncedPorts(void);
 void clusterUpdateMyselfHumanNodename(void);
 
 void clusterPropagatePublish(robj *channel, robj *message, int sharded);
+void clusterBroadcastPong(int target);
 
 unsigned long getClusterConnectionsCount(void);
 int isClusterHealthy(void);
@@ -118,6 +126,9 @@ void clearCachedClusterSlotsResponse(void);
 unsigned int countKeysInSlotForDb(unsigned int hashslot, serverDb *db);
 unsigned int countKeysInSlot(unsigned int hashslot);
 int getSlotOrReply(client *c, robj *o);
+int getNodeDefaultReplicationPort(clusterNode *node);
+bool isAnySlotInManualImportingState(void);
+bool isAnySlotInManualMigratingState(void);
 
 /* functions with shared implementations */
 int clusterNodeIsMyself(clusterNode *n);
@@ -137,4 +148,13 @@ long long getNodeReplicationOffset(clusterNode *node);
 sds aggregateClientOutputBuffer(client *c);
 void resetClusterStats(void);
 unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event);
+
+unsigned int propagateSlotDeletionByKeys(unsigned int hashslot);
+void clusterUpdateState(void);
+void clusterSaveConfigOrDie(int do_fsync);
+int clusterDelSlot(int slot);
+int clusterAddSlot(clusterNode *n, int slot);
+int clusterBumpConfigEpochWithoutConsensus(void);
+void clusterDoBeforeSleep(int flags);
+
 #endif /* __CLUSTER_H */