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

Add gcs service discovery to automatically detect GCS server restart (#8735)

Browse Source
This commit is contained in:
fangfengbin
2020-06-28 16:50:06 +08:00
committed by GitHub
parent d78128a69c
commit 1dc018d0bb
7 changed files with 176 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/ray_config_def.h
+5 -1
View File
@@ -297,9 +297,13 @@ 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, 600)
// Whether start the Plasma Store as a Raylet thread.
/// Whether start the Plasma Store as a Raylet thread.
RAY_CONFIG(bool, plasma_store_as_thread, false)
/// The interval at which the gcs client will check if the address of gcs service has
/// changed. When the address changed, we will resubscribe again.
RAY_CONFIG(int64_t, gcs_service_address_check_interval_milliseconds, 1000)
RAY_CONFIG(bool, gcs_actor_service_enabled,
getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") != nullptr &&
getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") == std::string("true"))
src/ray/gcs/gcs_client/service_based_gcs_client.cc
+113 -28
View File
@@ -39,15 +39,17 @@ Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
gcs_pub_sub_.reset(new GcsPubSub(redis_gcs_client_->GetRedisClient()));
// Get gcs service address.
auto get_server_address = [this]() {
std::pair<std::string, int> address;
GetGcsServerAddressFromRedis(redis_gcs_client_->primary_context()->sync_context(),
&address);
return address;
get_server_address_func_ = [this](std::pair<std::string, int> *address) {
return GetGcsServerAddressFromRedis(
redis_gcs_client_->primary_context()->sync_context(), address);
};
std::pair<std::string, int> address = get_server_address();
std::pair<std::string, int> address;
RAY_CHECK(GetGcsServerAddressFromRedis(
redis_gcs_client_->primary_context()->sync_context(), &address,
RayConfig::instance().gcs_service_connect_retries()))
<< "Failed to get gcs server address when init gcs client.";
auto re_subscribe = [this](bool is_pubsub_server_restarted) {
resubscribe_func_ = [this](bool is_pubsub_server_restarted) {
job_accessor_->AsyncResubscribe(is_pubsub_server_restarted);
actor_accessor_->AsyncResubscribe(is_pubsub_server_restarted);
node_accessor_->AsyncResubscribe(is_pubsub_server_restarted);
@@ -58,9 +60,9 @@ Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
// Connect to gcs service.
client_call_manager_.reset(new rpc::ClientCallManager(io_service));
gcs_rpc_client_.reset(new rpc::GcsRpcClient(address.first, address.second,
*client_call_manager_, get_server_address,
re_subscribe));
gcs_rpc_client_.reset(new rpc::GcsRpcClient(
address.first, address.second, *client_call_manager_,
[this](rpc::GcsServiceFailureType type) { GcsServiceFailureDetected(type); }));
job_accessor_.reset(new ServiceBasedJobInfoAccessor(this));
actor_accessor_.reset(new ServiceBasedActorInfoAccessor(this));
node_accessor_.reset(new ServiceBasedNodeInfoAccessor(this));
@@ -70,6 +72,10 @@ Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
error_accessor_.reset(new ServiceBasedErrorInfoAccessor(this));
worker_accessor_.reset(new ServiceBasedWorkerInfoAccessor(this));
// Init gcs service address check timer.
detect_timer_.reset(new boost::asio::deadline_timer(io_service));
PeriodicallyCheckGcsServerAddress();
is_connected_ = true;
RAY_LOG(INFO) << "ServiceBasedGcsClient Connected.";
@@ -79,18 +85,19 @@ Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
void ServiceBasedGcsClient::Disconnect() {
RAY_CHECK(is_connected_);
is_connected_ = false;
detect_timer_->cancel();
gcs_pub_sub_.reset();
redis_gcs_client_->Disconnect();
redis_gcs_client_.reset();
RAY_LOG(DEBUG) << "ServiceBasedGcsClient Disconnected.";
}
void ServiceBasedGcsClient::GetGcsServerAddressFromRedis(
redisContext *context, std::pair<std::string, int> *address) {
bool ServiceBasedGcsClient::GetGcsServerAddressFromRedis(
redisContext *context, std::pair<std::string, int> *address, int max_attempts) {
// Get gcs server address.
int num_attempts = 0;
redisReply *reply = nullptr;
while (num_attempts < RayConfig::instance().gcs_service_connect_retries()) {
while (num_attempts < max_attempts) {
reply = reinterpret_cast<redisReply *>(redisCommand(context, "GET GcsServerAddress"));
if (reply && reply->type != REDIS_REPLY_NIL) {
break;
@@ -98,23 +105,101 @@ void ServiceBasedGcsClient::GetGcsServerAddressFromRedis(
// Sleep for a little, and try again if the entry isn't there yet.
freeReplyObject(reply);
usleep(RayConfig::instance().internal_gcs_service_connect_wait_milliseconds() * 1000);
num_attempts++;
}
RAY_CHECK(num_attempts < RayConfig::instance().gcs_service_connect_retries())
<< "No entry found for GcsServerAddress";
RAY_CHECK(reply) << "Redis did not reply to GcsServerAddress. Is redis running?";
RAY_CHECK(reply->type == REDIS_REPLY_STRING)
<< "Expected string, found Redis type " << reply->type << " for GcsServerAddress";
std::string result(reply->str);
freeReplyObject(reply);
RAY_CHECK(!result.empty()) << "Gcs service address is empty";
size_t pos = result.find(':');
RAY_CHECK(pos != std::string::npos)
<< "Gcs service address format is erroneous: " << result;
address->first = result.substr(0, pos);
address->second = std::stoi(result.substr(pos + 1));
if (num_attempts < max_attempts) {
usleep(RayConfig::instance().internal_gcs_service_connect_wait_milliseconds() *
1000);
}
}
if (num_attempts < max_attempts) {
RAY_CHECK(reply) << "Redis did not reply to GcsServerAddress. Is redis running?";
RAY_CHECK(reply->type == REDIS_REPLY_STRING)
<< "Expected string, found Redis type " << reply->type << " for GcsServerAddress";
std::string result(reply->str);
freeReplyObject(reply);
RAY_CHECK(!result.empty()) << "Gcs service address is empty";
size_t pos = result.find(':');
RAY_CHECK(pos != std::string::npos)
<< "Gcs service address format is erroneous: " << result;
address->first = result.substr(0, pos);
address->second = std::stoi(result.substr(pos + 1));
return true;
}
return false;
}
void ServiceBasedGcsClient::PeriodicallyCheckGcsServerAddress() {
std::pair<std::string, int> address;
if (get_server_address_func_(&address)) {
if (address != current_gcs_server_address_) {
// If GCS server address has changed, invoke the `GcsServiceFailureDetected`
// callback.
current_gcs_server_address_ = address;
GcsServiceFailureDetected(rpc::GcsServiceFailureType::GCS_SERVER_RESTART);
}
}
auto check_period = boost::posix_time::milliseconds(
RayConfig::instance().gcs_service_address_check_interval_milliseconds());
detect_timer_->expires_from_now(check_period);
detect_timer_->async_wait([this](const boost::system::error_code &error) {
if (error == boost::system::errc::operation_canceled) {
// `operation_canceled` is set when `detect_timer_` is canceled or destroyed.
return;
}
RAY_CHECK(!error) << "Checking gcs server address failed with error: "
<< error.message();
PeriodicallyCheckGcsServerAddress();
});
}
void ServiceBasedGcsClient::GcsServiceFailureDetected(rpc::GcsServiceFailureType type) {
switch (type) {
case rpc::GcsServiceFailureType::RPC_DISCONNECT:
// If the GCS server address does not change, reconnect to GCS server.
ReconnectGcsServer();
break;
case rpc::GcsServiceFailureType::GCS_SERVER_RESTART:
// If GCS sever address has changed, reconnect to GCS server and redo
// subscription.
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.
resubscribe_func_(false);
break;
default:
RAY_LOG(FATAL) << "Unsupported failure type: " << type;
break;
}
}
void ServiceBasedGcsClient::ReconnectGcsServer() {
std::pair<std::string, int> address;
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;
if (Ping(address.first, address.second, 100)) {
RAY_LOG(INFO) << "Reconnected to GCS server: " << address.first << ":"
<< address.second;
break;
}
}
usleep(RayConfig::instance().ping_gcs_rpc_server_interval_milliseconds() * 1000);
}
if (index < RayConfig::instance().ping_gcs_rpc_server_max_retries()) {
gcs_rpc_client_->Reset(address.first, address.second, *client_call_manager_);
} else {
RAY_LOG(FATAL) << "Couldn't reconnect to GCS server. The last attempted GCS "
"server address was "
<< address.first << ":" << address.second;
}
}
} // namespace gcs
src/ray/gcs/gcs_client/service_based_gcs_client.h
+23 -2
View File
@@ -41,8 +41,23 @@ class RAY_EXPORT ServiceBasedGcsClient : public GcsClient {
///
/// \param context The context of redis.
/// \param address The address of gcs server.
void GetGcsServerAddressFromRedis(redisContext *context,
std::pair<std::string, int> *address);
/// \param max_attempts The maximum number of times to get gcs server rpc address.
/// \return Returns true if gcs server address is obtained, False otherwise.
bool GetGcsServerAddressFromRedis(redisContext *context,
std::pair<std::string, int> *address,
int max_attempts = 1);
/// Fire a periodic timer to check if GCS sever address has changed.
void PeriodicallyCheckGcsServerAddress();
/// This function is used to redo subscription and reconnect to GCS RPC server when gcs
/// service failure is detected.
///
/// \param type The type of GCS service failure.
void GcsServiceFailureDetected(rpc::GcsServiceFailureType type);
/// Reconnect to GCS RPC server.
void ReconnectGcsServer();
std::unique_ptr<RedisGcsClient> redis_gcs_client_;
@@ -51,6 +66,12 @@ class RAY_EXPORT ServiceBasedGcsClient : public GcsClient {
// Gcs rpc client
std::unique_ptr<rpc::GcsRpcClient> gcs_rpc_client_;
std::unique_ptr<rpc::ClientCallManager> client_call_manager_;
// A timer used to check if gcs server address changed.
std::unique_ptr<boost::asio::deadline_timer> detect_timer_;
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_;
};
} // namespace gcs
src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
-5
View File
@@ -902,11 +902,6 @@ TEST_F(ServiceBasedGcsClientTest, TestActorTableResubscribe) {
// Restart GCS server.
RestartGcsServer();
// We need to send a RPC to detect GCS server restart. Then GCS client will
// reconnect to GCS server and resubscribe.
ASSERT_TRUE(GetActor(actor_id).state() ==
rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
// When GCS client detects that GCS server has restarted, but the pub-sub server
// didn't restart, it will fetch data again from the GCS server. So we'll receive
// another notification of ALIVE state.
src/ray/gcs/pubsub/gcs_pub_sub.h
+1 -1
View File
@@ -140,7 +140,7 @@ class GcsPubSub {
GcsPubSub::Channel &channel)
EXCLUSIVE_LOCKS_REQUIRED(mutex_);
Status SubscribeInternal(const std::string &channel, const Callback &subscribe,
Status SubscribeInternal(const std::string &channel_name, const Callback &subscribe,
const StatusCallback &done,
const boost::optional<std::string> &id = boost::none);
src/ray/protobuf/gcs_service.proto
+5
View File
@@ -446,3 +446,8 @@ message CreateActorRequest {
message CreateActorReply {
GcsStatus status = 1;
}
enum GcsServiceFailureType {
RPC_DISCONNECT = 0;
GCS_SERVER_RESTART = 1;
}
src/ray/rpc/gcs_server/gcs_rpc_client.h
+29 -75
View File
@@ -63,7 +63,7 @@ class Executor {
callback(status, reply); \
delete executor; \
} else { \
Reconnect(); \
gcs_service_failure_detected_(GcsServiceFailureType::RPC_DISCONNECT); \
executor->Retry(); \
} \
}; \
@@ -82,18 +82,35 @@ class GcsRpcClient {
/// \param[in] address Address of gcs server.
/// \param[in] port Port of the gcs server.
/// \param[in] client_call_manager The `ClientCallManager` used for managing requests.
/// \param[in] get_server_address The function used for getting address when reconnect
/// rpc server.
GcsRpcClient(const std::string &address, const int port,
ClientCallManager &client_call_manager,
std::function<std::pair<std::string, int>()> get_server_address = nullptr,
std::function<void(bool)> reconnected_callback = nullptr)
: client_call_manager_(client_call_manager),
get_server_address_(std::move(get_server_address)),
reconnected_callback_(std::move(reconnected_callback)) {
Init(address, port, client_call_manager);
/// \param[in] gcs_service_failure_detected The function is used to redo subscription
/// and reconnect to GCS RPC server when gcs service failure is detected.
GcsRpcClient(
const std::string &address, const int port, ClientCallManager &client_call_manager,
std::function<void(GcsServiceFailureType)> gcs_service_failure_detected = nullptr)
: gcs_service_failure_detected_(std::move(gcs_service_failure_detected)) {
Reset(address, port, client_call_manager);
};
void Reset(const std::string &address, const int port,
ClientCallManager &client_call_manager) {
job_info_grpc_client_ = std::unique_ptr<GrpcClient<JobInfoGcsService>>(
new GrpcClient<JobInfoGcsService>(address, port, client_call_manager));
actor_info_grpc_client_ = std::unique_ptr<GrpcClient<ActorInfoGcsService>>(
new GrpcClient<ActorInfoGcsService>(address, port, client_call_manager));
node_info_grpc_client_ = std::unique_ptr<GrpcClient<NodeInfoGcsService>>(
new GrpcClient<NodeInfoGcsService>(address, port, client_call_manager));
object_info_grpc_client_ = std::unique_ptr<GrpcClient<ObjectInfoGcsService>>(
new GrpcClient<ObjectInfoGcsService>(address, port, client_call_manager));
task_info_grpc_client_ = std::unique_ptr<GrpcClient<TaskInfoGcsService>>(
new GrpcClient<TaskInfoGcsService>(address, port, client_call_manager));
stats_grpc_client_ = std::unique_ptr<GrpcClient<StatsGcsService>>(
new GrpcClient<StatsGcsService>(address, port, client_call_manager));
error_info_grpc_client_ = std::unique_ptr<GrpcClient<ErrorInfoGcsService>>(
new GrpcClient<ErrorInfoGcsService>(address, port, client_call_manager));
worker_info_grpc_client_ = std::unique_ptr<GrpcClient<WorkerInfoGcsService>>(
new GrpcClient<WorkerInfoGcsService>(address, port, client_call_manager));
}
/// Add job info to gcs server.
VOID_GCS_RPC_CLIENT_METHOD(JobInfoGcsService, AddJob, job_info_grpc_client_, )
@@ -215,70 +232,7 @@ class GcsRpcClient {
worker_info_grpc_client_, )
private:
void Init(const std::string &address, const int port,
ClientCallManager &client_call_manager) {
job_info_grpc_client_ = std::unique_ptr<GrpcClient<JobInfoGcsService>>(
new GrpcClient<JobInfoGcsService>(address, port, client_call_manager));
actor_info_grpc_client_ = std::unique_ptr<GrpcClient<ActorInfoGcsService>>(
new GrpcClient<ActorInfoGcsService>(address, port, client_call_manager));
node_info_grpc_client_ = std::unique_ptr<GrpcClient<NodeInfoGcsService>>(
new GrpcClient<NodeInfoGcsService>(address, port, client_call_manager));
object_info_grpc_client_ = std::unique_ptr<GrpcClient<ObjectInfoGcsService>>(
new GrpcClient<ObjectInfoGcsService>(address, port, client_call_manager));
task_info_grpc_client_ = std::unique_ptr<GrpcClient<TaskInfoGcsService>>(
new GrpcClient<TaskInfoGcsService>(address, port, client_call_manager));
stats_grpc_client_ = std::unique_ptr<GrpcClient<StatsGcsService>>(
new GrpcClient<StatsGcsService>(address, port, client_call_manager));
error_info_grpc_client_ = std::unique_ptr<GrpcClient<ErrorInfoGcsService>>(
new GrpcClient<ErrorInfoGcsService>(address, port, client_call_manager));
worker_info_grpc_client_ = std::unique_ptr<GrpcClient<WorkerInfoGcsService>>(
new GrpcClient<WorkerInfoGcsService>(address, port, client_call_manager));
}
void Reconnect() {
absl::MutexLock lock(&mutex_);
if (get_server_address_) {
std::pair<std::string, int> address;
int index = 0;
for (; index < RayConfig::instance().ping_gcs_rpc_server_max_retries(); ++index) {
address = get_server_address_();
RAY_LOG(DEBUG) << "Attempt to reconnect to GCS server: " << address.first << ":"
<< address.second;
if (Ping(address.first, address.second, 100)) {
RAY_LOG(INFO) << "Reconnected to GCS server: " << address.first << ":"
<< address.second;
break;
}
usleep(RayConfig::instance().ping_gcs_rpc_server_interval_milliseconds() * 1000);
}
if (index < RayConfig::instance().ping_gcs_rpc_server_max_retries()) {
Init(address.first, address.second, client_call_manager_);
if (reconnected_callback_) {
// TODO(ffbin): Once we separate the pubsub server and storage addresses, we can
// judge whether pubsub server is restarted. Currently, we only support the
// scenario where pubsub server does not restart.
reconnected_callback_(false);
}
} else {
RAY_LOG(FATAL) << "Couldn't reconnect to GCS server. The last attempted GCS "
"server address was "
<< address.first << ":" << address.second;
}
}
}
absl::Mutex mutex_;
ClientCallManager &client_call_manager_;
std::function<std::pair<std::string, int>()> get_server_address_;
/// The callback that will be called when we reconnect to GCS server.
/// Currently, we use this function to reestablish subscription to GCS.
/// Note, we use ping to detect whether the reconnection is successful. If the ping
/// succeeds but the RPC connection fails, this function might be called called again.
/// So it needs to be idempotent.
std::function<void(bool)> reconnected_callback_;
std::function<void(GcsServiceFailureType)> gcs_service_failure_detected_;
/// The gRPC-generated stub.
std::unique_ptr<GrpcClient<JobInfoGcsService>> job_info_grpc_client_;