Merge branch 'branch-0.5.0' into 'branch-0.5.0'

Branch 0.5.0

See merge request megasearch/milvus!588

Former-commit-id: eced8ff211797dcb4b914c2db5b92db2670e3249

cpp/CHANGELOG.md

@@ -6,6 +6,7 @@ Please mark all change in change log and use the ticket from JIRA.
 
 ## Bug
 - MS-568 - Fix gpuresource free error
+- MS-572 - Milvus crash when get SIGINT
 
 ## Improvement
 - MS-552 - Add and change the easylogging library

cpp/src/main.cpp

@@ -15,19 +15,20 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "server/Server.h"
-#include "version.h"
-
 #include <getopt.h>
 #include <libgen.h>
 #include <cstring>
 #include <string>
 #include <signal.h>
-#include "utils/easylogging++.h"
-#include "metrics/Metrics.h"
+#include <unistd.h>
 
+#include "utils/easylogging++.h"
 #include "utils/SignalUtil.h"
 #include "utils/CommonUtil.h"
+#include "metrics/Metrics.h"
+#include "server/Server.h"
+#include "version.h"
+
 
 INITIALIZE_EASYLOGGINGPP
 
@@ -40,12 +41,6 @@ main(int argc, char *argv[]) {
     std::cout << std::endl << "Welcome to use Milvus by Zilliz!" << std::endl;
     std::cout << "Milvus " << BUILD_TYPE << " version: v" << MILVUS_VERSION << " built at " << BUILD_TIME << std::endl;
 
-    signal(SIGINT, server::SignalUtil::HandleSignal);
-    signal(SIGSEGV, server::SignalUtil::HandleSignal);
-    signal(SIGUSR1, server::SignalUtil::HandleSignal);
-    signal(SIGUSR2, server::SignalUtil::HandleSignal);
-
-    std::string app_name = basename(argv[0]);
     static struct option long_options[] = {{"conf_file", required_argument, 0, 'c'},
                                            {"log_conf_file", required_argument, 0, 'l'},
                                            {"help", no_argument, 0, 'h'},
@@ -55,14 +50,12 @@ main(int argc, char *argv[]) {
 
     int option_index = 0;
     int64_t start_daemonized = 0;
-//    int pid_fd;
 
     std::string config_filename, log_config_file;
     std::string pid_filename;
+    std::string app_name = argv[0];
 
-    app_name = argv[0];
-
-    if(argc < 2) {
+    if (argc < 2) {
         print_help(app_name);
         std::cout << "Milvus server exit..." << std::endl;
         return EXIT_FAILURE;
@@ -109,14 +102,27 @@ main(int argc, char *argv[]) {
         }
     }
 
-    server::Server& server = server::Server::Instance();
+    server::Server &server = server::Server::Instance();
     server.Init(start_daemonized, pid_filename, config_filename, log_config_file);
-    return server.Start();
+    server.Start();
+
+    /* Handle Signal */
+    signal(SIGHUP, server::SignalUtil::HandleSignal);
+    signal(SIGINT, server::SignalUtil::HandleSignal);
+    signal(SIGUSR1, server::SignalUtil::HandleSignal);
+    signal(SIGSEGV, server::SignalUtil::HandleSignal);
+    signal(SIGUSR2, server::SignalUtil::HandleSignal);
+    signal(SIGTERM, server::SignalUtil::HandleSignal);
+
+    /* wait signal */
+    pause();
+
+    return 0;
 }
 
 void
 print_help(const std::string &app_name) {
-    std::cout << std::endl<< "Usage: " << app_name << " [OPTIONS]" << std::endl << std::endl;
+    std::cout << std::endl << "Usage: " << app_name << " [OPTIONS]" << std::endl << std::endl;
     std::cout << "  Options:" << std::endl;
     std::cout << "   -h --help                 Print this help" << std::endl;
     std::cout << "   -c --conf_file filename   Read configuration from the file" << std::endl;

cpp/src/server/Server.cpp

@@ -17,7 +17,7 @@
 
 #include <thread>
 #include "Server.h"
-#include "server/grpc_impl/GrpcMilvusServer.h"
+#include "server/grpc_impl/GrpcServer.h"
 #include "utils/Log.h"
 #include "utils/LogUtil.h"
 #include "utils/SignalUtil.h"
@@ -42,7 +42,7 @@ namespace zilliz {
 namespace milvus {
 namespace server {
 
-Server&
+Server &
 Server::Instance() {
     static Server server;
     return server;
@@ -154,71 +154,61 @@ Server::Daemonize() {
     }
 }
 
-int
+void
 Server::Start() {
-
     if (daemonized_) {
         Daemonize();
     }
 
-    do {
-        try {
-            // Read config file
-            if (LoadConfig() != SERVER_SUCCESS) {
-                return 1;
-            }
+    try {
+        /* Read config file */
+        if (LoadConfig() != SERVER_SUCCESS) {
+            std::cerr << "Milvus server fail to load config file" << std::endl;
+            return;
+        }
 
-            //log path is defined by LoadConfig, so InitLog must be called after LoadConfig
-            ServerConfig &config = ServerConfig::GetInstance();
-            ConfigNode server_config = config.GetConfig(CONFIG_SERVER);
+        /* log path is defined in Config file, so InitLog must be called after LoadConfig */
+        ServerConfig &config = ServerConfig::GetInstance();
+        ConfigNode server_config = config.GetConfig(CONFIG_SERVER);
 
-            std::string time_zone = server_config.GetValue(CONFIG_TIME_ZONE, "UTC+8");
-            if (time_zone.length() == 3) {
+        std::string time_zone = server_config.GetValue(CONFIG_TIME_ZONE, "UTC+8");
+        if (time_zone.length() == 3) {
+            time_zone = "CUT";
+        } else {
+            int time_bias = std::stoi(time_zone.substr(3, std::string::npos));
+            if (time_bias == 0)
                 time_zone = "CUT";
+            else if (time_bias > 0) {
+                time_zone = "CUT" + std::to_string(-time_bias);
             } else {
-                int time_bias = std::stoi(time_zone.substr(3, std::string::npos));
-                if (time_bias == 0)
-                    time_zone = "CUT";
-                else if (time_bias > 0) {
-                    time_zone = "CUT" + std::to_string(-time_bias);
-                } else {
-                    time_zone = "CUT+" + std::to_string(-time_bias);
-                }
-            }
-
-            if (setenv("TZ", time_zone.c_str(), 1) != 0) {
-                return -1;
+                time_zone = "CUT+" + std::to_string(-time_bias);
             }
-            tzset();
+        }
 
-            InitLog(log_config_file_);
+        if (setenv("TZ", time_zone.c_str(), 1) != 0) {
+            std::cerr << "Fail to setenv" << std::endl;
+            return;
+        }
+        tzset();
 
-            // Handle Signal
-            signal(SIGINT, SignalUtil::HandleSignal);
-            signal(SIGHUP, SignalUtil::HandleSignal);
-            signal(SIGTERM, SignalUtil::HandleSignal);
-            server::Metrics::GetInstance().Init();
-            server::SystemInfo::GetInstance().Init();
+        InitLog(log_config_file_);
 
-            std::cout << "Milvus server start successfully." << std::endl;
-            StartService();
+        server::Metrics::GetInstance().Init();
+        server::SystemInfo::GetInstance().Init();
 
-        } catch (std::exception &ex) {
-            std::cerr << "Milvus server encounter exception: " << std::string(ex.what())
-                      << "Is another server instance running?";
-            break;
-        }
-    } while (false);
+        std::cout << "Milvus server start successfully." << std::endl;
+        StartService();
 
-    Stop();
-    return 0;
+    } catch (std::exception &ex) {
+        std::cerr << "Milvus server encounter exception: " << ex.what();
+    }
 }
 
 void
 Server::Stop() {
     std::cerr << "Milvus server is going to shutdown ..." << std::endl;
 
-    // Unlock and close lockfile
+    /* Unlock and close lockfile */
     if (pid_fd != -1) {
         int ret = lockf(pid_fd, F_ULOCK, 0);
         if (ret != 0) {
@@ -232,7 +222,7 @@ Server::Stop() {
         }
     }
 
-    // Try to delete lockfile
+    /* delete lockfile */
     if (!pid_filename_.empty()) {
         int ret = unlink(pid_filename_.c_str());
         if (ret != 0) {
@@ -264,12 +254,12 @@ Server::StartService() {
     engine::KnowhereResource::Initialize();
     engine::StartSchedulerService();
     DBWrapper::GetInstance().StartService();
-    grpc::GrpcMilvusServer::StartService();
+    grpc::GrpcServer::GetInstance().Start();
 }
 
 void
 Server::StopService() {
-    grpc::GrpcMilvusServer::StopService();
+    grpc::GrpcServer::GetInstance().Stop();
     DBWrapper::GetInstance().StopService();
     engine::StopSchedulerService();
     engine::KnowhereResource::Finalize();

cpp/src/server/Server.h

@@ -22,24 +22,25 @@
 #include <cstdint>
 #include <string>
 
+
 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 Start();
     void Stop();
 
-private:
+ private:
     Server();
-
     ~Server();
 
     void Daemonize();
@@ -47,10 +48,9 @@ private:
     ErrorCode LoadConfig();
 
     void StartService();
-
     void StopService();
 
-private:
+ private:
     int64_t daemonized_ = 0;
     int pid_fd = -1;
     std::string pid_filename_;

cpp/src/server/grpc_impl/GrpcRequestTask.cpp

@@ -23,7 +23,7 @@
 #include "utils/ValidationUtil.h"
 #include "../DBWrapper.h"
 #include "version.h"
-#include "GrpcMilvusServer.h"
+#include "GrpcServer.h"
 #include "db/Utils.h"
 #include "scheduler/SchedInst.h"
 //#include <gperftools/profiler.h>

cpp/src/server/grpc_impl/GrpcMilvusServer.cpp → cpp/src/server/grpc_impl/GrpcServer.cpp

@@ -16,7 +16,7 @@
 // under the License.
 
 #include "milvus.grpc.pb.h"
-#include "GrpcMilvusServer.h"
+#include "GrpcServer.h"
 #include "server/ServerConfig.h"
 #include "server/DBWrapper.h"
 #include "utils/Log.h"
@@ -34,7 +34,6 @@
 #include <grpcpp/client_context.h>
 #include <grpcpp/create_channel.h>
 #include <grpcpp/security/credentials.h>
-#include <grpcpp/grpcpp.h>
 
 
 namespace zilliz {
@@ -42,7 +41,6 @@ namespace milvus {
 namespace server {
 namespace grpc {
 
-static std::unique_ptr<::grpc::Server> server;
 
 constexpr long MESSAGE_SIZE = -1;
 
@@ -53,18 +51,28 @@ class NoReusePortOption : public ::grpc::ServerBuilderOption {
         args->SetInt(GRPC_ARG_ALLOW_REUSEPORT, 0);
     }
 
-    void UpdatePlugins(std::vector<std::unique_ptr<::grpc::ServerBuilderPlugin>> *
-    plugins) override {}
+    void UpdatePlugins(std::vector<std::unique_ptr<::grpc::ServerBuilderPlugin>> *plugins) override {
+
+    }
 };
 
 
-Status
-GrpcMilvusServer::StartService() {
-    if (server != nullptr) {
-        std::cout << "stop service!\n";
-        StopService();
+void
+GrpcServer::Start() {
+    thread_ptr_ = std::make_shared<std::thread>(&GrpcServer::StartService, this);
+}
+
+void
+GrpcServer::Stop() {
+    StopService();
+    if (thread_ptr_) {
+        thread_ptr_->join();
+        thread_ptr_ = nullptr;
     }
+}
 
+Status
+GrpcServer::StartService() {
     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();
+GrpcServer::StopService() {
+    if (server_ptr_ != nullptr) {
+        server_ptr_->Shutdown();
     }
 
     return Status::OK();

cpp/src/server/grpc_impl/GrpcMilvusServer.h → cpp/src/server/grpc_impl/GrpcServer.h

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

cpp/src/utils/SignalUtil.cpp

@@ -22,28 +22,29 @@
 #include <signal.h>
 #include <execinfo.h>
 
+
 namespace zilliz {
 namespace milvus {
 namespace server {
 
-void SignalUtil::HandleSignal(int signum){
+void SignalUtil::HandleSignal(int signum) {
 
-    switch(signum){
+    switch (signum) {
         case SIGINT:
-        case SIGUSR2:{
-            SERVER_LOG_INFO << "Server received signal:" << std::to_string(signum);
+        case SIGUSR2: {
+            SERVER_LOG_INFO << "Server received signal: " << signum;
 
-            server::Server& server_ptr = server::Server::Instance();
-            server_ptr.Stop();
+            server::Server &server = server::Server::Instance();
+            server.Stop();
 
             exit(0);
         }
-        default:{
-            SERVER_LOG_INFO << "Server received critical signal:" << std::to_string(signum);
+        default: {
+            SERVER_LOG_INFO << "Server received critical signal: " << signum;
             SignalUtil::PrintStacktrace();
 
-            server::Server& server_ptr = server::Server::Instance();
-            server_ptr.Stop();
+            server::Server &server = server::Server::Instance();
+            server.Stop();
 
             exit(1);
         }
@@ -54,9 +55,9 @@ void SignalUtil::PrintStacktrace() {
     SERVER_LOG_INFO << "Call stack:";
 
     const int size = 32;
-    void* array[size];
+    void *array[size];
     int stack_num = backtrace(array, size);
-    char ** stacktrace = backtrace_symbols(array, stack_num);
+    char **stacktrace = backtrace_symbols(array, stack_num);
     for (int i = 0; i < stack_num; ++i) {
         std::string info = stacktrace[i];
         SERVER_LOG_INFO << info;
@@ -64,7 +65,6 @@ void SignalUtil::PrintStacktrace() {
     free(stacktrace);
 }
 
-
 }
 }
 }