diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 4627a7f29..5adc543b1 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -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")) diff --git a/src/ray/gcs/gcs_client/service_based_gcs_client.cc b/src/ray/gcs/gcs_client/service_based_gcs_client.cc index eede134bc..51399e786 100644 --- a/src/ray/gcs/gcs_client/service_based_gcs_client.cc +++ b/src/ray/gcs/gcs_client/service_based_gcs_client.cc @@ -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 address; - GetGcsServerAddressFromRedis(redis_gcs_client_->primary_context()->sync_context(), - &address); - return address; + get_server_address_func_ = [this](std::pair *address) { + return GetGcsServerAddressFromRedis( + redis_gcs_client_->primary_context()->sync_context(), address); }; - std::pair address = get_server_address(); + std::pair 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 *address) { +bool ServiceBasedGcsClient::GetGcsServerAddressFromRedis( + redisContext *context, std::pair *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(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 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 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 diff --git a/src/ray/gcs/gcs_client/service_based_gcs_client.h b/src/ray/gcs/gcs_client/service_based_gcs_client.h index 7295f4447..05fb4c467 100644 --- a/src/ray/gcs/gcs_client/service_based_gcs_client.h +++ b/src/ray/gcs/gcs_client/service_based_gcs_client.h @@ -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 *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 *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 redis_gcs_client_; @@ -51,6 +66,12 @@ class RAY_EXPORT ServiceBasedGcsClient : public GcsClient { // Gcs rpc client std::unique_ptr gcs_rpc_client_; std::unique_ptr client_call_manager_; + + // A timer used to check if gcs server address changed. + std::unique_ptr detect_timer_; + std::function *)> get_server_address_func_; + std::function resubscribe_func_; + std::pair current_gcs_server_address_; }; } // namespace gcs diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index ce6f51284..d5d9188a4 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -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. diff --git a/src/ray/gcs/pubsub/gcs_pub_sub.h b/src/ray/gcs/pubsub/gcs_pub_sub.h index 69a94b5fc..19cfaee6e 100644 --- a/src/ray/gcs/pubsub/gcs_pub_sub.h +++ b/src/ray/gcs/pubsub/gcs_pub_sub.h @@ -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 &id = boost::none); diff --git a/src/ray/protobuf/gcs_service.proto b/src/ray/protobuf/gcs_service.proto index 10e1f4cf3..4ffe4ee65 100644 --- a/src/ray/protobuf/gcs_service.proto +++ b/src/ray/protobuf/gcs_service.proto @@ -446,3 +446,8 @@ message CreateActorRequest { message CreateActorReply { GcsStatus status = 1; } + +enum GcsServiceFailureType { + RPC_DISCONNECT = 0; + GCS_SERVER_RESTART = 1; +} diff --git a/src/ray/rpc/gcs_server/gcs_rpc_client.h b/src/ray/rpc/gcs_server/gcs_rpc_client.h index f4ce9fcf0..2507d63dc 100644 --- a/src/ray/rpc/gcs_server/gcs_rpc_client.h +++ b/src/ray/rpc/gcs_server/gcs_rpc_client.h @@ -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()> get_server_address = nullptr, - std::function 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 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>( + new GrpcClient(address, port, client_call_manager)); + actor_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + node_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + object_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + task_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + stats_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + error_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(address, port, client_call_manager)); + worker_info_grpc_client_ = std::unique_ptr>( + new GrpcClient(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>( - new GrpcClient(address, port, client_call_manager)); - actor_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - node_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - object_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - task_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - stats_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - error_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - worker_info_grpc_client_ = std::unique_ptr>( - new GrpcClient(address, port, client_call_manager)); - } - - void Reconnect() { - absl::MutexLock lock(&mutex_); - if (get_server_address_) { - std::pair 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()> 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 reconnected_callback_; + std::function gcs_service_failure_detected_; /// The gRPC-generated stub. std::unique_ptr> job_info_grpc_client_;