feat(server): Support limiting the number of open connections.

src/facade/dragonfly_listener.cc

@@ -267,6 +267,10 @@ void Listener::OnConnectionClose(util::Connection* conn) {
   }
 }
 
+void Listener::OnMaxConnectionsReached(util::FiberSocketBase* sock) {
+  sock->Write(io::Buffer("-ERR max number of clients reached\r\n"));
+}
+
 // We can limit number of threads handling dragonfly connections.
 ProactorBase* Listener::PickConnectionProactor(util::FiberSocketBase* sock) {
   util::ProactorPool* pp = pool();

src/facade/dragonfly_listener.h

@@ -41,6 +41,7 @@ class Listener : public util::ListenerInterface {
 
   void OnConnectionStart(util::Connection* conn) final;
   void OnConnectionClose(util::Connection* conn) final;
+  void OnMaxConnectionsReached(util::FiberSocketBase* sock) final;
   void PreAcceptLoop(ProactorBase* pb) final;
 
   void PreShutdown() final;

src/server/config_registry.cc

@@ -39,6 +39,19 @@ bool ConfigRegistry::Set(std::string_view config_name, std::string_view value) {
   return cb(*flag);
 }
 
+std::optional<std::string> ConfigRegistry::Get(std::string_view config_name) {
+  unique_lock lk(mu_);
+  auto it = registry_.find(config_name);
+  if (it == registry_.end())
+    return {};
+  auto cb = it->second;
+  lk.unlock();
+
+  absl::CommandLineFlag* flag = absl::FindCommandLineFlag(config_name);
+  CHECK(flag);
+  return flag->CurrentValue();
+}
+
 void ConfigRegistry::Reset() {
   unique_lock lk(mu_);
   registry_.clear();

src/server/config_registry.h

@@ -19,6 +19,8 @@ class ConfigRegistry {
   // Returns true if the value was updated.
   bool Set(std::string_view config_name, std::string_view value);
 
+  std::optional<std::string> Get(std::string_view config_name);
+
   void Reset();
 
  private:

src/server/server_family.cc

@@ -59,6 +59,8 @@ ABSL_FLAG(string, dbfilename, "dump-{timestamp}", "the filename to save/load the
 ABSL_FLAG(string, requirepass, "",
           "password for AUTH authentication. "
           "If empty can also be set with DFLY_PASSWORD environment variable.");
+ABSL_FLAG(uint32_t, maxclients, 64000, "Maximum number of concurrent clients allowed.");
+
 ABSL_FLAG(string, save_schedule, "",
           "glob spec for the UTC time to save a snapshot which matches HH:MM 24h time");
 ABSL_FLAG(string, snapshot_cron, "",
@@ -569,6 +571,21 @@ void ServerFamily::Init(util::AcceptServer* acceptor, std::vector<facade::Listen
   listeners_ = std::move(listeners);
   dfly_cmd_ = make_unique<DflyCmd>(this);
 
+  for (auto* listener : listeners_) {
+    listener->SetMaxClients(absl::GetFlag(FLAGS_maxclients));
+  }
+  config_registry.Register("maxclients", [this](const absl::CommandLineFlag& flag) {
+    auto res = flag.TryGet<uint32_t>();
+    if (!res)
+      return false;
+
+    for (auto* listener : listeners_) {
+      listener->SetMaxClients(res.value());
+    }
+
+    return true;
+  });
+
   pb_task_ = shard_set->pool()->GetNextProactor();
   if (pb_task_->GetKind() == ProactorBase::EPOLL) {
     fq_threadpool_.reset(new FiberQueueThreadPool(absl::GetFlag(FLAGS_epoll_file_threads)));
@@ -1514,9 +1531,10 @@ void ServerFamily::Config(CmdArgList args, ConnectionContext* cntx) {
     if (param == "databases") {
       res.emplace_back(param);
       res.push_back(absl::StrCat(absl::GetFlag(FLAGS_dbnum)));
-    } else if (param == "maxmemory") {
+    } else if (auto value_from_registry = config_registry.Get(param);
+               value_from_registry.has_value()) {
       res.emplace_back(param);
-      res.push_back(absl::StrCat(max_memory_limit));
+      res.push_back(*value_from_registry);
     }
 
     return (*cntx)->SendStringArr(res, RedisReplyBuilder::MAP);

tests/dragonfly/config_test.py

@@ -0,0 +1,21 @@
+import pytest
+import redis
+from .utility import *
+from . import DflyStartException
+
+
+async def test_maxclients(df_factory):
+    # Needs some authentication
+    server = df_factory.create(port=1111, maxclients=1)
+    server.start()
+
+    async with server.client() as client1:
+        assert [b"maxclients", b"1"] == await client1.execute_command("CONFIG GET maxclients")
+
+        with pytest.raises(redis.exceptions.ConnectionError):
+            async with server.client() as client2:
+                await client2.get("test")
+
+        await client1.execute_command("CONFIG SET maxclients 3")
+        async with server.client() as client2:
+            await client2.get("test")