Steady state replication throttling POC

cmake/Modules/SourceFiles.cmake

@@ -123,7 +123,11 @@ set(VALKEY_SERVER_SRCS
     ${CMAKE_SOURCE_DIR}/src/vset.c
     ${CMAKE_SOURCE_DIR}/src/fifo.c
     ${CMAKE_SOURCE_DIR}/src/mutexqueue.c
-    ${CMAKE_SOURCE_DIR}/src/queues.c)
+    ${CMAKE_SOURCE_DIR}/src/queues.c
+    ${CMAKE_SOURCE_DIR}/src/throttle_token_bucket.c
+    ${CMAKE_SOURCE_DIR}/src/stat_calc.c
+    ${CMAKE_SOURCE_DIR}/src/throttle_repl.c
+    ${CMAKE_SOURCE_DIR}/src/throttle.c)
 
 
 # valkey-cli

sentinel.conf

@@ -242,7 +242,7 @@ sentinel failover-timeout mymaster 180000
 # If script exits with "1" the execution is retried later (up to a maximum
 # number of times currently set to 10).
 #
-# If script exits with "2" (or an higher value) the script execution is
+# If script exits with "2" (or a higher value) the script execution is
 # not retried.
 #
 # If script terminates because it receives a signal the behavior is the same

src/Makefile

@@ -583,7 +583,11 @@ ENGINE_SERVER_OBJ = \
     ziplist.o \
     zipmap.o \
     zmalloc.o \
-    queues.o
+	queues.o \
+	throttle_token_bucket.o \
+	stat_calc.o \
+	throttle_repl.o \
+	throttle.o
 ENGINE_SERVER_OBJ+=$(ENGINE_TRACE_OBJ)
 ENGINE_CLI_NAME=$(ENGINE_NAME)-cli$(PROG_SUFFIX)
 ENGINE_CLI_OBJ = \

src/acl.c

@@ -659,7 +659,7 @@ static void ACLChangeSelectorPerm(aclSelector *selector, struct serverCommand *c
 /* This is like ACLSetSelectorCommandBit(), but instead of setting the specified
  * ID, it will check all the commands in the category specified as argument,
  * and will set all the bits corresponding to such commands to the specified
- * value. Since the category passed by the user may be non existing, the
+ * value. Since the category passed by the user may be nonexistent, the
  * function returns C_ERR if the category was not found, or C_OK if it was
  * found and the operation was performed. */
 static void ACLSetSelectorCommandBitsForCategory(hashtable *commands, aclSelector *selector, uint64_t cflag, int value) {

src/aof.c

@@ -602,7 +602,7 @@ int persistAofManifest(aofManifest *am) {
     return ret;
 }
 
-/* Called in `loadAppendOnlyFiles` when we upgrade from a old version of the server.
+/* Called in `loadAppendOnlyFiles` when we upgrade from an old version of the server.
  *
  * 1) Create AOF directory use 'server.aof_dirname' as the name.
  * 2) Use 'server.aof_filename' to construct a BASE type aofInfo and add it to

src/bitops.c

@@ -1011,7 +1011,7 @@ void bitcountCommand(client *c) {
         return;
     }
 
-    /* Return 0 for non existing keys. */
+    /* Return 0 for nonexistent keys. */
     if (o == NULL) {
         addReply(c, shared.czero);
         return;
@@ -1122,8 +1122,8 @@ void bitposCommand(client *c) {
         return;
     }
 
-    /* For empty ranges (start > end) we return -1 as an empty range does
-     * not contain a 0 nor a 1. */
+    /* For empty ranges (start > end) we return -1 as an empty range contains
+     * neither 0 nor 1. */
     if (start > end) {
         addReplyLongLong(c, -1);
     } else {

src/blocked.c

@@ -165,6 +165,7 @@ void processUnblockedClients(void) {
         c = ln->value;
         listDelNode(server.unblocked_clients, ln);
         c->flag.unblocked = 0;
+        serverAssert(!c->flag.throttled);
 
         if (c->flag.module) {
             if (!c->flag.blocked) {
@@ -478,10 +479,10 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeo
         }
     }
     c->bstate->unblock_on_nokey = unblock_on_nokey;
-    /* Currently we assume key blocking will require reprocessing the command.
-     * However in case of modules, they have a different way to handle the reprocessing
-     * which does not require setting the pending command flag */
-    if (btype != BLOCKED_MODULE) c->flag.pending_command = 1;
+    /* Key-blocked clients require pending_command for reprocessing on unblock.
+     * The caller must have set it (processInputBuffer for real clients,
+     * RM_Call for module fake clients). */
+    serverAssert(c->flag.pending_command == 1);
     blockClient(c, btype);
 }
 
@@ -713,8 +714,7 @@ void blockPostponeClient(client *c) {
     listAddNodeTail(server.postponed_clients, c);
     serverAssert(c->bstate->postponed_list_node == NULL);
     c->bstate->postponed_list_node = listLast(server.postponed_clients);
-    /* Mark this client to execute its command */
-    c->flag.pending_command = 1;
+    serverAssert(c->flag.pending_command == 1);
 }
 
 /* Block client due to shutdown command */
@@ -745,7 +745,6 @@ static void unblockClientOnKey(client *c, robj *key) {
     /* In case this client was blocked on keys during command
      * we need to re process the command again */
     if (c->flag.pending_command) {
-        c->flag.pending_command = 0;
         c->flag.reexecuting_command = 1;
         /* We want the command processing and the unblock handler (see RM_Call 'K' option)
          * to run atomically, this is why we must enter the execution unit here before

src/cluster_legacy.c

@@ -1065,7 +1065,7 @@ int clusterLoadConfig(char *filename) {
  * of the POSIX filesystem semantics, so that if the server is stopped
  * or crashes during the write, we'll end with either the old file or the
  * new one. Since we have the full payload to write available we can use
- * a single write to write the whole file. If the pre-existing file was
+ * a single write to write the whole file. If the preexisting file was
  * bigger we pad our payload with newlines that are anyway ignored and truncate
  * the file afterward. */
 int clusterSaveConfig(int do_fsync) {
@@ -3358,7 +3358,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc
             serverLog(LL_NOTICE,
                       "My last slot was migrated to node %.40s (%s) in shard %.40s. I am now an empty primary.",
                       sender->name, humanNodename(sender), sender->shard_id);
-            /* We may still have dirty slots when we became a empty primary due to
+            /* We may still have dirty slots when we became an empty primary due to
              * a bad migration.
              *
              * In order to maintain a consistent state between keys and slots

src/commandlog.c

@@ -69,8 +69,11 @@ static commandlogEntry *commandlogCreateEntry(client *c, robj **argv, int argc,
     ce->time = time(NULL);
     ce->value = value;
     ce->id = server.commandlog[type].entry_id++;
-    ce->peerid = sdsnew(getClientPeerId(c));
-    ce->cname = c->name ? sdsnew(objectGetVal(c->name)) : sdsempty();
+    /* For commands executed from a script, the executing client is a fake
+     * client with no connection, so attribute the entry to the calling client. */
+    client *caller = scriptIsRunning() ? scriptGetCaller() : c;
+    ce->peerid = sdsnew(getClientPeerId(caller));
+    ce->cname = caller->name ? sdsnew(objectGetVal(caller->name)) : sdsempty();
     return ce;
 }
 
@@ -154,21 +157,13 @@ void commandlogPushCurrentCommand(client *c, struct serverCommand *cmd) {
     robj **argv = c->original_argv ? c->original_argv : c->argv;
     int argc = c->original_argv ? c->original_argc : c->argc;
 
-    /* In script, client will be replaced with its caller, so commandlog needs to use the metrics
-     * of the client that currently executing the command. */
-    long duration = c->duration;
-    unsigned long long net_input_bytes_curr_cmd = c->net_input_bytes_curr_cmd;
-    unsigned long long net_output_bytes_curr_cmd = c->net_output_bytes_curr_cmd;
-
-    /* If a script is currently running, the client passed in is a
-     * fake client. Or the client passed in is the original client
-     * if this is a EVAL or alike, doesn't matter. In this case,
-     * use the original client to get the client information. */
-    c = scriptIsRunning() ? scriptGetCaller() : c;
-
-    commandlogPushEntryIfNeeded(c, argv, argc, duration, COMMANDLOG_TYPE_SLOW);
-    commandlogPushEntryIfNeeded(c, argv, argc, net_input_bytes_curr_cmd, COMMANDLOG_TYPE_LARGE_REQUEST);
-    commandlogPushEntryIfNeeded(c, argv, argc, net_output_bytes_curr_cmd, COMMANDLOG_TYPE_LARGE_REPLY);
+    /* 'c' is the client that executed the command: for a command called from a
+     * script this is the fake client, whose argv, metrics and redaction bitmap
+     * describe the executed command. Entry creation resolves the calling client
+     * for the connection identity fields. */
+    commandlogPushEntryIfNeeded(c, argv, argc, c->duration, COMMANDLOG_TYPE_SLOW);
+    commandlogPushEntryIfNeeded(c, argv, argc, c->net_input_bytes_curr_cmd, COMMANDLOG_TYPE_LARGE_REQUEST);
+    commandlogPushEntryIfNeeded(c, argv, argc, c->net_output_bytes_curr_cmd, COMMANDLOG_TYPE_LARGE_REPLY);
 }
 
 /* The SLOWLOG command. Implements all the subcommands needed to handle the

src/commands.def

@@ -2370,7 +2370,7 @@ struct COMMAND_ARG MIGRATE_Args[] = {
 #ifndef SKIP_CMD_HISTORY_TABLE
 /* MOVE history */
 commandHistory MOVE_History[] = {
-{"10.0.0","added REPLACE option."},
+{"9.2.0","added REPLACE option."},
 };
 #endif
 
@@ -2390,7 +2390,7 @@ keySpec MOVE_Keyspecs[1] = {
 struct COMMAND_ARG MOVE_Args[] = {
 {MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)},
 {MAKE_ARG("db",ARG_TYPE_INTEGER,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)},
-{MAKE_ARG("replace",ARG_TYPE_PURE_TOKEN,-1,"REPLACE",NULL,"10.0.0",CMD_ARG_OPTIONAL,0,NULL)},
+{MAKE_ARG("replace",ARG_TYPE_PURE_TOKEN,-1,"REPLACE",NULL,"9.2.0",CMD_ARG_OPTIONAL,0,NULL)},
 };
 
 /********** OBJECT ENCODING ********************/
@@ -7670,7 +7670,7 @@ struct COMMAND_ARG CONFIG_SET_Args[] = {
 struct COMMAND_STRUCT CONFIG_Subcommands[] = {
 {MAKE_CMD("get","Returns the effective values of configuration parameters.","O(N) when N is the number of configuration parameters provided","2.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_GET_History,1,CONFIG_GET_Tips,0,configGetCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_GET_Keyspecs,0,NULL,1),.args=CONFIG_GET_Args},
 {MAKE_CMD("help","Returns helpful text about the different subcommands.","O(1)","5.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_HELP_History,0,CONFIG_HELP_Tips,0,configHelpCommand,2,CMD_LOADING|CMD_STALE,ACL_CATEGORY_SLOW,NULL,CONFIG_HELP_Keyspecs,0,NULL,0)},
-{MAKE_CMD("info","Returns information about configuration parameters matching the given patterns.","O(N) when N is the number of configuration parameters provided","10.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_INFO_History,0,CONFIG_INFO_Tips,0,configInfoCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_INFO_Keyspecs,0,NULL,1),.args=CONFIG_INFO_Args},
+{MAKE_CMD("info","Returns information about configuration parameters matching the given patterns.","O(N) when N is the number of configuration parameters provided","9.2.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_INFO_History,0,CONFIG_INFO_Tips,0,configInfoCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_INFO_Keyspecs,0,NULL,1),.args=CONFIG_INFO_Args},
 {MAKE_CMD("resetstat","Resets the server's statistics.","O(1)","2.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_RESETSTAT_History,0,CONFIG_RESETSTAT_Tips,2,configResetStatCommand,2,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_RESETSTAT_Keyspecs,0,NULL,0)},
 {MAKE_CMD("rewrite","Persists the effective configuration to file.","O(1)","2.8.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_REWRITE_History,0,CONFIG_REWRITE_Tips,2,configRewriteCommand,2,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_REWRITE_Keyspecs,0,NULL,0)},
 {MAKE_CMD("set","Sets configuration parameters in-flight.","O(N) when N is the number of configuration parameters provided","2.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,CONFIG_SET_History,1,CONFIG_SET_Tips,2,configSetCommand,-4,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_ADMIN|ACL_CATEGORY_DANGEROUS|ACL_CATEGORY_SLOW,NULL,CONFIG_SET_Keyspecs,0,NULL,1),.args=CONFIG_SET_Args},
@@ -12158,7 +12158,7 @@ struct COMMAND_STRUCT serverCommandTable[] = {
 {MAKE_CMD("substr","Returns a substring from a string value.","O(N) where N is the length of the returned string. The complexity is ultimately determined by the returned length, but because creating a substring from an existing string is very cheap, it can be considered O(1) for small strings.","1.0.0",CMD_DOC_NONE,NULL,NULL,"string",COMMAND_GROUP_STRING,SUBSTR_History,0,SUBSTR_Tips,0,getrangeCommand,4,CMD_READONLY,ACL_CATEGORY_READ|ACL_CATEGORY_SLOW|ACL_CATEGORY_STRING,NULL,SUBSTR_Keyspecs,1,NULL,3),.args=SUBSTR_Args},
 /* transactions */
 {MAKE_CMD("discard","Discards a transaction.","O(N), when N is the number of queued commands","2.0.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,DISCARD_History,0,DISCARD_Tips,0,discardCommand,1,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_FAST|CMD_ALLOW_BUSY,ACL_CATEGORY_FAST|ACL_CATEGORY_TRANSACTION,NULL,DISCARD_Keyspecs,0,NULL,0)},
-{MAKE_CMD("exec","Executes all commands in a transaction.","Depends on commands in the transaction","1.2.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,EXEC_History,0,EXEC_Tips,0,execCommand,1,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SKIP_COMMANDLOG,ACL_CATEGORY_SLOW|ACL_CATEGORY_TRANSACTION,NULL,EXEC_Keyspecs,0,NULL,0)},
+{MAKE_CMD("exec","Executes all commands in a transaction.","Depends on commands in the transaction","1.2.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,EXEC_History,0,EXEC_Tips,0,execCommand,1,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE,ACL_CATEGORY_SLOW|ACL_CATEGORY_TRANSACTION,NULL,EXEC_Keyspecs,0,NULL,0)},
 {MAKE_CMD("multi","Starts a transaction.","O(1)","1.2.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,MULTI_History,0,MULTI_Tips,0,multiCommand,1,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_FAST|CMD_NO_MULTI|CMD_ALLOW_BUSY,ACL_CATEGORY_FAST|ACL_CATEGORY_TRANSACTION,NULL,MULTI_Keyspecs,0,NULL,0)},
 {MAKE_CMD("unwatch","Forgets about watched keys of a transaction.","O(1)","2.2.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,UNWATCH_History,0,UNWATCH_Tips,0,unwatchCommand,1,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_FAST|CMD_ALLOW_BUSY,ACL_CATEGORY_FAST|ACL_CATEGORY_TRANSACTION,NULL,UNWATCH_Keyspecs,0,NULL,0)},
 {MAKE_CMD("watch","Monitors changes to keys to determine the execution of a transaction.","O(1) for every key.","2.2.0",CMD_DOC_NONE,NULL,NULL,"transactions",COMMAND_GROUP_TRANSACTIONS,WATCH_History,0,WATCH_Tips,0,watchCommand,-2,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_FAST|CMD_NO_MULTI|CMD_ALLOW_BUSY,ACL_CATEGORY_FAST|ACL_CATEGORY_TRANSACTION,NULL,WATCH_Keyspecs,1,NULL,1),.args=WATCH_Args},

src/commands/README.md

@@ -78,7 +78,7 @@ following keys. To be safe, assume all of them are optional.
   * `"TOUCHES_ARBITRARY_KEYS"`
   * `"WRITE"`
 * `"acl_categories"`: A list of ACL categories in uppercase. Note that the
-  effective ACL categies include "implicit ACL categories" explained below.
+  effective ACL categories include "implicit ACL categories" explained below.
   * `"ADMIN"`
   * `"BITMAP"`
   * `"CONNECTION"`
@@ -183,7 +183,7 @@ Each element in this array is an object with the following keys:
     command line. The first key is the argument after the keyword.
   * `{"unknown": null}`: Finding the keys of this command is too complicated to
     explain.
-* `"find_keys"`: How to find the remainnig keys of this key spec. It's an object
+* `"find_keys"`: How to find the remaining keys of this key spec. It's an object
   on one of these forms:
   * `{"range": {"lastkey": LAST, "step": STEP, "limit": LIMIT}}`: A range of keys.
     * LAST: If LAST is positive, it's the index of the last key relative to the
@@ -209,7 +209,7 @@ doesn't have an `"arguments"` key.
 Appendix
 --------
 
-How to list all the `group`, `command_flags` and `acl_categries`, etc. used in all these files:
+How to list all the `group`, `command_flags` and `acl_categories`, etc. used in all these files:
 
     cat *.json | jq '.[].group'             | grep -F '"' | sed 's/^ *//;s/, *$//;s/^/  * `/;s/$/`/' | sort | uniq
     cat *.json | jq '.[].command_flags'     | grep -F '"' | sed 's/^ *//;s/, *$//;s/^/  * `/;s/$/`/' | sort | uniq

src/commands/config-info.json

@@ -3,7 +3,7 @@
         "summary": "Returns information about configuration parameters matching the given patterns.",
         "complexity": "O(N) when N is the number of configuration parameters provided",
         "group": "server",
-        "since": "10.0.0",
+        "since": "9.2.0",
         "arity": -3,
         "container": "CONFIG",
         "function": "configInfoCommand",

src/commands/exec.json

@@ -9,8 +9,7 @@
         "command_flags": [
             "NOSCRIPT",
             "LOADING",
-            "STALE",
-            "SKIP_COMMANDLOG"
+            "STALE"
         ],
         "acl_categories": [
             "SLOW",

src/commands/move.json

@@ -8,7 +8,7 @@
         "function": "moveCommand",
         "history": [
             [
-                "10.0.0",
+                "9.2.0",
                 "added REPLACE option."
             ]
         ],
@@ -57,7 +57,7 @@
                 "name": "replace",
                 "token": "REPLACE",
                 "type": "pure-token",
-                "since": "10.0.0",
+                "since": "9.2.0",
                 "optional": true
             }
         ],

src/config.c

@@ -38,6 +38,7 @@
 #include "cluster_migrateslots.h"
 #include "eval.h"
 #include "lrulfu.h"
+#include "throttle_repl.h"
 
 #include <fcntl.h>
 #include <sys/stat.h>
@@ -3362,6 +3363,7 @@ standardConfig static_configs[] = {
     createBoolConfig("repl-mptcp", NULL, IMMUTABLE_CONFIG, server.repl_mptcp, 0, isValidMptcp, NULL),
     createBoolConfig("repl-diskless-sync", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.repl_diskless_sync, 1, NULL, NULL),
     createBoolConfig("dual-channel-replication-enabled", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.dual_channel_replication, 0, NULL, NULL),
+    createBoolConfig("repl-throttle-steady-state-enabled", NULL, MODIFIABLE_CONFIG, throttle_repl_config.repl_throttle_steady_state_enabled, 0, NULL, NULL),
     createBoolConfig("aof-rewrite-incremental-fsync", NULL, MODIFIABLE_CONFIG, server.aof_rewrite_incremental_fsync, 1, NULL, NULL),
     createBoolConfig("no-appendfsync-on-rewrite", NULL, MODIFIABLE_CONFIG, server.aof_no_fsync_on_rewrite, 0, NULL, NULL),
     createBoolConfig("cluster-require-full-coverage", NULL, MODIFIABLE_CONFIG, server.cluster_require_full_coverage, 1, NULL, updateClusterState),