merge host info

src/meta/ActiveHostsMan.cpp

@@ -27,7 +27,7 @@ ActiveHostsMan::ActiveHostsMan(int32_t intervalSeconds, int32_t expiredSeconds,
     CHECK(checkThread_.start());
     checkThread_.addTimerTask(intervalSeconds * 1000,
                               intervalSeconds * 1000,
-                              &ActiveHostsMan::cleanExpiredHosts,
+                              &ActiveHostsMan::handleHosts,
                               this);
 }
 
@@ -77,7 +77,12 @@ std::vector<HostAddr> ActiveHostsMan::getActiveHosts() {
     return hosts;
 }
 
-void ActiveHostsMan::loadHostMap() {
+void ActiveHostsMan::handleHosts() {
+    cleanExpiredHosts();
+    mergeHosts();
+}
+
+void ActiveHostsMan::mergeHosts() {
     if (kvstore_ == nullptr) {
         return;
     }
@@ -94,8 +99,11 @@ void ActiveHostsMan::loadHostMap() {
         HostInfo info;
         info.lastHBTimeInSec_ = time::TimeUtils::nowInSeconds();
         if (iter->val() == MetaServiceUtils::hostValOnline()) {
-            LOG(INFO) << "load host " << host.ip << ":" << host.port;
-            updateHostInfo({host.ip, host.port}, info);
+            bool found = hostsMap_.find({host.ip, host.port}) != hostsMap_.end();
+            if (!found) {
+                LOG(INFO) << "add active host " << host.ip << ":" << host.port;
+                updateHostInfo({host.ip, host.port}, info);
+            }
         }
         iter->next();
     }

src/meta/ActiveHostsMan.h

@@ -37,15 +37,14 @@ struct HostInfo {
 
 class ActiveHostsMan final {
     FRIEND_TEST(ActiveHostsManTest, NormalTest);
+    FRIEND_TEST(ActiveHostsMergeTest, NormalTest);
     FRIEND_TEST(ProcessorTest, ListHostsTest);
 
 public:
     static ActiveHostsMan* instance(kvstore::KVStore* kv = nullptr) {
         static auto activeHostsMan = std::unique_ptr<ActiveHostsMan>(
                 new ActiveHostsMan(FLAGS_expired_hosts_check_interval_sec,
                                    FLAGS_expired_threshold_sec, kv));
-        static std::once_flag initFlag;
-        std::call_once(initFlag, &ActiveHostsMan::loadHostMap, activeHostsMan.get());
         return activeHostsMan.get();
     }
 
@@ -70,7 +69,8 @@ class ActiveHostsMan final {
 private:
     ActiveHostsMan(int32_t intervalSeconds, int32_t expiredSeconds, kvstore::KVStore* kv = nullptr);
 
-    void loadHostMap();
+    void handleHosts();
+    void mergeHosts();
 
     void stopClean() {
         checkThread_.stop();

src/meta/test/ActiveHostsMergeTest.cpp

@@ -0,0 +1,54 @@
+
+/* Copyright (c) 2018 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+#include "base/Base.h"
+#include <gtest/gtest.h>
+#include "meta/ActiveHostsMan.h"
+#include "fs/TempDir.h"
+#include "meta/test/TestUtils.h"
+
+namespace nebula {
+namespace meta {
+
+TEST(ActiveHostsMergeTest, NormalTest) {
+    fs::TempDir rootPath("/tmp/ActiveHostsMergeTest.XXXXXX");
+    std::unique_ptr<kvstore::KVStore> kv(TestUtils::initKV(rootPath.path()));
+    auto now = time::TimeUtils::nowInSeconds();
+    ActiveHostsMan ahMan(1, 2, kv.get());
+    ahMan.updateHostInfo(HostAddr(0, 0), HostInfo(now));
+    ahMan.updateHostInfo(HostAddr(0, 1), HostInfo(now));
+    ahMan.updateHostInfo(HostAddr(0, 2), HostInfo(now));
+    ASSERT_EQ(3, ahMan.getActiveHosts().size());
+    sleep(4);
+    ASSERT_EQ(0, ahMan.getActiveHosts().size());
+    {
+        std::vector<kvstore::KV> data;
+        for (auto i = 0; i < 3; i++) {
+            data.emplace_back(MetaServiceUtils::hostKey(i, i),
+                MetaServiceUtils::hostValOnline());
+        }
+        kv->asyncMultiPut(kDefaultSpaceId, kDefaultPartId, std::move(data),
+            [] (kvstore::ResultCode code) {
+            CHECK_EQ(code, kvstore::ResultCode::SUCCEEDED);
+        });
+    }
+    sleep(2);
+    ASSERT_EQ(3, ahMan.getActiveHosts().size());
+}
+
+}  // namespace meta
+}  // namespace nebula
+
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+    folly::init(&argc, &argv, true);
+    google::SetStderrLogging(google::INFO);
+    return RUN_ALL_TESTS();
+}
+
+
+

src/meta/test/CMakeLists.txt

@@ -138,6 +138,36 @@ nebula_link_libraries(
 nebula_add_test(active_hosts_man_test)
 
 
+add_executable(
+    active_hosts_merge_test
+    ActiveHostsMergeTest.cpp
+    $<TARGET_OBJECTS:base_obj>
+    $<TARGET_OBJECTS:thread_obj>
+    $<TARGET_OBJECTS:time_obj>
+    $<TARGET_OBJECTS:fs_obj>
+    $<TARGET_OBJECTS:network_obj>
+    $<TARGET_OBJECTS:meta_service_handler>
+    $<TARGET_OBJECTS:meta_thrift_obj>
+    $<TARGET_OBJECTS:common_thrift_obj>
+    $<TARGET_OBJECTS:kvstore_obj>
+    $<TARGET_OBJECTS:meta_client>
+    $<TARGET_OBJECTS:thrift_obj>
+    $<TARGET_OBJECTS:time_obj>
+    $<TARGET_OBJECTS:schema_obj>
+    $<TARGET_OBJECTS:raftex_obj>
+    $<TARGET_OBJECTS:raftex_thrift_obj>
+    $<TARGET_OBJECTS:wal_obj>
+)
+nebula_link_libraries(
+    active_hosts_merge_test
+    ${ROCKSDB_LIBRARIES}
+    ${THRIFT_LIBRARIES}
+    wangle
+    gtest
+)
+nebula_add_test(active_hosts_merge_test)
+
+
 add_executable(
     meta_http_test
     MetaHttpHandlerTest.cpp