MS-572 fix Milvus crash when get SIGINT

Former-commit-id: 32b92068024731fcc3451271c7f66f8a37b68cbd

cpp/src/server/Server.cpp

@@ -42,7 +42,7 @@ namespace zilliz {
 namespace milvus {
 namespace server {
 
-Server&
+Server &
 Server::Instance() {
     static Server server;
     return server;
@@ -203,6 +203,10 @@ Server::Start() {
             std::cout << "Milvus server start successfully." << std::endl;
             StartService();
 
+            while (running_) {
+                sleep(10);
+            }
+
         } catch (std::exception &ex) {
             std::cerr << "Milvus server encounter exception: " << std::string(ex.what())
                       << "Is another server instance running?";
@@ -210,7 +214,6 @@ Server::Start() {
         }
     } while (false);
 
-    Stop();
     return 0;
 }
 
@@ -242,6 +245,10 @@ Server::Stop() {
     }
 
     StopService();
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        running_ = false;
+    }
 
     std::cerr << "Milvus server is closed!" << std::endl;
 }
@@ -264,12 +271,12 @@ Server::StartService() {
     engine::KnowhereResource::Initialize();
     engine::StartSchedulerService();
     DBWrapper::GetInstance().StartService();
-    grpc::GrpcMilvusServer::StartService();
+    grpc::GrpcMilvusServer::GetInstance().Start();
 }
 
 void
 Server::StopService() {
-    grpc::GrpcMilvusServer::StopService();
+    grpc::GrpcMilvusServer::GetInstance().Stop();
     DBWrapper::GetInstance().StopService();
     engine::StopSchedulerService();
     engine::KnowhereResource::Finalize();

cpp/src/server/Server.h

@@ -21,23 +21,27 @@
 
 #include <cstdint>
 #include <string>
+#include <mutex>
+
 
 namespace zilliz {
 namespace milvus {
 namespace server {
 
 class Server {
-public:
+ public:
     static Server &Instance();
 
-    void Init(int64_t daemonized, const std::string &pid_filename, const std::string &config_filename,
+    void Init(int64_t daemonized,
+              const std::string &pid_filename,
+              const std::string &config_filename,
               const std::string &log_config_file);
 
     int Start();
 
     void Stop();
 
-private:
+ private:
     Server();
 
     ~Server();
@@ -50,12 +54,15 @@ private:
 
     void StopService();
 
-private:
+ private:
     int64_t daemonized_ = 0;
     int pid_fd = -1;
     std::string pid_filename_;
     std::string config_filename_;
     std::string log_config_file_;
+
+    std::mutex mutex_;
+    bool running_ = true;
 };  // Server
 
 }   // server

cpp/src/server/grpc_impl/GrpcMilvusServer.cpp

@@ -42,7 +42,6 @@ namespace milvus {
 namespace server {
 namespace grpc {
 
-static std::unique_ptr<::grpc::Server> server;
 
 constexpr long MESSAGE_SIZE = -1;
 
@@ -58,13 +57,22 @@ class NoReusePortOption : public ::grpc::ServerBuilderOption {
 };
 
 
+void
+GrpcMilvusServer::Start() {
+    thread_ptr_ = std::make_shared<std::thread>(&GrpcMilvusServer::StartService, this);
+}
+
+void
+GrpcMilvusServer::Stop() {
+    StopService();
+    if (thread_ptr_) {
+        thread_ptr_->join();
+        thread_ptr_ = nullptr;
+    }
+}
+
 Status
 GrpcMilvusServer::StartService() {
-    if (server != nullptr) {
-        std::cout << "stop service!\n";
-        StopService();
-    }
-
     ServerConfig &config = ServerConfig::GetInstance();
     ConfigNode server_config = config.GetConfig(CONFIG_SERVER);
     ConfigNode engine_config = config.GetConfig(CONFIG_ENGINE);
@@ -87,16 +95,16 @@ GrpcMilvusServer::StartService() {
     builder.AddListeningPort(server_address, ::grpc::InsecureServerCredentials());
     builder.RegisterService(&service);
 
-    server = builder.BuildAndStart();
-    server->Wait();
+    server_ptr_ = builder.BuildAndStart();
+    server_ptr_->Wait();
 
     return Status::OK();
 }
 
 Status
 GrpcMilvusServer::StopService() {
-    if (server != nullptr) {
-        server->Shutdown();
+    if (server_ptr_ != nullptr) {
+        server_ptr_->Shutdown();
     }
 
     return Status::OK();

cpp/src/server/grpc_impl/GrpcMilvusServer.h

@@ -21,6 +21,8 @@
 
 #include <cstdint>
 #include <string>
+#include <thread>
+#include <grpcpp/grpcpp.h>
 
 namespace zilliz {
 namespace milvus {
@@ -28,12 +30,25 @@ namespace server {
 namespace grpc {
 
 class GrpcMilvusServer {
-public:
-    static Status
-    StartService();
+ public:
+    static GrpcMilvusServer& GetInstance() {
+        static GrpcMilvusServer grpc_server;
+        return grpc_server;
+    }
 
-    static Status
-    StopService();
+    void Start();
+    void Stop();
+
+ private:
+    GrpcMilvusServer() = default;
+    ~GrpcMilvusServer() = default;
+
+    Status StartService();
+    Status StopService();
+
+ private:
+    std::unique_ptr<::grpc::Server> server_ptr_;
+    std::shared_ptr<std::thread> thread_ptr_;
 };
 
 }

cpp/src/utils/SignalUtil.cpp

@@ -31,7 +31,7 @@ void SignalUtil::HandleSignal(int signum){
     switch(signum){
         case SIGINT:
         case SIGUSR2:{
-            SERVER_LOG_INFO << "Server received signal:" << std::to_string(signum);
+            SERVER_LOG_INFO << "Server received signal: " << signum;
 
             server::Server& server_ptr = server::Server::Instance();
             server_ptr.Stop();
@@ -39,7 +39,7 @@ void SignalUtil::HandleSignal(int signum){
             exit(0);
         }
         default:{
-            SERVER_LOG_INFO << "Server received critical signal:" << std::to_string(signum);
+            SERVER_LOG_INFO << "Server received critical signal: " << signum;
             SignalUtil::PrintStacktrace();
 
             server::Server& server_ptr = server::Server::Instance();