You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pegasus.apache.org by "GehaFearless (via GitHub)" <gi...@apache.org> on 2023/10/26 03:25:17 UTC
[PR] Liguohao fqdn2 [incubator-pegasus]
GehaFearless opened a new pull request, #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658
### What problem does this PR solve? <!--add issue link with summary if exists-->
### What is changed and how does it work?
### Checklist <!--REMOVE the items that are not applicable-->
##### Tests <!-- At least one of them must be included. -->
- Unit test
- Integration test
- Manual test (add detailed scripts or steps below)
- No code
##### Code changes
- Has exported function/method change
- Has exported variable/fields change
- Has interface methods change
- Has persistent data change
##### Side effects
- Possible performance regression
- Increased code complexity
- Breaking backward compatibility
##### Related changes
- Need to cherry-pick to the release branch
- Need to update the documentation
- Need to be included in the release note
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474354798
##########
src/meta/test/meta_duplication_service_test.cpp:
##########
@@ -635,7 +641,7 @@ TEST_F(meta_duplication_service_test, duplication_sync)
ce.confirmed_decree = 5;
confirm_list[gpid(app->app_id, 1)].push_back(ce);
- duplication_sync_response resp = duplication_sync(node, confirm_list);
+ duplication_sync_response resp = duplication_sync(addr, node, confirm_list);
Review Comment:
auto
##########
src/meta/test/meta_duplication_service_test.cpp:
##########
@@ -772,13 +778,13 @@ TEST_F(meta_duplication_service_test, fail_mode)
ASSERT_EQ(dup->status(), duplication_status::DS_PAUSE);
// ensure dup_sync will synchronize fail_mode
- std::vector<rpc_address> server_nodes = generate_node_list(3);
- rpc_address node = server_nodes[0];
+ auto node = generate_node_list(3)[0];
for (partition_configuration &pc : app->partitions) {
- pc.primary = server_nodes[0];
+ pc.primary = node.second;
+ pc.__set_hp_primary(node.first);
}
initialize_node_state();
- duplication_sync_response sync_resp = duplication_sync(node, {});
+ duplication_sync_response sync_resp = duplication_sync(node.second, node.first, {});
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#issuecomment-1989883247
Reopen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Support to communicate with servers and clients by host_port [incubator-pegasus]
Posted by "empiredan (via GitHub)" <gi...@apache.org>.
empiredan merged PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1421980046
##########
src/common/json_helper.h:
##########
@@ -423,6 +423,21 @@ inline bool json_decode(const dsn::json::JsonObject &in, dsn::rpc_address &addre
return address.from_string_ipv4(rpc_address_string.c_str());
}
+// json serialization for rpc address, we use the string representation of a address
+inline void json_encode(JsonWriter &out, const dsn::host_port &hp)
+{
+ json_encode(out, hp.to_string());
+}
+inline bool json_decode(const dsn::json::JsonObject &in, dsn::host_port &hp)
Review Comment:
`TEST(host_port_test, thrift_parser)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474330286
##########
src/meta/load_balance_policy.cpp:
##########
@@ -257,17 +271,17 @@ bool load_balance_policy::move_primary(std::unique_ptr<flow_path> path)
int current = path->_prev.back();
if (!calc_disk_load(
- nodes, apps, path->_app->app_id, address_vec[current], true, *current_load)) {
+ nodes, apps, path->_app->app_id, host_port_vec[current], true, *current_load)) {
LOG_WARNING("stop move primary as some replica infos aren't collected, node({}), app({})",
- address_vec[current],
+ host_port_vec[current].to_string(),
path->_app->get_logname());
return false;
}
int plan_moving = path->_flow.back();
while (path->_prev[current] != 0) {
- rpc_address from = address_vec[path->_prev[current]];
- rpc_address to = address_vec[current];
+ host_port from = host_port_vec[path->_prev[current]];
Review Comment:
auto & const
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474357905
##########
src/meta/test/meta_split_service_test.cpp:
##########
@@ -377,7 +379,8 @@ class meta_split_service_test : public meta_test_base
const int32_t PARENT_BALLOT = 3;
const int32_t PARENT_INDEX = 0;
const int32_t CHILD_INDEX = 4;
- const rpc_address NODE = rpc_address("127.0.0.1", 10086);
+ const host_port NODE = host_port("localhost", 10086);
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474134946
##########
src/failure_detector/failure_detector.cpp:
##########
@@ -302,28 +305,28 @@ void failure_detector::check_all_records()
}
}
-void failure_detector::add_allow_list(::dsn::rpc_address node)
+void failure_detector::add_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
_allow_list.insert(node);
}
-bool failure_detector::remove_from_allow_list(::dsn::rpc_address node)
+bool failure_detector::remove_from_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
return _allow_list.erase(node) > 0;
}
-void failure_detector::set_allow_list(const std::vector<std::string> &replica_addrs)
+void failure_detector::set_allow_list(const std::vector<std::string> &replica_hps)
{
CHECK(!_is_started, "FD is already started, the allow list should really not be modified");
- std::vector<rpc_address> nodes;
- for (auto &addr : replica_addrs) {
- rpc_address node;
- if (!node.from_string_ipv4(addr.c_str())) {
+ std::vector<host_port> nodes;
+ for (auto &hp : replica_hps) {
+ host_port node;
+ if (!node.from_string(hp.c_str())) {
Review Comment:
del c_str()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474342988
##########
src/meta/partition_guardian.cpp:
##########
@@ -231,42 +235,45 @@ pc_status partition_guardian::on_missing_primary(meta_view &view, const dsn::gpi
action.type = config_type::CT_INVALID;
// try to upgrade a secondary to primary if the primary is missing
- if (pc.secondaries.size() > 0) {
+ if (pc.hp_secondaries.size() > 0) {
action.node.set_invalid();
+ action.__set_hp_node(host_port());
- for (int i = 0; i < pc.secondaries.size(); ++i) {
- node_state *ns = get_node_state(*(view.nodes), pc.secondaries[i], false);
- CHECK_NOTNULL(ns, "invalid secondary address, address = {}", pc.secondaries[i]);
+ for (int i = 0; i < pc.hp_secondaries.size(); ++i) {
+ node_state *ns = get_node_state(*(view.nodes), pc.hp_secondaries[i], false);
Review Comment:
auto?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#issuecomment-1903363924
TODO:
- [ ] Just passn host_post parameter, don't pass rpc_address parameters for `generate_balancer_request` and `new_proposal_action`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474335868
##########
src/meta/meta_backup_service.cpp:
##########
@@ -1695,7 +1695,8 @@ void backup_service::start_backup_app(start_backup_app_rpc rpc)
});
int32_t app_id = request.app_id;
- std::shared_ptr<backup_engine> engine = std::make_shared<backup_engine>(this);
+ std::shared_ptr<backup_engine> engine =
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474352696
##########
src/meta/test/meta_backup_test.cpp:
##########
@@ -250,7 +252,7 @@ class backup_engine_test : public meta_test_base
int32_t progress)
{
gpid pid = gpid(_app_id, partition_index);
- rpc_address mock_primary_address = rpc_address("127.0.0.1", 10000 + partition_index);
+ host_port mock_primary_address = host_port("localhost", 10000 + partition_index);
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless closed pull request #1658: feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct
URL: https://github.com/apache/incubator-pegasus/pull/1658
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1436220888
##########
src/client/partition_resolver.cpp:
##########
@@ -124,7 +127,7 @@ void partition_resolver::call_task(const rpc_response_task_ptr &t)
}
hdr.gpid = result.pid;
}
- dsn_rpc_call(result.address, t.get());
+ dsn_rpc_call(this->_dns_resolver->resolve_address(result.hp), t.get());
Review Comment:
FLAGS_use_cache_resolve_hostname
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474128551
##########
src/common/replication_other_types.h:
##########
@@ -49,43 +51,55 @@ typedef int64_t decree;
#define invalid_offset (-1LL)
#define invalid_signature 0
-inline bool is_primary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_primary(const partition_configuration &pc, const host_port &node)
{
- return !node.is_invalid() && pc.primary == node;
+ return !node.is_invalid() && pc.hp_primary == node;
}
-inline bool is_secondary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_secondary(const partition_configuration &pc, const host_port &node)
{
return !node.is_invalid() &&
- std::find(pc.secondaries.begin(), pc.secondaries.end(), node) != pc.secondaries.end();
+ std::find(pc.hp_secondaries.begin(), pc.hp_secondaries.end(), node) !=
+ pc.hp_secondaries.end();
}
-inline bool is_member(const partition_configuration &pc, const rpc_address &node)
+inline bool is_member(const partition_configuration &pc, const host_port &node)
{
return is_primary(pc, node) || is_secondary(pc, node);
}
inline bool is_partition_config_equal(const partition_configuration &pc1,
const partition_configuration &pc2)
{
// secondaries no need to be same order
- for (const rpc_address &addr : pc1.secondaries)
+ for (const host_port &addr : pc1.hp_secondaries)
if (!is_secondary(pc2, addr))
return false;
// 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.secondaries.size() == pc2.secondaries.size() &&
+ 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;
}
class replica_helper
{
public:
- static bool remove_node(::dsn::rpc_address node,
- /*inout*/ std::vector<::dsn::rpc_address> &nodeList);
+ template <typename T>
+ static bool remove_node(const T node,
+ /*inout*/ std::vector<T> &nodes)
+ {
+ auto it = std::find(nodes.begin(), nodes.end(), node);
+ if (it != nodes.end()) {
+ nodes.erase(it);
+ return true;
+ } else {
Review Comment:
del else
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474344874
##########
src/meta/server_state.cpp:
##########
@@ -1042,6 +1066,7 @@ void server_state::init_app_partition_node(std::shared_ptr<app_state> &app,
std::string app_partition_path = get_partition_path(*app, pidx);
dsn::blob value =
dsn::json::json_forwarder<partition_configuration>::encode(app->partitions[pidx]);
+
Review Comment:
del
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474353186
##########
src/meta/test/meta_bulk_load_ingestion_test.cpp:
##########
@@ -88,7 +88,7 @@ class node_context_test : public meta_test_base
public:
ingestion_context::node_context _context;
- const rpc_address NODE = rpc_address("127.0.0.1", 10086);
+ const host_port NODE = host_port("localhost", 10086);
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1423542520
##########
run.sh:
##########
@@ -27,7 +27,7 @@ export REPORT_DIR="$ROOT/test_report"
export THIRDPARTY_ROOT=$ROOT/thirdparty
export LD_LIBRARY_PATH=$JAVA_HOME/jre/lib/amd64/server:${BUILD_LATEST_DIR}/output/lib:${THIRDPARTY_ROOT}/output/lib:${LD_LIBRARY_PATH}
# Disable AddressSanitizerOneDefinitionRuleViolation, see https://github.com/google/sanitizers/issues/1017 for details.
-export ASAN_OPTIONS=detect_odr_violation=0
+export ASAN_OPTIONS=detect_odr_violation=0:abort_on_error=1:disable_coredump=0:unmap_shadow_on_exit=1
Review Comment:
Add some comments to explain why add the new options.
And it would be better to output the ASAN_OPTIONS variable when running the script because it overwrite the enviroment variable.
##########
run.sh:
##########
@@ -467,7 +467,8 @@ function run_test()
m_count=3
if [ "${module}" == "recovery_test" ]; then
m_count=1
- opts="meta_state_service_type=meta_state_service_simple,distributed_lock_service_type=distributed_lock_service_simple"
+ fqdn=`hostname -f`
+ opts="server_list=$fqdn:34601,meta_state_service_type=meta_state_service_simple,distributed_lock_service_type=distributed_lock_service_simple"
Review Comment:
Add comments to explain why change the server_list only when run recovery_test.
##########
src/client/partition_resolver_simple.cpp:
##########
@@ -411,20 +414,21 @@ void partition_resolver_simple::handle_pending_requests(std::deque<request_conte
}
/*search in cache*/
-rpc_address partition_resolver_simple::get_address(const partition_configuration &config) const
+host_port partition_resolver_simple::get_host_port(const partition_configuration &config) const
{
+ const auto &pc = config;
Review Comment:
How about renaming the parameter `config` to `pc` directly?
##########
src/runtime/rpc/rpc_holder.h:
##########
@@ -111,7 +111,7 @@ class rpc_holder
bool is_initialized() const { return bool(_i); }
- const TRequest &request() const
+ TRequest &request() const
Review Comment:
Why remove 'const'?
##########
run.sh:
##########
@@ -487,6 +488,19 @@ function run_test()
run_start_zk
fi
pushd ${BUILD_LATEST_DIR}/bin/${module}
+ local function_tests=(
+ backup_restore_test
+ recovery_test
+ restore_test
+ base_api_test
+ throttle_test
+ bulk_load_test
+ detect_hotspot_test
+ partition_split_test
+ )
+ if [[ "${function_tests[@]}" =~ "${module}" ]]; then
Review Comment:
Add comments to explain why change LOCAL_HOSTNAME only when run function_tests.
##########
src/runtime/rpc/rpc_message.h:
##########
@@ -137,6 +138,8 @@ class message_ex : public ref_counter, public extensible_object<message_ex, 4>
rpc_session_ptr io_session; // send/recv session
rpc_address to_address; // always ipv4/v6 address, it is the to_node's net address
rpc_address server_address; // used by requests, and may be of uri/group address
+ host_port to_host_port;
Review Comment:
Add some comments please.
##########
src/runtime/rpc/network.h:
##########
@@ -327,6 +330,7 @@ class rpc_session : public ref_counter
// constant info
connection_oriented_network &_net;
dsn::rpc_address _remote_addr;
Review Comment:
Is it possible to remove `_remote_addr`?
##########
src/runtime/rpc/asio_net_provider.h:
##########
@@ -93,6 +95,7 @@ class asio_network_provider : public connection_oriented_network
std::vector<std::unique_ptr<boost::asio::io_service>> _io_services;
std::vector<std::shared_ptr<std::thread>> _workers;
::dsn::rpc_address _address;
Review Comment:
Is it possible to safe remove `_address`?
##########
src/common/replication_common.cpp:
##########
@@ -165,33 +168,46 @@ int32_t replication_options::app_mutation_2pc_min_replica_count(int32_t app_max_
}
}
+/*static*/ bool replica_helper::remove_node(::dsn::host_port node,
Review Comment:
The two functions can be replaced by a template function to remove duplicate code.
##########
src/runtime/rpc/rpc_host_port.h:
##########
@@ -67,6 +76,8 @@ class host_port
std::string to_string() const;
+ const char *get_char_ctr() const;
Review Comment:
what does `ctr` mean?
##########
src/runtime/rpc/network.sim.h:
##########
@@ -109,6 +111,7 @@ class sim_network_provider : public connection_oriented_network
private:
::dsn::rpc_address _address;
Review Comment:
Is it possible to remove `_address`?
##########
src/meta/load_balance_policy.h:
##########
@@ -283,10 +290,11 @@ class copy_primary_operation : public copy_replica_operation
copy_primary_operation(const std::shared_ptr<app_state> app,
const app_mapper &apps,
node_mapper &nodes,
- const std::vector<dsn::rpc_address> &address_vec,
- const std::unordered_map<dsn::rpc_address, int> &address_id,
+ const std::vector<dsn::host_port> &address_vec,
+ const std::unordered_map<dsn::host_port, int> &address_id,
Review Comment:
```suggestion
const std::vector<dsn::host_port> &host_port_vec,
const std::unordered_map<dsn::host_port, int> &host_port_id,
```
##########
src/runtime/api_layer1.h:
##########
@@ -215,6 +216,8 @@ replace the underneath implementation of the network (e.g., RDMA, simulated netw
extern dsn::rpc_address dsn_primary_address();
Review Comment:
Is it possible to safe remove `dsn_primary_address()`?
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -438,8 +442,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 = {}",
+ primary_hp.to_string(),
primary_addr.to_string(),
Review Comment:
```suggestion
primary_hp,
primary_addr,
```
##########
src/meta/load_balance_policy.h:
##########
@@ -268,11 +273,13 @@ class copy_replica_operation
const std::shared_ptr<app_state> _app;
const app_mapper &_apps;
node_mapper &_nodes;
- const std::vector<dsn::rpc_address> &_address_vec;
- const std::unordered_map<dsn::rpc_address, int> &_address_id;
- std::unordered_map<dsn::rpc_address, disk_load> _node_loads;
+ const std::vector<dsn::host_port> &_address_vec;
+ const std::unordered_map<dsn::host_port, int> &_address_id;
Review Comment:
```suggestion
const std::vector<dsn::host_port> &_host_port_vec;
const std::unordered_map<dsn::host_port, int> &_host_port_id;
```
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -450,6 +455,18 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
bulk_load_rpc rpc(std::move(req), RPC_BULK_LOAD, 0_ms, 0, pid.thread_hash());
rpc.call(primary_addr, _meta_svc->tracker(), [this, rpc](error_code err) mutable {
+
+ // fill host_port struct if needed
Review Comment:
Note the indent.
##########
src/meta/load_balance_policy.cpp:
##########
@@ -141,38 +146,40 @@ generate_balancer_request(const app_mapper &apps,
case balance_type::MOVE_PRIMARY:
ans = "move_primary";
result.balance_type = balancer_request_type::move_primary;
+ result.action_list.emplace_back(new_proposal_action(
+ from, from, hp_from, hp_from, config_type::CT_DOWNGRADE_TO_SECONDARY));
Review Comment:
Same, just to pass host ports parameters.
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -450,6 +455,18 @@ void bulk_load_service::partition_bulk_load(const std::string &app_name, const g
bulk_load_rpc rpc(std::move(req), RPC_BULK_LOAD, 0_ms, 0, pid.thread_hash());
rpc.call(primary_addr, _meta_svc->tracker(), [this, rpc](error_code err) mutable {
+
+ // fill host_port struct if needed
Review Comment:
The comment seems not enough, you need to mention the remote server doesn't support FQDN?
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -461,25 +478,28 @@ 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 rpc_address &primary_addr = request.primary_addr;
+ const auto &primary_addr = request.primary_addr;
+ const auto &primary_hp = request.hp_primary_addr;
if (err != ERR_OK) {
- LOG_ERROR(
- "app({}), partition({}) failed to receive bulk load response from node({}), error = {}",
- app_name,
- pid,
- primary_addr.to_string(),
- err.to_string());
+ LOG_ERROR("app({}), partition({}) failed to receive bulk load response from node({}({})), "
+ "error = {}",
+ app_name,
+ pid,
+ primary_hp.to_string(),
+ primary_addr.to_string(),
+ err.to_string());
Review Comment:
```suggestion
primary_hp,
primary_addr,
err);
```
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -461,25 +478,28 @@ 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 rpc_address &primary_addr = request.primary_addr;
+ const auto &primary_addr = request.primary_addr;
+ const auto &primary_hp = request.hp_primary_addr;
if (err != ERR_OK) {
- LOG_ERROR(
- "app({}), partition({}) failed to receive bulk load response from node({}), error = {}",
- app_name,
- pid,
- primary_addr.to_string(),
- err.to_string());
+ LOG_ERROR("app({}), partition({}) failed to receive bulk load response from node({}({})), "
+ "error = {}",
+ app_name,
+ pid,
+ primary_hp.to_string(),
+ primary_addr.to_string(),
+ err.to_string());
try_rollback_to_downloading(app_name, pid);
try_resend_bulk_load_request(app_name, pid);
return;
}
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.to_string(),
primary_addr.to_string(),
response.err.to_string());
Review Comment:
```suggestion
primary_hp,
primary_addr,
response.err);
```
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -1600,9 +1623,18 @@ void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
}
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) {
- response.bulk_load_states[kv.first.get_partition_index()] = kv.second;
+ auto gpid = kv.first.get_partition_index();
Review Comment:
gpid is a type name, it's better to avoid using it as a variable name.
```suggestion
auto pidx = kv.first.get_partition_index();
```
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -499,13 +520,15 @@ 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_addr.to_string(),
- response.err.to_string(),
- 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,
+ primary_hp.to_string(),
+ primary_addr.to_string(),
+ response.err.to_string(),
Review Comment:
```suggestion
primary_hp,
primary_addr,
response.err,
```
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -489,8 +509,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.to_string(),
primary_addr.to_string(),
Review Comment:
```suggestion
primary_hp,
primary_addr,
```
##########
src/client/partition_resolver.cpp:
##########
@@ -124,7 +127,7 @@ void partition_resolver::call_task(const rpc_response_task_ptr &t)
}
hdr.gpid = result.pid;
}
- dsn_rpc_call(result.address, t.get());
+ dsn_rpc_call(this->_dns_resolver->resolve_address(result.hp), t.get());
Review Comment:
@GehaFearless Have you improved the code according to my reply?
##########
src/meta/cluster_balance_policy.cpp:
##########
@@ -544,10 +545,21 @@ bool cluster_balance_policy::apply_move(const move_info &move,
// add into migration list and selected_pid
partition_configuration pc;
pc.pid = move.pid;
- pc.primary = primary_addr;
- list[move.pid] = generate_balancer_request(*_global_view->apps, pc, move.type, source, target);
+ pc.hp_primary = primary_hp;
+ std::shared_ptr<dsn::dns_resolver> resolver;
+ if (_svc == nullptr) {
+ resolver = std::make_shared<dsn::dns_resolver>();
Review Comment:
How about creating a valid meta_service there and pass it to this function, to ensure consistence.
##########
src/meta/load_balance_policy.h:
##########
@@ -244,8 +248,9 @@ class copy_replica_operation
copy_replica_operation(const std::shared_ptr<app_state> app,
const app_mapper &apps,
node_mapper &nodes,
- const std::vector<dsn::rpc_address> &address_vec,
- const std::unordered_map<dsn::rpc_address, int> &address_id);
+ const std::vector<dsn::host_port> &address_vec,
+ const std::unordered_map<dsn::host_port, int> &address_id,
Review Comment:
```suggestion
const std::vector<dsn::host_port> &host_port_vec,
const std::unordered_map<dsn::host_port, int> &host_port_id,
```
##########
src/meta/meta_data.cpp:
##########
@@ -151,12 +136,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.last_drops.empty(), "last_drops of partition({}) must be empty", pid);
+ CHECK(pc.hp_last_drops.empty(),
+ "last_drops of partition({}.{}) must be empty",
+ pid.get_app_id(),
+ pid.get_partition_index());
Review Comment:
```suggestion
"last_drops of partition({}) must be empty",
pid);
```
##########
src/meta/load_balance_policy.cpp:
##########
@@ -128,7 +131,9 @@ generate_balancer_request(const app_mapper &apps,
const partition_configuration &pc,
const balance_type &type,
const rpc_address &from,
- const rpc_address &to)
+ const rpc_address &to,
+ const host_port &hp_from,
+ const host_port &hp_to)
Review Comment:
Is it enough to pass just hp_from and hp_to? The ip addresses will be resolved internally.
##########
src/runtime/rpc/rpc_engine.h:
##########
@@ -202,6 +204,7 @@ class rpc_engine
std::unordered_map<int, std::vector<std::unique_ptr<network>>>
_server_nets; // <port, <CHANNEL, network*>>
::dsn::rpc_address _local_primary_address;
Review Comment:
Is it possible to remove `_local_primary_address`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1395128717
##########
src/client/replication_ddl_client.cpp:
##########
@@ -81,18 +82,51 @@ namespace replication {
using tp_output_format = ::dsn::utils::table_printer::output_format;
-replication_ddl_client::replication_ddl_client(const std::vector<dsn::rpc_address> &meta_servers)
+replication_ddl_client::replication_ddl_client(const std::vector<dsn::host_port> &meta_servers)
+ : _dns_resolver(new dns_resolver())
{
_meta_server.assign_group("meta-servers");
for (const auto &m : meta_servers) {
- if (!_meta_server.group_address()->add(m)) {
+ if (!_meta_server.group_host_port()->add(m)) {
LOG_WARNING("duplicate adress {}", m);
}
}
}
replication_ddl_client::~replication_ddl_client() { _tracker.cancel_outstanding_tasks(); }
+void replication_ddl_client::set_meta_servers_leader()
Review Comment:
We should selecting the leader host actively. All of `call()` we should transfer `host_port` to `rpc_address`，but not pointer passing should not change group leader by `on_recv_reply()`.
##########
src/client/replication_ddl_client.cpp:
##########
@@ -81,18 +82,51 @@ namespace replication {
using tp_output_format = ::dsn::utils::table_printer::output_format;
-replication_ddl_client::replication_ddl_client(const std::vector<dsn::rpc_address> &meta_servers)
+replication_ddl_client::replication_ddl_client(const std::vector<dsn::host_port> &meta_servers)
+ : _dns_resolver(new dns_resolver())
{
_meta_server.assign_group("meta-servers");
for (const auto &m : meta_servers) {
- if (!_meta_server.group_address()->add(m)) {
+ if (!_meta_server.group_host_port()->add(m)) {
LOG_WARNING("duplicate adress {}", m);
}
}
}
replication_ddl_client::~replication_ddl_client() { _tracker.cancel_outstanding_tasks(); }
+void replication_ddl_client::set_meta_servers_leader()
Review Comment:
We should selecting the leader host actively. All of `call()` we should transfer `host_port` to `rpc_address`，but not pointer passing should not change group leader by `on_recv_reply()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1475621378
##########
src/runtime/rpc/dns_resolver.cpp:
##########
@@ -53,11 +53,6 @@ dns_resolver::dns_resolver()
METRIC_VAR_INIT_server(dns_resolver_resolve_duration_ns),
METRIC_VAR_INIT_server(dns_resolver_resolve_by_dns_duration_ns)
{
-#ifndef MOCK_TEST
Review Comment:
It will crash if not remove it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474339352
##########
src/meta/meta_bulk_load_service.cpp:
##########
@@ -1257,7 +1279,7 @@ void bulk_load_service::partition_ingestion(const std::string &app_name, const g
return;
}
- rpc_address primary_addr = pconfig.primary;
+ host_port primary_addr = pconfig.hp_primary;
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1445719634
##########
src/runtime/test/rpc.cpp:
##########
@@ -238,7 +239,7 @@ TEST(core, send_to_invalid_address)
{
::dsn::rpc_address group = build_group();
/* here we assume 10.255.254.253:32766 is not assigned */
Review Comment:
The comment is incorrect now, remove it.
##########
src/meta/cluster_balance_policy.cpp:
##########
@@ -544,10 +545,17 @@ bool cluster_balance_policy::apply_move(const move_info &move,
// add into migration list and selected_pid
partition_configuration pc;
pc.pid = move.pid;
- pc.primary = primary_addr;
- list[move.pid] = generate_balancer_request(*_global_view->apps, pc, move.type, source, target);
+ pc.hp_primary = primary_hp;
+ std::shared_ptr<dsn::dns_resolver> resolver;
+ resolver = _svc->get_dns_resolver();
+ auto source_addr = resolver->resolve_address(source);
+ auto target_addr = resolver->resolve_address(target);
Review Comment:
```suggestion
auto source_addr = _svc->get_dns_resolver()->resolve_address(source);
auto target_addr = _svc->get_dns_resolver()->resolve_address(target);
```
##########
src/nfs/test/main.cpp:
##########
@@ -68,7 +69,8 @@ INSTANTIATE_TEST_CASE_P(, nfs_test, ::testing::Values(false, true));
TEST_P(nfs_test, basic)
{
- auto nfs = dsn::nfs_node::create();
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
Review Comment:
```suggestion
auto resolver = std::make_shared<dns_resolver>();
```
##########
src/replica/replica_2pc.cpp:
##########
@@ -306,9 +308,10 @@ void replica::init_prepare(mutation_ptr &mu, bool reconciliation, bool pop_all_c
// remote prepare
mu->set_prepare_ts();
- mu->set_left_secondary_ack_count((unsigned int)_primary_states.membership.secondaries.size());
- for (auto it = _primary_states.membership.secondaries.begin();
- it != _primary_states.membership.secondaries.end();
+ mu->set_left_secondary_ack_count(
+ (unsigned int)_primary_states.membership.hp_secondaries.size());
+ for (auto it = _primary_states.membership.hp_secondaries.begin();
+ it != _primary_states.membership.hp_secondaries.end();
++it) {
send_prepare_message(*it,
Review Comment:
```suggestion
for (const auto &secondary : hp_secondaries) {
send_prepare_message(secondary,
```
##########
src/replica/duplication/test/replica_follower_test.cpp:
##########
@@ -104,7 +106,8 @@ class replica_follower_test : public duplication_test_base
void init_nfs()
{
- stub->_nfs = nfs_node::create();
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
Review Comment:
Because we can infer the type from std::make_shared<xxx>, so we can use auto simply.
```suggestion
auto resolver = std::make_shared<dns_resolver>();
```
##########
src/redis_protocol/proxy_lib/proxy_layer.h:
##########
@@ -80,7 +80,8 @@ class proxy_session : public std::enable_shared_from_this<proxy_session>
// we need to backup one request to create a response struct.
dsn::message_ex *_backup_one_request;
// the client address for which this session served
- dsn::rpc_address _remote_address;
+ dsn::host_port _remote_host_port;
+ std::string _host_port;
Review Comment:
The two member names are confused, how about improving it by renaming _host_port to _log_prefix ?
##########
src/meta/test/copy_replica_operation_test.cpp:
##########
@@ -166,9 +168,11 @@ TEST(copy_primary_operation, can_select)
{
app_mapper apps;
node_mapper nodes;
- std::vector<dsn::rpc_address> address_vec;
- std::unordered_map<dsn::rpc_address, int> address_id;
- copy_primary_operation op(nullptr, apps, nodes, address_vec, address_id, false, false);
+ std::vector<dsn::host_port> host_port_vec;
+ std::unordered_map<dsn::host_port, int> host_port_id;
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
Review Comment:
```suggestion
auto resolver = std::make_shared<dns_resolver>();
```
##########
src/nfs/test/main.cpp:
##########
@@ -68,7 +69,8 @@ INSTANTIATE_TEST_CASE_P(, nfs_test, ::testing::Values(false, true));
TEST_P(nfs_test, basic)
{
- auto nfs = dsn::nfs_node::create();
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
+ std::unique_ptr<dsn::nfs_node> nfs(dsn::nfs_node::create(resolver));
Review Comment:
```suggestion
auto nfs = dsn::nfs_node::create(resolver);
```
##########
src/meta/test/copy_replica_operation_test.cpp:
##########
@@ -183,9 +187,11 @@ TEST(copy_primary_operation, only_copy_primary)
{
app_mapper apps;
node_mapper nodes;
- std::vector<dsn::rpc_address> address_vec;
- std::unordered_map<dsn::rpc_address, int> address_id;
- copy_primary_operation op(nullptr, apps, nodes, address_vec, address_id, false, false);
+ std::vector<dsn::host_port> host_port_vec;
+ std::unordered_map<dsn::host_port, int> host_port_id;
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
Review Comment:
```suggestion
auto resolver = std::make_shared<dns_resolver>();
```
##########
src/meta/test/meta_duplication_service_test.cpp:
##########
@@ -127,11 +129,13 @@ class meta_duplication_service_test : public meta_test_base
}
duplication_sync_response
- duplication_sync(const rpc_address &node,
+ duplication_sync(const rpc_address &addr,
Review Comment:
This is only used in tests, is it necessary to keep the rpc_address parameter?
##########
src/meta/test/copy_replica_operation_test.cpp:
##########
@@ -216,13 +222,14 @@ TEST(copy_secondary_operation, misc)
node_state ns3;
nodes[addr3] = ns3;
- std::vector<dsn::rpc_address> address_vec{addr1, addr2, addr3};
- std::unordered_map<dsn::rpc_address, int> address_id;
- address_id[addr1] = 0;
- address_id[addr2] = 1;
- address_id[addr3] = 2;
- copy_secondary_operation op(app, apps, nodes, address_vec, address_id, 0);
- op.init_ordered_address_ids();
+ std::vector<dsn::host_port> host_port_vec{addr1, addr2, addr3};
+ std::unordered_map<dsn::host_port, int> host_port_id;
+ host_port_id[addr1] = 0;
+ host_port_id[addr2] = 1;
+ host_port_id[addr3] = 2;
+ std::shared_ptr<dns_resolver> resolver = std::make_shared<dns_resolver>();
Review Comment:
```suggestion
auto resolver = std::make_shared<dns_resolver>();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1466108526
##########
src/runtime/rpc/asio_net_provider.cpp:
##########
@@ -356,7 +370,23 @@ void asio_udp_provider::do_receive()
return;
}
+ if (msg->header->from_address != remote_addr) {
Review Comment:
这里针对跨网环境的情景，如果 head 中 from 的 address 和实际 socket 连接的 address 不一致，非转发场景修正为能识别的 address
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474347681
##########
src/meta/test/cluster_balance_policy_test.cpp:
##########
@@ -180,48 +177,49 @@ TEST(cluster_balance_policy, get_node_migration_info)
all_apps[appid] = app;
node_state ns;
- ns.set_addr(address);
+ ns.set_hp(address);
gpid pid = gpid(appid, 0);
ns.put_partition(pid, true);
cluster_balance_policy::node_migration_info migration_info;
policy.get_node_migration_info(ns, all_apps, migration_info);
- ASSERT_EQ(migration_info.address, address);
+ ASSERT_EQ(migration_info.hp, address);
ASSERT_NE(migration_info.partitions.find(disk_tag), migration_info.partitions.end());
ASSERT_EQ(migration_info.partitions.at(disk_tag).size(), 1);
ASSERT_EQ(*migration_info.partitions.at(disk_tag).begin(), pid);
}
TEST(cluster_balance_policy, get_min_max_set)
{
- std::map<rpc_address, uint32_t> node_count_map;
- node_count_map.emplace(rpc_address(1, 10086), 1);
- node_count_map.emplace(rpc_address(2, 10086), 3);
- node_count_map.emplace(rpc_address(3, 10086), 5);
- node_count_map.emplace(rpc_address(4, 10086), 5);
+ std::map<host_port, uint32_t> node_count_map;
+ node_count_map.emplace(host_port("localhost", 10081), 1);
+ node_count_map.emplace(host_port("localhost", 10082), 3);
+ node_count_map.emplace(host_port("localhost", 10083), 5);
+ node_count_map.emplace(host_port("localhost", 10084), 5);
- std::set<rpc_address> min_set, max_set;
+ std::set<host_port> min_set, max_set;
get_min_max_set(node_count_map, min_set, max_set);
ASSERT_EQ(min_set.size(), 1);
- ASSERT_EQ(*min_set.begin(), rpc_address(1, 10086));
+ ASSERT_EQ(*min_set.begin(), host_port("localhost", 10081));
ASSERT_EQ(max_set.size(), 2);
- ASSERT_EQ(*max_set.begin(), rpc_address(3, 10086));
- ASSERT_EQ(*max_set.rbegin(), rpc_address(4, 10086));
+ ASSERT_EQ(*max_set.begin(), host_port("localhost", 10083));
+ ASSERT_EQ(*max_set.rbegin(), host_port("localhost", 10084));
}
TEST(cluster_balance_policy, get_disk_partitions_map)
{
- cluster_balance_policy policy(nullptr);
+ meta_service svc;
+ cluster_balance_policy policy(&svc);
cluster_balance_policy::cluster_migration_info cluster_info;
- rpc_address addr(1, 10086);
+ host_port addr("localhost", 10086);
Review Comment:
addr -> hp
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474123624
##########
src/client/replication_ddl_client.cpp:
##########
@@ -924,8 +961,8 @@ dsn::error_code replication_ddl_client::do_recovery(const std::vector<rpc_addres
out << "Skip lost partitions: " << (skip_lost_partitions ? "true" : "false") << std::endl;
out << "Node list:" << std::endl;
out << "=============================" << std::endl;
- for (auto &node : req->recovery_set) {
- out << node << std::endl;
+ for (auto &node : req->hp_recovery_set) {
+ out << node.to_string() << std::endl;
Review Comment:
del to_string()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474317687
##########
src/failure_detector/test/failure_detector.cpp:
##########
@@ -228,8 +237,8 @@ class test_master : public service_app
{
FLAGS_stable_rs_min_running_seconds = 10;
FLAGS_max_succssive_unstable_restart = 10;
-
- _master_fd = new master_fd_test();
+ std::shared_ptr<::dsn::dns_resolver> dns_resolver(new ::dsn::dns_resolver());
Review Comment:
auto resolver = std::make_shared<dsn::dns_resolver>();
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1418257488
##########
src/meta/cluster_balance_policy.cpp:
##########
@@ -544,10 +545,21 @@ bool cluster_balance_policy::apply_move(const move_info &move,
// add into migration list and selected_pid
partition_configuration pc;
pc.pid = move.pid;
- pc.primary = primary_addr;
- list[move.pid] = generate_balancer_request(*_global_view->apps, pc, move.type, source, target);
+ pc.hp_primary = primary_hp;
+ std::shared_ptr<dsn::dns_resolver> resolver;
+ if (_svc == nullptr) {
+ resolver = std::make_shared<dsn::dns_resolver>();
Review Comment:
This is a defensive coding, and there are cases in the test where `_svc` is nullptr.
`TEST(cluster_balance_policy, apply_move)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1439223636
##########
src/test/function_test/recovery/test_recovery.cpp:
##########
@@ -211,7 +216,7 @@ TEST_F(recovery_test, recovery)
std::this_thread::sleep_for(std::chrono::seconds(10));
// recovery only from 1 & 2
- std::vector<dsn::rpc_address> nodes = get_rpc_address_list({34801, 34802});
+ std::vector<dsn::host_port> nodes = get_rpc_host_port_list({34801, 34802});
Review Comment:
```suggestion
std::vector<dsn::host_port> nodes = get_rpc_host_port_list({34801, 34802});
```
```suggestion
auto nodes = get_rpc_host_port_list({34801, 34802});
```
##########
src/test/function_test/base_api/test_batch_get.cpp:
##########
@@ -48,8 +49,9 @@ class batch_get : public test_util
TEST_F(batch_get, set_and_then_batch_get)
{
- auto rrdb_client =
- new ::dsn::apps::rrdb_client(kClusterName.c_str(), meta_list_, table_name_.c_str());
+ std::shared_ptr<dsn::dns_resolver> dns_resolver = std::make_shared<dsn::dns_resolver>();
Review Comment:
```suggestion
auto dns_resolver = std::make_shared<dsn::dns_resolver>();
```
##########
src/client/replication_ddl_client.cpp:
##########
@@ -776,17 +813,17 @@ dsn::error_code replication_ddl_client::list_app(const std::string &app_name,
std::stringstream oss;
oss << replica_count << "/" << p.max_replica_count;
tp_details.append_data(oss.str());
- tp_details.append_data(
- (p.primary.is_invalid() ? "-" : host_name_resolve(resolve_ip,
- p.primary.to_std_string())));
+ tp_details.append_data((p.hp_primary.is_invalid()
+ ? "-"
+ : host_name_resolve(resolve_ip, p.hp_primary.to_string())));
Review Comment:
Because p.hp_primary is valid, we can use it directly, right?
##########
src/test/function_test/recovery/test_recovery.cpp:
##########
@@ -65,13 +66,17 @@ class recovery_test : public test_util
}
public:
- std::vector<dsn::rpc_address> get_rpc_address_list(const std::vector<int> ports)
+ recovery_test() : test_util(std::map<std::string, std::string>(), "single_master_cluster") {}
Review Comment:
It would be better to submit this change in a separated patch.
##########
src/shell/commands/detect_hotkey.cpp:
##########
@@ -100,10 +100,10 @@ bool detect_hotkey(command_executor *e, shell_context *sc, arguments args)
return false;
}
- dsn::rpc_address target_address;
+ dsn::host_port target_node;
Review Comment:
Unify to naming `target_hp`?
##########
src/client/replication_ddl_client.cpp:
##########
@@ -490,8 +525,10 @@ dsn::error_code replication_ddl_client::list_nodes(
return resp.err;
}
- for (const dsn::replication::node_info &n : resp.infos) {
- nodes[n.address] = n.status;
+ for (dsn::replication::node_info &n : resp.infos) {
Review Comment:
```suggestion
for (const auto &n : resp.infos) {
```
##########
src/meta/app_balance_policy.cpp:
##########
@@ -182,8 +187,8 @@ bool copy_secondary_operation::can_select(gpid pid, migration_list *result)
return false;
}
- int id_min = *_ordered_address_ids.begin();
- const node_state &min_ns = _nodes.at(_address_vec[id_min]);
+ int id_min = *_ordered_host_port_ids.begin();
+ const node_state &min_ns = _nodes.at(host_port(_host_port_vec[id_min]));
Review Comment:
```suggestion
const node_state &min_ns = _nodes.at(_host_port_vec[id_min]);
```
##########
src/meta/app_balance_policy.cpp:
##########
@@ -163,8 +168,8 @@ int copy_secondary_operation::get_partition_count(const node_state &ns) const
bool copy_secondary_operation::can_select(gpid pid, migration_list *result)
{
- int id_max = *_ordered_address_ids.rbegin();
- const node_state &max_ns = _nodes.at(_address_vec[id_max]);
+ int id_max = *_ordered_host_port_ids.rbegin();
+ const node_state &max_ns = _nodes.at(host_port(_host_port_vec[id_max]));
Review Comment:
```suggestion
const node_state &max_ns = _nodes.at(_host_port_vec[id_max]);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1386364196
##########
src/runtime/rpc/rpc_host_port.h:
##########
@@ -41,6 +41,31 @@ class TProtocol;
} // namespace thrift
} // namespace apache
+#define FILL_HP_OPTIONAL_SECTION(OBJ_NAME, SECTION_NAME) \
Review Comment:
Those macro define fuctions not use now. It will be deleted soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1401671098
##########
src/client/replication_ddl_client.cpp:
##########
@@ -490,8 +525,16 @@ dsn::error_code replication_ddl_client::list_nodes(
return resp.err;
}
- for (const dsn::replication::node_info &n : resp.infos) {
- nodes[n.address] = n.status;
+ for (dsn::replication::node_info &n : resp.infos) {
+ host_port hp;
+ if (n.__isset.hp_address) {
+ hp = n.hp_address;
+ LOG_ERROR("lgh: replica_server hp {}", hp);
Review Comment:
Remove the debug logs.
##########
run.sh:
##########
@@ -485,6 +486,19 @@ function run_test()
run_start_zk
fi
pushd ${BUILD_LATEST_DIR}/bin/${module}
+ local function_tests=(
+ backup_restore_test
+ recovery_test
Review Comment:
Keep alignment.
##########
src/client/partition_resolver_simple.cpp:
##########
@@ -411,20 +414,21 @@ void partition_resolver_simple::handle_pending_requests(std::deque<request_conte
}
/*search in cache*/
-rpc_address partition_resolver_simple::get_address(const partition_configuration &config) const
+host_port partition_resolver_simple::get_host_port(const partition_configuration &config) const
{
+ auto pc = config;
Review Comment:
```suggestion
const auto &pc = config;
```
##########
src/failure_detector/failure_detector.cpp:
##########
@@ -415,33 +431,47 @@ bool failure_detector::end_ping_internal(::dsn::error_code err, const beacon_ack
/*
* the caller of the end_ping_internal should lock necessarily!!!
*/
+ host_port hp_this_node, hp_primary_node;
+ if (ack.__isset.hp_this_node) {
+ hp_this_node = ack.hp_this_node;
+ } else {
+ hp_this_node = host_port(ack.this_node);
+ }
+ if (ack.__isset.hp_primary_node) {
+ hp_primary_node = ack.hp_primary_node;
+ } else {
+ hp_primary_node = host_port(ack.primary_node);
+ }
+
uint64_t beacon_send_time = ack.time;
- auto node = ack.this_node;
if (err != ERR_OK) {
LOG_WARNING("ping master({}) failed, timeout_ms = {}, err = {}",
- node,
+ hp_this_node,
_beacon_timeout_milliseconds,
err);
_recent_beacon_fail_count->increment();
}
- master_map::iterator itr = _masters.find(node);
+ master_map::iterator itr = _masters.find(hp_this_node);
if (itr == _masters.end()) {
LOG_WARNING("received beacon ack without corresponding master, ignore it, "
- "remote_master[{}], local_worker[{}]",
- node,
+ "remote_master[{}({})], local_worker[{}]",
+ hp_this_node,
+ ack.this_node,
dsn_primary_address());
Review Comment:
Also logging host_port by dsn_primary_host_port().
Other places are the same.
##########
src/meta/cluster_balance_policy.cpp:
##########
@@ -544,10 +545,21 @@ bool cluster_balance_policy::apply_move(const move_info &move,
// add into migration list and selected_pid
partition_configuration pc;
pc.pid = move.pid;
- pc.primary = primary_addr;
- list[move.pid] = generate_balancer_request(*_global_view->apps, pc, move.type, source, target);
+ pc.hp_primary = primary_hp;
+ std::shared_ptr<dsn::dns_resolver> resolver;
+ if (_svc == nullptr) {
+ resolver = std::make_shared<dsn::dns_resolver>();
Review Comment:
In which case the `_svc` is nullptr?
##########
src/meta/greedy_load_balancer.cpp:
##########
@@ -163,7 +163,7 @@ void greedy_load_balancer::score(meta_view view, double &primary_stddev, double
bool greedy_load_balancer::all_replica_infos_collected(const node_state &ns)
{
- dsn::rpc_address n = ns.addr();
+ auto n = ns.host_port();
Review Comment:
```suggestion
const auto &n = ns.host_port();
```
##########
src/common/json_helper.h:
##########
@@ -423,6 +423,21 @@ inline bool json_decode(const dsn::json::JsonObject &in, dsn::rpc_address &addre
return address.from_string_ipv4(rpc_address_string.c_str());
}
+// json serialization for rpc address, we use the string representation of a address
+inline void json_encode(JsonWriter &out, const dsn::host_port &hp)
+{
+ json_encode(out, hp.to_string());
+}
+inline bool json_decode(const dsn::json::JsonObject &in, dsn::host_port &hp)
Review Comment:
Add some unit tests for the two functions.
##########
src/common/replication_common.cpp:
##########
@@ -164,6 +167,18 @@ int32_t replication_options::app_mutation_2pc_min_replica_count(int32_t app_max_
}
}
+/*static*/ bool replica_helper::remove_node(::dsn::host_port node,
+ /*inout*/ std::vector<::dsn::host_port> &nodeList)
Review Comment:
```suggestion
/*inout*/ std::vector<::dsn::host_port> &nodes)
```
##########
src/client/replication_ddl_client.cpp:
##########
@@ -81,18 +82,51 @@ namespace replication {
using tp_output_format = ::dsn::utils::table_printer::output_format;
-replication_ddl_client::replication_ddl_client(const std::vector<dsn::rpc_address> &meta_servers)
+replication_ddl_client::replication_ddl_client(const std::vector<dsn::host_port> &meta_servers)
+ : _dns_resolver(new dns_resolver())
{
_meta_server.assign_group("meta-servers");
for (const auto &m : meta_servers) {
- if (!_meta_server.group_address()->add(m)) {
+ if (!_meta_server.group_host_port()->add(m)) {
LOG_WARNING("duplicate adress {}", m);
}
}
}
replication_ddl_client::~replication_ddl_client() { _tracker.cancel_outstanding_tasks(); }
+void replication_ddl_client::set_meta_servers_leader()
Review Comment:
Sorry I didn't get your point, add some comment for this newly added function (in header file) may help to comprehend.
##########
src/client/partition_resolver.cpp:
##########
@@ -124,7 +127,7 @@ void partition_resolver::call_task(const rpc_response_task_ptr &t)
}
hdr.gpid = result.pid;
}
- dsn_rpc_call(result.address, t.get());
+ dsn_rpc_call(this->_dns_resolver->resolve_address(result.hp), t.get());
Review Comment:
There will be performance degradation if resolving host name everytime, fortunately, you have added a cache for that. Even though, it's worse than using ip address directly.
A compromised way is to add an option to decide whether to resolve host name every time or just once, and make resolving once as default.
IMO, it's acceptable to resolve host name just once, because if a host changes it's ip address, it's reasonable to ask the administrators to restart the Pegasus servers and clients. If they don't accept this, they should know and accept the performance degradation caused by resolving host name every time.
##########
src/failure_detector/failure_detector.cpp:
##########
@@ -352,33 +355,46 @@ std::string failure_detector::get_allow_list(const std::vector<std::string> &arg
void failure_detector::on_ping_internal(const beacon_msg &beacon, /*out*/ beacon_ack &ack)
{
+ host_port hp_from_addr, hp_to_addr;
+ if (beacon.__isset.hp_from_addr) {
Review Comment:
Add a macro like this to simplify code:
```
#define HOST_PORT(obj, field, target) \
do { \
if (obj.__isset.hp_##field) { \
target = obj.hp_##field; \
} else { \
target = std::move(host_port(obj.field)); \
} \
} while (0)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1418264398
##########
src/client/replication_ddl_client.cpp:
##########
@@ -81,18 +82,51 @@ namespace replication {
using tp_output_format = ::dsn::utils::table_printer::output_format;
-replication_ddl_client::replication_ddl_client(const std::vector<dsn::rpc_address> &meta_servers)
+replication_ddl_client::replication_ddl_client(const std::vector<dsn::host_port> &meta_servers)
+ : _dns_resolver(new dns_resolver())
{
_meta_server.assign_group("meta-servers");
for (const auto &m : meta_servers) {
- if (!_meta_server.group_address()->add(m)) {
+ if (!_meta_server.group_host_port()->add(m)) {
LOG_WARNING("duplicate adress {}", m);
}
}
}
replication_ddl_client::~replication_ddl_client() { _tracker.cancel_outstanding_tasks(); }
+void replication_ddl_client::set_meta_servers_leader()
Review Comment:
`dsn_rpc_call(dns_resolver->resolve_address(hp), rpc_call);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#issuecomment-1989932719
Reopen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474031234
##########
idl/partition_split.thrift:
##########
@@ -112,10 +113,11 @@ struct notify_cacth_up_response
// primary parent -> child replicas to update partition count
struct update_child_group_partition_count_request
{
- 1:dsn.rpc_address target_address;
+ 1:dsn.rpc_address target;
Review Comment:
shouldn't be backward compatible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#issuecomment-1851706742
Some changes can be separated to another patch, for example, the changes in:
1. src/common/json_helper.h and its tests.
2. src/runtime/rpc/rpc_host_port.h (add some tests as well)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1474330286
##########
src/meta/load_balance_policy.cpp:
##########
@@ -257,17 +271,17 @@ bool load_balance_policy::move_primary(std::unique_ptr<flow_path> path)
int current = path->_prev.back();
if (!calc_disk_load(
- nodes, apps, path->_app->app_id, address_vec[current], true, *current_load)) {
+ nodes, apps, path->_app->app_id, host_port_vec[current], true, *current_load)) {
LOG_WARNING("stop move primary as some replica infos aren't collected, node({}), app({})",
- address_vec[current],
+ host_port_vec[current].to_string(),
path->_app->get_logname());
return false;
}
int plan_moving = path->_flow.back();
while (path->_prev[current] != 0) {
- rpc_address from = address_vec[path->_prev[current]];
- rpc_address to = address_vec[current];
+ host_port from = host_port_vec[path->_prev[current]];
Review Comment:
auto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1382871191
##########
src/client/partition_resolver.cpp:
##########
@@ -124,7 +127,7 @@ void partition_resolver::call_task(const rpc_response_task_ptr &t)
}
hdr.gpid = result.pid;
}
- dsn_rpc_call(result.address, t.get());
+ dsn_rpc_call(this->_dns_resolver->resolve_address(result.hp), t.get());
Review Comment:
Coz of design before, all of `dsn_rpc_call` shound be use host_port by `resolve_address.`
It will be performance problem, How could I improve it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1418257488
##########
src/meta/cluster_balance_policy.cpp:
##########
@@ -544,10 +545,21 @@ bool cluster_balance_policy::apply_move(const move_info &move,
// add into migration list and selected_pid
partition_configuration pc;
pc.pid = move.pid;
- pc.primary = primary_addr;
- list[move.pid] = generate_balancer_request(*_global_view->apps, pc, move.type, source, target);
+ pc.hp_primary = primary_hp;
+ std::shared_ptr<dsn::dns_resolver> resolver;
+ if (_svc == nullptr) {
+ resolver = std::make_shared<dsn::dns_resolver>();
Review Comment:
This is a defensive coding, and there are cases in the test where `_svc` is nullptr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1418264398
##########
src/client/replication_ddl_client.cpp:
##########
@@ -81,18 +82,51 @@ namespace replication {
using tp_output_format = ::dsn::utils::table_printer::output_format;
-replication_ddl_client::replication_ddl_client(const std::vector<dsn::rpc_address> &meta_servers)
+replication_ddl_client::replication_ddl_client(const std::vector<dsn::host_port> &meta_servers)
+ : _dns_resolver(new dns_resolver())
{
_meta_server.assign_group("meta-servers");
for (const auto &m : meta_servers) {
- if (!_meta_server.group_address()->add(m)) {
+ if (!_meta_server.group_host_port()->add(m)) {
LOG_WARNING("duplicate adress {}", m);
}
}
}
replication_ddl_client::~replication_ddl_client() { _tracker.cancel_outstanding_tasks(); }
+void replication_ddl_client::set_meta_servers_leader()
Review Comment:
`dsn_rpc_call(dns_resolver->resolve_address(hp), rpc_call);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1434027916
##########
src/meta/load_balance_policy.cpp:
##########
@@ -128,7 +131,9 @@ generate_balancer_request(const app_mapper &apps,
const partition_configuration &pc,
const balance_type &type,
const rpc_address &from,
- const rpc_address &to)
+ const rpc_address &to,
+ const host_port &hp_from,
+ const host_port &hp_to)
Review Comment:
That need another parameter `dns_resolver`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "GehaFearless (via GitHub)" <gi...@apache.org>.
GehaFearless commented on PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#issuecomment-1989882751
ropen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_port for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1519658474
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -61,13 +62,15 @@ namespace dsn {
namespace replication {
class meta_options;
-static void random_assign_partition_config(std::shared_ptr<app_state> &app,
- const std::vector<dsn::rpc_address> &server_list,
- int max_replica_count)
+static void random_assign_partition_config(
+ std::shared_ptr<app_state> &app,
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> &server_list,
Review Comment:
TODO: use a single host_port is enough, then use dns resolver to resolve it.
##########
src/meta/test/meta_bulk_load_service_test.cpp:
##########
@@ -506,6 +522,11 @@ class bulk_load_service_test : public meta_test_base
const rpc_address SECONDARY1 = rpc_address::from_ip_port("127.0.0.1", 10085);
Review Comment:
Resolve from the hosts below.
##########
src/meta/duplication/meta_duplication_service.cpp:
##########
@@ -427,6 +429,10 @@ void meta_duplication_service::check_follower_app_if_create_completed(
p.primary = rpc_address::from_ip_port("127.0.0.1", 34801);
p.secondaries.emplace_back(rpc_address::from_ip_port("127.0.0.2", 34801));
p.secondaries.emplace_back(rpc_address::from_ip_port("127.0.0.3", 34801));
+ p.__set_hp_primary(host_port("localhost", 34801));
+ p.__set_hp_secondaries(std::vector<host_port>());
+ p.hp_secondaries.emplace_back(host_port("localhost", 34802));
+ p.hp_secondaries.emplace_back(host_port("localhost", 34803));
Review Comment:
TODO: Should be the same to line 430~431
##########
src/common/replication_other_types.h:
##########
@@ -49,43 +51,54 @@ typedef int64_t decree;
#define invalid_offset (-1LL)
#define invalid_signature 0
-inline bool is_primary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_primary(const partition_configuration &pc, const host_port &node)
{
- return !node.is_invalid() && pc.primary == node;
+ return !node.is_invalid() && pc.hp_primary == node;
}
-inline bool is_secondary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_secondary(const partition_configuration &pc, const host_port &node)
{
return !node.is_invalid() &&
- std::find(pc.secondaries.begin(), pc.secondaries.end(), node) != pc.secondaries.end();
+ std::find(pc.hp_secondaries.begin(), pc.hp_secondaries.end(), node) !=
+ pc.hp_secondaries.end();
}
-inline bool is_member(const partition_configuration &pc, const rpc_address &node)
+inline bool is_member(const partition_configuration &pc, const host_port &node)
{
return is_primary(pc, node) || is_secondary(pc, node);
}
inline bool is_partition_config_equal(const partition_configuration &pc1,
const partition_configuration &pc2)
{
// secondaries no need to be same order
- for (const rpc_address &addr : pc1.secondaries)
+ for (const host_port &addr : pc1.hp_secondaries)
if (!is_secondary(pc2, addr))
return false;
// 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.secondaries.size() == pc2.secondaries.size() &&
+ 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;
}
class replica_helper
{
public:
- static bool remove_node(::dsn::rpc_address node,
- /*inout*/ std::vector<::dsn::rpc_address> &nodeList);
+ template <typename T>
+ static bool remove_node(const T node,
Review Comment:
TODO: make it as an util function
##########
src/meta/server_state.cpp:
##########
@@ -570,6 +573,9 @@ dsn::error_code server_state::sync_apps_from_remote_storage()
const blob &value) mutable {
if (ec == ERR_OK) {
partition_configuration pc;
+ pc.__isset.hp_secondaries = true;
Review Comment:
TODO: when upgrade from old versions, these fields are not filled.
##########
src/meta/meta_server_failure_detector.cpp:
##########
@@ -281,50 +295,61 @@ bool meta_server_failure_detector::update_stability_stat(const fd::beacon_msg &b
void meta_server_failure_detector::on_ping(const fd::beacon_msg &beacon,
rpc_replier<fd::beacon_ack> &reply)
{
- fd::beacon_ack ack;
- ack.time = beacon.time;
- ack.this_node = beacon.to_addr;
- ack.allowed = true;
+ host_port hp_from_node, hp_to_node;
+ GET_HOST_PORT(beacon, from_node, hp_from_node);
+ GET_HOST_PORT(beacon, to_node, hp_to_node);
if (beacon.__isset.start_time && !update_stability_stat(beacon)) {
- LOG_WARNING("{} is unstable, don't response to it's beacon", beacon.from_addr);
+ LOG_WARNING(
+ "{}({}) is unstable, don't response to it's beacon", beacon.from_node, hp_from_node);
Review Comment:
TODO: hostname fisrt
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -381,10 +392,14 @@ void meta_service_test_app::construct_apps_test()
std::shared_ptr<meta_service> svc(new meta_service());
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
std::string hint_message;
generate_node_list(nodes, 1, 1);
- svc->_state->construct_apps({resp}, nodes, hint_message);
+ std::vector<dsn::host_port> hps;
Review Comment:
TODO: optimize schema like "std::vector<dsn::host_port>"
##########
src/meta/test/misc/misc.cpp:
##########
@@ -382,19 +423,21 @@ void migration_check_and_apply(app_mapper &apps,
dsn::partition_configuration &pc =
the_app->partitions[proposal->gpid.get_partition_index()];
- CHECK(!pc.primary.is_invalid(), "");
- CHECK_EQ(pc.secondaries.size(), 2);
- for (auto &addr : pc.secondaries) {
- CHECK(!addr.is_invalid(), "");
+ CHECK(!pc.hp_primary.is_invalid(), "");
+ CHECK_EQ(pc.hp_secondaries.size(), 2);
+ for (auto &host_port : pc.hp_secondaries) {
Review Comment:
TODO: host_port -> hp
##########
src/replica/split/replica_split_manager.cpp:
##########
@@ -614,10 +618,12 @@ void replica_split_manager::child_notify_catch_up() // on child partition
request->parent_gpid = _replica->_split_states.parent_gpid;
request->child_gpid = get_gpid();
request->child_ballot = get_ballot();
- request->child_address = _stub->_primary_address;
+ request->child = _stub->primary_address();
+ request->__set_hp_child(_stub->primary_host_port());
- LOG_INFO_PREFIX("send notification to primary parent[{}@{}], ballot={}",
+ LOG_INFO_PREFIX("send notification to primary parent[{}@{}({})], ballot={}",
_replica->_split_states.parent_gpid,
+ _replica->_config.hp_primary,
Review Comment:
TODO: unify logs like this to a macro:
```
#define LOG_HOST_PORT_AND_IP(obj, field) \
fmt::format("{}({})", obj.field, obj.hp_##field)
```
##########
src/replica/replica_context.cpp:
##########
@@ -124,19 +125,21 @@ void primary_context::get_replica_config(partition_status::type st,
{
config.pid = membership.pid;
config.primary = membership.primary;
+ config.__set_hp_primary(membership.hp_primary);
config.ballot = membership.ballot;
config.status = st;
config.learner_signature = learner_signature;
}
-bool primary_context::check_exist(::dsn::rpc_address node, partition_status::type st)
+bool primary_context::check_exist(::dsn::host_port node, partition_status::type st)
{
switch (st) {
case partition_status::PS_PRIMARY:
- return membership.primary == node;
+ return membership.hp_primary == node;
case partition_status::PS_SECONDARY:
- return std::find(membership.secondaries.begin(), membership.secondaries.end(), node) !=
- membership.secondaries.end();
+ return std::find(membership.hp_secondaries.begin(),
Review Comment:
TODO: optimize this schema "std::find(...)"
##########
src/meta/test/meta_duplication_service_test.cpp:
##########
@@ -547,9 +551,10 @@ TEST_F(meta_duplication_service_test, duplication_sync)
// generate all primaries on node[0]
for (partition_configuration &pc : app->partitions) {
pc.ballot = random32(1, 10000);
- pc.primary = server_nodes[0];
- pc.secondaries.push_back(server_nodes[1]);
- pc.secondaries.push_back(server_nodes[2]);
+ pc.primary = addr;
+ pc.__set_hp_primary(server_nodes[0]);
+ pc.hp_secondaries.push_back(server_nodes[1]);
Review Comment:
Fill secondaries.
##########
src/meta/test/meta_test_base.cpp:
##########
@@ -161,12 +161,15 @@ std::vector<rpc_address> meta_test_base::ensure_enough_alive_nodes(int min_node_
return nodes;
}
- nodes = generate_node_list(min_node_count);
+ auto node_pairs = generate_node_list(min_node_count);
+ for (const auto &p : node_pairs) {
+ nodes.emplace_back(p.first);
Review Comment:
TODO: optimize it!
##########
src/client/partition_resolver_simple.cpp:
##########
@@ -137,12 +138,12 @@ void partition_resolver_simple::clear_all_pending_requests()
void partition_resolver_simple::on_timeout(request_context_ptr &&rc) const
{
- end_request(std::move(rc), ERR_TIMEOUT, rpc_address(), true);
+ end_request(std::move(rc), ERR_TIMEOUT, host_port(), true);
}
void partition_resolver_simple::end_request(request_context_ptr &&request,
error_code err,
- rpc_address addr,
+ host_port hp,
Review Comment:
TODO: Use constant reference
##########
src/meta/test/misc/misc.h:
##########
@@ -64,18 +65,22 @@ uint32_t random32(uint32_t min, uint32_t max);
// Generates a random number [min_count, max_count] of node addresses
// each node is given a random port value in range of [min_count, max_count]
-void generate_node_list(/*out*/ std::vector<dsn::rpc_address> &output_list,
- int min_count,
- int max_count);
+void generate_node_list(
+ /*out*/ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> &output_list,
+ int min_count,
+ int max_count);
// Generates `size` of node addresses, each with port value in range [start_port, start_port + size]
-inline std::vector<dsn::rpc_address> generate_node_list(size_t size, int start_port = 12321)
+inline std::vector<std::pair<dsn::host_port, dsn::rpc_address>>
+generate_node_list(size_t size, int start_port = 12321)
{
- std::vector<dsn::rpc_address> result;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> result;
result.resize(size);
- for (int i = 0; i < size; ++i)
- result[i] =
+ for (int i = 0; i < size; ++i) {
+ result[i].first = dsn::host_port("localhost", static_cast<uint16_t>(start_port + i + 1));
Review Comment:
TODO: optimize this.
##########
idl/meta_admin.thrift:
##########
@@ -278,8 +282,9 @@ struct query_app_manual_compact_response
struct node_info
{
- 1:node_status status = node_status.NS_INVALID;
- 2:dsn.rpc_address address;
+ 1:node_status status = node_status.NS_INVALID;
+ 2:dsn.rpc_address address;
+ 3:optional dsn.host_port hp_address;
Review Comment:
TODO: address -> node, hp_address -> hp_node
##########
src/client/replication_ddl_client.cpp:
##########
@@ -469,8 +502,10 @@ dsn::error_code replication_ddl_client::list_nodes(
return resp.err;
}
- for (const dsn::replication::node_info &n : resp.infos) {
- nodes[n.address] = n.status;
+ for (const auto &n : resp.infos) {
+ host_port hp;
+ GET_HOST_PORT(n, address, hp);
Review Comment:
TODO: rename vars in node_info, address -> node, hp_address -> hp_node
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -149,6 +150,8 @@ bool rpc_client_matcher::on_recv_reply(network *net, uint64_t key, message_ex *r
case GRPC_TO_LEADER:
if (req->server_address.group_address()->is_update_leader_automatically()) {
req->server_address.group_address()->set_leader(addr);
+ req->server_host_port.group_host_port()->set_leader(
+ host_port::from_address(addr));
Review Comment:
TODO: can it resolve a host_port from reply in line 136?
##########
src/replica/bulk_load/test/replica_bulk_loader_test.cpp:
##########
@@ -413,6 +421,9 @@ class replica_bulk_loader_test : public replica_test_base
rpc_address PRIMARY = rpc_address::from_ip_port("127.0.0.2", 34801);
rpc_address SECONDARY = rpc_address::from_ip_port("127.0.0.3", 34801);
rpc_address SECONDARY2 = rpc_address::from_ip_port("127.0.0.4", 34801);
+ const host_port PRIMARY_HP = host_port("localhost", 34801);
Review Comment:
TODO: use the resolved IPs from the hosts
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -209,4 +216,13 @@ error_s host_port::resolve_addresses(std::vector<rpc_address> &addresses) const
return error_s::ok();
}
+void host_port::fill_host_ports_from_addresses(const std::vector<rpc_address> &addr_v,
Review Comment:
TODO: make it as a util func
##########
src/failure_detector/failure_detector.cpp:
##########
@@ -302,28 +304,28 @@ void failure_detector::check_all_records()
}
}
-void failure_detector::add_allow_list(::dsn::rpc_address node)
+void failure_detector::add_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
_allow_list.insert(node);
}
-bool failure_detector::remove_from_allow_list(::dsn::rpc_address node)
+bool failure_detector::remove_from_allow_list(::dsn::host_port node)
{
zauto_lock l(_lock);
return _allow_list.erase(node) > 0;
}
-void failure_detector::set_allow_list(const std::vector<std::string> &replica_addrs)
+void failure_detector::set_allow_list(const std::vector<std::string> &replica_hps)
{
CHECK(!_is_started, "FD is already started, the allow list should really not be modified");
- std::vector<rpc_address> nodes;
- for (const auto &addr : replica_addrs) {
- const auto node = dsn::rpc_address::from_host_port(addr);
+ std::vector<host_port> nodes;
+ for (auto &hp : replica_hps) {
Review Comment:
TODO: search all to use `const auto &`
##########
src/runtime/rpc/serialization.h:
##########
@@ -101,4 +104,7 @@ inline void unmarshall(dsn::message_ex *msg, /*out*/ T &val)
unmarshall(reader, val, (dsn_msg_serialize_format)msg->header->context.u.serialize_format);
}
+template <>
+inline void unmarshall(dsn::message_ex *msg, /*out*/ partition_configuration &val);
Review Comment:
TODO: remove it?
##########
idl/bulk_load.thrift:
##########
@@ -89,15 +89,16 @@ struct partition_bulk_load_state
// meta server -> replica server
struct bulk_load_request
{
- 1:dsn.gpid pid;
- 2:string app_name;
- 3:dsn.rpc_address primary_addr;
- 4:string remote_provider_name;
- 5:string cluster_name;
- 6:i64 ballot;
- 7:bulk_load_status meta_bulk_load_status;
- 8:bool query_bulk_load_metadata;
- 9:string remote_root_path;
+ 1:dsn.gpid pid;
+ 2:string app_name;
+ 3:dsn.rpc_address primary_addr;
+ 4:string remote_provider_name;
+ 5:string cluster_name;
+ 6:i64 ballot;
+ 7:bulk_load_status meta_bulk_load_status;
+ 8:bool query_bulk_load_metadata;
+ 9:string remote_root_path;
+ 10:optional dsn.host_port hp_primary;
Review Comment:
TODO: primary_addr -> primary
##########
src/common/replication_other_types.h:
##########
@@ -49,43 +51,54 @@ typedef int64_t decree;
#define invalid_offset (-1LL)
#define invalid_signature 0
-inline bool is_primary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_primary(const partition_configuration &pc, const host_port &node)
{
- return !node.is_invalid() && pc.primary == node;
+ return !node.is_invalid() && pc.hp_primary == node;
}
-inline bool is_secondary(const partition_configuration &pc, const rpc_address &node)
+inline bool is_secondary(const partition_configuration &pc, const host_port &node)
{
return !node.is_invalid() &&
- std::find(pc.secondaries.begin(), pc.secondaries.end(), node) != pc.secondaries.end();
+ std::find(pc.hp_secondaries.begin(), pc.hp_secondaries.end(), node) !=
+ pc.hp_secondaries.end();
}
-inline bool is_member(const partition_configuration &pc, const rpc_address &node)
+inline bool is_member(const partition_configuration &pc, const host_port &node)
{
return is_primary(pc, node) || is_secondary(pc, node);
}
inline bool is_partition_config_equal(const partition_configuration &pc1,
const partition_configuration &pc2)
{
// secondaries no need to be same order
- for (const rpc_address &addr : pc1.secondaries)
+ for (const host_port &addr : pc1.hp_secondaries)
Review Comment:
TODO: const auto ...
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -195,8 +204,12 @@ void meta_partition_guardian_test::cure_test()
ASSERT_TRUE(state->spin_wait_staging(20));
svc->_started = true;
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
Review Comment:
TODO: duplicate in many places.
##########
src/meta/server_state.cpp:
##########
@@ -1850,34 +1954,40 @@ void server_state::downgrade_secondary_to_inactive(std::shared_ptr<app_state> &a
void server_state::downgrade_stateless_nodes(std::shared_ptr<app_state> &app,
int pidx,
- const rpc_address &address)
+ const host_port &address)
{
std::shared_ptr<configuration_update_request> req =
std::make_shared<configuration_update_request>();
req->info = *app;
req->type = config_type::CT_REMOVE;
- req->host_node = address;
+ req->host_node = dsn::dns_resolver::instance().resolve_address(address);
req->node.set_invalid();
+ req->hp_node.reset();
req->config = app->partitions[pidx];
config_context &cc = app->helpers->contexts[pidx];
partition_configuration &pc = req->config;
unsigned i = 0;
- for (; i < pc.secondaries.size(); ++i) {
- if (pc.secondaries[i] == address) {
+ for (; i < pc.hp_secondaries.size(); ++i) {
+ if (pc.hp_secondaries[i] == address) {
req->node = pc.last_drops[i];
+ req->__set_hp_node(pc.hp_last_drops[i]);
break;
}
}
CHECK(!req->node.is_invalid(), "invalid node address, address = {}", req->node);
// remove host_node & node from secondaries/last_drops, as it will be sync to remote storage
- for (++i; i < pc.secondaries.size(); ++i) {
+ for (++i; i < pc.hp_secondaries.size(); ++i) {
Review Comment:
TODO: check the size of hp_secondaries and secondaries equals.
##########
src/meta/server_state.cpp:
##########
@@ -1443,39 +1472,84 @@ void server_state::request_check(const partition_configuration &old,
switch (request.type) {
case config_type::CT_ASSIGN_PRIMARY:
- CHECK_NE(old.primary, request.node);
- CHECK(std::find(old.secondaries.begin(), old.secondaries.end(), request.node) ==
- old.secondaries.end(),
- "");
+ if (request.__isset.hp_node) {
+ CHECK_NE(old.hp_primary, request.hp_node);
+ CHECK(std::find(old.hp_secondaries.begin(),
Review Comment:
TODO: Add an util function
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
Review Comment:
TODO: reduce the parameters, remove rpc_addresses
##########
src/failure_detector/test/failure_detector.cpp:
##########
@@ -236,7 +241,8 @@ class test_master : public service_app
utils::split_args(args[2].c_str(), ports, ',');
for (auto &port : ports) {
rpc_address addr(network::get_local_ipv4(), std::stoi(port));
- _master_fd->add_allow_list(addr);
+ const auto hp = ::dsn::host_port::from_address(addr);
Review Comment:
TODO: use local host directly.
##########
src/meta/test/misc/misc.cpp:
##########
@@ -58,12 +61,15 @@ uint32_t random32(uint32_t min, uint32_t max)
return res + min;
}
-void generate_node_list(std::vector<dsn::rpc_address> &output_list, int min_count, int max_count)
+void generate_node_list(std::vector<std::pair<dsn::host_port, dsn::rpc_address>> &output_list,
+ int min_count,
+ int max_count)
{
const auto count = random32(min_count, max_count);
output_list.resize(count);
- for (auto i = 0; i < count; ++i) {
- output_list[i] = dsn::rpc_address::from_ip_port("127.0.0.1", i + 1);
+ for (int i = 0; i < count; ++i) {
+ output_list[i] = std::make_pair(dsn::host_port("localhost", i + 1),
Review Comment:
TODO: use resolved IP
##########
src/failure_detector/test/failure_detector.cpp:
##########
@@ -652,19 +658,21 @@ TEST(fd, update_stability)
dsn::rpc_replier<beacon_ack> r(create_fake_rpc_response());
beacon_msg msg;
- msg.from_addr = rpc_address::from_host_port("localhost", 123);
- msg.to_addr = rpc_address::from_host_port("localhost", MPORT_START);
+ msg.from_node = rpc_address::from_host_port("localhost", 123);
Review Comment:
TODO: resolve and reuse the hosts
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
+ nodes_list[0].second,
+ dsn::host_port(),
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 3: test invalid proposal: invalid node" << std::endl;
- cpa2 =
- new_proposal_action(node_list[0], dsn::rpc_address(), config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ dsn::rpc_address(),
+ nodes_list[0].first,
+ dsn::host_port(),
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 4: test invalid proposal: dead target" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[0], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[0], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(true);
std::cerr << "Case 5: test invalid proposal: dead node" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[1], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[1], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(true);
std::cerr << "Case 6: test invalid proposal: already have priamry but assign" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_ASSIGN_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_ASSIGN_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[1];
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 7: test invalid proposal: upgrade non-secondary" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
pc.primary.set_invalid();
+ pc.hp_primary.reset();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 8: test invalid proposal: add exist secondary" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries = {node_list[1]};
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
+ pc.secondaries = {nodes_list[1].second};
+ pc.__set_hp_secondaries({nodes_list[1].first});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 9: test invalid proposal: downgrade non member" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_REMOVE);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_REMOVE);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
+ pc.hp_secondaries.clear();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 10: test abnormal learning detect" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_ADD_SECONDARY);
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
+ pc.hp_secondaries.clear();
Review Comment:
TODO: secondaries.clear()
##########
src/meta/test/meta_partition_guardian_test.cpp:
##########
@@ -859,69 +897,109 @@ void meta_partition_guardian_test::from_proposal_test()
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 2: test invalid proposal: invalid target" << std::endl;
- cpa2 =
- new_proposal_action(dsn::rpc_address(), node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(dsn::rpc_address(),
+ nodes_list[0].second,
+ dsn::host_port(),
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 3: test invalid proposal: invalid node" << std::endl;
- cpa2 =
- new_proposal_action(node_list[0], dsn::rpc_address(), config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ dsn::rpc_address(),
+ nodes_list[0].first,
+ dsn::host_port(),
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 4: test invalid proposal: dead target" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[0], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[0], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[0].first, false)->set_alive(true);
std::cerr << "Case 5: test invalid proposal: dead node" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- get_node_state(nodes, node_list[1], false)->set_alive(false);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(false);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
- get_node_state(nodes, node_list[1], false)->set_alive(true);
+ get_node_state(nodes, nodes_list[1].first, false)->set_alive(true);
std::cerr << "Case 6: test invalid proposal: already have priamry but assign" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_ASSIGN_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_ASSIGN_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[1];
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 7: test invalid proposal: upgrade non-secondary" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[0], config_type::CT_UPGRADE_TO_PRIMARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[0].second,
+ nodes_list[0].first,
+ nodes_list[0].first,
+ config_type::CT_UPGRADE_TO_PRIMARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
pc.primary.set_invalid();
+ pc.hp_primary.reset();
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 8: test invalid proposal: add exist secondary" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_ADD_SECONDARY);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_ADD_SECONDARY);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries = {node_list[1]};
+ pc.primary = nodes_list[1].second;
+ pc.__set_hp_primary(nodes_list[1].first);
+ pc.secondaries = {nodes_list[1].second};
+ pc.__set_hp_secondaries({nodes_list[1].first});
ASSERT_FALSE(guardian.from_proposals(mv, p, cpa));
ASSERT_EQ(config_type::CT_INVALID, cpa.type);
std::cerr << "Case 9: test invalid proposal: downgrade non member" << std::endl;
- cpa2 = new_proposal_action(node_list[0], node_list[1], config_type::CT_REMOVE);
+ cpa2 = new_proposal_action(nodes_list[0].second,
+ nodes_list[1].second,
+ nodes_list[0].first,
+ nodes_list[1].first,
+ config_type::CT_REMOVE);
cc.lb_actions.assign_balancer_proposals({cpa2});
- pc.primary = node_list[0];
- pc.secondaries.clear();
+ pc.primary = nodes_list[0].second;
+ pc.__set_hp_primary(nodes_list[0].first);
Review Comment:
TODO: secondaries.clear()
##########
src/meta/test/meta_split_service_test.cpp:
##########
@@ -377,7 +378,8 @@ class meta_split_service_test : public meta_test_base
const int32_t PARENT_BALLOT = 3;
const int32_t PARENT_INDEX = 0;
const int32_t CHILD_INDEX = 4;
- const rpc_address NODE = rpc_address::from_ip_port("127.0.0.1", 10086);
+ const host_port NODE = host_port("localhost", 10086);
+ const rpc_address NODE_ADDR = rpc_address::from_host_port("127.0.0.1", 10086);
Review Comment:
TODO: use the resolved IP.
##########
src/meta/test/update_configuration_test.cpp:
##########
@@ -238,52 +247,54 @@ void meta_service_test_app::update_configuration_test()
ss->_all_apps.emplace(1, app);
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
Review Comment:
TODO: optimize schema like "std::vector<std::pair<dsn::host_port, dsn::rpc_address>>".
##########
src/redis_protocol/proxy_lib/proxy_layer.h:
##########
@@ -79,8 +79,9 @@ class proxy_session : public std::enable_shared_from_this<proxy_session>
// when get message from raw parser, request & response of "dsn::message_ex*" are not in couple.
// we need to backup one request to create a response struct.
dsn::message_ex *_backup_one_request;
- // the client address for which this session served
- dsn::rpc_address _remote_address;
+ // the client for which this session served
Review Comment:
TODO: add host_port
##########
src/nfs/nfs_client_impl.cpp:
##########
@@ -347,14 +355,17 @@ void nfs_client_impl::end_copy(::dsn::error_code err,
METRIC_VAR_INCREMENT(nfs_client_copy_failed_requests);
if (!fc->user_req->is_finished) {
+ host_port hp = fc->user_req->file_size_req.hp_source;
Review Comment:
TODO: unify style
##########
src/replica/replica_learn.cpp:
##########
@@ -1096,19 +1142,21 @@ void replica::on_copy_remote_state_completed(error_code err,
auto start_ts = dsn_now_ns();
err = apply_learned_state_from_private_log(lstate);
if (err == ERR_OK) {
- LOG_INFO_PREFIX("on_copy_remote_state_completed[{:#018x}]: learnee = {}, "
+ LOG_INFO_PREFIX("on_copy_remote_state_completed[{:#018x}]: learnee = {}({}), "
"learn_duration = {} ms, apply_log_duration = {} ns, apply learned "
"state from private log succeed, app_committed_decree = {}",
req.signature,
+ resp.config.hp_primary,
resp.config.primary,
_potential_secondary_states.duration_ms(),
dsn_now_ns() - start_ts,
_app->last_committed_decree());
} else {
- LOG_ERROR_PREFIX("on_copy_remote_state_completed[{:#018x}]: learnee = {}, "
+ LOG_ERROR_PREFIX("on_copy_remote_state_completed[{:#018x}]: learnee = {}({}), "
"learn_duration = {} ms, apply_log_duration = {} ns, apply "
"learned state from private log failed, err = {}",
req.signature,
+ resp.config.hp_primary,
Review Comment:
TODO: reverse the host and IP order
##########
src/replica/split/test/replica_split_test.cpp:
##########
@@ -525,9 +535,12 @@ class replica_split_test : public replica_test_base
const int32_t APP_ID = 2;
const int32_t OLD_PARTITION_COUNT = 8;
const int32_t NEW_PARTITION_COUNT = 16;
- const rpc_address PRIMARY = rpc_address::from_ip_port("127.0.0.1", 18230);
- const rpc_address SECONDARY = rpc_address::from_ip_port("127.0.0.2", 10058);
- const rpc_address SECONDARY2 = rpc_address::from_ip_port("127.0.0.3", 10805);
+ const host_port PRIMARY = host_port("localhost", 18230);
Review Comment:
TODO: use resolve ones
##########
src/replica/duplication/replica_follower.h:
##########
@@ -78,9 +79,12 @@ class replica_follower : replica_base
std::string master_replica_name()
{
std::string app_info = fmt::format("{}.{}", _master_cluster_name, _master_app_name);
- if (_master_replica_config.primary != rpc_address::s_invalid_address) {
- return fmt::format(
- "{}({}|{})", app_info, _master_replica_config.primary, _master_replica_config.pid);
+ if (_master_replica_config.hp_primary != host_port::s_invalid_host_port) {
Review Comment:
TODO: s_invalid_host_port vs is_invalid()
##########
src/runtime/rpc/network.sim.h:
##########
@@ -91,6 +92,7 @@ class sim_network_provider : public connection_oriented_network
virtual error_code start(rpc_channel channel, int port, bool client_only);
virtual ::dsn::rpc_address address() { return _address; }
Review Comment:
TODO: const & xxx() const
##########
src/replica/replica_config.cpp:
##########
@@ -257,15 +267,20 @@ void replica::downgrade_to_secondary_on_primary(configuration_update_request &pr
return;
CHECK_EQ(proposal.config.pid, _primary_states.membership.pid);
- CHECK_EQ(proposal.config.primary, _primary_states.membership.primary);
- CHECK(proposal.config.secondaries == _primary_states.membership.secondaries, "");
- CHECK_EQ(proposal.node, proposal.config.primary);
+ CHECK_EQ(proposal.config.hp_primary, _primary_states.membership.hp_primary);
+ CHECK(proposal.config.hp_secondaries == _primary_states.membership.hp_secondaries, "");
+ CHECK_EQ(proposal.hp_node, proposal.config.hp_primary);
proposal.config.primary.set_invalid();
+ proposal.config.__set_hp_primary(host_port());
proposal.config.secondaries.push_back(proposal.node);
+ if (!proposal.config.__isset.hp_secondaries) {
Review Comment:
TODO: refactor this schema one by one
1. find out new newly added hp_xxx fields
2. check the usage
##########
src/replica/test/replica_test.cpp:
##########
@@ -271,6 +271,7 @@ TEST_P(replica_test, write_size_limited)
auto write_request = dsn::message_ex::create_request(RPC_TEST);
auto cleanup = dsn::defer([=]() { delete write_request; });
+ header.context.u.is_forwarded = false;
Review Comment:
TODO: why change it?
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -177,6 +180,8 @@ bool rpc_client_matcher::on_recv_reply(network *net, uint64_t key, message_ex *r
req->server_address.group_address()->is_update_leader_automatically()) {
req->server_address.group_address()->set_leader(
reply->header->from_address);
+ req->server_host_port.group_host_port()->set_leader(
+ host_port::from_address(reply->header->from_address));
Review Comment:
TODO: add a from_host_port to header?
##########
src/replica/storage/simple_kv/simple_kv.app.example.h:
##########
@@ -45,8 +45,8 @@ class simple_kv_client_app : public ::dsn::service_app
return ::dsn::ERR_INVALID_PARAMETERS;
printf("%s %s %s\n", args[1].c_str(), args[2].c_str(), args[3].c_str());
- const auto meta = rpc_address::from_host_port(args[2]);
- _simple_kv_client.reset(new simple_kv_client(args[1].c_str(), {meta}, args[3].c_str()));
+ const auto hp = host_port::from_string(args[2].c_str());
Review Comment:
TODO: remove c_str
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -42,8 +43,11 @@ const host_port host_port::s_invalid_host_port;
host_port::host_port(std::string host, uint16_t port)
: _host(std::move(host)), _port(port), _type(HOST_TYPE_IPV4)
{
- // ipv4_from_host may be slow, just call it in DEBUG version.
- DCHECK_OK(rpc_address::ipv4_from_host(_host, nullptr), "invalid hostname: {}", _host);
+ // Solve the problem of not translating "0.0.0.0"
+ if (_host != "0.0.0.0") {
Review Comment:
TODO: check it
##########
src/runtime/rpc/rpc_host_port.cpp:
##########
@@ -169,11 +173,14 @@ error_s host_port::resolve_addresses(std::vector<rpc_address> &addresses) const
__builtin_unreachable();
}
- // 1. Try to resolve hostname in the form of "localhost:80" or "192.168.0.1:8080".
- const auto rpc_addr = rpc_address::from_ip_port(this->to_string());
- if (rpc_addr) {
- addresses.emplace_back(rpc_addr);
- return error_s::ok();
+ // 1. Try to resolve hostname in the form of "192.168.0.1:8080".
+ uint32_t ip_addr;
+ if (inet_pton(AF_INET, this->_host.c_str(), &ip_addr)) {
Review Comment:
TODO: check it
##########
src/runtime/rpc/rpc_host_port.h:
##########
@@ -67,6 +77,8 @@ class host_port
[[nodiscard]] bool is_invalid() const { return _type == HOST_TYPE_INVALID; }
+ operator bool() const { return !is_invalid(); }
Review Comment:
TODO: consider to remove it
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -517,9 +522,11 @@ error_code rpc_engine::start(const service_app_spec &aspec)
_local_primary_address = _client_nets[NET_HDR_DSN][0]->address();
_local_primary_address.set_port(aspec.ports.size() > 0 ? *aspec.ports.begin() : aspec.id);
+ _local_primary_host_port = host_port::from_address(_local_primary_address);
Review Comment:
TODO: Is it equal to _client_nets[NET_HDR_DSN][0]->host_port() ?
##########
src/test/function_test/recovery/test_recovery.cpp:
##########
@@ -71,12 +70,15 @@ class recovery_test : public test_util
// cluster has only one meta server, while "onebox" means the cluster has 3 meta servers.
recovery_test() : test_util(std::map<std::string, std::string>(), "single_master_cluster") {}
- std::vector<dsn::rpc_address> get_rpc_address_list(const std::vector<uint16_t> &ports)
+ std::vector<dsn::host_port> get_rpc_host_port_list(const std::vector<int> ports)
{
- std::vector<dsn::rpc_address> result;
+ std::vector<dsn::host_port> result;
result.reserve(ports.size());
for (const auto &port : ports) {
- result.emplace_back(dsn::rpc_address(global_env::instance()._host_ip, port));
+ char hostname[1024];
+ gethostname(hostname, 1024);
Review Comment:
TODO: change _host_ip and use it
##########
src/shell/command_utils.cpp:
##########
@@ -22,29 +22,29 @@
#include "client/replication_ddl_client.h"
#include "command_executor.h"
#include "meta_admin_types.h"
-#include "runtime/rpc/rpc_address.h"
+#include "runtime/rpc/rpc_host_port.h"
#include "utils/error_code.h"
bool validate_ip(shell_context *sc,
const std::string &ip_str,
- dsn::rpc_address &target_address,
+ dsn::host_port &target_hp,
std::string &err_info)
{
- target_address = dsn::rpc_address::from_ip_port(ip_str);
- if (!target_address) {
- err_info = fmt::format("invalid ip:port={}, can't transform it into rpc_address", ip_str);
+ target_hp = dsn::host_port::from_string(ip_str);
Review Comment:
TODO: use rpc_address or change the func name to validate_host_port
##########
src/test/function_test/utils/test_util.cpp:
##########
@@ -89,6 +89,7 @@ void test_util::SetUp()
ddl_client_ = std::make_shared<replication_ddl_client>(meta_list_);
ASSERT_TRUE(ddl_client_ != nullptr);
ddl_client_->set_max_wait_app_ready_secs(120);
+ ddl_client_->set_meta_servers_leader();
Review Comment:
TODO: is it necessary?
##########
src/server/config.min.ini:
##########
@@ -43,7 +43,7 @@
logging_start_level = LOG_LEVEL_INFO
[network]
- primary_interface = lo
+ primary_interface = @LOCAL_HOSTNAME@
Review Comment:
TODO: support not change configs when upgrade version
##########
src/shell/commands/table_management.cpp:
##########
@@ -378,11 +377,11 @@ bool app_disk(command_executor *e, shell_context *sc, arguments args)
}
std::stringstream oss;
std::string hostname;
- std::string ip = p.primary.to_string();
+ const auto &ip = p.hp_primary.to_string();
Review Comment:
TODO: fix the resolve_ip
##########
src/server/pegasus_server_impl_init.cpp:
##########
@@ -629,7 +629,7 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_positives),
METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_true_positives)
{
- _primary_address = dsn::rpc_address(dsn_primary_address()).to_string();
+ _primary_address = dsn_primary_host_port().to_string();
Review Comment:
TODO: _primary_address -> _primary_host_port
##########
src/shell/commands/rebalance.cpp:
##########
@@ -112,12 +114,12 @@ bool propose(command_executor *e, shell_context *sc, arguments args)
proposal_type += optarg;
break;
case 't':
- target = dsn::rpc_address::from_host_port(optarg);
- PRINT_AND_RETURN_FALSE_IF_NOT(target, "parse {} as target_address failed\n", optarg);
+ target = dsn::host_port::from_string(optarg);
Review Comment:
TODO: check both hostname and IP are accepted
##########
src/shell/commands/node_management.cpp:
##########
@@ -666,9 +666,9 @@ bool remote_command(command_executor *e, shell_context *sc, arguments args)
node_desc &n = node_list[i];
std::string hostname;
if (resolve_ip) {
- dsn::utils::hostname_from_ip_port(n.address.to_string(), &hostname);
+ hostname = dsn::dns_resolver::instance().resolve_address(n.hp).to_string();
Review Comment:
TODO: fix the reversed
##########
src/shell/commands/detect_hotkey.cpp:
##########
@@ -100,10 +100,10 @@ bool detect_hotkey(command_executor *e, shell_context *sc, arguments args)
return false;
}
- dsn::rpc_address target_address;
+ dsn::host_port target_hp;
std::string err_info;
std::string ip_str = cmd({"-d", "--address"}).str();
- if (!validate_ip(sc, ip_str, target_address, err_info)) {
+ if (!validate_ip(sc, ip_str, target_hp, err_info)) {
Review Comment:
TODO: it means only host_port is accepted?
##########
src/shell/commands/rebalance.cpp:
##########
@@ -194,20 +201,31 @@ bool balance(command_executor *e, shell_context *sc, arguments args)
std::vector<configuration_proposal_action> &actions = request.action_list;
actions.reserve(4);
+ const auto &from_addr = dsn::dns_resolver::instance().resolve_address(from);
+ const auto &to_addr = dsn::dns_resolver::instance().resolve_address(to);
if (balance_type == "move_pri") {
+ actions.emplace_back(new_proposal_action(
Review Comment:
TODO: reduce parameters
##########
src/test/function_test/config.ini:
##########
@@ -75,8 +75,8 @@ rpc_timeout_milliseconds = 5000
lb_interval_ms = 3000
[pegasus.clusters]
-onebox = 127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603
-single_master_cluster = 127.0.0.1:34601
+onebox = @LOCAL_HOSTNAME@:34601,@LOCAL_HOSTNAME@:34602,@LOCAL_HOSTNAME@:34603
Review Comment:
TODO: check ip is acceptable as well
##########
src/runtime/test/rpc.cpp:
##########
@@ -224,8 +225,8 @@ TEST(core, group_address_no_response_2)
TEST(core, send_to_invalid_address)
Review Comment:
TODO: add test to send to invalid hp group
##########
src/runtime/rpc/asio_net_provider.cpp:
##########
@@ -356,7 +370,23 @@ void asio_udp_provider::do_receive()
return;
}
+ if (msg->header->from_address != remote_addr) {
Review Comment:
TODO: check it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Support to communicate with servers and clients by host_port [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1521183995
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -381,10 +392,14 @@ void meta_service_test_app::construct_apps_test()
std::shared_ptr<meta_service> svc(new meta_service());
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
std::string hint_message;
generate_node_list(nodes, 1, 1);
- svc->_state->construct_apps({resp}, nodes, hint_message);
+ std::vector<dsn::host_port> hps;
Review Comment:
TODO: optimize schema like "std::vector <dsn::host_port>"
##########
src/meta/test/state_sync_test.cpp:
##########
@@ -381,10 +392,14 @@ void meta_service_test_app::construct_apps_test()
std::shared_ptr<meta_service> svc(new meta_service());
- std::vector<dsn::rpc_address> nodes;
+ std::vector<std::pair<dsn::host_port, dsn::rpc_address>> nodes;
std::string hint_message;
generate_node_list(nodes, 1, 1);
- svc->_state->construct_apps({resp}, nodes, hint_message);
+ std::vector<dsn::host_port> hps;
Review Comment:
TODO: optimize schema like "std::vector\<dsn::host_port>"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
Re: [PR] feat(FQDN): Replace rpc_address to host_prot for function & Add host_port on thrift struct [incubator-pegasus]
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1658:
URL: https://github.com/apache/incubator-pegasus/pull/1658#discussion_r1464343255
##########
src/runtime/rpc/asio_net_provider.cpp:
##########
@@ -356,7 +370,23 @@ void asio_udp_provider::do_receive()
return;
}
+ if (msg->header->from_address != remote_addr) {
Review Comment:
Why add this? It seems not related to FQDN.
##########
src/server/test/config.ini:
##########
@@ -493,15 +493,15 @@ profiler::cancelled = false
;profiler::timeout.qps = false
[meta_server]
-server_list = 0.0.0.0:34701
+server_list = 127.0.0.1:34701
Review Comment:
Why change this?
If the default behavior has changed to treat the server_list as hostnames, and it doesn't work if they are IP addresses, how about introducing a new config, e.g. `hostname_server_list`, to make it compatiable when user upgrade Pegasus from old versions, so they don't needed to change configs when upgrade.
Other places are the same.
##########
src/runtime/rpc/rpc_engine.cpp:
##########
@@ -131,8 +133,10 @@ bool rpc_client_matcher::on_recv_reply(network *net, uint64_t key, message_ex *r
// if this is pure client (no server port assigned), we can only do fake forwarding,
// in this case, the server will return ERR_FORWARD_TO_OTHERS
if (err == ERR_FORWARD_TO_OTHERS) {
- rpc_address addr;
- ::dsn::unmarshall((dsn::message_ex *)reply, addr);
+ host_port hp;
Review Comment:
What happen if use rpc_address here and in line 857 (I guess they are in pair, right?) ?
##########
src/runtime/rpc/rpc_message.h:
##########
@@ -137,6 +138,8 @@ class message_ex : public ref_counter, public extensible_object<message_ex, 4>
rpc_session_ptr io_session; // send/recv session
rpc_address to_address; // always ipv4/v6 address, it is the to_node's net address
rpc_address server_address; // used by requests, and may be of uri/group address
+ host_port to_host_port; // fqdn from to_address, older versions use fqdn as net
Review Comment:
What does "older versions use fqdn as net" mean?
##########
src/server/config.ini:
##########
@@ -94,6 +94,7 @@
io_service_worker_count = 4
; how many connections can be established from one ip address to a server(both replica and meta), 0 means no threshold
conn_threshold_per_ip = 0
+ use_cache_resolve_hostname = true
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org