fix(interactive): make the server exit cleanly (#4566)

Fixes

Author: liulx20

flex/bin/interactive_server.cc

@@ -54,15 +54,6 @@ std::string parse_codegen_dir(const bpo::variables_map& vm) {
   return codegen_dir;
 }
 
-void blockSignal(int sig) {
-  sigset_t set;
-  sigemptyset(&set);
-  sigaddset(&set, sig);
-  if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0) {
-    perror("pthread_sigmask");
-  }
-}
-
 // When graph_schema is not specified, codegen proxy will use the running graph
 // schema in graph_db_service
 void init_codegen_proxy(const bpo::variables_map& vm,

flex/bin/rt_server.cc

@@ -13,11 +13,11 @@
  * limitations under the License.
  */
 
-#include "grape/util.h"
-
 #include "flex/engines/graph_db/database/graph_db.h"
 #include "flex/engines/http_server/graph_db_service.h"
 #include "flex/engines/http_server/options.h"
+#include "flex/utils/service_utils.h"
+#include "grape/util.h"
 
 #include <boost/program_options.hpp>
 #include <seastar/core/alien.hh>
@@ -77,6 +77,9 @@ int main(int argc, char** argv) {
   setenv("TZ", "Asia/Shanghai", 1);
   tzset();
 
+  gs::blockSignal(SIGINT);
+  gs::blockSignal(SIGTERM);
+
   double t0 = -grape::GetCurrentTime();
   auto& db = gs::GraphDB::get();
   std::string graph_schema_path = data_path + "/graph.yaml";
@@ -107,6 +110,7 @@ int main(int argc, char** argv) {
   service_config.start_compiler = false;
   service_config.set_sharding_mode(vm["sharding-mode"].as<std::string>());
   server::GraphDBService::get().init(service_config);
+
   server::GraphDBService::get().run_and_wait_for_exit();
 
   return 0;

flex/engines/graph_db/app/cypher_app_utils.cc

@@ -110,10 +110,16 @@ bool generate_plan(const std::string& query, const std::string& statistics,
 
   // call compiler to generate plan
   {
+    int pipefd[2];
+    if (pipe(pipefd) == -1) {
+      LOG(ERROR) << "pipe failed!" << strerror(errno);
+      exit(EXIT_FAILURE);
+    }
+
     pid_t pid = fork();
 
     if (pid == -1) {
-      std::cerr << "Fork failed!" << std::endl;
+      LOG(ERROR) << "fork failed!" << strerror(errno);
       return false;
     } else if (pid == 0) {
       const char* const args[] = {
@@ -127,22 +133,45 @@ bool generate_plan(const std::string& query, const std::string& statistics,
           "temp.cypher.yaml",
           nullptr  // execvp expects a null-terminated array
       };
+
+      close(pipefd[0]);
+
+      if (dup2(pipefd[1], STDERR_FILENO) == -1) {
+        LOG(ERROR) << "dup2 failed!" << strerror(errno);
+        exit(EXIT_FAILURE);
+      }
+
+      close(pipefd[1]);
+
       execvp(args[0], const_cast<char* const*>(args));
 
       std::cerr << "Exec failed!" << std::endl;
       return false;
     } else {
+      close(pipefd[1]);
+
+      ssize_t count;
+      constexpr size_t BUFFSIZ = 4096;
+      char buffer[BUFFSIZ];
+      std::string error_message;
+      while ((count = read(pipefd[0], buffer, sizeof(buffer) - 1)) > 0) {
+        buffer[count] = '\0';
+        error_message += buffer;
+      }
+
       int status;
       waitpid(pid, &status, 0);
       if (WIFEXITED(status)) {
         VLOG(1) << "Child exited with status " << WEXITSTATUS(status)
                 << std::endl;
       }
+      close(pipefd[0]);
 
       {
         std::ifstream file(output_file, std::ios::binary);
 
         if (!file.is_open()) {
+          LOG(ERROR) << "Compiler message: " << error_message;
           return false;
         }
 
@@ -157,6 +186,7 @@ bool generate_plan(const std::string& query, const std::string& statistics,
 
         file.close();
         if (!plan.ParseFromString(std::string(buffer))) {
+          LOG(ERROR) << "Compiler message: " << error_message;
           return false;
         }
       }

flex/engines/graph_db/database/graph_db.cc

@@ -59,7 +59,7 @@ GraphDB::~GraphDB() {
     compact_thread_.join();
   }
   if (contexts_ != nullptr) {
-    showAppMetrics();
+    // showAppMetrics();
     for (int i = 0; i < thread_num_; ++i) {
       contexts_[i].~SessionLocalContext();
     }
@@ -478,16 +478,16 @@ void GraphDB::openWalAndCreateContexts(const GraphDBConfig& config,
                                  data_dir);
   }
   VLOG(1) << "Using wal uri: " << wal_uri;
+
+  auto wal_parser = WalParserFactory::CreateWalParser(wal_uri);
+  ingestWals(*wal_parser, data_dir, thread_num_);
   for (int i = 0; i < thread_num_; ++i) {
     new (&contexts_[i])
         SessionLocalContext(*this, data_dir, i, allocator_strategy,
                             WalWriterFactory::CreateWalWriter(wal_uri));
     contexts_[i].logger->open(wal_uri, i);
   }
 
-  auto wal_parser = WalParserFactory::CreateWalParser(wal_uri);
-  ingestWals(*wal_parser, data_dir, thread_num_);
-
   initApps(graph_.schema().GetPlugins());
   VLOG(1) << "Successfully restore load plugins";
 }

flex/engines/graph_db/database/wal/wal.cc

@@ -29,8 +29,8 @@ std::string get_wal_uri_scheme(const std::string& uri) {
     scheme = uri.substr(0, pos);
   }
   if (scheme.empty()) {
-    LOG(INFO) << "No scheme found in wal uri: " << uri
-              << ", using default scheme: file";
+    VLOG(1) << "No scheme found in wal uri: " << uri
+            << ", using default scheme: file";
     scheme = "file";
   }
   return scheme;

flex/engines/graph_db/runtime/utils/cypher_runner_impl.cc

@@ -39,13 +39,21 @@ bool CypherRunnerImpl::gen_plan(const GraphDB& db, const std::string& query,
   }
 
   physical::PhysicalPlan plan;
-  if (!generate_plan(query, statistics, compiler_path, compiler_yaml, tmp_dir,
-                     plan)) {
-    LOG(ERROR) << "Generate plan failed for query: " << query;
-    return false;
+  {
+    // avoid multiple threads to generate plan for the same query
+    std::unique_lock<std::mutex> lock(mutex_);
+    if (plan_cache.get(query, plan_str)) {
+      return true;
+    }
+    if (!generate_plan(query, statistics, compiler_path, compiler_yaml, tmp_dir,
+                       plan)) {
+      LOG(ERROR) << "Generate plan failed for query: " << query;
+      return false;
+    }
+    plan_str = plan.SerializeAsString();
+    plan_cache.put(query, plan_str);
   }
-  plan_str = plan.SerializeAsString();
-  plan_cache.put(query, plan_str);
+
   return true;
 }
 

flex/engines/graph_db/runtime/utils/cypher_runner_impl.h

@@ -71,6 +71,7 @@ class CypherRunnerImpl {
   CypherRunnerImpl(const CypherRunnerImpl&) = delete;
   CypherRunnerImpl& operator=(const CypherRunnerImpl&) = delete;
   PlanCache plan_cache_;
+  std::mutex mutex_;
 };
 }  // namespace runtime
 }  // namespace gs

flex/storages/rt_mutable_graph/mutable_property_fragment.cc

@@ -197,7 +197,7 @@ void MutablePropertyFragment::Open(const std::string& work_dir,
     // We will reserve the at least 4096 slots for each vertex label
     size_t vertex_capacity =
         std::max(lf_indexers_[i].capacity(), (size_t) 4096);
-    if (vertex_capacity >= lf_indexers_[i].capacity()) {
+    if (vertex_capacity > lf_indexers_[i].capacity()) {
       lf_indexers_[i].reserve(vertex_capacity);
     }
     vertex_data_[i].resize(vertex_capacity);

flex/utils/service_utils.h

@@ -46,6 +46,16 @@ namespace gs {
 static constexpr const char* CODEGEN_BIN = "load_plan_and_gen.sh";
 
 /// Util functions.
+
+inline void blockSignal(int sig) {
+  sigset_t set;
+  sigemptyset(&set);
+  sigaddset(&set, sig);
+  if (pthread_sigmask(SIG_BLOCK, &set, NULL) != 0) {
+    perror("pthread_sigmask");
+  }
+}
+
 inline int64_t GetCurrentTimeStamp() {
   return std::chrono::duration_cast<std::chrono::milliseconds>(
              std::chrono::system_clock::now().time_since_epoch())