No checksum performed on diskless replication (#14851)

In #14575, we disable rdb compression when using diskless replication.
Actually rdb checksum also costs much CPU, and it is used to check RDB
disk file instead of replication stream, so when using diskless
replication (`diskless sync` is enabled on master, `diskless load` is
enabled on replica), we don't store/load any rdb file into/from disk, so
we can skip the checksum in this scenario to accelerate rdb delivery.

People may argue the checksum for replication stream can recognize the
network data corruption, but actually we can not recognize the network
data corruption issue for incremental replication-stream/commands, so it
is limited, we should use `tls` feature to ensure the data correctness.

I tested on AWS, ec2 type is c8g.2xlarge, data size is 12.37GB, all keys
are string type, keys size is 1024.
- If master and replica are different ec2 instances but in the same
placement group, the full synchronization time reduced from 16s to 11s.
- If the master and replica are in the same ec2 instance, the full
synchronization time reduced from 15s to 10s.
- If master and replica are different ec2 instances but not in the same
placement group, there is no change.

It seems network latency is the bottleneck now.

And after #14575 and this PR, the full synchronization time reduced from
35s to 11s when master and replica instances are in the same placement
group. A decrease of nearly 68.5%!

src/rdb.c

@@ -3550,6 +3550,7 @@ void updateLoadingFileName(char* filename) {
 void stopLoading(int success) {
     server.loading = 0;
     server.async_loading = 0;
+    server.loading_skip_checksum = 0;
     blockingOperationEnds();
     rdbFileBeingLoaded = NULL;
 
@@ -3587,7 +3588,7 @@ void stopSaving(int success) {
 /* Track loading progress in order to serve client's from time to time
    and if needed calculate rdb checksum  */
 void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
-    if (server.rdb_checksum)
+    if (server.rdb_checksum && !server.loading_skip_checksum)
         rioGenericUpdateChecksum(r, buf, len);
     if (server.loading_process_events_interval_bytes &&
         (r->processed_bytes + len)/server.loading_process_events_interval_bytes > r->processed_bytes/server.loading_process_events_interval_bytes)
@@ -4017,7 +4018,7 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
         uint64_t cksum, expected = rdb->cksum;
 
         if (rioRead(rdb,&cksum,8) == 0) goto eoferr;
-        if (server.rdb_checksum && !server.skip_checksum_validation) {
+        if (server.rdb_checksum && !server.loading_skip_checksum && !server.skip_checksum_validation) {
             memrev64ifbe(&cksum);
             if (cksum == 0) {
                 serverLog(LL_NOTICE,"RDB file was saved with checksum disabled: no check performed.");
@@ -4301,9 +4302,12 @@ int rdbSaveToSlavesSockets(int req, rdbSaveInfo *rsi) {
         redisSetProcTitle("redis-rdb-to-slaves");
         redisSetCpuAffinity(server.bgsave_cpulist);
 
-        /* Disable RDB compression if requested. */
+        /* Disable RDB compression and checksum in the fork child if requested.
+         * The parent's configuration is not affected. */
         if (req & SLAVE_REQ_RDB_NO_COMPRESS)
             server.rdb_compression = 0;
+        if (req & SLAVE_REQ_RDB_NO_CHECKSUM)
+            server.rdb_checksum = 0;
 
         if (req & SLAVE_REQ_SLOTS_SNAPSHOT) {
             /* Slots snapshot is required */

src/replication.c

@@ -1549,9 +1549,11 @@ void replconfCommand(client *c) {
                 return;
             }
             c->main_ch_client_id = (uint64_t)client_id;
-            /* Inherit the rdb-no-compress request from the main channel. */
+            /* Inherit the rdb-no-compress and rdb-no-checksum request from the main channel. */
             if (main_ch->slave_req & SLAVE_REQ_RDB_NO_COMPRESS)
                 c->slave_req |= SLAVE_REQ_RDB_NO_COMPRESS;
+            if (main_ch->slave_req & SLAVE_REQ_RDB_NO_CHECKSUM)
+                c->slave_req |= SLAVE_REQ_RDB_NO_CHECKSUM;
         } else if (!strcasecmp(c->argv[j]->ptr, "rdb-no-compress")) {
             long rdb_no_compress = 0;
             if (getRangeLongFromObjectOrReply(c, c->argv[j + 1], 0, 1, &rdb_no_compress, NULL) != C_OK)
@@ -1561,6 +1563,15 @@ void replconfCommand(client *c) {
             } else {
                 c->slave_req &= ~SLAVE_REQ_RDB_NO_COMPRESS;
             }
+        } else if (!strcasecmp(c->argv[j]->ptr, "rdb-no-checksum")) {
+            long rdb_no_checksum = 0;
+            if (getRangeLongFromObjectOrReply(c, c->argv[j + 1], 0, 1, &rdb_no_checksum, NULL) != C_OK)
+                return;
+            if (rdb_no_checksum == 1) {
+                c->slave_req |= SLAVE_REQ_RDB_NO_CHECKSUM;
+            } else {
+                c->slave_req &= ~SLAVE_REQ_RDB_NO_CHECKSUM;
+            }
         } else {
             addReplyErrorFormat(c,"Unrecognized REPLCONF option: %s",
                 (char*)c->argv[j]->ptr);
@@ -2429,6 +2440,11 @@ void readSyncBulkPayload(connection *conn) {
         rioInitWithConn(&rdb,conn,server.repl_transfer_size);
         disklessLoadingRio = &rdb;
 
+        /* Disable checksum verification when diskless on both master and replica.
+         * The RDB checksum is designed to detect disk corruption, but if the data
+         * never touched disk, we can skip verification. */
+        if (usemark) server.loading_skip_checksum = 1;
+
         /* Empty db */
         loadingSetFlags(NULL, server.repl_transfer_size, asyncLoading);
         if (server.repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
@@ -3008,7 +3024,7 @@ void syncWithMaster(connection *conn) {
     char tmpfile[256], *err = NULL;
     int dfd = -1, maxtries = 5;
     int psync_result;
-    static int replconf_rdb_no_compress = 0;
+    static int no_compress_checksum = 0;
 
     /* If this event fired after the user turned the instance into a master
      * with SLAVEOF NO ONE we must just return ASAP. */
@@ -3108,11 +3124,13 @@ void syncWithMaster(connection *conn) {
         }
 
         /* If we are not going to save the RDB to disk, request that RDB
-         * compression be disabled, which speeds up RDB delivery. */
-        replconf_rdb_no_compress = 0;
+         * compression and checksum be disabled, which speeds up RDB delivery
+         * and loading. */
+        no_compress_checksum = 0;
         if (useDisklessLoad()) {
-            replconf_rdb_no_compress = 1;
-            err = sendCommand(conn, "REPLCONF", "rdb-no-compress", "1", NULL);
+            no_compress_checksum = 1;
+            err = sendCommand(conn, "REPLCONF", "rdb-no-compress", "1",
+                                    "rdb-no-checksum", "1", NULL);
             if (err) goto write_error;
         }
 
@@ -3166,7 +3184,7 @@ void syncWithMaster(connection *conn) {
     }
 
     if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY && !server.slave_announce_ip)
-        server.repl_state = REPL_STATE_RECEIVE_COMP_REPLY;
+        server.repl_state = REPL_STATE_RECEIVE_REQ_REPLY;
 
     /* Receive REPLCONF ip-address reply. */
     if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) {
@@ -3179,22 +3197,22 @@ void syncWithMaster(connection *conn) {
                                 "REPLCONF ip-address: %s", err);
         }
         sdsfree(err);
-        server.repl_state = REPL_STATE_RECEIVE_COMP_REPLY;
+        server.repl_state = REPL_STATE_RECEIVE_REQ_REPLY;
         return;
     }
 
-    if (server.repl_state == REPL_STATE_RECEIVE_COMP_REPLY && !replconf_rdb_no_compress)
+    if (server.repl_state == REPL_STATE_RECEIVE_REQ_REPLY && !no_compress_checksum)
         server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;
 
-    /* Receive REPLCONF rdb-no-compress reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_COMP_REPLY) {
+    /* Receive REPLCONF REQUEST reply (rdb-no-compress and rdb-no-checksum). */
+    if (server.repl_state == REPL_STATE_RECEIVE_REQ_REPLY) {
         err = receiveSynchronousResponse(conn);
         if (err == NULL) goto no_response_error;
         /* Ignore the error if any, not all the Redis versions support
-         * REPLCONF rdb-no-compress. */
+         * REPLCONF rdb-no-compress and rdb-no-checksum. */
         if (err[0] == '-') {
             serverLog(LL_NOTICE,"(Non critical) Master does not understand "
-                                "REPLCONF rdb-no-compress: %s", err);
+                                "REPLCONF rdb-no-compress/checksum: %s", err);
         }
         sdsfree(err);
         server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;

src/server.h

@@ -520,7 +520,7 @@ typedef enum {
     REPL_STATE_RECEIVE_AUTH_REPLY,  /* Wait for AUTH reply */
     REPL_STATE_RECEIVE_PORT_REPLY,  /* Wait for REPLCONF reply */
     REPL_STATE_RECEIVE_IP_REPLY,    /* Wait for REPLCONF reply */
-    REPL_STATE_RECEIVE_COMP_REPLY,  /* Wait for REPLCONF reply */
+    REPL_STATE_RECEIVE_REQ_REPLY,   /* Wait for REPLCONF reply */
     REPL_STATE_RECEIVE_CAPA_REPLY,  /* Wait for REPLCONF reply */
     REPL_STATE_SEND_PSYNC,          /* Send PSYNC */
     REPL_STATE_RECEIVE_PSYNC_REPLY, /* Wait for PSYNC reply */
@@ -580,13 +580,17 @@ typedef enum {
 #define SLAVE_CAPA_PSYNC2           (1<<1) /* Supports PSYNC2 protocol. */
 #define SLAVE_CAPA_RDB_CHANNEL_REPL (1<<2) /* Supports rdb channel replication during full sync */
 
-/* Slave requirements */
+/* Slave requirements. NO_COMPRESS and NO_CHECKSUM are hints rather than strict
+ * requirements - the replica can handle compressed/checksummed RDB either way,
+ * but prefers to skip them for diskless loading since they become redundant.
+ * We reuse the REQ mechanism for simplicity, avoiding a separate HINT bitfield. */
 #define SLAVE_REQ_NONE                  0
 #define SLAVE_REQ_RDB_EXCLUDE_DATA      (1 << 0) /* Exclude data from RDB */
 #define SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS (1 << 1) /* Exclude functions from RDB */
 #define SLAVE_REQ_SLOTS_SNAPSHOT        (1 << 2) /* Only slots snapshot is required */
 #define SLAVE_REQ_RDB_CHANNEL           (1 << 3) /* Use rdb channel replication, transfer RDB background */
 #define SLAVE_REQ_RDB_NO_COMPRESS       (1 << 4) /* Don't enable RDB compression */
+#define SLAVE_REQ_RDB_NO_CHECKSUM       (1 << 5) /* Don't enable RDB checksum */
 /* Mask of all bits in the slave requirements bitfield that represent non-standard (filtered) RDB requirements */
 #define SLAVE_REQ_RDB_MASK (SLAVE_REQ_RDB_EXCLUDE_DATA | SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS | SLAVE_REQ_SLOTS_SNAPSHOT)
 
@@ -2033,6 +2037,7 @@ struct redisServer {
     off_t loading_loaded_bytes;
     time_t loading_start_time;
     off_t loading_process_events_interval_bytes;
+    int loading_skip_checksum; /* Skip checksum verification during diskless loading */
     /* Fields used only for stats */
     time_t stat_starttime;          /* Server start time */
     long long stat_numcommands;     /* Number of processed commands */