From a0dc02042b6585b2476da597efc33e04b978dcd8 Mon Sep 17 00:00:00 2001 From: micafan <550435771@qq.com> Date: Wed, 15 Jan 2020 13:38:58 +0800 Subject: [PATCH] [GCS] Add ErrorInfoAccessor to GCS Client (#6749) --- src/ray/gcs/accessor.h | 26 ++++++++++++++++++++++ src/ray/gcs/gcs_client.h | 8 +++++++ src/ray/gcs/pb_util.h | 12 +++++++++++ src/ray/gcs/redis_accessor.cc | 16 ++++++++++++++ src/ray/gcs/redis_accessor.h | 16 ++++++++++++++ src/ray/gcs/redis_gcs_client.cc | 1 + src/ray/gcs/redis_gcs_client.h | 6 ++++-- src/ray/gcs/tables.cc | 10 --------- src/ray/gcs/tables.h | 18 +--------------- src/ray/raylet/monitor.cc | 8 ++++--- src/ray/raylet/node_manager.cc | 38 ++++++++++++++++++--------------- src/ray/raylet/worker_pool.cc | 6 ++++-- 12 files changed, 114 insertions(+), 51 deletions(-) diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index bdf36f50c..33d37fdd0 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -473,6 +473,32 @@ class NodeInfoAccessor { NodeInfoAccessor() = default; }; +/// \class ErrorInfoAccessor +/// `ErrorInfoAccessor` is a sub-interface of `GcsClient`. +/// This class includes all the methods that are related to accessing +/// error information in the GCS. +class ErrorInfoAccessor { + public: + virtual ~ErrorInfoAccessor() = default; + + /// Report a job error to GCS asynchronously. + /// The error message will be pushed to the driver of a specific if it is + /// a job internal error, or broadcast to all drivers if it is a system error. + /// + /// TODO(rkn): We need to make sure that the errors are unique because + /// duplicate messages currently cause failures (the GCS doesn't allow it). A + /// natural way to do this is to have finer-grained time stamps. + /// + /// \param data_ptr The error message that will be reported to GCS. + /// \param callback Callback that will be called when report is complete. + /// \return Status + virtual Status AsyncReportJobError(const std::shared_ptr &data_ptr, + const StatusCallback &callback) = 0; + + protected: + ErrorInfoAccessor() = default; +}; + /// \class StatsInfoAccessor /// `StatsInfoAccessor` is a sub-interface of `GcsClient`. /// This class includes all the methods that are related to accessing diff --git a/src/ray/gcs/gcs_client.h b/src/ray/gcs/gcs_client.h index d28027c77..3d554f6e2 100644 --- a/src/ray/gcs/gcs_client.h +++ b/src/ray/gcs/gcs_client.h @@ -95,6 +95,13 @@ class GcsClient : public std::enable_shared_from_this { return *task_accessor_; } + /// Get the sub-interface for accessing error information in GCS. + /// This function is thread safe. + ErrorInfoAccessor &Errors() { + RAY_CHECK(error_accessor_ != nullptr); + return *error_accessor_; + } + /// Get the sub-interface for accessing stats in GCS. /// This function is thread safe. StatsInfoAccessor &Stats() { @@ -118,6 +125,7 @@ class GcsClient : public std::enable_shared_from_this { std::unique_ptr object_accessor_; std::unique_ptr node_accessor_; std::unique_ptr task_accessor_; + std::unique_ptr error_accessor_; std::unique_ptr stats_accessor_; }; diff --git a/src/ray/gcs/pb_util.h b/src/ray/gcs/pb_util.h index 9fe979f79..3a6cbf224 100644 --- a/src/ray/gcs/pb_util.h +++ b/src/ray/gcs/pb_util.h @@ -30,6 +30,18 @@ inline std::shared_ptr CreateJobTableData( return job_info_ptr; } +/// Helper function to produce error table data. +inline std::shared_ptr CreateErrorTableData( + const std::string &error_type, const std::string &error_msg, double timestamp, + const JobID &job_id = JobID::Nil()) { + auto error_info_ptr = std::make_shared(); + error_info_ptr->set_type(error_type); + error_info_ptr->set_error_message(error_msg); + error_info_ptr->set_timestamp(timestamp); + error_info_ptr->set_job_id(job_id.Binary()); + return error_info_ptr; +} + /// Helper function to produce actor table data. inline std::shared_ptr CreateActorTableData( const TaskSpecification &task_spec, const ray::rpc::Address &address, diff --git a/src/ray/gcs/redis_accessor.cc b/src/ray/gcs/redis_accessor.cc index 06e681c1b..871f8a935 100644 --- a/src/ray/gcs/redis_accessor.cc +++ b/src/ray/gcs/redis_accessor.cc @@ -589,6 +589,22 @@ Status RedisNodeInfoAccessor::AsyncSubscribeToResources( return resource_sub_executor_.AsyncSubscribeAll(ClientID::Nil(), subscribe, done); } +RedisErrorInfoAccessor::RedisErrorInfoAccessor(RedisGcsClient *client_impl) + : client_impl_(client_impl) {} + +Status RedisErrorInfoAccessor::AsyncReportJobError( + const std::shared_ptr &data_ptr, const StatusCallback &callback) { + ErrorTable::WriteCallback on_done = nullptr; + if (callback != nullptr) { + on_done = [callback](RedisGcsClient *client, const JobID &job_id, + const ErrorTableData &data) { callback(Status::OK()); }; + } + + JobID job_id = JobID::FromBinary(data_ptr->job_id()); + ErrorTable &error_table = client_impl_->error_table(); + return error_table.Append(job_id, job_id, data_ptr, on_done); +} + RedisStatsInfoAccessor::RedisStatsInfoAccessor(RedisGcsClient *client_impl) : client_impl_(client_impl) {} diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index 70eb6f532..9efcda2bf 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -290,6 +290,22 @@ class RedisNodeInfoAccessor : public NodeInfoAccessor { HeartbeatBatchSubscriptionExecutor heartbeat_batch_sub_executor_; }; +/// \class RedisErrorInfoAccessor +/// RedisErrorInfoAccessor is an implementation of `ErrorInfoAccessor` +/// that uses Redis as the backend storage. +class RedisErrorInfoAccessor : public ErrorInfoAccessor { + public: + explicit RedisErrorInfoAccessor(RedisGcsClient *client_impl); + + virtual ~RedisErrorInfoAccessor() = default; + + Status AsyncReportJobError(const std::shared_ptr &data_ptr, + const StatusCallback &callback) override; + + private: + RedisGcsClient *client_impl_{nullptr}; +}; + /// \class RedisStatsInfoAccessor /// RedisStatsInfoAccessor is an implementation of `StatsInfoAccessor` /// that uses Redis as the backend storage. diff --git a/src/ray/gcs/redis_gcs_client.cc b/src/ray/gcs/redis_gcs_client.cc index f4b5e910d..0cb255dfb 100644 --- a/src/ray/gcs/redis_gcs_client.cc +++ b/src/ray/gcs/redis_gcs_client.cc @@ -147,6 +147,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) { object_accessor_.reset(new RedisObjectInfoAccessor(this)); node_accessor_.reset(new RedisNodeInfoAccessor(this)); task_accessor_.reset(new RedisTaskInfoAccessor(this)); + error_accessor_.reset(new RedisErrorInfoAccessor(this)); stats_accessor_.reset(new RedisStatsInfoAccessor(this)); is_connected_ = true; diff --git a/src/ray/gcs/redis_gcs_client.h b/src/ray/gcs/redis_gcs_client.h index 62c589c5b..e67c13a10 100644 --- a/src/ray/gcs/redis_gcs_client.h +++ b/src/ray/gcs/redis_gcs_client.h @@ -25,6 +25,7 @@ class RAY_EXPORT RedisGcsClient : public GcsClient { friend class RedisTaskInfoAccessor; friend class RedisNodeInfoAccessor; friend class RedisObjectInfoAccessor; + friend class RedisErrorInfoAccessor; friend class RedisStatsInfoAccessor; friend class SubscriptionExecutorTest; friend class LogSubscribeTestHelper; @@ -64,8 +65,6 @@ class RAY_EXPORT RedisGcsClient : public GcsClient { /// Disconnect with GCS Service. Non-thread safe. void Disconnect(); - // TODO: Some API for getting the error on the driver - ErrorTable &error_table(); WorkerFailureTable &worker_failure_table(); // We also need something to export generic code to run on workers from the @@ -106,6 +105,9 @@ class RAY_EXPORT RedisGcsClient : public GcsClient { raylet::TaskTable &raylet_task_table(); TaskLeaseTable &task_lease_table(); TaskReconstructionLog &task_reconstruction_log(); + /// This method will be deprecated, use method Errors() instead. + // TODO: Some API for getting the error on the driver + ErrorTable &error_table(); /// This method will be deprecated, use method Stats() instead. ProfileTable &profile_table(); diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index aa60f2dc7..ea093d8e7 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -515,16 +515,6 @@ Status Hash::Subscribe(const JobID &job_id, const ClientID &client_id, return Status::OK(); } -Status ErrorTable::PushErrorToDriver(const JobID &job_id, const std::string &type, - const std::string &error_message, double timestamp) { - auto data = std::make_shared(); - data->set_job_id(job_id.Binary()); - data->set_type(type); - data->set_error_message(error_message); - data->set_timestamp(timestamp); - return Append(job_id, job_id, data, /*done_callback=*/nullptr); -} - std::string ErrorTable::DebugString() const { return Log::DebugString(); } diff --git a/src/ray/gcs/tables.h b/src/ray/gcs/tables.h index 27bcbc8ea..9ecdab665 100644 --- a/src/ray/gcs/tables.h +++ b/src/ray/gcs/tables.h @@ -823,7 +823,7 @@ class TaskTable : public Table { } // namespace raylet -class ErrorTable : private Log { +class ErrorTable : public Log { public: ErrorTable(const std::vector> &contexts, RedisGcsClient *client) @@ -832,22 +832,6 @@ class ErrorTable : private Log { prefix_ = TablePrefix::ERROR_INFO; }; - /// Push an error message for the driver of a specific. - /// - /// TODO(rkn): We need to make sure that the errors are unique because - /// duplicate messages currently cause failures (the GCS doesn't allow it). A - /// natural way to do this is to have finer-grained time stamps. - /// - /// \param job_id The ID of the job that generated the error. If the error - /// should be pushed to all drivers, then this should be nil. - /// \param type The type of the error. - /// \param error_message The error message to push. - /// \param timestamp The timestamp of the error. - /// \return Status. - // TODO(qwang): refactor this API to implement broadcast. - Status PushErrorToDriver(const JobID &job_id, const std::string &type, - const std::string &error_message, double timestamp); - /// Returns debug string for class. /// /// \return string. diff --git a/src/ray/raylet/monitor.cc b/src/ray/raylet/monitor.cc index af367669f..d6566d331 100644 --- a/src/ray/raylet/monitor.cc +++ b/src/ray/raylet/monitor.cc @@ -2,6 +2,7 @@ #include "ray/common/ray_config.h" #include "ray/common/status.h" +#include "ray/gcs/pb_util.h" #include "ray/util/util.h" namespace ray { @@ -67,9 +68,10 @@ void Monitor::Tick() { error_message << "The node with client ID " << node_id << " has been marked dead because the monitor" << " has missed too many heartbeats from it."; - // We use the nil JobID to broadcast the message to all drivers. - RAY_CHECK_OK(gcs_client_.error_table().PushErrorToDriver( - JobID::Nil(), type, error_message.str(), current_time_ms())); + auto error_data_ptr = + gcs::CreateErrorTableData(type, error_message.str(), current_time_ms()); + RAY_CHECK_OK( + gcs_client_.Errors().AsyncReportJobError(error_data_ptr, nullptr)); } }; RAY_CHECK_OK(gcs_client_.Nodes().AsyncGetAll(lookup_callback)); diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 657eb839b..0af594445 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -408,9 +408,10 @@ void NodeManager::WarnResourceDeadlock() { << "To resolve the issue, consider creating fewer actors or increase the " << "resources available to this Ray cluster. You can ignore this message " << "if this Ray cluster is expected to auto-scale."; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - exemplar.GetTaskSpecification().JobId(), "resource_deadlock", error_message.str(), - current_time_ms())); + auto error_data_ptr = gcs::CreateErrorTableData( + "resource_deadlock", error_message.str(), current_time_ms(), + exemplar.GetTaskSpecification().JobId()); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); resource_deadlock_warned_ = true; } } @@ -1177,8 +1178,9 @@ void NodeManager::ProcessDisconnectClientMessage( std::ostringstream error_message; error_message << "A worker died or was killed while executing task " << task_id << "."; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - job_id, type, error_message.str(), current_time_ms())); + auto error_data_ptr = gcs::CreateErrorTableData(type, error_message.str(), + current_time_ms(), job_id); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); } } @@ -1342,13 +1344,12 @@ void NodeManager::ProcessWaitForDirectActorCallArgsRequestMessage( void NodeManager::ProcessPushErrorRequestMessage(const uint8_t *message_data) { auto message = flatbuffers::GetRoot(message_data); - JobID job_id = from_flatbuf(*message->job_id()); auto const &type = string_from_flatbuf(*message->type()); auto const &error_message = string_from_flatbuf(*message->error_message()); double timestamp = message->timestamp(); - - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(job_id, type, error_message, - timestamp)); + JobID job_id = from_flatbuf(*message->job_id()); + auto error_data_ptr = gcs::CreateErrorTableData(type, error_message, timestamp, job_id); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); } void NodeManager::ProcessPrepareActorCheckpointRequest( @@ -1807,9 +1808,10 @@ void NodeManager::ScheduleTasks( << "provide the requested resources. To resolve this issue, consider " << "reducing the resource requests of this task or add nodes that " << "can fit the task."; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - task.GetTaskSpecification().JobId(), type, error_message.str(), - current_time_ms())); + auto error_data_ptr = + gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(), + task.GetTaskSpecification().JobId()); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); } // Assert that this placeable task is not feasible locally (necessary but not // sufficient). @@ -1885,8 +1887,9 @@ void NodeManager::MarkObjectsAsFailed(const ErrorType &error_type, << " object may hang forever."; std::string error_message = stream.str(); RAY_LOG(WARNING) << error_message; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - job_id, "task", error_message, current_time_ms())); + auto error_data_ptr = + gcs::CreateErrorTableData("task", error_message, current_time_ms(), job_id); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); } } } @@ -2662,9 +2665,10 @@ void NodeManager::ResubmitTask(const Task &task, const ObjectID &required_object error_message << "The task with ID " << task.GetTaskSpecification().TaskId() << " is a driver task and so the object created by ray.put " << "could not be reconstructed."; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - task.GetTaskSpecification().JobId(), type, error_message.str(), - current_time_ms())); + auto error_data_ptr = + gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(), + task.GetTaskSpecification().JobId()); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); MarkObjectsAsFailed(ErrorType::OBJECT_UNRECONSTRUCTABLE, {required_object_id.ToPlasmaId()}, task.GetTaskSpecification().JobId()); diff --git a/src/ray/raylet/worker_pool.cc b/src/ray/raylet/worker_pool.cc index b1df7e37b..220ead748 100644 --- a/src/ray/raylet/worker_pool.cc +++ b/src/ray/raylet/worker_pool.cc @@ -12,6 +12,7 @@ #include "ray/common/constants.h" #include "ray/common/ray_config.h" #include "ray/common/status.h" +#include "ray/gcs/pb_util.h" #include "ray/stats/stats.h" #include "ray/util/logging.h" #include "ray/util/util.h" @@ -487,8 +488,9 @@ void WorkerPool::WarnAboutSize() { << "using nested tasks " << "(see https://github.com/ray-project/ray/issues/3644) for " << "some a discussion of workarounds."; - RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( - JobID::Nil(), "worker_pool_large", warning_message.str(), current_time_ms())); + auto error_data_ptr = gcs::CreateErrorTableData( + "worker_pool_large", warning_message.str(), current_time_ms()); + RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr)); } } }