wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-22 13:00:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

[GCS]Optimize gcs client reconnection (#12878)

Browse Source 
* [GCS]Optimize gcs client reconnection

* fix review comment

* fix review comment

* add part code

Co-authored-by: 灵洵 <fengbin.ffb@antgroup.com>
This commit is contained in:
fangfengbin
2020-12-17 21:57:37 -08:00
committed by GitHub
co-authored by 灵洵
parent cfefd7c70e
commit a442cd17e0
4 changed files with 59 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/ray_config_def.h
+3
View File
@@ -266,6 +266,9 @@ RAY_CONFIG(int64_t, ping_gcs_rpc_server_interval_milliseconds, 1000)
/// Maximum number of times to retry ping gcs rpc server when gcs server restarts.
RAY_CONFIG(int32_t, ping_gcs_rpc_server_max_retries, 1)
/// Minimum interval between reconnecting gcs rpc server when gcs server restarts.
RAY_CONFIG(int32_t, minimum_gcs_reconnect_interval_milliseconds, 5000)
/// Whether start the Plasma Store as a Raylet thread.
RAY_CONFIG(bool, plasma_store_as_thread, false)
src/ray/gcs/gcs_client/service_based_accessor.cc
+28 -9
View File
@@ -90,17 +90,23 @@ Status ServiceBasedJobInfoAccessor::AsyncSubscribeAll(
void ServiceBasedJobInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
RAY_LOG(DEBUG) << "Reestablishing subscription for job info.";
auto fetch_all_done = [](const Status &status) {
RAY_LOG(INFO) << "Finished fetching all job information from gcs server after gcs "
"server or pub-sub server is restarted.";
};
// If only the GCS sever has restarted, we only need to fetch data from the GCS server.
// If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
// server first, then fetch data from the GCS server.
if (is_pubsub_server_restarted) {
if (subscribe_operation_ != nullptr) {
RAY_CHECK_OK(subscribe_operation_(
[this](const Status &status) { fetch_all_data_operation_(nullptr); }));
RAY_CHECK_OK(subscribe_operation_([this, fetch_all_done](const Status &status) {
fetch_all_data_operation_(fetch_all_done);
}));
}
} else {
if (fetch_all_data_operation_ != nullptr) {
fetch_all_data_operation_(nullptr);
fetch_all_data_operation_(fetch_all_done);
}
}
}
@@ -301,14 +307,20 @@ Status ServiceBasedActorInfoAccessor::AsyncUnsubscribe(const ActorID &actor_id)
void ServiceBasedActorInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
RAY_LOG(DEBUG) << "Reestablishing subscription for actor info.";
auto fetch_all_done = [](const Status &status) {
RAY_LOG(INFO) << "Finished fetching all actor information from gcs server after gcs "
"server or pub-sub server is restarted.";
};
// If only the GCS sever has restarted, we only need to fetch data from the GCS server.
// If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
// server first, then fetch data from the GCS server.
absl::MutexLock lock(&mutex_);
if (is_pubsub_server_restarted) {
if (subscribe_all_operation_ != nullptr) {
RAY_CHECK_OK(subscribe_all_operation_(
[this](const Status &status) { fetch_all_data_operation_(nullptr); }));
RAY_CHECK_OK(subscribe_all_operation_([this, fetch_all_done](const Status &status) {
fetch_all_data_operation_(fetch_all_done);
}));
}
for (auto &item : subscribe_operations_) {
auto &actor_id = item.first;
@@ -325,7 +337,7 @@ void ServiceBasedActorInfoAccessor::AsyncResubscribe(bool is_pubsub_server_resta
}
} else {
if (fetch_all_data_operation_ != nullptr) {
fetch_all_data_operation_(nullptr);
fetch_all_data_operation_(fetch_all_done);
}
for (auto &item : fetch_data_operations_) {
item.second(nullptr);
@@ -651,20 +663,27 @@ void ServiceBasedNodeInfoAccessor::HandleNotification(const GcsNodeInfo &node_in
void ServiceBasedNodeInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
RAY_LOG(DEBUG) << "Reestablishing subscription for node info.";
auto fetch_all_done = [](const Status &status) {
RAY_LOG(INFO) << "Finished fetching all node information from gcs server after gcs "
"server or pub-sub server is restarted.";
};
// If only the GCS sever has restarted, we only need to fetch data from the GCS server.
// If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
// server first, then fetch data from the GCS server.
if (is_pubsub_server_restarted) {
if (subscribe_node_operation_ != nullptr) {
RAY_CHECK_OK(subscribe_node_operation_(
[this](const Status &status) { fetch_node_data_operation_(nullptr); }));
RAY_CHECK_OK(
subscribe_node_operation_([this, fetch_all_done](const Status &status) {
fetch_node_data_operation_(fetch_all_done);
}));
}
if (subscribe_batch_resource_usage_operation_ != nullptr) {
RAY_CHECK_OK(subscribe_batch_resource_usage_operation_(nullptr));
}
} else {
if (fetch_node_data_operation_ != nullptr) {
fetch_node_data_operation_(nullptr);
fetch_node_data_operation_(fetch_all_done);
}
}
}
src/ray/gcs/gcs_client/service_based_gcs_client.cc
+26 -6
View File
@@ -25,7 +25,9 @@ namespace ray {
namespace gcs {
ServiceBasedGcsClient::ServiceBasedGcsClient(const GcsClientOptions &options)
: GcsClient(options) {}
: GcsClient(options),
last_reconnect_timestamp_ms_(0),
last_reconnect_address_(std::make_pair("", -1)) {}
Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
RAY_CHECK(!is_connected_);
@@ -175,7 +177,7 @@ void ServiceBasedGcsClient::GcsServiceFailureDetected(rpc::GcsServiceFailureType
ReconnectGcsServer();
// NOTE(ffbin): Currently we don't support the case where the pub-sub server restarts,
// because we use the same Redis server for both GCS storage and pub-sub. So the
// following flag is alway false.
// following flag is always false.
resubscribe_func_(false);
// Resend resource usage after reconnected, needed by resource view in GCS.
node_accessor_->AsyncReReportResourceUsage();
@@ -191,11 +193,27 @@ void ServiceBasedGcsClient::ReconnectGcsServer() {
int index = 0;
for (; index < RayConfig::instance().ping_gcs_rpc_server_max_retries(); ++index) {
if (get_server_address_func_(&address)) {
RAY_LOG(DEBUG) << "Attemptting to reconnect to GCS server: " << address.first << ":"
<< address.second;
// After GCS is restarted, the gcs client will reestablish the connection. At
// present, every failed RPC request will trigger `ReconnectGcsServer`. In order to
// avoid repeated connections in a short period of time, we add a protection
// mechanism: if the address does not change (meaning gcs server doesn't restart),
// the connection can be made at most once in
// `minimum_gcs_reconnect_interval_milliseconds` milliseconds.
if (last_reconnect_address_ == address &&
(current_sys_time_ms() - last_reconnect_timestamp_ms_) <
RayConfig::instance().minimum_gcs_reconnect_interval_milliseconds()) {
RAY_LOG(INFO)
<< "Repeated reconnection in "
<< RayConfig::instance().minimum_gcs_reconnect_interval_milliseconds()
<< "milliseconds, return directly.";
return;
}
RAY_LOG(INFO) << "Attemptting to reconnect to GCS server: " << address.first << ":"
<< address.second;
if (Ping(address.first, address.second, 100)) {
RAY_LOG(DEBUG) << "Reconnected to GCS server: " << address.first << ":"
<< address.second;
RAY_LOG(INFO) << "Reconnected to GCS server: " << address.first << ":"
<< address.second;
break;
}
}
@@ -205,6 +223,8 @@ void ServiceBasedGcsClient::ReconnectGcsServer() {
if (index < RayConfig::instance().ping_gcs_rpc_server_max_retries()) {
gcs_rpc_client_->Reset(address.first, address.second, *client_call_manager_);
last_reconnect_address_ = address;
last_reconnect_timestamp_ms_ = current_sys_time_ms();
} else {
RAY_LOG(FATAL) << "Couldn't reconnect to GCS server. The last attempted GCS "
"server address was "
src/ray/gcs/gcs_client/service_based_gcs_client.h
+2
View File
@@ -72,6 +72,8 @@ class RAY_EXPORT ServiceBasedGcsClient : public GcsClient {
std::function<bool(std::pair<std::string, int> *)> get_server_address_func_;
std::function<void(bool)> resubscribe_func_;
std::pair<std::string, int> current_gcs_server_address_;
int64_t last_reconnect_timestamp_ms_;
std::pair<std::string, int> last_reconnect_address_;
};
} // namespace gcs