Integrate raft into storage

CMakeLists.txt

@@ -304,8 +304,8 @@ macro(nebula_link_libraries target)
  ${OPENSSL_CRYPTO_LIBRARY}
  ${KRB5_LIBRARIES}
  ${COMPRESSION_LIBRARIES}
- ${JEMALLOC_LIB}
  ${LIBUNWIND_LIBRARIES}
+ ${JEMALLOC_LIB}
  dl
  ${GETTIME_LIB}
  -pthread

cmake/FindNCURSES.cmake

@@ -24,7 +24,7 @@ find_path(
 
 find_path(
  NCURSES_INCLUDE_DIR
- NAMES ncurses/curses.h
+ NAMES ncurses.h curses.h ncurses/curses.h
  HINTS ${NCURSES_ROOT_DIR}/include
 )
 

src/common/base/EitherOr.h

@@ -12,8 +12,6 @@
 
 namespace nebula {
 
-namespace { // NOLINT
-
 using LeftType = std::true_type;
 using RightType = std::false_type;
 
@@ -23,8 +21,6 @@ enum class State : int16_t {
  RIGHT_TYPE = 2,
 };
 
-} // Anonymous namespace
-
 static constexpr LeftType* kConstructLeft = nullptr;
 static constexpr RightType* kConstructRight = nullptr;
 

src/common/network/NetworkUtils.cpp

@@ -83,6 +83,7 @@ StatusOr<std::unordered_map<std::string, std::string>> NetworkUtils::listDeviceA
  return dev2ipv4s;
 }
 
+
 bool NetworkUtils::getDynamicPortRange(uint16_t& low, uint16_t& high) {
  FILE* pipe = popen("cat /proc/sys/net/ipv4/ip_local_port_range", "r");
  if (!pipe) {

src/common/test/ServerContext.h

@@ -23,39 +23,39 @@ struct ServerContext {
  if (thread_ != nullptr) {
  thread_->join();
  }
- KVStore_ = nullptr;
+ kvStore_ = nullptr;
  server_ = nullptr;
  thread_ = nullptr;
  VLOG(3) << "~ServerContext";
  }
 
+
+ void mockCommon(const std::string& name,
+ uint16_t port,
+ std::shared_ptr<apache::thrift::ServerInterface> handler) {
+ server_ = std::make_unique<apache::thrift::ThriftServer>();
+ server_->setInterface(std::move(handler));
+ server_->setPort(port);
+ thread_ = std::make_unique<thread::NamedThread>(name, [this, name] {
+ server_->serve();
+ LOG(INFO) << "The " << name << " server has stopped";
+ });
+
+ while (!server_->getServeEventBase() ||
+ !server_->getServeEventBase()->isRunning()) {
+ usleep(10000);
+ }
+ port_ = server_->getAddress().getPort();
+ }
+
+
  std::unique_ptr<apache::thrift::ThriftServer> server_{nullptr};
  std::unique_ptr<thread::NamedThread> thread_{nullptr};
  // To keep meta and storage's KVStore
- std::unique_ptr<kvstore::KVStore> KVStore_{nullptr};
+ std::unique_ptr<kvstore::KVStore> kvStore_{nullptr};
  uint16_t port_{0};
 };
 
-static void mockCommon(test::ServerContext *sc,
- const std::string &name,
- uint16_t port,
- std::shared_ptr<apache::thrift::ServerInterface> handler) {
- if (nullptr == sc) {
- LOG(ERROR) << "ServerContext is nullptr";
- return;
- }
- sc->server_ = std::make_unique<apache::thrift::ThriftServer>();
- sc->server_->setInterface(std::move(handler));
- sc->server_->setPort(port);
- sc->thread_ = std::make_unique<thread::NamedThread>(name, [&]() {
- sc->server_->serve();
- LOG(INFO) << "Stop the server...";
- });
- while (!sc->server_->getServeEventBase() ||
- !sc->server_->getServeEventBase()->isRunning()) {
- }
- sc->port_ = sc->server_->getAddress().getPort();
-}
 
 } // namespace test
 } // namespace nebula

src/common/test/TestServerContext.cpp

@@ -24,7 +24,7 @@ class TestServer final : public apache::thrift::ServerInterface {
 TEST(ServerContext, mockCommon) {
  auto sc = std::make_unique<ServerContext>();
  auto handler = std::make_shared<TestServer>();
- test::mockCommon(sc.get(), "test", 0, handler);
+ sc->mockCommon("test", 0, handler);
 }
 
 } // namespace test

src/daemons/CMakeLists.txt

@@ -45,10 +45,13 @@ add_executable(
  $<TARGET_OBJECTS:meta_client>
  $<TARGET_OBJECTS:meta_thrift_obj>
  $<TARGET_OBJECTS:common_thrift_obj>
- $<TARGET_OBJECTS:thrift_obj>
+ $<TARGET_OBJECTS:raftex_obj>
+ $<TARGET_OBJECTS:raftex_thrift_obj>
+ $<TARGET_OBJECTS:wal_obj>
  $<TARGET_OBJECTS:dataman_obj>
  $<TARGET_OBJECTS:schema_obj>
  $<TARGET_OBJECTS:base_obj>
+ $<TARGET_OBJECTS:thrift_obj>
  $<TARGET_OBJECTS:thread_obj>
  $<TARGET_OBJECTS:time_obj>
  $<TARGET_OBJECTS:fs_obj>
@@ -77,9 +80,11 @@ add_executable(
  $<TARGET_OBJECTS:meta_client>
  $<TARGET_OBJECTS:meta_thrift_obj>
  $<TARGET_OBJECTS:common_thrift_obj>
- $<TARGET_OBJECTS:thrift_obj>
- $<TARGET_OBJECTS:network_obj>
+ $<TARGET_OBJECTS:raftex_obj>
+ $<TARGET_OBJECTS:raftex_thrift_obj>
+ $<TARGET_OBJECTS:wal_obj>
  $<TARGET_OBJECTS:base_obj>
+ $<TARGET_OBJECTS:thrift_obj>
  $<TARGET_OBJECTS:network_obj>
  $<TARGET_OBJECTS:thread_obj>
  $<TARGET_OBJECTS:time_obj>

src/daemons/MetaDaemon.cpp

@@ -11,7 +11,9 @@
 #include "webservice/WebService.h"
 #include "network/NetworkUtils.h"
 #include "process/ProcessUtils.h"
+#include "thread/GenericThreadPool.h"
 #include "kvstore/PartManager.h"
+#include "kvstore/NebulaStore.h"
 
 using nebula::ProcessUtils;
 using nebula::Status;
@@ -23,6 +25,8 @@ DEFINE_string(peers, "", "It is a list of IPs split by comma,"
  "the ips number equals replica number."
  "If empty, it means replica is 1");
 DEFINE_string(local_ip, "", "Local ip speicified for NetworkUtils::getLocalIP");
+DEFINE_int32(num_workers, 4, "Number of worker threads");
+DEFINE_int32(num_io_threads, 16, "Number of IO threads");
 DECLARE_string(part_man_type);
 
 DEFINE_string(pid_file, "pids/nebula-metad.pid", "File to hold the process id");
@@ -89,7 +93,7 @@ int main(int argc, char *argv[]) {
  LOG(ERROR) << "Bad local host addr, status:" << hostAddrRet.status();
  return EXIT_FAILURE;
  }
- auto& localHost = hostAddrRet.value();
+ auto& localhost = hostAddrRet.value();
 
  auto peersRet = nebula::network::NetworkUtils::toHosts(FLAGS_peers);
  if (!peersRet.ok()) {
@@ -108,22 +112,32 @@ int main(int argc, char *argv[]) {
  // The meta server has only one space, one part.
  partMan->addPart(0, 0, std::move(peersRet.value()));
 
+ // Generic thread pool
+ auto workers = std::make_shared<nebula::thread::GenericThreadPool>();
+ workers->start(FLAGS_num_workers);
+
+ // folly IOThreadPoolExecutor
+ auto ioPool = std::make_shared<folly::IOThreadPoolExecutor>(FLAGS_num_io_threads);
+
  nebula::kvstore::KVOptions options;
- options.local_ = localHost;
  options.dataPaths_ = {FLAGS_data_path};
  options.partMan_ = std::move(partMan);
- std::unique_ptr<nebula::kvstore::KVStore> kvstore(
- nebula::kvstore::KVStore::instance(std::move(options)));
+ std::unique_ptr<nebula::kvstore::KVStore> kvstore =
+ std::make_unique<nebula::kvstore::NebulaStore>(std::move(options),
+ ioPool,
+ workers,
+ localhost);
 
  auto handler = std::make_shared<nebula::meta::MetaServiceHandler>(kvstore.get());
 
- nebula::operator<<(operator<<(LOG(INFO), "The meta deamon start on "), localHost);
+ nebula::operator<<(operator<<(LOG(INFO), "The meta deamon start on "), localhost);
  try {
  gServer = std::make_unique<apache::thrift::ThriftServer>();
  gServer->setInterface(std::move(handler));
  gServer->setPort(FLAGS_port);
  gServer->setReusePort(FLAGS_reuse_port);
  gServer->setIdleTimeout(std::chrono::seconds(0)); // No idle timeout on client connection
+ gServer->setIOThreadPool(ioPool);
  gServer->serve(); // Will wait until the server shuts down
  } catch (const std::exception &e) {
  LOG(ERROR) << "Exception thrown: " << e.what();

src/daemons/StorageDaemon.cpp

@@ -7,9 +7,10 @@
 #include "base/Base.h"
 #include <thrift/lib/cpp2/server/ThriftServer.h>
 #include "network/NetworkUtils.h"
+#include "thread/GenericThreadPool.h"
 #include "storage/StorageServiceHandler.h"
 #include "storage/StorageHttpHandler.h"
-#include "kvstore/KVStore.h"
+#include "kvstore/NebulaStore.h"
 #include "kvstore/PartManager.h"
 #include "process/ProcessUtils.h"
 #include "storage/test/TestUtils.h"
@@ -21,13 +22,16 @@ DEFINE_int32(port, 44500, "Storage daemon listening port");
 DEFINE_bool(reuse_port, true, "Whether to turn on the SO_REUSEPORT option");
 DEFINE_string(data_path, "", "Root data path, multi paths should be split by comma."
  "For rocksdb engine, one path one instance.");
-DEFINE_string(local_ip, "", "Local ip speicified for NetworkUtils::getLocalIP");
 DEFINE_bool(mock_server, true, "start mock server");
 DEFINE_bool(daemonize, true, "Whether to run the process as a daemon");
 DEFINE_string(pid_file, "pids/nebula-storaged.pid", "File to hold the process id");
 DEFINE_string(meta_server_addrs, "", "list of meta server addresses,"
  "the format looks like ip1:port1, ip2:port2, ip3:port3");
-DEFINE_int32(io_handlers, 10, "io handlers");
+DEFINE_string(store_type, "nebula",
+ "Which type of KVStore to be used by the storage daemon."
+ " Options can be \"nebula\", \"hbase\", etc.");
+DEFINE_int32(num_workers, 4, "Number of worker threads");
+DEFINE_int32(num_io_threads, 16, "Number of IO threads");
 
 using nebula::Status;
 using nebula::HostAddr;
@@ -44,6 +48,33 @@ static void signalHandler(int sig);
 static Status setupSignalHandler();
 
 
+std::unique_ptr<nebula::kvstore::KVStore> getStoreInstance(
+ HostAddr localhost,
+ std::vector<std::string> paths,
+ std::shared_ptr<folly::IOThreadPoolExecutor> ioPool,
+ std::shared_ptr<nebula::thread::GenericThreadPool> workers,
+ nebula::meta::MetaClient* metaClient) {
+ nebula::kvstore::KVOptions options;
+ options.dataPaths_ = std::move(paths);
+ options.partMan_ = std::make_unique<nebula::kvstore::MetaServerBasedPartManager>(
+ localhost,
+ metaClient);
+
+ if (FLAGS_store_type == "nebula") {
+ return std::make_unique<nebula::kvstore::NebulaStore>(std::move(options),
+ ioPool,
+ workers,
+ localhost);
+ } else if (FLAGS_store_type == "hbase") {
+ LOG(FATAL) << "HBase store has not been implemented";
+ } else {
+ LOG(FATAL) << "Unknown store type \"" << FLAGS_store_type << "\"";
+ }
+
+ return nullptr;
+}
+
+
 int main(int argc, char *argv[]) {
  google::SetVersionString(nebula::versionString());
  folly::init(&argc, &argv, true);
@@ -81,18 +112,17 @@ int main(int argc, char *argv[]) {
  return EXIT_FAILURE;
  }
 
- auto result = nebula::network::NetworkUtils::getLocalIP(FLAGS_local_ip);
+ auto result = nebula::network::NetworkUtils::getLocalIP();
  if (!result.ok()) {
- LOG(ERROR) << "Get localIp failed, ip " << FLAGS_local_ip
- << ", status:" << result.status();
+ LOG(ERROR) << "Get localIp failed, status:" << result.status();
  return EXIT_FAILURE;
  }
  auto hostRet = nebula::network::NetworkUtils::toHostAddr(result.value(), FLAGS_port);
  if (!hostRet.ok()) {
  LOG(ERROR) << "Bad local host addr, status:" << hostRet.status();
  return EXIT_FAILURE;
  }
- auto& localHost = hostRet.value();
+ auto& localhost = hostRet.value();
  auto metaAddrsRet = nebula::network::NetworkUtils::toHosts(FLAGS_meta_server_addrs);
  if (!metaAddrsRet.ok() || metaAddrsRet.value().empty()) {
  LOG(ERROR) << "Can't get metaServer address, status:" << metaAddrsRet.status()
@@ -110,19 +140,25 @@ int main(int argc, char *argv[]) {
  return EXIT_FAILURE;
  }
 
- auto ioThreadPool = std::make_shared<folly::IOThreadPoolExecutor>(FLAGS_io_handlers);
+ // Generic thread pool
+ auto workers = std::make_shared<nebula::thread::GenericThreadPool>();
+ workers->start(FLAGS_num_workers);
+
+ // folly IOThreadPoolExecutor
+ auto ioThreadPool = std::make_shared<folly::IOThreadPoolExecutor>(FLAGS_num_io_threads);
+
+ // Meta client
  auto metaClient = std::make_unique<nebula::meta::MetaClient>(ioThreadPool,
  std::move(metaAddrsRet.value()),
  true);
  metaClient->init();
 
- nebula::kvstore::KVOptions options;
- options.local_ = localHost;
- options.dataPaths_ = std::move(paths);
- options.partMan_ = std::make_unique<nebula::kvstore::MetaServerBasedPartManager>(
- options.local_, metaClient.get());
- std::unique_ptr<nebula::kvstore::KVStore> kvstore(
- nebula::kvstore::KVStore::instance(std::move(options)));
+ std::unique_ptr<KVStore> kvstore = getStoreInstance(localhost,
+ std::move(paths),
+ ioThreadPool,
+ workers,
+ metaClient.get());
+
  auto schemaMan = nebula::meta::SchemaManager::create();
  schemaMan->init(metaClient.get());
 
@@ -136,16 +172,18 @@ int main(int argc, char *argv[]) {
  LOG(ERROR) << "Failed to start web service: " << status;
  return EXIT_FAILURE;
  }
+
  // Setup the signal handlers
  status = setupSignalHandler();
  if (!status.ok()) {
  LOG(ERROR) << status;
  return EXIT_FAILURE;
  }
 
+ // Thrift server
  auto handler = std::make_shared<StorageServiceHandler>(kvstore.get(), std::move(schemaMan));
  try {
- nebula::operator<<(operator<<(LOG(INFO), "The storage deamon start on "), localHost);
+ nebula::operator<<(operator<<(LOG(INFO), "The storage deamon start on "), localhost);
  gServer = std::make_unique<apache::thrift::ThriftServer>();
  gServer->setInterface(std::move(handler));
  gServer->setPort(FLAGS_port);