fix listener may crash when remove

conf/nebula-storaged-listener.conf.default

@@ -3,13 +3,13 @@
 # Whether to run as a daemon process
 --daemonize=true
 # The file to host the process id
---pid_file=pids_listener/nebula-storaged.pid
+--pid_file=pids/nebula-storaged-listener.pid
 # Whether to use the configuration obtained from the configuration file
 --local_config=true
 
 ########## logging ##########
 # The directory to host logging files
---log_dir=logs_listener
+--log_dir=logs_storaged_listener
 # Log level, 0, 1, 2, 3 for INFO, WARNING, ERROR, FATAL respectively
 --minloglevel=0
 # Verbose log level, 1, 2, 3, 4, the higher of the level, the more verbose of the logging
@@ -19,8 +19,8 @@
 # Whether to redirect stdout and stderr to separate output files
 --redirect_stdout=true
 # Destination filename of stdout and stderr, which will also reside in log_dir.
---stdout_log_file=storaged-stdout.log
---stderr_log_file=storaged-stderr.log
+--stdout_log_file=storaged-listener-stdout.log
+--stderr_log_file=storaged-listener-stderr.log
 # Copy log messages at or above this level to stderr in addition to logfiles. The numbers of severity levels INFO, WARNING, ERROR, and FATAL are 0, 1, 2, and 3, respectively.
 --stderrthreshold=2
 # Wether logging files' name contain timestamp.

conf/nebula-storaged-listener.conf.production

@@ -3,13 +3,13 @@
 # Whether to run as a daemon process
 --daemonize=true
 # The file to host the process id
---pid_file=pids_listener/nebula-storaged-listener.pid
+--pid_file=pids/nebula-storaged-listener.pid
 # Whether to use the configuration obtained from the configuration file
 --local_config=true
 
 ########## logging ##########
 # The directory to host logging files
---log_dir=logs_storage_listener
+--log_dir=logs_storaged_listener
 # Log level, 0, 1, 2, 3 for INFO, WARNING, ERROR, FATAL respectively
 --minloglevel=0
 # Verbose log level, 1, 2, 3, 4, the higher of the level, the more verbose of the logging

src/graph/executor/admin/SpaceExecutor.cpp

@@ -258,9 +258,9 @@ folly::Future<Status> ShowCreateSpaceExecutor::execute() {
         row.values.emplace_back(properties.get_space_name());
         auto fmt = properties.comment_ref().has_value()
                        ? "CREATE SPACE `%s` (partition_num = %d, replica_factor = %d, "
-                         "charset = %s, collate = %s, vid_type = %s) ON %s comment = '%s'"
+                         "charset = %s, collate = %s, vid_type = %s) comment = '%s'"
                        : "CREATE SPACE `%s` (partition_num = %d, replica_factor = %d, "
-                         "charset = %s, collate = %s, vid_type = %s) ON %s";
+                         "charset = %s, collate = %s, vid_type = %s)";
         auto zoneNames = folly::join(",", properties.get_zone_names());
         if (properties.comment_ref().has_value()) {
           row.values.emplace_back(
@@ -271,7 +271,6 @@ folly::Future<Status> ShowCreateSpaceExecutor::execute() {
                                   properties.get_charset_name().c_str(),
                                   properties.get_collate_name().c_str(),
                                   SchemaUtil::typeToString(properties.get_vid_type()).c_str(),
-                                  zoneNames.c_str(),
                                   properties.comment_ref()->c_str()));
         } else {
           row.values.emplace_back(
@@ -281,8 +280,7 @@ folly::Future<Status> ShowCreateSpaceExecutor::execute() {
                                   properties.get_replica_factor(),
                                   properties.get_charset_name().c_str(),
                                   properties.get_collate_name().c_str(),
-                                  SchemaUtil::typeToString(properties.get_vid_type()).c_str(),
-                                  zoneNames.c_str()));
+                                  SchemaUtil::typeToString(properties.get_vid_type()).c_str()));
         }
         dataSet.rows.emplace_back(std::move(row));
         return finish(ResultBuilder()

src/kvstore/listener/Listener.cpp

@@ -77,8 +77,11 @@ void Listener::start(std::vector<HostAddr>&& peers, bool) {
   status_ = Status::RUNNING;
   role_ = Role::LEARNER;
 
-  size_t delayMS = 100 + folly::Random::rand32(900);
-  bgWorkers_->addDelayTask(delayMS, &Listener::doApply, this);
+  applyPool_ = std::make_unique<folly::IOThreadPoolExecutor>(
+      1,
+      std::make_shared<folly::NamedThreadFactory>(
+          folly::stringPrintf("ListenerApplyPool-%d-%d", spaceId_, partId_)));
+  applyPool_->add(std::bind(&Listener::doApply, this));
 }
 
 void Listener::stop() {
@@ -88,6 +91,8 @@ void Listener::stop() {
     status_ = Status::STOPPED;
     leader_ = {"", 0};
   }
+  applyPool_->stop();
+  applyPool_->join();
 }
 
 bool Listener::preProcessLog(LogID logId,
@@ -138,21 +143,14 @@ std::tuple<nebula::cpp2::ErrorCode, LogID, TermID> Listener::commitLogs(
 }
 
 void Listener::doApply() {
-  if (isStopped()) {
-    return;
-  }
-
-  if (needToCleanupSnapshot()) {
-    cleanupSnapshot();
-  }
-  // todo(doodle): only put is handled, all remove is ignored for now
-  folly::via(executor_.get(), [this] {
-    SCOPE_EXIT {
-      bgWorkers_->addDelayTask(
-          FLAGS_listener_commit_interval_secs * 1000, &Listener::doApply, this);
-    };
+  while (!isStopped()) {
+    if (needToCleanupSnapshot()) {
+      cleanupSnapshot();
+    }
+    // todo(doodle): only put is handled, all remove is ignored for now
     processLogs();
-  });
+    sleep(FLAGS_listener_commit_interval_secs);
+  }
 }
 
 void Listener::resetListener() {

src/kvstore/listener/Listener.h

@@ -275,6 +275,7 @@ class Listener : public raftex::RaftPart {
   LogID leaderCommitId_ = 0;
   LogID lastApplyLogId_ = 0;
   std::set<HostAddr> peers_;
+  std::unique_ptr<folly::IOThreadPoolExecutor> applyPool_;
 };
 
 }  // namespace kvstore