fix: ipc server crash when client leave

Author: TzuHuanTai

src/ipc/unix_socket_server.cpp

@@ -1,5 +1,7 @@
 #include "ipc/unix_socket_server.h"
 
+#include "common/logging.h"
+
 std::shared_ptr<UnixSocketServer> UnixSocketServer::Create(const std::string &socket_path) {
     return std::make_shared<UnixSocketServer>(socket_path);
 }
@@ -43,11 +45,13 @@ void UnixSocketServer::Start() {
 
     if (bind(server_fd_, (sockaddr *)&addr, sizeof(addr)) == -1) {
         perror("bind");
+        close(server_fd_);
         return;
     }
 
     if (listen(server_fd_, 16) == -1) {
         perror("listen");
+        close(server_fd_);
         return;
     }
 
@@ -57,48 +61,85 @@ void UnixSocketServer::Start() {
 
 void UnixSocketServer::Stop() {
     running_ = false;
+
+    if (server_fd_ >= 0) {
+        shutdown(server_fd_, SHUT_RDWR);
+        close(server_fd_);
+        server_fd_ = -1;
+    }
+
     if (accept_thread_.joinable())
         accept_thread_.join();
 
-    for (auto &[fd, t] : client_threads_) {
+    std::unordered_map<int, std::thread> local_clients;
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        for (const auto &[fd, _] : client_threads_) {
+            shutdown(fd, SHUT_RDWR);
+        }
+
+        local_clients.swap(client_threads_);
+    }
+
+    for (auto &[fd, thread] : local_clients) {
+        if (thread.joinable())
+            thread.join(); // already detached ones won't be in here
         close(fd);
-        if (t.joinable())
-            t.join();
     }
 
-    if (server_fd_ >= 0)
-        close(server_fd_);
     unlink(socket_path_.c_str());
 }
 
 void UnixSocketServer::AcceptLoop() {
     while (running_) {
         int client_fd = accept(server_fd_, nullptr, nullptr);
-        if (client_fd < 0)
+        if (client_fd < 0) {
+            if (running_) {
+                perror("accept");
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            }
             continue;
-        std::lock_guard<std::mutex> lock(mutex_);
-        client_threads_[client_fd] = std::thread(&UnixSocketServer::HandleClient, this, client_fd);
+        }
+
+        std::thread t(&UnixSocketServer::HandleClient, this, client_fd);
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            client_threads_[client_fd] = std::move(t);
+        }
     }
 }
 
 void UnixSocketServer::HandleClient(int client_fd) {
     char buffer[1024];
     while (running_) {
         int n = read(client_fd, buffer, sizeof(buffer));
-        if (n <= 0)
+        if (n <= 0) {
             break;
+        }
 
         std::string msg(buffer, n);
-        std::cout << "[IPC] Received: " << msg << std::endl;
+        DEBUG_PRINT("[%d] Received: %s", client_fd, msg.c_str());
 
         std::lock_guard<std::mutex> lock(mutex_);
-        for (auto &[_, callback] : peer_callbacks_) {
+        for (const auto &[_, callback] : peer_callbacks_) {
             if (callback) {
                 callback(msg);
             }
         }
     }
+
     close(client_fd);
-    std::lock_guard<std::mutex> lock(mutex_);
-    client_threads_.erase(client_fd);
+
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        auto it = client_threads_.find(client_fd);
+        if (it != client_threads_.end()) {
+            if (std::this_thread::get_id() == it->second.get_id()) {
+                it->second.detach();
+            }
+            client_threads_.erase(it);
+        }
+    }
+
+    DEBUG_PRINT("[%d] leaved!", client_fd);
 }