refactor(FQDN): feather refactor on client and host_port.h

Author: Samunroyu

src/common/replication_common.cpp

@@ -180,12 +180,16 @@ int32_t replication_options::app_mutation_2pc_min_replica_count(int32_t app_max_
     rc.learner_signature = invalid_signature;
     SET_OBJ_IP_AND_HOST_PORT(rc, primary, pc, primary);
 
-    if (node == pc.hp_primary) {
+    host_port primary;
+    GET_HOST_PORT(pc, primary, primary);
+    if (node == primary) {
         rc.status = partition_status::PS_PRIMARY;
         return true;
     }
 
-    if (utils::contains(pc.hp_secondaries, node)) {
+    std::vector<host_port> secondaries;
+    GET_HOST_PORTS(pc, secondaries, secondaries);
+    if (utils::contains(secondaries, node)) {
         rc.status = partition_status::PS_SECONDARY;
         return true;
     }

src/common/replication_other_types.h

@@ -36,6 +36,7 @@
 #include "consensus_types.h"
 #include "replica_admin_types.h"
 #include "common/replication_enums.h"
+#include "rpc/dns_resolver.h"
 #include "rpc/rpc_address.h"
 #include "rpc/rpc_host_port.h"
 
@@ -78,18 +79,33 @@ inline bool is_member(const partition_configuration &pc, const rpc_address &node
 inline bool is_partition_config_equal(const partition_configuration &pc1,
                                       const partition_configuration &pc2)
 {
-    // secondaries no need to be same order
-    for (const auto &pc1_secondary : pc1.hp_secondaries) {
+    if (pc1.ballot != pc2.ballot || pc1.pid != pc2.pid ||
+        pc1.max_replica_count != pc2.max_replica_count ||
+        pc1.last_committed_decree != pc2.last_committed_decree) {
+        return false;
+    }
+
+    host_port pc1_primary;
+    GET_HOST_PORT(pc1, primary, pc1_primary);
+    host_port pc2_primary;
+    GET_HOST_PORT(pc2, primary, pc2_primary);
+    if (pc1_primary != pc2_primary) {
+        return false;
+    }
+
+    // secondaries no need to be in the same order.
+    std::vector<host_port> pc1_secondaries;
+    GET_HOST_PORTS(pc1, secondaries, pc1_secondaries);
+    for (const auto &pc1_secondary : pc1_secondaries) {
         if (!is_secondary(pc2, pc1_secondary)) {
             return false;
         }
     }
+
+    std::vector<host_port> pc2_secondaries;
+    GET_HOST_PORTS(pc2, secondaries, pc2_secondaries);
     // last_drops is not considered into equality check
-    return pc1.ballot == pc2.ballot && pc1.pid == pc2.pid &&
-           pc1.max_replica_count == pc2.max_replica_count && pc1.primary == pc2.primary &&
-           pc1.hp_primary == pc2.hp_primary && pc1.secondaries.size() == pc2.secondaries.size() &&
-           pc1.hp_secondaries.size() == pc2.hp_secondaries.size() &&
-           pc1.last_committed_decree == pc2.last_committed_decree;
+    return pc1_secondaries.size() == pc2_secondaries.size();
 }
 
 class replica_helper

src/meta/cluster_balance_policy.cpp

@@ -229,12 +229,17 @@ bool cluster_balance_policy::get_app_migration_info(std::shared_ptr<app_state> a
     info.partitions.reserve(app->pcs.size());
     for (const auto &pc : app->pcs) {
         std::map<host_port, partition_status::type> pstatus_map;
-        pstatus_map[pc.hp_primary] = partition_status::PS_PRIMARY;
-        if (pc.hp_secondaries.size() != pc.max_replica_count - 1) {
+        host_port primary;
+        GET_HOST_PORT(pc, primary, primary);
+        pstatus_map[primary] = partition_status::PS_PRIMARY;
+
+        std::vector<host_port> secondaries;
+        GET_HOST_PORTS(pc, secondaries, secondaries);
+        if (secondaries.size() != pc.max_replica_count - 1) {
             // partition is unhealthy
             return false;
         }
-        for (const auto &secondary : pc.hp_secondaries) {
+        for (const auto &secondary : secondaries) {
             pstatus_map[secondary] = partition_status::PS_SECONDARY;
         }
         info.partitions.push_back(std::move(pstatus_map));

src/meta/load_balance_policy.cpp

@@ -35,6 +35,7 @@
 #include "meta_admin_types.h"
 #include "rpc/dns_resolver.h" // IWYU pragma: keep
 #include "rpc/rpc_address.h"
+#include "rpc/rpc_host_port.h"
 #include "utils/command_manager.h"
 #include "utils/fail_point.h"
 #include "utils/flags.h"
@@ -172,14 +173,17 @@ generate_balancer_request(const app_mapper &apps,
             new_proposal_action(to, to, config_type::CT_UPGRADE_TO_PRIMARY));
         result.action_list.emplace_back(new_proposal_action(to, from, config_type::CT_REMOVE));
         break;
-    case balance_type::COPY_SECONDARY:
+    case balance_type::COPY_SECONDARY: {
         ans = "copy_secondary";
         result.balance_type = balancer_request_type::copy_secondary;
+        host_port primary;
+        GET_HOST_PORT(pc, primary, primary);
         result.action_list.emplace_back(
-            new_proposal_action(pc.hp_primary, to, config_type::CT_ADD_SECONDARY_FOR_LB));
+            new_proposal_action(primary, to, config_type::CT_ADD_SECONDARY_FOR_LB));
         result.action_list.emplace_back(
-            new_proposal_action(pc.hp_primary, from, config_type::CT_REMOVE));
+            new_proposal_action(primary, from, config_type::CT_REMOVE));
         break;
+    }
     default:
         CHECK(false, "");
     }
@@ -566,7 +570,9 @@ void ford_fulkerson::update_decree(int node_id, const node_state &ns)
 {
     ns.for_each_primary(_app->app_id, [&, this](const gpid &pid) {
         const auto &pc = _app->pcs[pid.get_partition_index()];
-        for (const auto &secondary : pc.hp_secondaries) {
+        std::vector<host_port> secondaries;
+        GET_HOST_PORTS(pc, secondaries, secondaries);
+        for (const auto &secondary : secondaries) {
             auto i = _host_port_id.find(secondary);
             CHECK(i != _host_port_id.end(), "invalid secondary: {}", secondary);
             _network[node_id][i->second]++;

src/meta/meta_data.cpp

@@ -538,6 +538,10 @@ app_state::app_state(const app_info &info) : app_info(info), helpers(new app_sta
     CLEAR_IP_AND_HOST_PORT(pc, secondaries);
     CLEAR_IP_AND_HOST_PORT(pc, last_drops);
 
+    // TODO(yujingwei): use marco simplify the code, and the logical may should change
+    pc.__set_hp_secondaries({});
+    pc.__set_hp_last_drops({});
+
     pcs.assign(app_info::partition_count, pc);
     for (int i = 0; i != app_info::partition_count; ++i) {
         pcs[i].pid.set_partition_index(i);

src/rpc/rpc_host_port.h

@@ -132,7 +132,6 @@ class TProtocol;
         auto &_obj = (obj);                                                                        \
         _obj.field.set_invalid();                                                                  \
         _obj.hp_##field.reset();                                                                   \
-        _obj.__isset.hp_##field = false;                                                           \
     } while (0)
 
 // Clear the '<field>' and optional 'hp_<field>' of 'obj'. The types of the fields are std::vector
@@ -141,8 +140,8 @@ class TProtocol;
     do {                                                                                           \
         auto &_obj = (obj);                                                                        \
         _obj.field.clear();                                                                        \
-        _obj.hp_##field.clear();                                                                   \
-        _obj.__isset.hp_##field = false;                                                           \
+	_obj.hp_##field.clear();                                                                   \
+	_obj.__isset.hp_##field = false;                                                           \
     } while (0)
 
 // Add 'addr' and 'hp' to the vector '<field>' and optional vector 'hp_<field>' of 'obj'. The types