You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pegasus.apache.org by la...@apache.org on 2024/04/17 02:19:32 UTC
(incubator-pegasus) branch master updated: refactor(FQDN): Add some convenient macros and update idl/bulk_load.thrift related code (#1949)
This is an automated email from the ASF dual-hosted git repository.
laiyingchun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
The following commit(s) were added to refs/heads/master by this push:
new 09e3e44a4 refactor(FQDN): Add some convenient macros and update idl/bulk_load.thrift related code (#1949)
09e3e44a4 is described below
commit 09e3e44a4a97754416cded3e67bd3846144498a5
Author: Yingchun Lai <la...@apache.org>
AuthorDate: Wed Apr 17 10:19:27 2024 +0800
refactor(FQDN): Add some convenient macros and update idl/bulk_load.thrift related code (#1949)
- Add a new macros SET_IP_AND_HOST_PORT, SET_VALUE_FROM_IP_AND_HOST_PORT,
SET_VALUE_FROM_HOST_PORT and FMT_HOST_PORT_AND_IP
- Add fill_partition_bulk_load_states() function to simplify code in
src/meta/test/meta_bulk_load_service_test.cpp
---
idl/bulk_load.thrift | 2 +-
src/meta/meta_bulk_load_service.cpp | 70 ++++++++++-----------
src/meta/test/meta_bulk_load_service_test.cpp | 63 +++++++++----------
src/replica/bulk_load/replica_bulk_loader.cpp | 73 ++++++++++------------
.../bulk_load/test/replica_bulk_loader_test.cpp | 3 +-
src/runtime/rpc/rpc_host_port.h | 34 ++++++++++
src/shell/commands/bulk_load.cpp | 3 +
7 files changed, 135 insertions(+), 113 deletions(-)
diff --git a/idl/bulk_load.thrift b/idl/bulk_load.thrift
index cb2e77c4b..92c9cb404 100644
--- a/idl/bulk_load.thrift
+++ b/idl/bulk_load.thrift
@@ -91,7 +91,7 @@ struct bulk_load_request
{
1:dsn.gpid pid;
2:string app_name;
- 3:dsn.rpc_address primary_addr;
+ 3:dsn.rpc_address primary;
4:string remote_provider_name;
5:string cluster_name;
6:i64 ballot;
diff --git a/src/meta/meta_bulk_load_service.cpp b/src/meta/meta_bulk_load_service.cpp
index 26f51c5a0..8b68993b5 100644
--- a/src/meta/meta_bulk_load_service.cpp
+++ b/src/meta/meta_bulk_load_service.cpp
@@ -24,7 +24,9 @@
#include <algorithm>
#include <chrono>
#include <cstdint>
+#include <type_traits>
+#include "absl/strings/string_view.h"
#include "block_service/block_service.h"
#include "block_service/block_service_manager.h"
#include "common/replica_envs.h"
@@ -51,7 +53,6 @@
#include "utils/fail_point.h"
#include "utils/fmt_logging.h"
#include "utils/string_conv.h"
-#include "absl/strings/string_view.h"
DSN_DEFINE_uint32(meta_server,
bulk_load_max_rollback_times,
@@ -430,8 +431,7 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
const app_bulk_load_info &ainfo = _app_bulk_load_info[pid.get_app_id()];
req->pid = pid;
req->app_name = app_name;
- req->primary_addr = pconfig.primary;
- req->__set_hp_primary(pconfig.hp_primary);
+ SET_IP_AND_HOST_PORT(*req, primary, pconfig.primary, pconfig.hp_primary);
req->remote_provider_name = ainfo.file_provider_type;
req->cluster_name = ainfo.cluster_name;
req->meta_bulk_load_status = get_partition_bulk_load_status_unlocked(pid);
@@ -440,10 +440,9 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
req->remote_root_path = ainfo.remote_root_path;
}
- LOG_INFO("send bulk load request to node({}({})), app({}), partition({}), partition "
+ LOG_INFO("send bulk load request to node({}), app({}), partition({}), partition "
"status = {}, remote provider = {}, cluster_name = {}, remote_root_path = {}",
- pconfig.hp_primary,
- pconfig.primary,
+ FMT_HOST_PORT_AND_IP(pconfig, primary),
app_name,
pid,
dsn::enum_to_string(req->meta_bulk_load_status),
@@ -458,9 +457,8 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
auto &bulk_load_resp = rpc.response();
if (!bulk_load_resp.__isset.hp_group_bulk_load_state) {
bulk_load_resp.__set_hp_group_bulk_load_state({});
- for (const auto &kv : bulk_load_resp.group_bulk_load_state) {
- bulk_load_resp.hp_group_bulk_load_state[host_port::from_address(kv.first)] =
- kv.second;
+ for (const auto & [ addr, pbls ] : bulk_load_resp.group_bulk_load_state) {
+ bulk_load_resp.hp_group_bulk_load_state[host_port::from_address(addr)] = pbls;
}
}
@@ -475,16 +473,15 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
{
const std::string &app_name = request.app_name;
const gpid &pid = request.pid;
- const auto &primary_addr = request.primary_addr;
+ const auto &primary_addr = request.primary;
const auto &primary_hp = request.hp_primary;
if (err != ERR_OK) {
- LOG_ERROR("app({}), partition({}) failed to receive bulk load response from node({}({})), "
+ LOG_ERROR("app({}), partition({}) failed to receive bulk load response from node({}), "
"error = {}",
app_name,
pid,
- primary_hp,
- primary_addr,
+ FMT_HOST_PORT_AND_IP(request, primary),
err);
try_rollback_to_downloading(app_name, pid);
try_resend_bulk_load_request(app_name, pid);
@@ -493,11 +490,10 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
if (response.err == ERR_OBJECT_NOT_FOUND || response.err == ERR_INVALID_STATE) {
LOG_ERROR(
- "app({}), partition({}) doesn't exist or has invalid state on node({}({})), error = {}",
+ "app({}), partition({}) doesn't exist or has invalid state on node({}), error = {}",
app_name,
pid,
- primary_hp,
- primary_addr,
+ FMT_HOST_PORT_AND_IP(request, primary),
response.err);
try_rollback_to_downloading(app_name, pid);
try_resend_bulk_load_request(app_name, pid);
@@ -506,10 +502,9 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
if (response.err == ERR_BUSY) {
LOG_WARNING(
- "node({}({})) has enough replicas downloading, wait for next round to send bulk load "
+ "node({}) has enough replicas downloading, wait for next round to send bulk load "
"request for app({}), partition({})",
- primary_hp,
- primary_addr,
+ FMT_HOST_PORT_AND_IP(request, primary),
app_name,
pid);
try_resend_bulk_load_request(app_name, pid);
@@ -517,15 +512,13 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
}
if (response.err != ERR_OK) {
- LOG_ERROR(
- "app({}), partition({}) from node({}({})) handle bulk load response failed, error = "
- "{}, primary status = {}",
- app_name,
- pid,
- primary_hp,
- primary_addr,
- response.err,
- dsn::enum_to_string(response.primary_bulk_load_status));
+ LOG_ERROR("app({}), partition({}) from node({}) handle bulk load response failed, error = "
+ "{}, primary status = {}",
+ app_name,
+ pid,
+ FMT_HOST_PORT_AND_IP(request, primary),
+ response.err,
+ dsn::enum_to_string(response.primary_bulk_load_status));
handle_bulk_load_failed(pid.get_app_id(), response.err);
try_resend_bulk_load_request(app_name, pid);
return;
@@ -544,7 +537,7 @@ void bulk_load_service::on_partition_bulk_load_reply(error_code err,
LOG_WARNING(
"receive out-date response from node({}), app({}), partition({}), request ballot = "
"{}, current ballot= {}",
- primary_addr,
+ FMT_HOST_PORT_AND_IP(request, primary),
app_name,
pid,
request.ballot,
@@ -1614,19 +1607,20 @@ void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
}
}
+ // Fill bulk_load_states and hp_bulk_load_states fileds.
response.bulk_load_states.resize(partition_count);
response.__set_hp_bulk_load_states(
std::vector<std::map<host_port, partition_bulk_load_state>>(partition_count));
- for (const auto &kv : _partitions_bulk_load_state) {
- if (kv.first.get_app_id() == app_id) {
- auto pidx = kv.first.get_partition_index();
- response.hp_bulk_load_states[pidx] = kv.second;
-
- std::map<rpc_address, partition_bulk_load_state> addr_pbls;
- for (const auto &bls : kv.second) {
- addr_pbls[dsn::dns_resolver::instance().resolve_address(bls.first)] = bls.second;
+ for (const auto & [ pid, pbls_by_hps ] : _partitions_bulk_load_state) {
+ if (pid.get_app_id() == app_id) {
+ auto pidx = pid.get_partition_index();
+ response.hp_bulk_load_states[pidx] = pbls_by_hps;
+
+ std::map<rpc_address, partition_bulk_load_state> pbls_by_addrs;
+ for (const auto & [ hp, pbls ] : pbls_by_hps) {
+ pbls_by_addrs[dsn::dns_resolver::instance().resolve_address(hp)] = pbls;
}
- response.bulk_load_states[pidx] = addr_pbls;
+ response.bulk_load_states[pidx] = pbls_by_addrs;
}
}
diff --git a/src/meta/test/meta_bulk_load_service_test.cpp b/src/meta/test/meta_bulk_load_service_test.cpp
index dedb28d73..2126f74c0 100644
--- a/src/meta/test/meta_bulk_load_service_test.cpp
+++ b/src/meta/test/meta_bulk_load_service_test.cpp
@@ -17,6 +17,7 @@
#include <boost/lexical_cast.hpp>
#include <string.h>
+#include <type_traits>
#include <algorithm>
#include <atomic>
#include <cstdint>
@@ -792,9 +793,8 @@ public:
_req.ballot = BALLOT;
_req.cluster_name = CLUSTER;
_req.pid = gpid(_app_id, _pidx);
- _req.primary_addr = PRIMARY;
+ SET_IP_AND_HOST_PORT(_req, primary, PRIMARY, PRIMARY_HP);
_req.meta_bulk_load_status = status;
- _req.__set_hp_primary(PRIMARY_HP);
}
void create_basic_response(error_code err, bulk_load_status::type status)
@@ -805,6 +805,19 @@ public:
_resp.primary_bulk_load_status = status;
}
+ void fill_partition_bulk_load_states(
+ const std::map<std::pair<::dsn::rpc_address, ::dsn::host_port>, partition_bulk_load_state>
+ &state_by_hosts)
+ {
+ if (!_resp.__isset.hp_group_bulk_load_state) {
+ _resp.__set_hp_group_bulk_load_state({});
+ }
+ for (const auto & [ addr_and_hp, state ] : state_by_hosts) {
+ _resp.group_bulk_load_state[addr_and_hp.first] = state;
+ _resp.hp_group_bulk_load_state[addr_and_hp.second] = state;
+ }
+ }
+
void mock_response_progress(error_code progress_err, bool finish_download)
{
create_basic_response(ERR_OK, bulk_load_status::BLS_DOWNLOADING);
@@ -817,15 +830,11 @@ public:
state2.__set_download_status(progress_err);
state2.__set_download_progress(secondary2_progress);
- _resp.group_bulk_load_state[PRIMARY] = state;
- _resp.group_bulk_load_state[SECONDARY1] = state;
- _resp.group_bulk_load_state[SECONDARY2] = state2;
- _resp.__set_total_download_progress(total_progress);
+ fill_partition_bulk_load_states({{{PRIMARY, PRIMARY_HP}, state},
+ {{SECONDARY1, SECONDARY1_HP}, state},
+ {{SECONDARY2, SECONDARY2_HP}, state2}});
- _resp.__set_hp_group_bulk_load_state({});
- _resp.hp_group_bulk_load_state[PRIMARY_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY1_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY2_HP] = state2;
+ _resp.__set_total_download_progress(total_progress);
}
void mock_response_bulk_load_metadata()
@@ -853,15 +862,11 @@ public:
state.__set_ingest_status(ingestion_status::IS_SUCCEED);
state2.__set_ingest_status(secondary_istatus);
- _resp.group_bulk_load_state[PRIMARY] = state;
- _resp.group_bulk_load_state[SECONDARY1] = state;
- _resp.group_bulk_load_state[SECONDARY2] = state2;
- _resp.__set_is_group_ingestion_finished(secondary_istatus == ingestion_status::IS_SUCCEED);
+ fill_partition_bulk_load_states({{{PRIMARY, PRIMARY_HP}, state},
+ {{SECONDARY1, SECONDARY1_HP}, state},
+ {{SECONDARY2, SECONDARY2_HP}, state2}});
- _resp.__set_hp_group_bulk_load_state({});
- _resp.hp_group_bulk_load_state[PRIMARY_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY1_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY2_HP] = state2;
+ _resp.__set_is_group_ingestion_finished(secondary_istatus == ingestion_status::IS_SUCCEED);
set_app_ingesting_count(_app_id, ingestion_count);
}
@@ -872,16 +877,13 @@ public:
partition_bulk_load_state state, state2;
state.__set_is_cleaned_up(true);
- _resp.group_bulk_load_state[PRIMARY] = state;
- _resp.group_bulk_load_state[SECONDARY1] = state;
- _resp.__set_hp_group_bulk_load_state({});
- _resp.hp_group_bulk_load_state[PRIMARY_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY1_HP] = state;
+ fill_partition_bulk_load_states(
+ {{{PRIMARY, PRIMARY_HP}, state}, {{SECONDARY1, SECONDARY1_HP}, state}});
state2.__set_is_cleaned_up(all_cleaned_up);
- _resp.group_bulk_load_state[SECONDARY2] = state2;
- _resp.hp_group_bulk_load_state[SECONDARY2_HP] = state2;
+ fill_partition_bulk_load_states({{{SECONDARY2, SECONDARY2_HP}, state2}});
+
_resp.__set_is_group_bulk_load_context_cleaned_up(all_cleaned_up);
}
@@ -893,14 +895,9 @@ public:
state.__set_is_paused(true);
state2.__set_is_paused(is_group_paused);
- _resp.group_bulk_load_state[PRIMARY] = state;
- _resp.group_bulk_load_state[SECONDARY1] = state;
- _resp.group_bulk_load_state[SECONDARY2] = state2;
-
- _resp.__set_hp_group_bulk_load_state({});
- _resp.hp_group_bulk_load_state[PRIMARY_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY1_HP] = state;
- _resp.hp_group_bulk_load_state[SECONDARY2_HP] = state2;
+ fill_partition_bulk_load_states({{{PRIMARY, PRIMARY_HP}, state},
+ {{SECONDARY1, SECONDARY1_HP}, state},
+ {{SECONDARY2, SECONDARY2_HP}, state2}});
_resp.__set_is_group_bulk_load_paused(is_group_paused);
}
diff --git a/src/replica/bulk_load/replica_bulk_loader.cpp b/src/replica/bulk_load/replica_bulk_loader.cpp
index 9ae6f0598..fb583ccb2 100644
--- a/src/replica/bulk_load/replica_bulk_loader.cpp
+++ b/src/replica/bulk_load/replica_bulk_loader.cpp
@@ -195,8 +195,7 @@ void replica_bulk_loader::broadcast_group_bulk_load(const bulk_load_request &met
auto request = std::make_unique<group_bulk_load_request>();
request->app_name = _replica->_app_info.app_name;
const auto &addr = dsn::dns_resolver::instance().resolve_address(hp);
- request->target = addr;
- request->__set_hp_target(hp);
+ SET_IP_AND_HOST_PORT(*request, target, addr, hp);
_replica->_primary_states.get_replica_config(partition_status::PS_SECONDARY,
request->config);
request->cluster_name = meta_req.cluster_name;
@@ -286,31 +285,27 @@ void replica_bulk_loader::on_group_bulk_load_reply(error_code err,
_replica->_primary_states.group_bulk_load_pending_replies.erase(req.hp_target);
if (err != ERR_OK) {
- LOG_ERROR_PREFIX("failed to receive group_bulk_load_reply from {}({}), error = {}",
- req.hp_target,
- req.target,
+ LOG_ERROR_PREFIX("failed to receive group_bulk_load_reply from {}, error = {}",
+ FMT_HOST_PORT_AND_IP(req, target),
err);
_replica->_primary_states.reset_node_bulk_load_states(req.hp_target);
return;
}
if (resp.err != ERR_OK) {
- LOG_ERROR_PREFIX("receive group_bulk_load response from {}({}) failed, error = {}",
- req.hp_target,
- req.target,
+ LOG_ERROR_PREFIX("receive group_bulk_load response from {} failed, error = {}",
+ FMT_HOST_PORT_AND_IP(req, target),
resp.err);
_replica->_primary_states.reset_node_bulk_load_states(req.hp_target);
return;
}
if (req.config.ballot != get_ballot()) {
- LOG_ERROR_PREFIX(
- "recevied wrong group_bulk_load response from {}({}), request ballot = {}, "
- "current ballot = {}",
- req.hp_target,
- req.target,
- req.config.ballot,
- get_ballot());
+ LOG_ERROR_PREFIX("recevied wrong group_bulk_load response from {}, request ballot = {}, "
+ "current ballot = {}",
+ FMT_HOST_PORT_AND_IP(req, target),
+ req.config.ballot,
+ get_ballot());
_replica->_primary_states.reset_node_bulk_load_states(req.hp_target);
return;
}
@@ -933,9 +928,11 @@ void replica_bulk_loader::report_group_download_progress(/*out*/ bulk_load_respo
primary_state.__set_download_progress(_download_progress.load());
primary_state.__set_download_status(_download_status.load());
}
- response.group_bulk_load_state[_replica->_primary_states.membership.primary] = primary_state;
- response.hp_group_bulk_load_state[_replica->_primary_states.membership.hp_primary] =
- primary_state;
+ SET_VALUE_FROM_IP_AND_HOST_PORT(response,
+ group_bulk_load_state,
+ _replica->_primary_states.membership.primary,
+ _replica->_primary_states.membership.hp_primary,
+ primary_state);
LOG_INFO_PREFIX("primary = {}({}), download progress = {}%, status = {}",
_replica->_primary_states.membership.hp_primary,
_replica->_primary_states.membership.primary,
@@ -952,9 +949,7 @@ void replica_bulk_loader::report_group_download_progress(/*out*/ bulk_load_respo
secondary_state.__isset.download_status ? secondary_state.download_status : ERR_OK;
LOG_INFO_PREFIX(
"secondary = {}, download progress = {}%, status={}", target_hp, s_progress, s_status);
- response.group_bulk_load_state[dsn::dns_resolver::instance().resolve_address(target_hp)] =
- secondary_state;
- response.hp_group_bulk_load_state[target_hp] = secondary_state;
+ SET_VALUE_FROM_HOST_PORT(response, group_bulk_load_state, target_hp, secondary_state);
total_progress += s_progress;
}
@@ -976,9 +971,11 @@ void replica_bulk_loader::report_group_ingestion_status(/*out*/ bulk_load_respon
partition_bulk_load_state primary_state;
primary_state.__set_ingest_status(_replica->_app->get_ingestion_status());
- response.group_bulk_load_state[_replica->_primary_states.membership.primary] = primary_state;
- response.hp_group_bulk_load_state[_replica->_primary_states.membership.hp_primary] =
- primary_state;
+ SET_VALUE_FROM_IP_AND_HOST_PORT(response,
+ group_bulk_load_state,
+ _replica->_primary_states.membership.primary,
+ _replica->_primary_states.membership.hp_primary,
+ primary_state);
LOG_INFO_PREFIX("primary = {}({}), ingestion status = {}",
_replica->_primary_states.membership.hp_primary,
_replica->_primary_states.membership.primary,
@@ -996,9 +993,7 @@ void replica_bulk_loader::report_group_ingestion_status(/*out*/ bulk_load_respon
: ingestion_status::IS_INVALID;
LOG_INFO_PREFIX(
"secondary = {}, ingestion status={}", target_hp, enum_to_string(ingest_status));
- response.group_bulk_load_state[dsn::dns_resolver::instance().resolve_address(target_hp)] =
- secondary_state;
- response.hp_group_bulk_load_state[target_hp] = secondary_state;
+ SET_VALUE_FROM_HOST_PORT(response, group_bulk_load_state, target_hp, secondary_state);
is_group_ingestion_finish &= (ingest_status == ingestion_status::IS_SUCCEED);
}
response.__set_is_group_ingestion_finished(is_group_ingestion_finish);
@@ -1024,9 +1019,11 @@ void replica_bulk_loader::report_group_cleaned_up(bulk_load_response &response)
partition_bulk_load_state primary_state;
primary_state.__set_is_cleaned_up(is_cleaned_up());
- response.group_bulk_load_state[_replica->_primary_states.membership.primary] = primary_state;
- response.hp_group_bulk_load_state[_replica->_primary_states.membership.hp_primary] =
- primary_state;
+ SET_VALUE_FROM_IP_AND_HOST_PORT(response,
+ group_bulk_load_state,
+ _replica->_primary_states.membership.primary,
+ _replica->_primary_states.membership.hp_primary,
+ primary_state);
LOG_INFO_PREFIX("primary = {}, bulk load states cleaned_up = {}",
_replica->_primary_states.membership.primary,
primary_state.is_cleaned_up);
@@ -1041,9 +1038,7 @@ void replica_bulk_loader::report_group_cleaned_up(bulk_load_response &response)
secondary_state.__isset.is_cleaned_up ? secondary_state.is_cleaned_up : false;
LOG_INFO_PREFIX(
"secondary = {}, bulk load states cleaned_up = {}", target_hp, is_cleaned_up);
- response.group_bulk_load_state[dsn::dns_resolver::instance().resolve_address(target_hp)] =
- secondary_state;
- response.hp_group_bulk_load_state[target_hp] = secondary_state;
+ SET_VALUE_FROM_HOST_PORT(response, group_bulk_load_state, target_hp, secondary_state);
group_flag &= is_cleaned_up;
}
LOG_INFO_PREFIX("group bulk load states cleaned_up = {}", group_flag);
@@ -1063,9 +1058,11 @@ void replica_bulk_loader::report_group_is_paused(bulk_load_response &response)
partition_bulk_load_state primary_state;
primary_state.__set_is_paused(_status == bulk_load_status::BLS_PAUSED);
- response.group_bulk_load_state[_replica->_primary_states.membership.primary] = primary_state;
- response.hp_group_bulk_load_state[_replica->_primary_states.membership.hp_primary] =
- primary_state;
+ SET_VALUE_FROM_IP_AND_HOST_PORT(response,
+ group_bulk_load_state,
+ _replica->_primary_states.membership.primary,
+ _replica->_primary_states.membership.hp_primary,
+ primary_state);
LOG_INFO_PREFIX("primary = {}({}), bulk_load is_paused = {}",
_replica->_primary_states.membership.hp_primary,
_replica->_primary_states.membership.primary,
@@ -1079,9 +1076,7 @@ void replica_bulk_loader::report_group_is_paused(bulk_load_response &response)
_replica->_primary_states.secondary_bulk_load_states[target_hp];
bool is_paused = secondary_state.__isset.is_paused ? secondary_state.is_paused : false;
LOG_INFO_PREFIX("secondary = {}, bulk_load is_paused = {}", target_hp, is_paused);
- response.group_bulk_load_state[dsn::dns_resolver::instance().resolve_address(target_hp)] =
- secondary_state;
- response.hp_group_bulk_load_state[target_hp] = secondary_state;
+ SET_VALUE_FROM_HOST_PORT(response, group_bulk_load_state, target_hp, secondary_state);
group_is_paused &= is_paused;
}
LOG_INFO_PREFIX("group bulk load is_paused = {}", group_is_paused);
diff --git a/src/replica/bulk_load/test/replica_bulk_loader_test.cpp b/src/replica/bulk_load/test/replica_bulk_loader_test.cpp
index 1375549bf..95dbad2cb 100644
--- a/src/replica/bulk_load/test/replica_bulk_loader_test.cpp
+++ b/src/replica/bulk_load/test/replica_bulk_loader_test.cpp
@@ -220,8 +220,7 @@ public:
_group_req.meta_bulk_load_status = status;
_group_req.config.status = partition_status::PS_SECONDARY;
_group_req.config.ballot = b;
- _group_req.target = SECONDARY;
- _group_req.__set_hp_target(SECONDARY_HP);
+ SET_IP_AND_HOST_PORT(_group_req, target, PRIMARY, PRIMARY_HP);
}
void mock_replica_config(partition_status::type status)
diff --git a/src/runtime/rpc/rpc_host_port.h b/src/runtime/rpc/rpc_host_port.h
index cc1cd70e0..00ca8dd1c 100644
--- a/src/runtime/rpc/rpc_host_port.h
+++ b/src/runtime/rpc/rpc_host_port.h
@@ -44,6 +44,8 @@ class TProtocol;
} // namespace thrift
} // namespace apache
+// Get host_port from 'obj', the result is filled in 'target', the source is from host_port type
+// field 'hp_<field>' if it is set, otherwise, reverse resolve from the rpc_address '<field>'.
#define GET_HOST_PORT(obj, field, target) \
do { \
const auto &_obj = (obj); \
@@ -54,6 +56,38 @@ class TProtocol;
} \
} while (0)
+// Set 'addr' and 'hp' to the '<field>' and optional 'hp_<field>' of 'obj'. The type of the
+// fields are rpc_address and host_port, respectively.
+#define SET_IP_AND_HOST_PORT(obj, field, addr, hp) \
+ do { \
+ auto &_obj = (obj); \
+ _obj.field = addr; \
+ _obj.__set_hp_##field(hp); \
+ } while (0)
+
+// Set 'value' to the '<field>' map and optional 'hp_<field>' map of 'obj'. The key of the
+// maps are rpc_address and host_port type and indexed by 'addr' and 'hp', respectively.
+#define SET_VALUE_FROM_IP_AND_HOST_PORT(obj, field, addr, hp, value) \
+ do { \
+ auto &_obj = (obj); \
+ _obj.field[addr] = value; \
+ if (!_obj.__isset.hp_##field) { \
+ _obj.__set_hp_##field({}); \
+ } \
+ _obj.hp_##field[hp] = value; \
+ } while (0)
+
+// Set 'value' to the '<field>' map and optional 'hp_<field>' map of 'obj'. The key of the
+// maps are rpc_address and host_port type and indexed by 'addr' and reverse resolve result of
+// 'addr', respectively.
+#define SET_VALUE_FROM_HOST_PORT(obj, field, hp, value) \
+ do { \
+ const auto addr = dsn::dns_resolver::instance().resolve_address(hp); \
+ SET_VALUE_FROM_IP_AND_HOST_PORT(obj, field, addr, hp, value); \
+ } while (0)
+
+#define FMT_HOST_PORT_AND_IP(obj, field) fmt::format("{}({})", (obj).hp_##field, (obj).field)
+
namespace dsn {
class rpc_group_host_port;
diff --git a/src/shell/commands/bulk_load.cpp b/src/shell/commands/bulk_load.cpp
index 17153dcfb..e5b0265e3 100644
--- a/src/shell/commands/bulk_load.cpp
+++ b/src/shell/commands/bulk_load.cpp
@@ -331,6 +331,7 @@ bool query_bulk_load_status(command_executor *e, shell_context *sc, arguments ar
if (print_download_progress) {
for (auto i = 0; i < partition_count; ++i) {
auto progress = 0;
+ // The 'bulk_load_states' must be set whatever the version of the server is.
for (const auto &kv : resp.bulk_load_states[i]) {
progress += kv.second.download_progress;
}
@@ -357,6 +358,7 @@ bool query_bulk_load_status(command_executor *e, shell_context *sc, arguments ar
}
for (auto i = 0; i < partition_count; ++i) {
+ // The 'bulk_load_states' must be set whatever the version of the server is.
auto states = resp.bulk_load_states[i];
tp_all.add_row(i);
tp_all.append_data(get_short_status(resp.partitions_status[i]));
@@ -408,6 +410,7 @@ bool query_bulk_load_status(command_executor *e, shell_context *sc, arguments ar
tp_single.add_column("is_paused");
}
+ // The 'bulk_load_states' must be set whatever the version of the server is.
auto states = resp.bulk_load_states[pidx];
for (auto iter = states.begin(); iter != states.end(); ++iter) {
tp_single.add_row(pidx);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org