use ThrfitServer::steup/cleanUp when setup-up/tear-down, rebased

src/kvstore/NebulaStore.cpp

@@ -33,11 +33,9 @@ namespace nebula {
 namespace kvstore {
 
 NebulaStore::~NebulaStore() {
+  stop();
   LOG(INFO) << "Cut off the relationship with meta client";
   options_.partMan_.reset();
-  raftService_->stop();
-  LOG(INFO) << "Waiting for the raft service stop...";
-  raftService_->waitUntilStop();
   spaces_.clear();
   spaceListeners_.clear();
   bgWorkers_->stop();
@@ -55,7 +53,7 @@ bool NebulaStore::init() {
   CHECK(storeWorker_->start());
   snapshot_.reset(new NebulaSnapshotManager(this));
   raftService_ = raftex::RaftexService::createService(ioPool_, workers_, raftAddr_.port);
-  if (!raftService_->start()) {
+  if (raftService_ == nullptr) {
     LOG(ERROR) << "Start the raft service failed";
     return false;
   }
@@ -272,6 +270,7 @@ void NebulaStore::stop() {
   LOG(INFO) << "Stop the raft service...";
   raftService_->stop();
 
+  LOG(INFO) << "Stop kv engine...";
   for (const auto& space : spaces_) {
     for (const auto& engine : space.second->engines_) {
       engine->stop();

src/kvstore/NebulaStore.h

@@ -111,7 +111,7 @@ class NebulaStore : public KVStore, public Handler {
   bool init();
 
   /**
-   * @brief Stop the raft service and kv engine
+   * @brief Stop the engine
    */
   void stop() override;
 

src/kvstore/raftex/RaftPart.cpp

@@ -341,6 +341,8 @@ void RaftPart::stop() {
 
   VLOG(1) << idStr_ << "Invoked stop() on all peer hosts";
 
+  // Host::waitForStop will wait a callback executed in ioThreadPool, so make sure the
+  // RaftPart::stop SHOULD NOT be executed in the same ioThreadPool
   for (auto& h : hosts) {
     VLOG(1) << idStr_ << "Waiting " << h->idStr() << " to stop";
     h->waitForStop();

src/kvstore/raftex/RaftexService.cpp

@@ -21,93 +21,37 @@ namespace raftex {
  *
  ******************************************************/
 std::shared_ptr<RaftexService> RaftexService::createService(
-    std::shared_ptr<folly::IOThreadPoolExecutor> pool,
+    std::shared_ptr<folly::IOThreadPoolExecutor> ioPool,
     std::shared_ptr<folly::Executor> workers,
     uint16_t port) {
-  auto svc = std::shared_ptr<RaftexService>(new RaftexService());
-  CHECK(svc != nullptr) << "Failed to create a raft service";
-
-  svc->server_ = std::make_unique<apache::thrift::ThriftServer>();
-  CHECK(svc->server_ != nullptr) << "Failed to create a thrift server";
-  svc->server_->setInterface(svc);
-
-  svc->initThriftServer(pool, workers, port);
-  return svc;
-}
-
-RaftexService::~RaftexService() {}
-
-bool RaftexService::start() {
-  serverThread_.reset(new std::thread([&] { serve(); }));
-
-  waitUntilReady();
-
-  // start failed, reclaim resource
-  if (status_.load() != STATUS_RUNNING) {
-    waitUntilStop();
-    return false;
-  }
-
-  return true;
-}
-
-void RaftexService::waitUntilReady() {
-  while (status_.load() == STATUS_NOT_RUNNING) {
-    std::this_thread::sleep_for(std::chrono::microseconds(100));
-  }
-}
-
-void RaftexService::initThriftServer(std::shared_ptr<folly::IOThreadPoolExecutor> pool,
-                                     std::shared_ptr<folly::Executor> workers,
-                                     uint16_t port) {
-  LOG(INFO) << "Init thrift server for raft service, port: " << port;
-  if (FLAGS_enable_ssl) {
-    server_->setSSLConfig(nebula::sslContextConfig());
-  }
-  server_->setPort(port);
-  server_->setIdleTimeout(std::chrono::seconds(0));
-  if (pool != nullptr) {
-    server_->setIOThreadPool(pool);
-  }
-  if (workers != nullptr) {
-    server_->setThreadManager(
-        std::dynamic_pointer_cast<apache::thrift::concurrency::ThreadManager>(workers));
-  }
-  server_->setStopWorkersOnStopListening(false);
-}
-
-bool RaftexService::setup() {
   try {
-    server_->setup();
-    serverPort_ = server_->getAddress().getPort();
-
-    LOG(INFO) << "Starting the Raftex Service on " << serverPort_;
+    auto svc = std::shared_ptr<RaftexService>(new RaftexService());
+    auto server = std::make_unique<apache::thrift::ThriftServer>();
+    server->setPort(port);
+    server->setIdleTimeout(std::chrono::seconds(0));
+    if (ioPool != nullptr) {
+      server->setIOThreadPool(ioPool);
+    }
+    if (workers != nullptr) {
+      server->setThreadManager(
+          std::dynamic_pointer_cast<apache::thrift::concurrency::ThreadManager>(workers));
+    }
+    if (FLAGS_enable_ssl) {
+      server->setSSLConfig(nebula::sslContextConfig());
+    }
+    server->setInterface(svc);
+    server->setup();
+    svc->server_ = std::move(server);
+    svc->serverPort_ = svc->server_->getAddress().getPort();
+    LOG(INFO) << "Start raft service on " << svc->serverPort_;
+    return svc;
   } catch (const std::exception& e) {
-    LOG(ERROR) << "Setup the Raftex Service failed, error: " << e.what();
-    return false;
+    LOG(ERROR) << "Start raft service failed: " << e.what();
+    return nullptr;
+  } catch (...) {
+    LOG(ERROR) << "Start raft service failed";
+    return nullptr;
   }
-
-  return true;
-}
-
-void RaftexService::serve() {
-  LOG(INFO) << "Starting the Raftex Service";
-
-  if (!setup()) {
-    status_.store(STATUS_SETUP_FAILED);
-    return;
-  }
-
-  SCOPE_EXIT {
-    server_->cleanUp();
-  };
-
-  status_.store(STATUS_RUNNING);
-  LOG(INFO) << "Start the Raftex Service successfully";
-  server_->getEventBaseManager()->getEventBase()->loopForever();
-
-  status_.store(STATUS_NOT_RUNNING);
-  LOG(INFO) << "The Raftex Service stopped";
 }
 
 std::shared_ptr<folly::IOThreadPoolExecutor> RaftexService::getIOThreadPool() const {
@@ -119,11 +63,6 @@ std::shared_ptr<folly::Executor> RaftexService::getThreadManager() {
 }
 
 void RaftexService::stop() {
-  int expected = STATUS_RUNNING;
-  if (!status_.compare_exchange_strong(expected, STATUS_NOT_RUNNING)) {
-    return;
-  }
-
   // stop service
   LOG(INFO) << "Stopping the raftex service on port " << serverPort_;
   std::unordered_map<std::pair<GraphSpaceID, PartitionID>, std::shared_ptr<RaftPart>> parts;
@@ -139,16 +78,6 @@ void RaftexService::stop() {
   server_->stop();
 }
 
-void RaftexService::waitUntilStop() {
-  if (serverThread_) {
-    serverThread_->join();
-    serverThread_.reset();
-    server_.reset();
-    LOG(INFO) << "Server thread has stopped. Service on port " << serverPort_
-              << " is ready to be destroyed";
-  }
-}
-
 void RaftexService::addPartition(std::shared_ptr<RaftPart> part) {
   // todo(doodle): If we need to start both listener and normal replica on same
   // hosts, this class need to be aware of type.

src/kvstore/raftex/RaftexService.h

@@ -27,20 +27,20 @@ class RaftexService : public cpp2::RaftexServiceSvIf {
   /**
    * @brief Create a raft service
    *
-   * @param pool IOThreadPool to use
+   * @param ioPool IOThreadPool to use
    * @param workers Worker thread pool to use
    * @param port Listen port of thrift server
    * @return std::shared_ptr<RaftexService>
    */
   static std::shared_ptr<RaftexService> createService(
-      std::shared_ptr<folly::IOThreadPoolExecutor> pool,
+      std::shared_ptr<folly::IOThreadPoolExecutor> ioPool,
       std::shared_ptr<folly::Executor> workers,
       uint16_t port = 0);
 
   /**
    * @brief Destroy the Raftex Service
    */
-  virtual ~RaftexService();
+  virtual ~RaftexService() = default;
 
   /**
    * @brief Return the raft thrift server port
@@ -64,22 +64,10 @@ class RaftexService : public cpp2::RaftexServiceSvIf {
   std::shared_ptr<folly::Executor> getThreadManager();
 
   /**
-   * @brief Start the raft thrift server
-   *
-   * @return Whether start succeed
-   */
-  bool start();
-
-  /**
-   * @brief Set the state to stopped
+   * @brief Stop all partitions and wait thrift server stops
    */
   void stop();
 
-  /**
-   * @brief Wait until the thrift server has been stopped
-   */
-  void waitUntilStop();
-
   /**
    * @brief Handle leader election request in worker thread
    *
@@ -143,40 +131,11 @@ class RaftexService : public cpp2::RaftexServiceSvIf {
   std::shared_ptr<RaftPart> findPart(GraphSpaceID spaceId, PartitionID partId);
 
  private:
-  /**
-   * @brief Start the thrift server
-   *
-   * @param pool IO thread pool
-   * @param workers Worker thread pool
-   * @param port Thrift port to listener
-   */
-  void initThriftServer(std::shared_ptr<folly::IOThreadPoolExecutor> pool,
-                        std::shared_ptr<folly::Executor> workers,
-                        uint16_t port = 0);
-
-  /**
-   * @brief Prepare the setup of thrift server
-   *
-   * @return Return whether succeed
-   */
-  bool setup();
-  void serve();
-
-  /**
-   * @brief Wait until the service is ready to serve
-   */
-  void waitUntilReady();
-
   RaftexService() = default;
 
- private:
   std::unique_ptr<apache::thrift::ThriftServer> server_;
-  std::unique_ptr<std::thread> serverThread_;
   uint32_t serverPort_;
 
-  enum RaftServiceStatus { STATUS_NOT_RUNNING = 0, STATUS_SETUP_FAILED = 1, STATUS_RUNNING = 2 };
-  std::atomic_int status_{STATUS_NOT_RUNNING};
-
   folly::RWSpinLock partsLock_;
   std::unordered_map<std::pair<GraphSpaceID, PartitionID>, std::shared_ptr<RaftPart>> parts_;
 };

src/kvstore/raftex/test/RaftexTestBase.cpp

@@ -161,7 +161,6 @@ void setupRaft(int32_t numCopies,
   // Set up services
   for (int i = 0; i < numCopies; ++i) {
     services.emplace_back(RaftexService::createService(nullptr, nullptr));
-    if (!services.back()->start()) return;
     uint16_t port = services.back()->getServerPort();
     allHosts.emplace_back(ipStr, port);
   }
@@ -217,7 +216,7 @@ void finishRaft(std::vector<std::shared_ptr<RaftexService>>& services,
   workers->wait();
   LOG(INFO) << "Waiting for all service stopped";
   for (auto& svc : services) {
-    svc->waitUntilStop();
+    svc->stop();
   }
 }
 

src/mock/MockCluster.h

@@ -34,8 +34,6 @@ class MockCluster {
 
   ~MockCluster() {
     stop();
-    storageAdminServer_.reset();
-    graphStorageServer_.reset();
   }
 
   void startAll();
@@ -94,6 +92,16 @@ class MockCluster {
   }
 
   void stop() {
+    if (storageKV_) {
+      storageKV_->stop();
+    }
+    if (metaKV_) {
+      metaKV_->stop();
+    }
+
+    storageAdminServer_.reset();
+    graphStorageServer_.reset();
+
     if (metaClient_) {
       metaClient_->notifyStop();
       metaClient_->stop();
@@ -102,12 +110,6 @@ class MockCluster {
       metaClient_->notifyStop();
       lMetaClient_->stop();
     }
-    if (metaKV_) {
-      metaKV_->stop();
-    }
-    if (storageKV_) {
-      storageKV_->stop();
-    }
     if (esListener_) {
       esListener_->stop();
     }