Contribute threadsave

Author: Nitai Caro

src/Makefile

@@ -561,7 +561,8 @@ ENGINE_SERVER_OBJ = \
     vset.o \
     ziplist.o \
     zipmap.o \
-    zmalloc.o
+    zmalloc.o \
+	threadsave.o
 ENGINE_SERVER_OBJ+=$(ENGINE_TRACE_OBJ)
 ENGINE_CLI_NAME=$(ENGINE_NAME)-cli$(PROG_SUFFIX)
 ENGINE_CLI_OBJ = \

src/bgiteration.c

@@ -54,10 +54,6 @@ static bool isDeleteCmd(struct serverCommand *cmd) {
 }
 
 
-static bool onValkeyMainThread(void) {
-    return (pthread_equal(server.main_thread_id, pthread_self()) != 0);
-}
-
 /* Parse a parameters robj, extracting a valid DBID.
  * Returns FALSE if DBID isn't valid.
  */

src/commands.def

@@ -7170,6 +7170,7 @@ struct COMMAND_STRUCT ACL_Subcommands[] = {
 commandHistory BGSAVE_History[] = {
 {"3.2.2","Added the `SCHEDULE` option."},
 {"8.1.0","Added the `CANCEL` option."},
+{"10.0.0","Added the `FORK` and `THREAD` options. `SCHEDULE` can be combined with `FORK` or `THREAD`."},
 };
 #endif
 
@@ -7183,15 +7184,17 @@ commandHistory BGSAVE_History[] = {
 #define BGSAVE_Keyspecs NULL
 #endif
 
-/* BGSAVE operation argument table */
-struct COMMAND_ARG BGSAVE_operation_Subargs[] = {
-{MAKE_ARG("schedule",ARG_TYPE_PURE_TOKEN,-1,"SCHEDULE",NULL,"3.2.2",CMD_ARG_NONE,0,NULL)},
-{MAKE_ARG("cancel",ARG_TYPE_PURE_TOKEN,-1,"CANCEL",NULL,"8.1.0",CMD_ARG_NONE,0,NULL)},
+/* BGSAVE save_type argument table */
+struct COMMAND_ARG BGSAVE_save_type_Subargs[] = {
+{MAKE_ARG("fork",ARG_TYPE_PURE_TOKEN,-1,"FORK",NULL,NULL,CMD_ARG_NONE,0,NULL)},
+{MAKE_ARG("thread",ARG_TYPE_PURE_TOKEN,-1,"THREAD",NULL,NULL,CMD_ARG_NONE,0,NULL)},
 };
 
 /* BGSAVE argument table */
 struct COMMAND_ARG BGSAVE_Args[] = {
-{MAKE_ARG("operation",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,2,NULL),.subargs=BGSAVE_operation_Subargs},
+{MAKE_ARG("schedule",ARG_TYPE_PURE_TOKEN,-1,"SCHEDULE",NULL,"3.2.2",CMD_ARG_OPTIONAL,0,NULL)},
+{MAKE_ARG("save-type",ARG_TYPE_ONEOF,-1,NULL,NULL,"10.0.0",CMD_ARG_OPTIONAL,2,NULL),.subargs=BGSAVE_save_type_Subargs},
+{MAKE_ARG("cancel",ARG_TYPE_PURE_TOKEN,-1,"CANCEL",NULL,"8.1.0",CMD_ARG_OPTIONAL,0,NULL)},
 };
 
 /********** COMMAND COUNT ********************/
@@ -11992,7 +11995,7 @@ struct COMMAND_STRUCT serverCommandTable[] = {
 /* server */
 {MAKE_CMD("acl","A container for Access List Control commands.","Depends on subcommand.","6.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,ACL_History,0,ACL_Tips,0,NULL,-2,CMD_SENTINEL,ACL_CATEGORY_SLOW,NULL,ACL_Keyspecs,0,NULL,0),.subcommands=ACL_Subcommands},
 {MAKE_CMD("bgrewriteaof","Asynchronously rewrites the append-only file to disk.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,BGREWRITEAOF_History,0,BGREWRITEAOF_Tips,0,bgrewriteaofCommand,1,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,BGREWRITEAOF_Keyspecs,0,NULL,0)},
-{MAKE_CMD("bgsave","Asynchronously saves the database(s) to disk.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,BGSAVE_History,2,BGSAVE_Tips,0,bgsaveCommand,-1,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,BGSAVE_Keyspecs,0,NULL,1),.args=BGSAVE_Args},
+{MAKE_CMD("bgsave","Asynchronously saves the database(s) to disk.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,BGSAVE_History,3,BGSAVE_Tips,0,bgsaveCommand,-1,CMD_NO_ASYNC_LOADING|CMD_ADMIN|CMD_NOSCRIPT,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,BGSAVE_Keyspecs,0,NULL,3),.args=BGSAVE_Args},
 {MAKE_CMD("command","Returns detailed information about all commands.","O(N) where N is the total number of commands","2.8.13",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,COMMAND_History,0,COMMAND_Tips,1,commandCommand,-1,CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_CONNECTION|ACL_CATEGORY_SLOW,NULL,COMMAND_Keyspecs,0,NULL,0),.subcommands=COMMAND_Subcommands},
 {MAKE_CMD("commandlog","A container for command log commands.","Depends on subcommand.","8.1.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,COMMANDLOG_History,0,COMMANDLOG_Tips,0,NULL,-2,0,ACL_CATEGORY_SLOW,NULL,COMMANDLOG_Keyspecs,0,NULL,0),.subcommands=COMMANDLOG_Subcommands},
 {MAKE_CMD("config","A container for server configuration commands.","Depends on subcommand.","2.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_History,0,CONFIG_Tips,0,NULL,-2,0,ACL_CATEGORY_SLOW,NULL,CONFIG_Keyspecs,0,NULL,0),.subcommands=CONFIG_Subcommands},

src/commands/bgsave.json

@@ -14,6 +14,10 @@
             [
                 "8.1.0",
                 "Added the `CANCEL` option."
+            ],
+            [
+                "10.0.0",
+                "Added the `FORK` and `THREAD` options. `SCHEDULE` can be combined with `FORK` or `THREAD`."
             ]
         ],
         "command_flags": [
@@ -23,29 +27,45 @@
         ],
         "arguments": [
             {
-                "name": "operation",
+                "name": "schedule",
+                "token": "SCHEDULE",
+                "type": "pure-token",
+                "optional": true,
+                "since": "3.2.2"
+            },
+            {
+                "name": "save-type",
                 "type": "oneof",
                 "optional": true,
+                "since": "10.0.0",
                 "arguments": [
                     {
-                        "name": "schedule",
-                        "token": "SCHEDULE",
-                        "type": "pure-token",
-                        "since": "3.2.2"
+                        "name": "fork",
+                        "token": "FORK",
+                        "type": "pure-token"
                     },
                     {
-                        "name": "cancel",
-                        "token": "CANCEL",
-                        "type": "pure-token",
-                        "since": "8.1.0"
+                        "name": "thread",
+                        "token": "THREAD",
+                        "type": "pure-token"
                     }
                 ]
+            },
+            {
+                "name": "cancel",
+                "token": "CANCEL",
+                "type": "pure-token",
+                "optional": true,
+                "since": "8.1.0"
             }
         ],
         "reply_schema": {
             "oneOf": [
                 {
-                    "const": "Background saving started"
+                    "const": "Background saving (fork) started"
+                },
+                {
+                    "const": "Background saving (thread) started"
                 },
                 {
                     "const": "Background saving scheduled"

src/config.c

@@ -3256,6 +3256,7 @@ standardConfig static_configs[] = {
     createBoolConfig("rdb-del-sync-files", NULL, MODIFIABLE_CONFIG, server.rdb_del_sync_files, 0, NULL, NULL),
     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),
+    createBoolConfig("threadsave-enabled-for-backup", NULL, MODIFIABLE_CONFIG, server.threadsave_enabled_for_backup, 0, 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

@@ -38,6 +38,7 @@
 #include "vector.h"
 #include "expire.h"
 #include "bgiteration.h"
+#include "threadsave.h"
 
 /*-----------------------------------------------------------------------------
  * C-level DB API
@@ -823,7 +824,8 @@ int getFlushCommandFlags(client *c, int *flags) {
 /* Flushes the whole server data set. */
 void flushAllDataAndResetRDB(int flags) {
     server.dirty += emptyData(-1, flags, NULL);
-    if (server.child_type == CHILD_TYPE_RDB) killRDBChild();
+    if (isForkBgsaveInProgress()) killRDBChild();
+    if (isThreadBgsaveInProgress()) threadsaveCancel();
     if (server.child_type == CHILD_TYPE_SLOT_MIGRATION) killSlotMigrationChild();
     if (server.saveparamslen > 0) {
         rdbSaveInfo rsi, *rsiptr;

src/module.c

@@ -71,6 +71,7 @@
 #include "scripting_engine.h"
 #include "cluster_migrateslots.h"
 #include "bgiteration.h"
+#include "threadsave.h"
 #include <dlfcn.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
@@ -13746,7 +13747,8 @@ int VM_RdbLoad(ValkeyModuleCtx *ctx, ValkeyModuleRdbStream *stream, int flags) {
 
     /* Kill existing RDB fork as it is saving outdated data. Also killing it
      * will prevent COW memory issue. */
-    if (server.child_type == CHILD_TYPE_RDB) killRDBChild();
+    if (isForkBgsaveInProgress()) killRDBChild();
+    if (isThreadBgsaveInProgress()) threadsaveCancel();
 
     /* Kill existing slot migration fork as it is saving outdated data. Also killing it
      * will prevent COW memory issue. */