Add main-thread incremental snapshots (snapshot-mode forkless)

src/Makefile

@@ -604,6 +604,7 @@ ENGINE_SERVER_OBJ = \
     crdt/hlc.o \
     crdt/merge.o \
     crdt/peer.o \
+    rdb_forkless.o \
     rdma.o \
     release.o \
     replication.o \

src/config.c

@@ -189,6 +189,10 @@ configEnum bgsave_method_enum[] = {{"fork", RDB_BGSAVE_TYPE_FORK},
                                    {"forkless", RDB_BGSAVE_TYPE_FORKLESS},
                                    {NULL, 0}};
 
+configEnum snapshot_mode_enum[] = {{"fork", SNAPSHOT_MODE_FORK},
+                                   {"forkless", SNAPSHOT_MODE_FORKLESS},
+                                   {NULL, 0}};
+
 configEnum cluster_replica_no_failover_enum[] = {{"no", CLUSTER_REPLICA_NO_FAILOVER_NO},
                                                  {"yes", CLUSTER_REPLICA_NO_FAILOVER_YES},
                                                  {"if-empty", CLUSTER_REPLICA_NO_FAILOVER_IF_EMPTY},
@@ -2475,8 +2479,8 @@ static void numericConfigRewrite(standardConfig *config, const char *name, struc
     }
 
 #define createSpecialConfig(name, alias, modifiable, setfn, getfn, rewritefn, applyfn) \
-    {.type = SPECIAL_CONFIG,                                                           \
-     embedCommonConfig(name, alias, modifiable) embedConfigInterface(NULL, setfn, getfn, rewritefn, applyfn)}
+    { .type = SPECIAL_CONFIG,                                                          \
+      embedCommonConfig(name, alias, modifiable) embedConfigInterface(NULL, setfn, getfn, rewritefn, applyfn) }
 
 static int isValidBgsaveDefaultMethod(int val, const char **err) {
     /* During startup config parsing the directives are applied one by one, so
@@ -3449,6 +3453,8 @@ standardConfig static_configs[] = {
     createBoolConfig("activerehashing", NULL, MODIFIABLE_CONFIG, server.activerehashing, 1, NULL, NULL),
     createBoolConfig("stop-writes-on-bgsave-error", NULL, MODIFIABLE_CONFIG, server.stop_writes_on_bgsave_err, 1, NULL, NULL),
     createEnumConfig("bgsave-default-method", NULL, MODIFIABLE_CONFIG, bgsave_method_enum, server.bgsave_default_method, RDB_BGSAVE_TYPE_FORK, isValidBgsaveDefaultMethod, NULL),
+    createEnumConfig("snapshot-mode", NULL, MODIFIABLE_CONFIG, snapshot_mode_enum, server.snapshot_mode, SNAPSHOT_MODE_FORK, NULL, NULL),
+    createIntConfig("snapshot-forkless-slice-keys", NULL, MODIFIABLE_CONFIG, 1, 1000000, server.snapshot_forkless_slice_keys, 64, INTEGER_CONFIG, NULL, NULL),
     createBoolConfig("set-proc-title", NULL, IMMUTABLE_CONFIG, server.set_proc_title, 1, NULL, NULL), /* Should setproctitle be used? */
     createBoolConfig("lazyfree-lazy-eviction", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_eviction, 1, NULL, NULL),
     createBoolConfig("lazyfree-lazy-expire", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_expire, 1, NULL, NULL),

src/db.c

@@ -29,6 +29,7 @@
 
 #include "server.h"
 #include "crdt/peer.h"
+#include "rdb_forkless.h"
 #include "listpack.h"
 #include "hotkeys.h"
 #include "ordered_index.h"
@@ -86,6 +87,8 @@ static robj *dbFindWithDictIndex(serverDb *db, sds key, int dict_index);
  * expired on replicas even if the primary is lagging expiring our key via DELs
  * in the replication link. */
 robj *lookupKey(serverDb *db, robj *key, int flags) {
+    /* Capture the pre-image before expiry or the caller mutates the value. */
+    if ((flags & LOOKUP_WRITE) && snapshotForklessInProgress()) snapshotForklessBeforeWrite(db, key);
     int dict_index = getKVStoreIndexUsingCachedSlot(objectGetVal(key));
     robj *val = dbFindWithDictIndex(db, objectGetVal(key), dict_index);
     if (val) {
@@ -546,6 +549,7 @@ int dbGenericDeleteWithDictIndex(serverDb *db, robj *key, int async, int flags,
 
 /* Helper for sync and async delete. */
 int dbGenericDelete(serverDb *db, robj *key, int async, int flags) {
+    snapshotForklessBeforeWrite(db, key);
     int dict_index = getKVStoreIndexForKey(objectGetVal(key));
     return dbGenericDeleteWithDictIndex(db, key, async, flags, dict_index);
 }
@@ -858,6 +862,7 @@ void flushAllDataAndResetRDB(int flags) {
     server.dirty += emptyData(-1, flags, NULL);
     if (isForkBgsaveInProgress()) killRDBChild();
     if (isForklessSaveInProgress()) forklessSaveCancel();
+    if (snapshotForklessInProgress()) snapshotForklessCancel();
     if (server.child_type == CHILD_TYPE_SLOT_MIGRATION) killSlotMigrationChild();
     if (server.saveparamslen > 0) {
         rdbSaveInfo rsi, *rsiptr;

src/rdb.c

@@ -50,6 +50,7 @@
 #include "compression.h"
 #include "compression_stream.h"
 #include "forkless.h"
+#include "rdb_forkless.h"
 
 #include <math.h>
 #include <fcntl.h>
@@ -1813,6 +1814,10 @@ int rdbSaveToFile(const char *filename) {
 /* Save the DB on disk. Return C_ERR on error, C_OK on success. */
 int rdbSave(int req, char *filename, rdbSaveInfo *rsi, int rdbflags) {
     char tmpfile[256];
+    /* A synchronous save replaces an in-progress incremental snapshot.
+     * The incremental file has not been renamed yet, so the previous dump
+     * is still the live file until this save finishes. */
+    if (snapshotForklessInProgress()) snapshotForklessCancel();
     char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */
 
     startSaving(rdbflags);
@@ -1855,7 +1860,7 @@ int isForkBgsaveInProgress(void) {
 }
 
 int isSaveInProgress(void) {
-    return isForkBgsaveInProgress() || isForklessSaveInProgress();
+    return isForkBgsaveInProgress() || isForklessSaveInProgress() || snapshotForklessInProgress();
 }
 
 /* Choose the background save method based on configuration. Returns forkless
@@ -1879,6 +1884,7 @@ int resolveBgsaveType(void) {
 
 /* Start a background save, choosing fork or forkless based on bgsave_type. */
 int rdbStartBgsave(int bgsave_type) {
+    if (server.snapshot_mode == SNAPSHOT_MODE_FORKLESS) return snapshotForklessStart(server.rdb_filename);
     if (bgsave_type == RDB_BGSAVE_TYPE_FORKLESS) {
         return forklessSaveToDisk(server.rdb_filename);
     } else {
@@ -4541,6 +4547,10 @@ void bgsaveCommand(client *c) {
                 serverLog(LL_NOTICE, "Background saving (forkless) will be aborted due to user request");
                 forklessSaveCancel();
                 addReplyStatus(c, "Background saving cancelled");
+            } else if (snapshotForklessInProgress()) {
+                serverLog(LL_NOTICE, "Background saving (forkless snapshot) will be aborted due to user request");
+                snapshotForklessCancel();
+                addReplyStatus(c, "Background saving cancelled");
             } else if (server.rdb_bgsave_scheduled != RDB_BGSAVE_TYPE_NONE) {
                 serverLog(LL_NOTICE, "Scheduled background saving will be cancelled due to user request");
                 server.rdb_bgsave_scheduled = RDB_BGSAVE_TYPE_NONE;
@@ -4576,6 +4586,12 @@ void bgsaveCommand(client *c) {
                              "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever "
                              "possible.");
         }
+    } else if (server.snapshot_mode == SNAPSHOT_MODE_FORKLESS) {
+        if (snapshotForklessStart(server.rdb_filename) == C_OK) {
+            addReplyStatus(c, "Background saving started");
+        } else {
+            addReplyErrorObject(c, shared.err);
+        }
     } else if (chosen_save_type == RDB_BGSAVE_TYPE_FORKLESS) {
         if (forklessSaveToDisk(server.rdb_filename) == C_OK) {
             addReplyStatus(c, "Background saving started");