diff --git a/data_substrate b/data_substrate index ea8b3c46..1b44e27a 160000 --- a/data_substrate +++ b/data_substrate @@ -1 +1 @@ -Subproject commit ea8b3c464e4b8fffda5fd2854dec803a8f6296ba +Subproject commit 1b44e27a96cab42fc5e32dcf2a004d9711c2abf1 diff --git a/docs/02-command-processing.md b/docs/02-command-processing.md index 4be77b39..71e4afdb 100644 --- a/docs/02-command-processing.md +++ b/docs/02-command-processing.md @@ -47,7 +47,8 @@ service layer only; engine internals are in `data_substrate/docs/` (esp. `02-thr listen address, starts the metrics collector thread and namespace GC daemon. In `bootstrap` mode it exits the process right here after table creation (`src/redis_service.cpp:704-721`). 6. brpc `Server::Start()` with `server_options.redis_service = redis_service_impl` (ownership - transfers to the server) and optional SSL options; then `RunUntilAskedToQuit()` + transfers to the server), the public listener declared Redis-only, `redis_max_connections` set + from `maxclients`, and optional force-SSL settings; then `RunUntilAskedToQuit()` (`src/redis_server.cpp:502-548`). ## 2. Request path end-to-end @@ -99,7 +100,18 @@ key ≤ 32 MB (2 KB for the EloqStore backend) and object ≤ 256 MB ## 3. Connection state machine -One `RedisConnectionContext` per socket, created by `NewConnectionContext` +The brpc acceptor enforces `maxclients` only on the Redis-only public listener, immediately after +the kernel accepts a TCP connection and before it creates a brpc `Socket`. Over-limit plaintext +connections receive `-ERR max number of clients reached` and are closed; SSL-capable listeners +close over-limit connections before TLS authentication or handshake work. Other EloqKV brpc/RPC +servers are not subject to this limit. The acceptor reserves connection slots atomically before +socket creation, so idle clients count toward the limit and concurrent accepts cannot overshoot it. +`CONFIG GET maxclients` reports the active limit and `CONFIG SET maxclients ` atomically +changes it for subsequent accepts. Lowering the limit does not disconnect established clients. +The runtime value is not written back to disk; a restart loads `maxclients` from the `[local]` +section of `eloqkv.ini` (or its gflag override) again. + +One `RedisConnectionContext` per admitted socket, created by `NewConnectionContext` (`src/redis_service.cpp:5917`) and owned by brpc's per-socket parsing context. Key fields (`include/redis_connection_context.h:61-170`): @@ -304,8 +316,9 @@ the special MOVED/READONLY translations of §6. ## 9. Stats, INFO, slow log, metrics - `RedisStats` (`src/redis_stats.cpp`) exposes brpc bvars when `--enable_redis_stats` (default - on): connections received/rejected/closed, blocked clients, read/write/multi-object command - counters. Incremented in the connection ctor/dtor and in `ExecuteTxRequest`/ + on): connections received/closed, blocked clients, and read/write/multi-object command + counters. Rejected connections come from the Redis listener's brpc admission counter. The + remaining counters are incremented in the connection ctor/dtor and in `ExecuteTxRequest`/ `ExecuteMultiObjTxRequest` (4486-4496, 4528-4536). `INFO` is a `DirectCommand` (`include/redis_command.h:811`) assembled from these counters plus service fields captured at Init (OS info, exe path, memory; `src/redis_service.cpp:587-630`). diff --git a/eloqkv.ini b/eloqkv.ini index 8c634cfa..d0214373 100644 --- a/eloqkv.ini +++ b/eloqkv.ini @@ -24,6 +24,12 @@ ip = 127.0.0.1 # changed once the server has been launched. port = 6379 +# Maximum number of simultaneous Redis client connections. This limits only +# the Redis listener and does not change the process open-file limit. CONFIG +# SET maxclients updates the running process; after restart this file is read +# again. +maxclients = 500000 + # EloqKV data directory. By default, the transaction service data, log service data, and local key-value storage data # are all stored in the eloq_data_path directory. eloq_data_path = eloq_data diff --git a/include/redis_command.h b/include/redis_command.h index dabb2bca..d0f49d53 100644 --- a/include/redis_command.h +++ b/include/redis_command.h @@ -1321,6 +1321,7 @@ struct ConfigCommand : public DirectCommand std::vector keys_; std::vector values_; std::vector results_; + std::string error_message_; }; struct TimeCommand : public DirectCommand diff --git a/include/redis_service.h b/include/redis_service.h index 9f6b6a49..36852cea 100644 --- a/include/redis_service.h +++ b/include/redis_service.h @@ -24,8 +24,10 @@ #include #include +#include #include #include +#include #include #include //std::unique_ptr #include @@ -156,6 +158,7 @@ class RedisServiceImpl; class NamespaceStorage; const std::unordered_set redis_config_keys = { + "maxclients", "slowlog-log-slower-than", "slowlog-max-len", }; @@ -543,6 +546,8 @@ class RedisServiceImpl : public brpc::RedisService void ResizeSlowLog(uint32_t len); + // Returns the active Redis listener limit. CONFIG SET may update this + // value without changing the startup configuration in DataSubstrate. size_t MaxConnectionCount() const; static bool SendTxRequest(TransactionExecution *txm, @@ -612,6 +617,15 @@ class RedisServiceImpl : public brpc::RedisService std::atomic_bool config_accessing_{false}; std::unordered_map config_; + // The Server outlives the service because it owns this RedisServiceImpl + // after Start(). The pointer is used only by CONFIG SET to update the + // Redis-only public acceptor's atomic admission limit. + brpc::Server *brpc_server_{nullptr}; + // Race-free cache for startup and INFO reads. config_accessing_ serializes + // CONFIG mutations; relaxed access here does not synchronize the brpc + // acceptor update, which uses its own atomic state. + std::atomic max_connection_count_{0}; + bool enable_redis_stats_; bool skip_kv_; bool skip_wal_; diff --git a/src/redis_command.cpp b/src/redis_command.cpp index 0f1d4292..2a20daaa 100644 --- a/src/redis_command.cpp +++ b/src/redis_command.cpp @@ -1788,11 +1788,19 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl, max_connection_count_ = redis_impl->MaxConnectionCount(); active_extern_txms_ = redis_impl->ActiveExternTxCount(); + conn_rejected_count_ = + server_acceptor == nullptr + ? 0 + : server_acceptor->RejectedRedisConnectionCount(); + // The acceptor owns the maxclients slots, so use the same source for + // connected_clients. This includes idle sockets that have not sent a + // first Redis command yet. + connecting_count_ = + server_acceptor == nullptr ? 0 : server_acceptor->ConnectionCount(); + if (redis_impl->IsEnableRedisStats()) { conn_received_count_ = RedisStats::GetConnReceivedCount(); - conn_rejected_count_ = RedisStats::GetConnRejectedCount(); - connecting_count_ = RedisStats::GetConnectingCount(); blocked_clients_count_ = RedisStats::GetBlockedClientsCount(); cmd_read_count_ = RedisStats::GetReadCommandsCount(); @@ -2827,6 +2835,11 @@ void ConfigCommand::OutputResult(OutputHandler *reply) const { if (flag_ == CONFIG_SET) { + if (!error_message_.empty()) + { + reply->OnError(error_message_); + return; + } reply->OnStatus("OK"); return; } diff --git a/src/redis_server.cpp b/src/redis_server.cpp index 834eeb17..4d4f3428 100644 --- a/src/redis_server.cpp +++ b/src/redis_server.cpp @@ -505,10 +505,17 @@ int main(int argc, char *argv[]) // Notice: redis_service_impl will be deleted in server's destructor. server_options.redis_service = redis_service_impl.release(); server_options.has_builtin_services = false; + // This listener is exclusively RESP. Declaring it as such lets brpc + // enforce maxclients immediately after accept, before any optional TLS + // handshake, without applying the limit to EloqKV's other RPC servers. + server_options.enabled_protocols = "redis"; + server_options.redis_max_connections = + redis_service_ptr->MaxConnectionCount(); // Configure TLS if enabled if (redis_service_ptr->IsTlsEnabled()) { + server_options.force_ssl = true; brpc::ServerSSLOptions *ssl_options = server_options.mutable_ssl_options(); diff --git a/src/redis_service.cpp b/src/redis_service.cpp index 05c1bb88..fb38832a 100644 --- a/src/redis_service.cpp +++ b/src/redis_service.cpp @@ -42,6 +42,7 @@ #include #include #include +#include #include #include #include @@ -232,6 +233,7 @@ RedisServiceImpl::RedisServiceImpl(const std::string &config_file, bool RedisServiceImpl::Init(brpc::Server &brpc_server) { + brpc_server_ = &brpc_server; INIReader config_reader(config_file_); if (!config_file_.empty() && config_reader.ParseError() != 0) @@ -243,6 +245,10 @@ bool RedisServiceImpl::Init(brpc::Server &brpc_server) // Engine registration: EloqKv auto &ds = DataSubstrate::Instance(); + max_connection_count_.store(ds.GetCoreConfig().maxclients, + std::memory_order_relaxed); + config_.try_emplace("maxclients", std::to_string(MaxConnectionCount())); + databases = config_reader.GetInteger("local", "databases", 16); // Prebuilt Redis tables (same names as today: data_table_0 .. @@ -4733,13 +4739,58 @@ void RedisServiceImpl::ExecuteSetConfig(ConfigCommand *cmd) } for (size_t i = 0; i < cmd->keys_.size(); ++i) { - config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]); - if (cmd->keys_[i] == "slowlog-log-slower-than") + if (cmd->keys_[i] == "maxclients") + { + uint64_t value = 0; + const std::string_view text = cmd->values_[i]; + if (!text.empty() && text.front() == '-') + { + cmd->error_message_ = + "ERR CONFIG SET failed (possibly related to argument " + "'maxclients') - argument must be between 1 and " + "4294967295 inclusive"; + break; + } + const auto [end, error] = + std::from_chars(text.data(), text.data() + text.size(), value); + if (error == std::errc::invalid_argument || + end != text.data() + text.size()) + { + cmd->error_message_ = + "ERR CONFIG SET failed (possibly related to argument " + "'maxclients') - argument couldn't be parsed into an " + "integer"; + break; + } + if (error == std::errc::result_out_of_range || value == 0 || + value > std::numeric_limits::max()) + { + cmd->error_message_ = + "ERR CONFIG SET failed (possibly related to argument " + "'maxclients') - argument must be between 1 and " + "4294967295 inclusive"; + break; + } + const uint32_t maxclients = static_cast(value); + if (brpc_server_ == nullptr || + brpc_server_->SetRedisMaxConnections(maxclients) != 0) + { + cmd->error_message_ = + "ERR CONFIG SET failed (possibly related to argument " + "'maxclients') - unable to update the Redis listener"; + break; + } + max_connection_count_.store(maxclients, std::memory_order_relaxed); + config_["maxclients"] = std::to_string(maxclients); + } + else if (cmd->keys_[i] == "slowlog-log-slower-than") { + config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]); slow_log_threshold_ = std::stoul(std::string(cmd->values_[i])); } else if (cmd->keys_[i] == "slowlog-max-len") { + config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]); ResizeSlowLog(std::stoul(std::string(cmd->values_[i]))); } } @@ -6514,8 +6565,7 @@ metrics::Meter *RedisServiceImpl::GetMeter(std::size_t core_id) const size_t RedisServiceImpl::MaxConnectionCount() const { - auto &ds = DataSubstrate::Instance(); - return ds.GetCoreConfig().maxclients; + return max_connection_count_.load(std::memory_order_relaxed); } } // namespace EloqKV diff --git a/tests/unit/eloq/maxclients.tcl b/tests/unit/eloq/maxclients.tcl new file mode 100644 index 00000000..76fe5aa4 --- /dev/null +++ b/tests/unit/eloq/maxclients.tcl @@ -0,0 +1,31 @@ +start_server {tags {"maxclients network"}} { + test {CONFIG GET and SET expose the live maxclients limit} { + set original [lindex [r config get maxclients] 1] + + assert_equal {OK} [r config set maxclients 1] + assert_equal {1} [lindex [r config get maxclients] 1] + + # The connection issuing CONFIG SET remains established when the new + # limit is below the current connection count. + assert_equal {PONG} [r ping] + + if {$::tls} { + set expected_rejection {*I/O error*} + } else { + set expected_rejection {*ERR max*reached*} + } + set rejected [catch {redis_deferring_client} rejection] + assert_equal {1} $rejected + assert_match $expected_rejection $rejection + + assert_error {*argument must be between 1 and 4294967295 inclusive*} { + r config set maxclients 0 + } + assert_error {*argument couldn't be parsed into an integer*} { + r config set maxclients invalid + } + assert_equal {1} [lindex [r config get maxclients] 1] + + r config set maxclients $original + } +}