diff --git a/src/ray/common/test_util.cc b/src/ray/common/test_util.cc index 436565dbc..4edb590f8 100644 --- a/src/ray/common/test_util.cc +++ b/src/ray/common/test_util.cc @@ -47,6 +47,16 @@ void RedisServiceManagerForTest::TearDownTestCase() { usleep(100 * 1000); } +void RedisServiceManagerForTest::FlushAll() { + std::string flush_all_redis_command = + REDIS_CLIENT_EXEC_PATH + " -p " + std::to_string(REDIS_SERVER_PORT) + " flushall"; + RAY_LOG(INFO) << "Cleaning up redis with command: " << flush_all_redis_command; + if (system(flush_all_redis_command.c_str()) != 0) { + RAY_LOG(WARNING) << "Failed to flush redis. The redis process may no longer exist."; + } + usleep(100 * 1000); +} + bool WaitForCondition(std::function condition, int timeout_ms) { int wait_time = 0; while (true) { diff --git a/src/ray/common/test_util.h b/src/ray/common/test_util.h index 342462f5a..b55d5f16c 100644 --- a/src/ray/common/test_util.h +++ b/src/ray/common/test_util.h @@ -62,6 +62,7 @@ class RedisServiceManagerForTest : public ::testing::Test { public: static void SetUpTestCase(); static void TearDownTestCase(); + static void FlushAll(); }; } // namespace ray diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index 2299d4fac..5459d1707 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -32,6 +32,12 @@ class ActorInfoAccessor { public: virtual ~ActorInfoAccessor() = default; + /// Get all actor specification from GCS synchronously. + /// + /// \param actor_table_data_list The container to hold the actor specification. + /// \return Status + virtual Status GetAll(std::vector *actor_table_data_list) = 0; + /// Get actor specification from GCS asynchronously. /// /// \param actor_id The ID of actor to look up in the GCS. diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc index 51a752206..ebf66fbcf 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.cc +++ b/src/ray/gcs/gcs_client/service_based_accessor.cc @@ -77,9 +77,14 @@ Status ServiceBasedJobInfoAccessor::AsyncSubscribeToFinishedJobs( ServiceBasedActorInfoAccessor::ServiceBasedActorInfoAccessor( ServiceBasedGcsClient *client_impl) - : client_impl_(client_impl), - subscribe_id_(ClientID::FromRandom()), - actor_sub_executor_(client_impl->GetRedisGcsClient().actor_table()) {} + : subscribe_id_(ClientID::FromRandom()), + client_impl_(client_impl), + actor_sub_executor_(client_impl->GetRedisGcsClient().log_based_actor_table()) {} + +Status ServiceBasedActorInfoAccessor::GetAll( + std::vector *actor_table_data_list) { + return Status::Invalid("Not implemented"); +} Status ServiceBasedActorInfoAccessor::AsyncGet( const ActorID &actor_id, const OptionalItemCallback &callback) { diff --git a/src/ray/gcs/gcs_client/service_based_accessor.h b/src/ray/gcs/gcs_client/service_based_accessor.h index d2e627074..c428b5054 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.h +++ b/src/ray/gcs/gcs_client/service_based_accessor.h @@ -58,6 +58,8 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor { virtual ~ServiceBasedActorInfoAccessor() = default; + Status GetAll(std::vector *actor_table_data_list) override; + Status AsyncGet(const ActorID &actor_id, const OptionalItemCallback &callback) override; @@ -89,12 +91,13 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor { const ActorID &actor_id, const OptionalItemCallback &callback) override; + protected: + ClientID subscribe_id_; + private: ServiceBasedGcsClient *client_impl_; - ClientID subscribe_id_; - - typedef SubscriptionExecutor + typedef SubscriptionExecutor ActorSubscriptionExecutor; ActorSubscriptionExecutor actor_sub_executor_; 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 a3cbbdba4..2d1eb3407 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 @@ -64,6 +64,7 @@ class ServiceBasedGcsGcsClientTest : public RedisServiceManagerForTest { thread_io_service_->join(); thread_gcs_server_->join(); gcs_client_->Disconnect(); + FlushAll(); } bool AddJob(const std::shared_ptr &job_table_data) { @@ -313,6 +314,7 @@ class ServiceBasedGcsGcsClientTest : public RedisServiceManagerForTest { rpc::GcsNodeInfo gcs_node_info; gcs_node_info.set_node_id(node_id); gcs_node_info.set_state(rpc::GcsNodeInfo_GcsNodeState_ALIVE); + gcs_node_info.set_node_manager_address("127.0.0.1"); return gcs_node_info; } diff --git a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc index a3fe63136..766ddf31e 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc @@ -410,6 +410,8 @@ class GcsServerTest : public RedisServiceManagerForTest { rpc::GcsNodeInfo GenGcsNodeInfo(const std::string &node_id) { rpc::GcsNodeInfo gcs_node_info; gcs_node_info.set_node_id(node_id); + gcs_node_info.set_node_manager_address("127.0.0.1"); + gcs_node_info.set_node_manager_port(6000); gcs_node_info.set_state(rpc::GcsNodeInfo_GcsNodeState_ALIVE); return gcs_node_info; } diff --git a/src/ray/gcs/redis_accessor.cc b/src/ray/gcs/redis_accessor.cc index 77451f77f..3e7611e06 100644 --- a/src/ray/gcs/redis_accessor.cc +++ b/src/ray/gcs/redis_accessor.cc @@ -22,10 +22,32 @@ namespace ray { namespace gcs { -RedisActorInfoAccessor::RedisActorInfoAccessor(RedisGcsClient *client_impl) - : client_impl_(client_impl), actor_sub_executor_(client_impl_->actor_table()) {} +RedisLogBasedActorInfoAccessor::RedisLogBasedActorInfoAccessor( + RedisGcsClient *client_impl) + : client_impl_(client_impl), + log_based_actor_sub_executor_(client_impl_->log_based_actor_table()) {} -Status RedisActorInfoAccessor::AsyncGet( +std::vector RedisLogBasedActorInfoAccessor::GetAllActorID() const { + return client_impl_->log_based_actor_table().GetAllActorID(); +} + +Status RedisLogBasedActorInfoAccessor::Get(const ActorID &actor_id, + ActorTableData *actor_table_data) const { + return client_impl_->log_based_actor_table().Get(actor_id, actor_table_data); +} + +Status RedisLogBasedActorInfoAccessor::GetAll( + std::vector *actor_table_data_list) { + RAY_CHECK(actor_table_data_list); + auto actor_id_list = GetAllActorID(); + actor_table_data_list->resize(actor_id_list.size()); + for (size_t i = 0; i < actor_id_list.size(); ++i) { + RAY_CHECK_OK(Get(actor_id_list[i], &(*actor_table_data_list)[i])); + } + return Status::OK(); +} + +Status RedisLogBasedActorInfoAccessor::AsyncGet( const ActorID &actor_id, const OptionalItemCallback &callback) { RAY_CHECK(callback != nullptr); auto on_done = [callback](RedisGcsClient *client, const ActorID &actor_id, @@ -37,10 +59,11 @@ Status RedisActorInfoAccessor::AsyncGet( callback(Status::OK(), result); }; - return client_impl_->actor_table().Lookup(actor_id.JobId(), actor_id, on_done); + return client_impl_->log_based_actor_table().Lookup(actor_id.JobId(), actor_id, + on_done); } -Status RedisActorInfoAccessor::AsyncRegister( +Status RedisLogBasedActorInfoAccessor::AsyncRegister( const std::shared_ptr &data_ptr, const StatusCallback &callback) { auto on_success = [callback](RedisGcsClient *client, const ActorID &actor_id, const ActorTableData &data) { @@ -57,12 +80,12 @@ Status RedisActorInfoAccessor::AsyncRegister( }; ActorID actor_id = ActorID::FromBinary(data_ptr->actor_id()); - return client_impl_->actor_table().AppendAt(actor_id.JobId(), actor_id, data_ptr, - on_success, on_failure, - /*log_length*/ 0); + return client_impl_->log_based_actor_table().AppendAt(actor_id.JobId(), actor_id, + data_ptr, on_success, on_failure, + /*log_length*/ 0); } -Status RedisActorInfoAccessor::AsyncUpdate( +Status RedisLogBasedActorInfoAccessor::AsyncUpdate( const ActorID &actor_id, const std::shared_ptr &data_ptr, const StatusCallback &callback) { // The actor log starts with an ALIVE entry. This is followed by 0 to N pairs @@ -100,30 +123,32 @@ Status RedisActorInfoAccessor::AsyncUpdate( } }; - return client_impl_->actor_table().AppendAt(actor_id.JobId(), actor_id, data_ptr, - on_success, on_failure, log_length); + return client_impl_->log_based_actor_table().AppendAt( + actor_id.JobId(), actor_id, data_ptr, on_success, on_failure, log_length); } -Status RedisActorInfoAccessor::AsyncSubscribeAll( +Status RedisLogBasedActorInfoAccessor::AsyncSubscribeAll( const SubscribeCallback &subscribe, const StatusCallback &done) { RAY_CHECK(subscribe != nullptr); - return actor_sub_executor_.AsyncSubscribeAll(ClientID::Nil(), subscribe, done); + return log_based_actor_sub_executor_.AsyncSubscribeAll(ClientID::Nil(), subscribe, + done); } -Status RedisActorInfoAccessor::AsyncSubscribe( +Status RedisLogBasedActorInfoAccessor::AsyncSubscribe( const ActorID &actor_id, const SubscribeCallback &subscribe, const StatusCallback &done) { RAY_CHECK(subscribe != nullptr); - return actor_sub_executor_.AsyncSubscribe(subscribe_id_, actor_id, subscribe, done); + return log_based_actor_sub_executor_.AsyncSubscribe(subscribe_id_, actor_id, subscribe, + done); } -Status RedisActorInfoAccessor::AsyncUnsubscribe(const ActorID &actor_id, - const StatusCallback &done) { - return actor_sub_executor_.AsyncUnsubscribe(subscribe_id_, actor_id, done); +Status RedisLogBasedActorInfoAccessor::AsyncUnsubscribe(const ActorID &actor_id, + const StatusCallback &done) { + return log_based_actor_sub_executor_.AsyncUnsubscribe(subscribe_id_, actor_id, done); } -Status RedisActorInfoAccessor::AsyncAddCheckpoint( +Status RedisLogBasedActorInfoAccessor::AsyncAddCheckpoint( const std::shared_ptr &data_ptr, const StatusCallback &callback) { ActorID actor_id = ActorID::FromBinary(data_ptr->actor_id()); @@ -143,7 +168,7 @@ Status RedisActorInfoAccessor::AsyncAddCheckpoint( return actor_cp_table.Add(actor_id.JobId(), checkpoint_id, data_ptr, on_add_data_done); } -Status RedisActorInfoAccessor::AsyncGetCheckpoint( +Status RedisLogBasedActorInfoAccessor::AsyncGetCheckpoint( const ActorCheckpointID &checkpoint_id, const ActorID &actor_id, const OptionalItemCallback &callback) { RAY_CHECK(callback != nullptr); @@ -164,7 +189,7 @@ Status RedisActorInfoAccessor::AsyncGetCheckpoint( return actor_cp_table.Lookup(actor_id.JobId(), checkpoint_id, on_success, on_failure); } -Status RedisActorInfoAccessor::AsyncGetCheckpointID( +Status RedisLogBasedActorInfoAccessor::AsyncGetCheckpointID( const ActorID &actor_id, const OptionalItemCallback &callback) { RAY_CHECK(callback != nullptr); @@ -183,7 +208,7 @@ Status RedisActorInfoAccessor::AsyncGetCheckpointID( return cp_id_table.Lookup(actor_id.JobId(), actor_id, on_success, on_failure); } -Status RedisActorInfoAccessor::AsyncAddCheckpointID( +Status RedisLogBasedActorInfoAccessor::AsyncAddCheckpointID( const ActorID &actor_id, const ActorCheckpointID &checkpoint_id, const StatusCallback &callback) { ActorCheckpointIdTable::WriteCallback on_done = nullptr; @@ -196,6 +221,79 @@ Status RedisActorInfoAccessor::AsyncAddCheckpointID( return cp_id_table.AddCheckpointId(actor_id.JobId(), actor_id, checkpoint_id, on_done); } +RedisActorInfoAccessor::RedisActorInfoAccessor(RedisGcsClient *client_impl) + : RedisLogBasedActorInfoAccessor(client_impl), + actor_sub_executor_(client_impl_->actor_table()) {} + +std::vector RedisActorInfoAccessor::GetAllActorID() const { + return client_impl_->actor_table().GetAllActorID(); +} + +Status RedisActorInfoAccessor::Get(const ActorID &actor_id, + ActorTableData *actor_table_data) const { + return client_impl_->actor_table().Get(actor_id, actor_table_data); +} + +Status RedisActorInfoAccessor::AsyncGet( + const ActorID &actor_id, const OptionalItemCallback &callback) { + RAY_CHECK(callback != nullptr); + auto on_done = [callback](RedisGcsClient *client, const ActorID &actor_id, + const ActorTableData &data) { callback(Status::OK(), data); }; + + auto on_failure = [callback](RedisGcsClient *client, const ActorID &actor_id) { + if (callback != nullptr) { + callback(Status::Invalid("Get actor failed."), boost::none); + } + }; + + return client_impl_->actor_table().Lookup(JobID::Nil(), actor_id, on_done, on_failure); +} + +Status RedisActorInfoAccessor::AsyncRegister( + const std::shared_ptr &data_ptr, const StatusCallback &callback) { + auto on_register_done = [callback](RedisGcsClient *client, const ActorID &actor_id, + const ActorTableData &data) { + if (callback != nullptr) { + callback(Status::OK()); + } + }; + ActorID actor_id = ActorID::FromBinary(data_ptr->actor_id()); + return client_impl_->actor_table().Add(JobID::Nil(), actor_id, data_ptr, + on_register_done); +} + +Status RedisActorInfoAccessor::AsyncUpdate( + const ActorID &actor_id, const std::shared_ptr &data_ptr, + const StatusCallback &callback) { + auto on_update_done = [callback](RedisGcsClient *client, const ActorID &actor_id, + const ActorTableData &data) { + if (callback != nullptr) { + callback(Status::OK()); + } + }; + return client_impl_->actor_table().Add(JobID::Nil(), actor_id, data_ptr, + on_update_done); +} + +Status RedisActorInfoAccessor::AsyncSubscribeAll( + const SubscribeCallback &subscribe, + const StatusCallback &done) { + RAY_CHECK(subscribe != nullptr); + return actor_sub_executor_.AsyncSubscribeAll(ClientID::Nil(), subscribe, done); +} + +Status RedisActorInfoAccessor::AsyncSubscribe( + const ActorID &actor_id, const SubscribeCallback &subscribe, + const StatusCallback &done) { + RAY_CHECK(subscribe != nullptr); + return actor_sub_executor_.AsyncSubscribe(subscribe_id_, actor_id, subscribe, done); +} + +Status RedisActorInfoAccessor::AsyncUnsubscribe(const ActorID &actor_id, + const StatusCallback &done) { + return actor_sub_executor_.AsyncUnsubscribe(subscribe_id_, actor_id, done); +} + RedisJobInfoAccessor::RedisJobInfoAccessor(RedisGcsClient *client_impl) : client_impl_(client_impl), job_sub_executor_(client_impl->job_table()) {} diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index b0b95af20..e2ba41f36 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -27,14 +27,16 @@ namespace gcs { class RedisGcsClient; -/// \class RedisActorInfoAccessor -/// `RedisActorInfoAccessor` is an implementation of `ActorInfoAccessor` +/// \class RedisLogBasedActorInfoAccessor +/// `RedisLogBasedActorInfoAccessor` is an implementation of `ActorInfoAccessor` /// that uses Redis as the backend storage. -class RedisActorInfoAccessor : public ActorInfoAccessor { +class RedisLogBasedActorInfoAccessor : public ActorInfoAccessor { public: - explicit RedisActorInfoAccessor(RedisGcsClient *client_impl); + explicit RedisLogBasedActorInfoAccessor(RedisGcsClient *client_impl); - virtual ~RedisActorInfoAccessor() {} + virtual ~RedisLogBasedActorInfoAccessor() {} + + Status GetAll(std::vector *actor_table_data_list) override; Status AsyncGet(const ActorID &actor_id, const OptionalItemCallback &callback) override; @@ -66,6 +68,10 @@ class RedisActorInfoAccessor : public ActorInfoAccessor { const ActorID &actor_id, const OptionalItemCallback &callback) override; + protected: + virtual std::vector GetAllActorID() const; + virtual Status Get(const ActorID &actor_id, ActorTableData *actor_table_data) const; + private: /// Add checkpoint id to GCS asynchronously. /// @@ -76,7 +82,7 @@ class RedisActorInfoAccessor : public ActorInfoAccessor { const ActorCheckpointID &checkpoint_id, const StatusCallback &callback); - private: + protected: RedisGcsClient *client_impl_{nullptr}; // Use a random ClientID for actor subscription. Because: // If we use ClientID::Nil, GCS will still send all actors' updates to this GCS Client. @@ -86,6 +92,45 @@ class RedisActorInfoAccessor : public ActorInfoAccessor { // TODO(micafan): Remove this random id, once GCS becomes a service. ClientID subscribe_id_{ClientID::FromRandom()}; + private: + typedef SubscriptionExecutor + ActorSubscriptionExecutor; + ActorSubscriptionExecutor log_based_actor_sub_executor_; +}; + +/// \class RedisActorInfoAccessor +/// `RedisActorInfoAccessor` is an implementation of `ActorInfoAccessor` +/// that uses Redis as the backend storage. +class RedisActorInfoAccessor : public RedisLogBasedActorInfoAccessor { + public: + explicit RedisActorInfoAccessor(RedisGcsClient *client_impl); + + virtual ~RedisActorInfoAccessor() {} + + Status AsyncGet(const ActorID &actor_id, + const OptionalItemCallback &callback) override; + + Status AsyncRegister(const std::shared_ptr &data_ptr, + const StatusCallback &callback) override; + + Status AsyncUpdate(const ActorID &actor_id, + const std::shared_ptr &data_ptr, + const StatusCallback &callback) override; + + Status AsyncSubscribeAll(const SubscribeCallback &subscribe, + const StatusCallback &done) override; + + Status AsyncSubscribe(const ActorID &actor_id, + const SubscribeCallback &subscribe, + const StatusCallback &done) override; + + Status AsyncUnsubscribe(const ActorID &actor_id, const StatusCallback &done) override; + + protected: + std::vector GetAllActorID() const override; + Status Get(const ActorID &actor_id, ActorTableData *actor_table_data) const override; + + private: typedef SubscriptionExecutor ActorSubscriptionExecutor; ActorSubscriptionExecutor actor_sub_executor_; diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc index 35e9a6a6d..ea34b20ef 100644 --- a/src/ray/gcs/redis_context.cc +++ b/src/ray/gcs/redis_context.cc @@ -88,7 +88,6 @@ CallbackReply::CallbackReply(redisReply *redis_reply) : reply_type_(redis_reply- break; } case REDIS_REPLY_ARRAY: { - // Array replies are only used for pub-sub messages. Parse the published message. redisReply *message_type = redis_reply->element[0]; if (strcmp(message_type->str, "subscribe") == 0) { // If the message is for the initial subscription call, return the empty @@ -100,7 +99,13 @@ CallbackReply::CallbackReply(redisReply *redis_reply) : reply_type_(redis_reply- string_reply_ = std::string(message->str, message->len); RAY_CHECK(!string_reply_.empty()) << "Empty message received on subscribe channel."; } else { - RAY_LOG(FATAL) << "This is not a pubsub reply: data=" << message_type->str; + string_array_reply_.reserve(redis_reply->elements); + for (size_t i = 0; i < redis_reply->elements; ++i) { + auto *entry = redis_reply->element[i]; + RAY_CHECK(REDIS_REPLY_STRING == entry->type) + << "Unexcepted type: " << entry->type; + string_array_reply_.emplace_back(std::string(entry->str, entry->len)); + } } break; } @@ -122,16 +127,21 @@ Status CallbackReply::ReadAsStatus() const { return status_reply_; } -std::string CallbackReply::ReadAsString() const { +const std::string &CallbackReply::ReadAsString() const { RAY_CHECK(reply_type_ == REDIS_REPLY_STRING) << "Unexpected type: " << reply_type_; return string_reply_; } -std::string CallbackReply::ReadAsPubsubData() const { +const std::string &CallbackReply::ReadAsPubsubData() const { RAY_CHECK(reply_type_ == REDIS_REPLY_ARRAY) << "Unexpected type: " << reply_type_; return string_reply_; } +const std::vector &CallbackReply::ReadAsStringArray() const { + RAY_CHECK(reply_type_ == REDIS_REPLY_ARRAY) << "Unexpected type: " << reply_type_; + return string_array_reply_; +} + // This is a global redis callback which will be registered for every // asynchronous redis call. It dispatches the appropriate callback // that was registered with the RedisCallbackManager. @@ -271,6 +281,27 @@ Status RedisContext::Connect(const std::string &address, int port, bool sharding return Status::OK(); } +std::unique_ptr RedisContext::RunArgvSync( + const std::vector &args) { + RAY_CHECK(context_); + // Build the arguments. + std::vector argv; + std::vector argc; + for (const auto &arg : args) { + argv.push_back(arg.data()); + argc.push_back(arg.size()); + } + auto redis_reply = reinterpret_cast( + ::redisCommandArgv(context_, args.size(), argv.data(), argc.data())); + if (redis_reply == nullptr) { + RAY_LOG(ERROR) << "Failed to send redis command (sync)."; + return nullptr; + } + std::unique_ptr callback_reply(new CallbackReply(redis_reply)); + freeReplyObject(redis_reply); + return callback_reply; +} + Status RedisContext::RunArgvAsync(const std::vector &args) { RAY_CHECK(redis_async_context_); // Build the arguments. diff --git a/src/ray/gcs/redis_context.h b/src/ray/gcs/redis_context.h index 0735cb2d2..5573d92f2 100644 --- a/src/ray/gcs/redis_context.h +++ b/src/ray/gcs/redis_context.h @@ -62,10 +62,13 @@ class CallbackReply { /// /// Note that this will return an empty string if /// the type of this reply is `nil` or `status`. - std::string ReadAsString() const; + const std::string &ReadAsString() const; /// Read this reply data as pub-sub data. - std::string ReadAsPubsubData() const; + const std::string &ReadAsPubsubData() const; + + /// Read this reply data as a string array. + const std::vector &ReadAsStringArray() const; private: /// Flag indicating the type of reply this represents. @@ -77,9 +80,12 @@ class CallbackReply { /// Reply data if reply_type_ is REDIS_REPLY_STATUS. Status status_reply_; - /// Reply data if reply_type_ is REDIS_REPLY_STRING or REDIS_REPLY_ARRAY. - /// Note that REDIS_REPLY_ARRAY is only used for pub-sub data. + /// Reply data if reply_type_ is REDIS_REPLY_STRING. std::string string_reply_; + + /// Reply data if reply_type_ is REDIS_REPLY_ARRAY. + /// Note this is used when get actor table data. + std::vector string_array_reply_; }; /// Every callback should take in a vector of the results from the Redis @@ -168,6 +174,12 @@ class RedisContext { const TablePubsub pubsub_channel, int log_length = -1); + /// Run an arbitrary Redis command synchronously. + /// + /// \param args The vector of command args to pass to Redis. + /// \return CallbackReply(The reply from redis). + std::unique_ptr RunArgvSync(const std::vector &args); + /// Run an operation on some table key. /// /// \param command The command to run. This must match a registered Ray Redis diff --git a/src/ray/gcs/redis_gcs_client.cc b/src/ray/gcs/redis_gcs_client.cc index 26e39fa85..6d527e1f9 100644 --- a/src/ray/gcs/redis_gcs_client.cc +++ b/src/ray/gcs/redis_gcs_client.cc @@ -133,6 +133,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) { Attach(io_service); + log_based_actor_table_.reset(new LogBasedActorTable({primary_context_}, this)); actor_table_.reset(new ActorTable({primary_context_}, this)); // TODO(micafan) Modify ClientTable' Constructor(remove ClientID) in future. @@ -156,7 +157,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) { resource_table_.reset(new DynamicResourceTable({primary_context_}, this)); worker_failure_table_.reset(new WorkerFailureTable(shard_contexts_, this)); - actor_accessor_.reset(new RedisActorInfoAccessor(this)); + actor_accessor_.reset(new RedisLogBasedActorInfoAccessor(this)); job_accessor_.reset(new RedisJobInfoAccessor(this)); object_accessor_.reset(new RedisObjectInfoAccessor(this)); node_accessor_.reset(new RedisNodeInfoAccessor(this)); @@ -198,6 +199,7 @@ std::string RedisGcsClient::DebugString() const { std::stringstream result; result << "RedisGcsClient:"; result << "\n- TaskTable: " << raylet_task_table_->DebugString(); + result << "\n- LogBasedActorTable: " << log_based_actor_table_->DebugString(); result << "\n- ActorTable: " << actor_table_->DebugString(); result << "\n- TaskReconstructionLog: " << task_reconstruction_log_->DebugString(); result << "\n- TaskLeaseTable: " << task_lease_table_->DebugString(); @@ -213,6 +215,10 @@ ObjectTable &RedisGcsClient::object_table() { return *object_table_; } raylet::TaskTable &RedisGcsClient::raylet_task_table() { return *raylet_task_table_; } +LogBasedActorTable &RedisGcsClient::log_based_actor_table() { + return *log_based_actor_table_; +} + ActorTable &RedisGcsClient::actor_table() { return *actor_table_; } WorkerFailureTable &RedisGcsClient::worker_failure_table() { diff --git a/src/ray/gcs/redis_gcs_client.h b/src/ray/gcs/redis_gcs_client.h index 8fde4007a..55c51a90a 100644 --- a/src/ray/gcs/redis_gcs_client.h +++ b/src/ray/gcs/redis_gcs_client.h @@ -78,6 +78,7 @@ class RAY_EXPORT RedisGcsClient : public GcsClient { /// The following xxx_table methods implement the Accessor interfaces. /// Implements the Actors() interface. + LogBasedActorTable &log_based_actor_table(); ActorTable &actor_table(); ActorCheckpointTable &actor_checkpoint_table(); ActorCheckpointIdTable &actor_checkpoint_id_table(); @@ -113,6 +114,7 @@ class RAY_EXPORT RedisGcsClient : public GcsClient { std::unique_ptr object_table_; std::unique_ptr raylet_task_table_; + std::unique_ptr log_based_actor_table_; std::unique_ptr actor_table_; std::unique_ptr task_reconstruction_log_; std::unique_ptr task_lease_table_; diff --git a/src/ray/gcs/subscription_executor.cc b/src/ray/gcs/subscription_executor.cc index 90df8eaba..85a294ede 100644 --- a/src/ray/gcs/subscription_executor.cc +++ b/src/ray/gcs/subscription_executor.cc @@ -198,6 +198,7 @@ Status SubscriptionExecutor::AsyncUnsubscribe( return table_.CancelNotifications(JobID::Nil(), id, client_id, on_done); } +template class SubscriptionExecutor; template class SubscriptionExecutor; template class SubscriptionExecutor; template class SubscriptionExecutor; diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index ebc6aec34..9a833af15 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -746,6 +746,84 @@ Status TaskLeaseTable::Subscribe(const JobID &job_id, const ClientID &client_id, return Table::Subscribe(job_id, client_id, on_subscribe, done); } +std::vector SyncGetAllActorID(redisContext *redis_context, + const std::string &table_prefix) { + std::unordered_set actor_id_set; + size_t cursor = 0; + do { + auto r = redisCommand(redis_context, "SCAN %d match %s* count 100", cursor, + table_prefix.c_str()); + auto reply = reinterpret_cast(r); + RAY_CHECK(reply != nullptr && reply->type == REDIS_REPLY_ARRAY); + RAY_CHECK(reply->elements == 2); + + // current cursor + redisReply *cursor_reply = reply->element[0]; + RAY_CHECK(cursor_reply != nullptr && cursor_reply->type == REDIS_REPLY_STRING); + cursor = std::stoi(std::string(cursor_reply->str, cursor_reply->len)); + + // actor ids + redisReply *array_reply = reply->element[1]; + RAY_CHECK(array_reply != nullptr && array_reply->type == REDIS_REPLY_ARRAY); + for (size_t i = 0; i < array_reply->elements; ++i) { + redisReply *id_reply = array_reply->element[i]; + RAY_CHECK(id_reply != nullptr && id_reply->type == REDIS_REPLY_STRING); + auto id_with_prefix = std::string(id_reply->str, id_reply->len); + // The key of actor_checkpoint table and actor_checkpoint_id table have the same + // prefix of `ACTOR`, so we should check the length of the key to filter them. + if (id_with_prefix.size() == table_prefix.size() + ActorID::Size()) { + auto id = ActorID::FromBinary(id_with_prefix.substr(table_prefix.size())); + actor_id_set.emplace(id); + } + } + } while (cursor != 0); + std::vector actor_id_list; + actor_id_list.reserve(actor_id_set.size()); + actor_id_list.insert(actor_id_list.end(), actor_id_set.begin(), actor_id_set.end()); + return actor_id_list; +} + +std::vector LogBasedActorTable::GetAllActorID() { + auto redis_context = client_->primary_context()->sync_context(); + return SyncGetAllActorID(redis_context, TablePrefix_Name(prefix_)); +} + +Status LogBasedActorTable::Get(const ray::ActorID &actor_id, + ray::rpc::ActorTableData *actor_table_data) { + RAY_CHECK(actor_table_data != nullptr); + auto key = TablePrefix_Name(prefix_) + actor_id.Binary(); + auto reply = GetRedisContext(actor_id)->RunArgvSync({"LRANGE", key, "-1", "-1"}); + if (!reply || reply->IsNil()) { + return Status::IOError("Failed to get actor data by actor_id " + actor_id.Hex()); + } + + const auto &data_list = reply->ReadAsStringArray(); + if (data_list.empty()) { + return Status::IOError("Failed to get actor data by actor_id " + actor_id.Hex()); + } + + RAY_CHECK(data_list.size() == 1); + actor_table_data->ParseFromString(data_list.front()); + return Status::OK(); +} + +std::vector ActorTable::GetAllActorID() { + auto redis_context = client_->primary_context()->sync_context(); + return SyncGetAllActorID(redis_context, TablePrefix_Name(prefix_)); +} + +Status ActorTable::Get(const ray::ActorID &actor_id, + ray::rpc::ActorTableData *actor_table_data) { + RAY_CHECK(actor_table_data != nullptr); + auto key = TablePrefix_Name(prefix_) + actor_id.Binary(); + auto reply = GetRedisContext(actor_id)->RunArgvSync({"GET", key}); + if (!reply || reply->IsNil()) { + return Status::IOError("Failed to get actor data by actor_id " + actor_id.Hex()); + } + actor_table_data->ParseFromString(reply->ReadAsString()); + return Status::OK(); +} + Status ActorCheckpointIdTable::AddCheckpointId(const JobID &job_id, const ActorID &actor_id, const ActorCheckpointID &checkpoint_id, @@ -795,6 +873,7 @@ template class Log; template class Table; template class Table; template class Table; +template class Table; template class Log; template class Hash; diff --git a/src/ray/gcs/tables.h b/src/ray/gcs/tables.h index 89b80f6b2..b071fc9e1 100644 --- a/src/ray/gcs/tables.h +++ b/src/ray/gcs/tables.h @@ -716,19 +716,44 @@ class JobTable : public Log { virtual ~JobTable() {} }; -/// Actor table starts with an ALIVE entry, which represents the first time the actor -/// is created. This may be followed by 0 or more pairs of RECONSTRUCTING, ALIVE entries, -/// which represent each time the actor fails (RECONSTRUCTING) and gets recreated (ALIVE). -/// These may be followed by a DEAD entry, which means that the actor has failed and will -/// not be reconstructed. -class ActorTable : public Log { +/// Log-based Actor table starts with an ALIVE entry, which represents the first time the +/// actor is created. This may be followed by 0 or more pairs of RECONSTRUCTING, ALIVE +/// entries, which represent each time the actor fails (RECONSTRUCTING) and gets recreated +/// (ALIVE). These may be followed by a DEAD entry, which means that the actor has failed +/// and will not be reconstructed. +class LogBasedActorTable : public Log { public: - ActorTable(const std::vector> &contexts, - RedisGcsClient *client) + LogBasedActorTable(const std::vector> &contexts, + RedisGcsClient *client) : Log(contexts, client) { pubsub_channel_ = TablePubsub::ACTOR_PUBSUB; prefix_ = TablePrefix::ACTOR; } + + /// Get all actor id synchronously. + std::vector GetAllActorID(); + + /// Get actor table data by actor id synchronously. + Status Get(const ActorID &actor_id, ActorTableData *actor_table_data); +}; + +/// Actor table. +/// This table is only used for GCS-based actor management. And when completely migrate to +/// GCS service, the log-based actor table could be removed. +class ActorTable : public Table { + public: + ActorTable(const std::vector> &contexts, + RedisGcsClient *client) + : Table(contexts, client) { + pubsub_channel_ = TablePubsub::ACTOR_PUBSUB; + prefix_ = TablePrefix::ACTOR; + } + + /// Get all actor id synchronously. + std::vector GetAllActorID(); + + /// Get actor table data by actor id synchronously. + Status Get(const ActorID &actor_id, ActorTableData *actor_table_data); }; class WorkerFailureTable : public Table { diff --git a/src/ray/gcs/test/redis_actor_info_accessor_test.cc b/src/ray/gcs/test/redis_actor_info_accessor_test.cc index 22781c729..6c4879903 100644 --- a/src/ray/gcs/test/redis_actor_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_actor_info_accessor_test.cc @@ -79,7 +79,7 @@ TEST_F(ActorInfoAccessorTest, RegisterAndGet) { WaitPendingDone(wait_pending_timeout_); - // get + // async get for (const auto &elem : id_to_data_) { ++pending_count_; RAY_CHECK_OK(actor_accessor.AsyncGet( @@ -93,6 +93,15 @@ TEST_F(ActorInfoAccessorTest, RegisterAndGet) { } WaitPendingDone(wait_pending_timeout_); + + // sync get + std::vector actor_table_data_list; + RAY_CHECK_OK(actor_accessor.GetAll(&actor_table_data_list)); + ASSERT_EQ(id_to_data_.size(), actor_table_data_list.size()); + for (auto &data : actor_table_data_list) { + ActorID actor_id = ActorID::FromBinary(data.actor_id()); + ASSERT_TRUE(id_to_data_.count(actor_id) != 0); + } } TEST_F(ActorInfoAccessorTest, Subscribe) { diff --git a/src/ray/gcs/test/subscription_executor_test.cc b/src/ray/gcs/test/subscription_executor_test.cc index c74da6382..c4b6bc28b 100644 --- a/src/ray/gcs/test/subscription_executor_test.cc +++ b/src/ray/gcs/test/subscription_executor_test.cc @@ -25,12 +25,13 @@ namespace gcs { class SubscriptionExecutorTest : public AccessorTestBase { public: - typedef SubscriptionExecutor ActorSubExecutor; + typedef SubscriptionExecutor + ActorSubExecutor; virtual void SetUp() { AccessorTestBase::SetUp(); - actor_sub_executor_.reset(new ActorSubExecutor(gcs_client_->actor_table())); + actor_sub_executor_.reset(new ActorSubExecutor(gcs_client_->log_based_actor_table())); subscribe_ = [this](const ActorID &id, const ActorTableData &data) { const auto it = id_to_data_.find(id);