refactor(FQDN): feather refactor on meta server side

Author: Samunroyu

src/client/partition_resolver_simple.cpp

@@ -413,14 +413,18 @@ void partition_resolver_simple::handle_pending_requests(std::deque<request_conte
 host_port partition_resolver_simple::get_host_port(const partition_configuration &pc) const
 {
     if (_app_is_stateful) {
-        return pc.hp_primary;
+        host_port primary;
+        GET_HOST_PORT(pc, primary, primary);
+        return primary;
     }
 
-    if (pc.hp_last_drops.empty()) {
+    std::vector<host_port> last_drops;
+    GET_HOST_PORTS(pc, last_drops, last_drops);
+    if (last_drops.empty()) {
         return host_port();
     }
 
-    return pc.hp_last_drops[rand::next_u32(0, pc.last_drops.size() - 1)];
+    return last_drops[rand::next_u32(0, last_drops.size() - 1)];
 }
 
 error_code partition_resolver_simple::get_host_port(int partition_index, /*out*/ host_port &hp)

src/client/replication_ddl_client.cpp

@@ -168,9 +168,19 @@ dsn::error_code replication_ddl_client::wait_app_ready(const std::string &app_na
         int ready_count = 0;
         for (int i = 0; i < partition_count; i++) {
             const auto &pc = query_resp.partitions[i];
-            if (pc.hp_primary && (pc.hp_secondaries.size() + 1 >= max_replica_count)) {
-                ready_count++;
+            host_port primary;
+            GET_HOST_PORT(pc, primary, primary);
+            if (!primary) {
+                continue;
+            }
+
+            std::vector<host_port> secondaries;
+            GET_HOST_PORTS(pc, secondaries, secondaries);
+            if (secondaries.size() + 1 < max_replica_count) {
+                continue;
             }
+
+            ready_count++;
         }
         if (ready_count == partition_count) {
             std::cout << app_name << " is ready now: (" << ready_count << "/" << partition_count
@@ -429,11 +439,16 @@ error_s replication_ddl_client::list_apps(bool detailed,
             int read_unhealthy = 0;
             for (const auto &pc : pcs) {
                 int replica_count = 0;
-                if (pc.hp_primary) {
+                host_port primary;
+                GET_HOST_PORT(pc, primary, primary);
+                if (primary) {
                     replica_count++;
                 }
-                replica_count += pc.hp_secondaries.size();
-                if (pc.hp_primary) {
+
+                std::vector<host_port> secondaries;
+                GET_HOST_PORTS(pc, secondaries, secondaries);
+                replica_count += static_cast<int>(secondaries.size());
+                if (primary) {
                     if (replica_count >= pc.max_replica_count) {
                         fully_healthy++;
                     } else if (replica_count < 2) {

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;
     }
@@ -383,15 +387,19 @@ void add_app_info(const std::string &app_name,
     int read_unhealthy = 0;
     for (const auto &pc : pcs) {
         int replica_count = 0;
-        if (pc.hp_primary) {
+        host_port primary;
+        GET_HOST_PORT(pc, primary, primary);
+        if (primary) {
             ++replica_count;
-            ++node_stats[pc.hp_primary].primary_count;
+            ++node_stats[primary].primary_count;
             ++total_prim_count;
         }
-        replica_count += static_cast<int>(pc.hp_secondaries.size());
-        total_sec_count += static_cast<int>(pc.hp_secondaries.size());
+        std::vector<host_port> secondaries;
+        GET_HOST_PORTS(pc, secondaries, secondaries);
+        replica_count += static_cast<int>(secondaries.size());
+        total_sec_count += static_cast<int>(secondaries.size());
 
-        if (pc.hp_primary) {
+        if (primary) {
             if (replica_count >= pc.max_replica_count) {
                 ++fully_healthy;
             } else if (replica_count < 2) {
@@ -405,10 +413,10 @@ void add_app_info(const std::string &app_name,
         partitions_printer.add_row(pc.pid.get_partition_index());
         partitions_printer.append_data(pc.ballot);
         partitions_printer.append_data(fmt::format("{}/{}", replica_count, pc.max_replica_count));
-        partitions_printer.append_data(pc.hp_primary ? pc.hp_primary.to_string() : "-");
-        partitions_printer.append_data(fmt::format("[{}]", fmt::join(pc.hp_secondaries, ",")));
+        partitions_printer.append_data(primary ? primary.to_string() : "-");
+        partitions_printer.append_data(fmt::format("[{}]", fmt::join(secondaries, ",")));
 
-        for (const auto &secondary : pc.hp_secondaries) {
+        for (const auto &secondary : secondaries) {
             ++node_stats[secondary].secondary_count;
         }
     }

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,16 @@ 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));
-        result.action_list.emplace_back(
-            new_proposal_action(pc.hp_primary, from, config_type::CT_REMOVE));
+            new_proposal_action(primary, to, config_type::CT_ADD_SECONDARY_FOR_LB));
+        result.action_list.emplace_back(new_proposal_action(primary, from, config_type::CT_REMOVE));
         break;
+    }
     default:
         CHECK(false, "");
     }
@@ -566,7 +569,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

@@ -24,8 +24,8 @@
  * THE SOFTWARE.
  */
 
-#include <boost/lexical_cast.hpp>
 #include <algorithm>
+#include <boost/lexical_cast.hpp>
 #include <cstdint>
 #include <ostream>
 
@@ -134,13 +134,16 @@ bool construct_replica(meta_view view, const gpid &pid, int max_replica_count)
     // we put max_replica_count-1 recent replicas to last_drops, in case of the DDD-state when the
     // only primary dead
     // when add node to pc.last_drops, we don't remove it from our cc.drop_list
-    CHECK(pc.hp_last_drops.empty(), "last_drops of partition({}) must be empty", pid);
+    std::vector<host_port> last_drops;
+    GET_HOST_PORTS(pc, last_drops, last_drops);
+    CHECK(last_drops.empty(), "last_drops of partition({}) must be empty", pid);
     for (auto iter = drop_list.rbegin(); iter != drop_list.rend(); ++iter) {
-        if (pc.hp_last_drops.size() + 1 >= max_replica_count) {
+        // hp_last_drops is added in the steps bellow.
+        if (last_drops.size() + 1 >= max_replica_count) {
             break;
         }
         // similar to cc.drop_list, pc.last_drop is also a stack structure
-        HEAD_INSERT_IP_AND_HOST_PORT_BY_DNS(pc, last_drops, iter->node);
+        HEAD_INSERT_IP_AND_HOST_PORT_BY_DNS(pc, last_drops, iter->node, last_drops);
         LOG_INFO("construct for ({}), select {} into last_drops, ballot({}), "
                  "committed_decree({}), prepare_decree({})",
                  pid,
@@ -538,6 +541,11 @@ 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/meta/meta_data.h

@@ -26,9 +26,9 @@
 
 #pragma once
 
-#include <stdint.h>
 #include <algorithm>
 #include <atomic>
+#include <cstdint>
 #include <functional>
 #include <map>
 #include <memory>
@@ -265,12 +265,31 @@ class config_context
 struct partition_configuration_stateless
 {
     partition_configuration &pc;
-    partition_configuration_stateless(partition_configuration &_pc) : pc(_pc) {}
-    std::vector<dsn::host_port> &workers() { return pc.hp_last_drops; }
-    std::vector<dsn::host_port> &hosts() { return pc.hp_secondaries; }
-    bool is_host(const host_port &node) const { return utils::contains(pc.hp_secondaries, node); }
-    bool is_worker(const host_port &node) const { return utils::contains(pc.hp_last_drops, node); }
-    bool is_member(const host_port &node) const { return is_host(node) || is_worker(node); }
+    explicit partition_configuration_stateless(partition_configuration &_pc) : pc(_pc) {}
+    std::vector<dsn::host_port> &workers()
+    {
+        DCHECK(pc.__isset.hp_last_drops, "");
+        return pc.hp_last_drops;
+    }
+    std::vector<dsn::host_port> &hosts()
+    {
+        DCHECK(pc.__isset.hp_secondaries, "");
+        return pc.hp_secondaries;
+    }
+    [[nodiscard]] bool is_host(const host_port &node) const
+    {
+        DCHECK(pc.__isset.hp_secondaries, "");
+        return utils::contains(pc.hp_secondaries, node);
+    }
+    [[nodiscard]] bool is_worker(const host_port &node) const
+    {
+        DCHECK(pc.__isset.hp_last_drops, "");
+        return utils::contains(pc.hp_last_drops, node);
+    }
+    [[nodiscard]] bool is_member(const host_port &node) const
+    {
+        return is_host(node) || is_worker(node);
+    }
 };
 
 struct restore_state
@@ -294,30 +313,25 @@ struct restore_state
 // in `status`.
 struct split_state
 {
-    int32_t splitting_count;
+    int32_t splitting_count{0};
     // partition_index -> split_status
     std::map<int32_t, split_status::type> status;
-    split_state() : splitting_count(0) {}
+    split_state() = default;
 };
 
 class app_state;
 
 class app_state_helper
 {
 public:
-    app_state *owner;
-    std::atomic_int partitions_in_progress;
+    app_state *owner{nullptr};
+    std::atomic_int partitions_in_progress{0};
     std::vector<config_context> contexts;
-    dsn::message_ex *pending_response;
+    dsn::message_ex *pending_response{nullptr};
     std::vector<restore_state> restore_states;
     split_state split_states;
 
-public:
-    app_state_helper() : owner(nullptr), partitions_in_progress(0)
-    {
-        contexts.clear();
-        pending_response = nullptr;
-    }
+    app_state_helper() = default;
     void on_init_partitions();
     void clear_proposals()
     {

src/meta/meta_http_service.cpp

@@ -15,12 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <fmt/core.h>
-#include <rapidjson/ostreamwrapper.h>
 #include <algorithm>
 #include <cstddef>
+#include <fmt/core.h>
 #include <map>
 #include <memory>
+#include <rapidjson/ostreamwrapper.h>
 #include <set>
 #include <sstream>
 #include <string>
@@ -251,11 +251,15 @@ void meta_http_service::list_app_handler(const http_request &req, http_response
             int read_unhealthy = 0;
             for (const auto &pc : response.partitions) {
                 int replica_count = 0;
-                if (pc.hp_primary) {
+                host_port primary;
+                GET_HOST_PORT(pc, primary, primary);
+                if (primary) {
                     replica_count++;
                 }
-                replica_count += pc.hp_secondaries.size();
-                if (pc.hp_primary) {
+                std::vector<host_port> secondaries;
+                GET_HOST_PORTS(pc, secondaries, secondaries);
+                replica_count += static_cast<int>(secondaries.size());
+                if (primary) {
                     if (replica_count >= pc.max_replica_count) {
                         fully_healthy++;
                     } else if (replica_count < 2) {