feat: add tcp keep alive (#110)

add max tcp keep alive time

etc/conf/kiwi.conf

@@ -16,6 +16,10 @@ ip 127.0.0.1
 # Close the connection after a client is idle for N seconds (0 to disable)
 timeout 0
 
+# Enable the TCP keep-alive function. Set the value to 0 to disable keep-alive.
+# A non-zero value represents the idle time (in seconds) before the first keep-alive probe is sent.
+tcp-keepalive 300
+
 # Directory to store the data of kiwi.
 db-path ./db/
 

src/config.cc

@@ -137,6 +137,10 @@ Config::Config() {
   AddNumberWithLimit<uint16_t>("port", false, &port, PORT_LIMIT_MIN, PORT_LIMIT_MAX);
   AddNumber("raft-port-offset", true, &raft_port_offset);
   AddNumber("timeout", true, &timeout);
+  AddNumber("tcp-keepalive", true, &tcp_keepalive);
+  AddString("db-path", false, &db_path);
+  AddStringWithFunc("loglevel", &CheckLogLevel, false, &log_level);
+  AddString("logfile", false, &log_dir);
   AddString("db-path", false, &db_path);
   AddStringWithFunc("loglevel", &CheckLogLevel, false, &log_level);
   AddString("logfile", false, &log_dir);

src/config.h

@@ -209,6 +209,13 @@ class Config {
    * 0 means no timeout limit, otherwise it is the specified number of seconds.
    */
   uint32_t timeout = 0;
+
+  /*
+   *Enable the tcp keep-alive function.
+   *0 indicates that the TCP keep-alive function is disabled
+   */
+  uint32_t tcp_keepalive = 300;
+
   /*
    * Client connect to kiwi server may need password
    */

src/kiwi.cc

@@ -205,6 +205,9 @@ bool KiwiDB::Init() {
     PREPL.SetMasterAddr(g_config.master_ip.c_str(), g_config.master_port);
   }
 
+  auto tcpKeepAlive = g_config.tcp_keepalive;
+  options_.SetOpTcpKeepAlive(tcpKeepAlive);
+
   options_.SetRwSeparation(true);
 
   event_server_ = std::make_unique<net::EventServer<std::shared_ptr<PClient>>>(options_);

src/net/base_socket.cc

@@ -31,6 +31,7 @@ void BaseSocket::OnCreate() {
   SetNonBlock(true);
 #endif
   SetNodelay();
+  SetTcpKeepAlive();
   SetSndBuf();
   SetRcvBuf();
 }
@@ -59,6 +60,49 @@ void BaseSocket::SetNodelay() {
   }
 }
 
+void BaseSocket::SetTcpKeepAlive() {
+  if (tcp_keep_alive_ == 0) {
+    return;
+  }
+
+  int enabled = 1;
+  uint32_t idle = tcp_keep_alive_;
+  uint32_t intvl = idle / 3;
+  int cnt = 3;
+
+  if (setsockopt(Fd(), SOL_SOCKET, SO_KEEPALIVE, &enabled, sizeof(enabled)) == -1) {
+    WARN("SetTcpKeepAlive fd:{} error:{}", Fd(), errno);
+    return;
+  }
+
+#ifdef TCP_KEEPIDLE
+  if (setsockopt(Fd(), IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle))) {
+    WARN("SetTcpKeepAlive fd:{} error:{}", Fd(), errno);
+    return;
+  }
+#elif defined(TCP_KEEPALIVE)
+  /* support MacOS */
+  if (setsockopt(Fd(), IPPROTO_TCP, TCP_KEEPALIVE, &idle, sizeof(idle))) {
+    WARN("SetTcpKeepAlive fd:{} error:{}", Fd(), errno);
+    return;
+  }
+#endif
+
+#ifdef TCP_KEEPINTVL
+  if (setsockopt(Fd(), IPPROTO_TCP, TCP_KEEPINTVL, &intvl, sizeof(intvl))) {
+    WARN("SetTcpKeepAlive fd:{} error:{}", Fd(), errno);
+    return;
+  }
+#endif
+
+#ifdef TCP_KEEPCNT
+  if (setsockopt(Fd(), IPPROTO_TCP, TCP_KEEPCNT, &cnt, sizeof(cnt))) {
+    WARN("SetTcpKeepAlive fd:{} error:{}", Fd(), errno);
+    return;
+  }
+#endif
+}
+
 void BaseSocket::SetSndBuf(socklen_t winsize) {
   if (::setsockopt(Fd(), SOL_SOCKET, SO_SNDBUF, reinterpret_cast<const char *>(&winsize), sizeof(winsize)) == -1) {
     WARN("SetSndBuf fd:{} error:{}", Fd(), errno);

src/net/base_socket.h

@@ -49,6 +49,8 @@ class BaseSocket : public NetEvent {
 
   void SetNodelay();
 
+  void SetTcpKeepAlive();
+
   void SetSndBuf(socklen_t size = SOCKET_WIN_SIZE);
 
   void SetRcvBuf(socklen_t size = SOCKET_WIN_SIZE);
@@ -65,12 +67,15 @@ class BaseSocket : public NetEvent {
 
   void SetSocketType(int type) { type_ = type; }
 
+  inline void SetBSTcpKeepAlive(uint32_t keepAlive) { tcp_keep_alive_ = keepAlive; }
+
  protected:
   bool NoBlock() const { return noBlock_; }
 
  private:
   int type_ = SOCKET_NONE;  // socket type (TCP/UDP)
   bool noBlock_ = true;
+  uint32_t tcp_keep_alive_ = 300;  // TCP keepalive
 };
 
 }  // namespace net

src/net/event_server.h

@@ -244,9 +244,10 @@ template <typename T>
 requires HasSetFdFunction<T>
 int EventServer<T>::StartThreadManager(bool serverMode) {
   std::shared_ptr<ListenSocket> listen(ListenSocket::CreateTCPListen());
+  auto tcpKeepAlive = opt_.GetOpTcpKeepAlive();
   if (serverMode) {
     listen->SetListenAddr(listenAddrs_);
-
+    listen->SetBSTcpKeepAlive(tcpKeepAlive);
     if (auto ret = listen->Init() != static_cast<int>(NetListen::OK)) {
       return ret;
     }
@@ -257,6 +258,7 @@ int EventServer<T>::StartThreadManager(bool serverMode) {
     if (i > 0 && ListenSocket::REUSE_PORT && serverMode) {
       listen.reset(ListenSocket::CreateTCPListen());
       listen->SetListenAddr(listenAddrs_);
+      listen->SetBSTcpKeepAlive(tcpKeepAlive);
       if (auto ret = listen->Init() != static_cast<int>(NetListen::OK)) {
         return ret;
       }

src/net/net_options.h

@@ -22,10 +22,16 @@ class NetOptions {
 
   bool GetRwSeparation() const { return rwSeparation_; }
 
+  void SetOpTcpKeepAlive(uint32_t tcpKeepAlive) { tcpKeepAlive_ = tcpKeepAlive; }
+
+  uint32_t GetOpTcpKeepAlive() const { return tcpKeepAlive_; }
+
  private:
   bool rwSeparation_ = true;  // Whether to separate read and write
 
   int8_t threadNum_ = 1;  // The number of threads
+
+  uint32_t tcpKeepAlive_ = 300;  // The timeout of the keepalive connection in seconds
 };
 
 }  // namespace net

src/net/thread_manager.h

@@ -92,6 +92,7 @@ class ThreadManager {
 
  private:
   const int8_t index_ = 0;            // The index of the thread
+  uint32_t tcpKeepAlive_ = 300;       // The timeout of the keepalive connection in seconds
   std::atomic<bool> running_ = true;  // Whether the thread is running
 
   NetOptions netOptions_;