feat(worker): Allow passing a socket FD to dup and listen on

src/cli/main.cc

@@ -30,7 +30,6 @@
 #include <iomanip>
 #include <ostream>
 
-#include "config.h"
 #include "daemon_util.h"
 #include "io_util.h"
 #include "pid_util.h"
@@ -40,9 +39,7 @@
 #include "storage/storage.h"
 #include "string_util.h"
 #include "time_util.h"
-#include "unique_fd.h"
 #include "vendor/crc64.h"
-#include "version.h"
 #include "version_util.h"
 
 Server *srv = nullptr;
@@ -136,14 +133,19 @@ int main(int argc, char *argv[]) {
     std::cout << "Failed to load config. Error: " << s.Msg() << std::endl;
     return 1;
   }
+  const auto socket_fd_exit = MakeScopeExit([&config] {
+    if (config.socket_fd != -1) {
+      close(config.socket_fd);
+    }
+  });
 
   crc64_init();
   InitGoogleLog(&config);
   LOG(INFO) << "kvrocks " << PrintVersion;
   // Tricky: We don't expect that different instances running on the same port,
   // but the server use REUSE_PORT to support the multi listeners. So we connect
   // the listen port to check if the port has already listened or not.
-  if (!config.binds.empty()) {
+  if (config.socket_fd != -1 && !config.binds.empty()) {
     uint32_t ports[] = {config.port, config.tls_port, 0};
     for (uint32_t *port = ports; *port; ++port) {
       if (util::IsPortInUse(*port)) {

src/config/config.cc

@@ -110,6 +110,7 @@ Config::Config() {
       {"daemonize", true, new YesNoField(&daemonize, false)},
       {"bind", true, new StringField(&binds_str_, "")},
       {"port", true, new UInt32Field(&port, kDefaultPort, 1, PORT_LIMIT)},
+      {"socket-fd", true, new IntField(&socket_fd, -1, -1, 1 << 16)},
 #ifdef ENABLE_OPENSSL
       {"tls-port", true, new UInt32Field(&tls_port, 0, 0, PORT_LIMIT)},
       {"tls-cert-file", false, new StringField(&tls_cert_file, "")},

src/config/config.h

@@ -70,6 +70,7 @@ struct Config {
   Config();
   ~Config() = default;
   uint32_t port = 0;
+  int socket_fd = -1;
 
   uint32_t tls_port = 0;
   std::string tls_cert_file;

src/server/worker.cc

@@ -22,6 +22,7 @@
 
 #include <event2/util.h>
 #include <glog/logging.h>
+#include <unistd.h>
 
 #include <stdexcept>
 #include <string>
@@ -30,7 +31,6 @@
 #include "io_util.h"
 #include "scope_exit.h"
 #include "thread_util.h"
-#include "time_util.h"
 
 #ifdef ENABLE_OPENSSL
 #include <event2/bufferevent_ssl.h>
@@ -44,7 +44,6 @@
 #include <sys/un.h>
 
 #include <algorithm>
-#include <list>
 #include <utility>
 
 #include "redis_connection.h"
@@ -59,17 +58,22 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
   timeval tm = {10, 0};
   evtimer_add(timer_.get(), &tm);
 
-  uint32_t ports[3] = {config->port, config->tls_port, 0};
-  auto binds = config->binds;
-
-  for (uint32_t *port = ports; *port; ++port) {
-    for (const auto &bind : binds) {
-      Status s = listenTCP(bind, *port, config->backlog);
-      if (!s.IsOK()) {
-        LOG(ERROR) << "[worker] Failed to listen on: " << bind << ":" << *port << ". Error: " << s.Msg();
-        exit(1);
+  if (config->socket_fd != -1) {
+    if (const Status s = listenFD(config->socket_fd, config->port, config->backlog); !s.IsOK()) {
+      LOG(ERROR) << "[worker] Failed to listen to socket with fd: " << config->socket_fd << ". Error: " << s.Msg();
+      exit(1);
+    }
+  } else {
+    const uint32_t ports[3] = {config->port, config->tls_port, 0};
+
+    for (const uint32_t *port = ports; *port; ++port) {
+      for (const auto &bind : config->binds) {
+        if (const Status s = listenTCP(bind, *port, config->backlog); !s.IsOK()) {
+          LOG(ERROR) << "[worker] Failed to listen on: " << bind << ":" << *port << ". Error: " << s.Msg();
+          exit(1);
+        }
+        LOG(INFO) << "[worker] Listening on: " << bind << ":" << *port;
       }
-      LOG(INFO) << "[worker] Listening on: " << bind << ":" << *port;
     }
   }
   lua_ = lua::CreateState(srv);
@@ -216,6 +220,22 @@ void Worker::newUnixSocketConnection(evconnlistener *listener, evutil_socket_t f
   }
 }
 
+Status Worker::listenFD(int fd, uint32_t expected_port, int backlog) {
+  const uint32_t port = util::GetLocalPort(fd);
+  if (port != expected_port) {
+    return {Status::NotOK, "The port of the provided socket fd doesn't match the configured port"};
+  }
+  const int dup_fd = dup(fd);
+  if (dup_fd == -1) {
+    return {Status::NotOK, evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR())};
+  }
+  evconnlistener *lev =
+      NewEvconnlistener<&Worker::newTCPConnection>(base_, LEV_OPT_THREADSAFE | LEV_OPT_CLOSE_ON_FREE, backlog, dup_fd);
+  listen_events_.emplace_back(lev);
+  LOG(INFO) << "Listening on dup'ed fd: " << dup_fd;
+  return Status::OK();
+}
+
 Status Worker::listenTCP(const std::string &host, uint32_t port, int backlog) {
   bool ipv6_used = strchr(host.data(), ':');
 

src/server/worker.h

@@ -27,7 +27,6 @@
 
 #include <cstdint>
 #include <cstring>
-#include <iostream>
 #include <lua.hpp>
 #include <map>
 #include <memory>
@@ -36,9 +35,9 @@
 #include <utility>
 #include <vector>
 
+#include "config/config.h"
 #include "event_util.h"
 #include "redis_connection.h"
-#include "storage/storage.h"
 
 class Server;
 
@@ -79,6 +78,7 @@ class Worker : EventCallbackBase<Worker>, EvconnlistenerBase<Worker> {
   Server *srv;
 
  private:
+  Status listenFD(int fd, uint32_t expected_port, int backlog);
   Status listenTCP(const std::string &host, uint32_t port, int backlog);
   void newTCPConnection(evconnlistener *listener, evutil_socket_t fd, sockaddr *address, int socklen);
   void newUnixSocketConnection(evconnlistener *listener, evutil_socket_t fd, sockaddr *address, int socklen);