diff --git a/BUILD.bazel b/BUILD.bazel index 84db79b10..e2739bff7 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -1600,27 +1600,6 @@ cc_test( ], ) -cc_test( - name = "subscription_executor_test", - srcs = ["src/ray/gcs/test/subscription_executor_test.cc"], - args = [ - "$(location redis-server)", - "$(location redis-cli)", - "$(location libray_redis_module.so)", - ], - copts = COPTS, - data = [ - "//:libray_redis_module.so", - "//:redis-cli", - "//:redis-server", - ], - deps = [ - ":gcs", - ":gcs_test_util_lib", - "@com_google_googletest//:gtest_main", - ], -) - cc_test( name = "redis_job_info_accessor_test", srcs = ["src/ray/gcs/test/redis_job_info_accessor_test.cc"], diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index ef7c6c1dc..27e360b48 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -83,27 +83,6 @@ class ActorInfoAccessor { virtual Status AsyncCreateActor(const TaskSpecification &task_spec, const StatusCallback &callback) = 0; - /// Register an actor to GCS asynchronously. - /// - /// \param data_ptr The actor that will be registered to the GCS. - /// \param callback Callback that will be called after actor has been registered - /// to the GCS. - /// \return Status - virtual Status AsyncRegister(const std::shared_ptr &data_ptr, - const StatusCallback &callback) = 0; - - /// Update dynamic states of actor in GCS asynchronously. - /// - /// \param actor_id ID of the actor to update. - /// \param data_ptr Data of the actor to update. - /// \param callback Callback that will be called after update finishes. - /// \return Status - /// TODO(micafan) Don't expose the whole `ActorTableData` and only allow - /// updating dynamic states. - virtual Status AsyncUpdate(const ActorID &actor_id, - const std::shared_ptr &data_ptr, - const StatusCallback &callback) = 0; - /// Subscribe to any register or update operations of actors. /// /// \param subscribe Callback that will be called each time when an actor is registered diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc index 345322f0e..2524c4c31 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.cc +++ b/src/ray/gcs/gcs_client/service_based_accessor.cc @@ -211,58 +211,6 @@ Status ServiceBasedActorInfoAccessor::AsyncCreateActor( return Status::OK(); } -Status ServiceBasedActorInfoAccessor::AsyncRegister( - const std::shared_ptr &data_ptr, - const StatusCallback &callback) { - ActorID actor_id = ActorID::FromBinary(data_ptr->actor_id()); - RAY_LOG(DEBUG) << "Registering actor info, actor id = " << actor_id; - rpc::RegisterActorInfoRequest request; - request.mutable_actor_table_data()->CopyFrom(*data_ptr); - - auto operation = [this, request, actor_id, - callback](const SequencerDoneCallback &done_callback) { - client_impl_->GetGcsRpcClient().RegisterActorInfo( - request, [actor_id, callback, done_callback]( - const Status &status, const rpc::RegisterActorInfoReply &reply) { - if (callback) { - callback(status); - } - RAY_LOG(DEBUG) << "Finished registering actor info, status = " << status - << ", actor id = " << actor_id; - done_callback(); - }); - }; - - sequencer_.Post(actor_id, operation); - return Status::OK(); -} - -Status ServiceBasedActorInfoAccessor::AsyncUpdate( - const ActorID &actor_id, const std::shared_ptr &data_ptr, - const StatusCallback &callback) { - RAY_LOG(DEBUG) << "Updating actor info, actor id = " << actor_id; - rpc::UpdateActorInfoRequest request; - request.set_actor_id(actor_id.Binary()); - request.mutable_actor_table_data()->CopyFrom(*data_ptr); - - auto operation = [this, request, actor_id, - callback](const SequencerDoneCallback &done_callback) { - client_impl_->GetGcsRpcClient().UpdateActorInfo( - request, [actor_id, callback, done_callback]( - const Status &status, const rpc::UpdateActorInfoReply &reply) { - if (callback) { - callback(status); - } - RAY_LOG(DEBUG) << "Finished updating actor info, status = " << status - << ", actor id = " << actor_id; - done_callback(); - }); - }; - - sequencer_.Post(actor_id, operation); - return Status::OK(); -} - Status ServiceBasedActorInfoAccessor::AsyncSubscribeAll( const SubscribeCallback &subscribe, const StatusCallback &done) { diff --git a/src/ray/gcs/gcs_client/service_based_accessor.h b/src/ray/gcs/gcs_client/service_based_accessor.h index dd9ae406e..d5f67ffbd 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.h +++ b/src/ray/gcs/gcs_client/service_based_accessor.h @@ -88,13 +88,6 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor { Status AsyncCreateActor(const TaskSpecification &task_spec, const StatusCallback &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; 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 79b390a99..36d69604e 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 @@ -178,15 +178,6 @@ class ServiceBasedGcsClientTest : public ::testing::Test { return WaitReady(promise.get_future(), timeout_ms_); } - bool UpdateActor(const ActorID &actor_id, - const std::shared_ptr &actor_table_data) { - std::promise promise; - RAY_CHECK_OK(gcs_client_->Actors().AsyncUpdate( - actor_id, actor_table_data, - [&promise](Status status) { promise.set_value(status.ok()); })); - return WaitReady(promise.get_future(), timeout_ms_); - } - rpc::ActorTableData GetActor(const ActorID &actor_id) { std::promise promise; rpc::ActorTableData actor_table_data; diff --git a/src/ray/gcs/redis_accessor.cc b/src/ray/gcs/redis_accessor.cc index 0287af67b..c2d945bb1 100644 --- a/src/ray/gcs/redis_accessor.cc +++ b/src/ray/gcs/redis_accessor.cc @@ -82,59 +82,6 @@ Status RedisLogBasedActorInfoAccessor::AsyncCreateActor( return Status::Invalid(error_msg); } -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) { - if (callback != nullptr) { - callback(Status::OK()); - } - }; - - auto on_failure = [callback](RedisGcsClient *client, const ActorID &actor_id, - const ActorTableData &data) { - if (callback != nullptr) { - callback(Status::Invalid("Adding actor failed.")); - } - }; - - ActorID actor_id = ActorID::FromBinary(data_ptr->actor_id()); - return client_impl_->log_based_actor_table().AppendAt(actor_id.JobId(), actor_id, - data_ptr, on_success, on_failure, - /*log_length*/ 0); -} - -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 - // of (RESTARTING, ALIVE) entries, where N is the maximum number of - // reconstructions. This is followed optionally by a DEAD entry. - int log_length = 2 * (data_ptr->num_restarts()); - if (data_ptr->state() != ActorTableData::ALIVE) { - // RESTARTING or DEAD entries have an odd index. - log_length += 1; - } - RAY_LOG(DEBUG) << "AsyncUpdate actor state to " << data_ptr->state() - << ", actor id: " << actor_id << ", log_length: " << log_length; - auto on_success = [callback](RedisGcsClient *client, const ActorID &actor_id, - const ActorTableData &data) { - if (callback != nullptr) { - callback(Status::OK()); - } - }; - - auto on_failure = [callback](RedisGcsClient *client, const ActorID &actor_id, - const ActorTableData &data) { - if (callback != nullptr) { - callback(Status::Invalid("Updating actor failed.")); - } - }; - - return client_impl_->log_based_actor_table().AppendAt( - actor_id.JobId(), actor_id, data_ptr, on_success, on_failure, log_length); -} - Status RedisLogBasedActorInfoAccessor::AsyncSubscribeAll( const SubscribeCallback &subscribe, const StatusCallback &done) { @@ -285,32 +232,6 @@ Status RedisActorInfoAccessor::AsyncGetAll( return Status::OK(); } -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) { diff --git a/src/ray/gcs/redis_accessor.h b/src/ray/gcs/redis_accessor.h index 9d93ec2c5..f3a3eb4cb 100644 --- a/src/ray/gcs/redis_accessor.h +++ b/src/ray/gcs/redis_accessor.h @@ -58,13 +58,6 @@ class RedisLogBasedActorInfoAccessor : public ActorInfoAccessor { Status AsyncCreateActor(const TaskSpecification &task_spec, const StatusCallback &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; @@ -139,13 +132,6 @@ class RedisActorInfoAccessor : public RedisLogBasedActorInfoAccessor { "RedisActorInfoAccessor does not support named detached actors."); } - 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; 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 b7f91740e..2f2700f7a 100644 --- a/src/ray/gcs/test/redis_actor_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_actor_info_accessor_test.cc @@ -65,45 +65,6 @@ class ActorInfoAccessorTest : public AccessorTestBase { size_t checkpoint_number_{2}; }; -TEST_F(ActorInfoAccessorTest, RegisterAndGet) { - ActorInfoAccessor &actor_accessor = gcs_client_->Actors(); - // register - for (const auto &elem : id_to_data_) { - const auto &actor = elem.second; - ++pending_count_; - RAY_CHECK_OK(actor_accessor.AsyncRegister(actor, [this](Status status) { - RAY_CHECK_OK(status); - --pending_count_; - })); - } - - WaitPendingDone(wait_pending_timeout_); - - // async get - for (const auto &elem : id_to_data_) { - ++pending_count_; - RAY_CHECK_OK(actor_accessor.AsyncGet( - elem.first, [this](Status status, const boost::optional &data) { - ASSERT_TRUE(data); - ActorID actor_id = ActorID::FromBinary(data->actor_id()); - auto it = id_to_data_.find(actor_id); - ASSERT_TRUE(it != id_to_data_.end()); - --pending_count_; - })); - } - - 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) { ActorInfoAccessor &actor_accessor = gcs_client_->Actors(); // subscribe @@ -124,24 +85,6 @@ TEST_F(ActorInfoAccessorTest, Subscribe) { RAY_CHECK_OK(actor_accessor.AsyncSubscribeAll(subscribe, done)); // Wait until subscribe finishes. WaitPendingDone(do_sub_pending_count, wait_pending_timeout_); - - // register - std::atomic register_pending_count(0); - for (const auto &elem : id_to_data_) { - const auto &actor = elem.second; - ++sub_pending_count; - ++register_pending_count; - RAY_CHECK_OK( - actor_accessor.AsyncRegister(actor, [®ister_pending_count](Status status) { - RAY_CHECK_OK(status); - --register_pending_count; - })); - } - // Wait until register finishes. - WaitPendingDone(register_pending_count, wait_pending_timeout_); - - // Wait for all subscribe notifications. - WaitPendingDone(sub_pending_count, wait_pending_timeout_); } TEST_F(ActorInfoAccessorTest, GetActorCheckpointTest) { diff --git a/src/ray/gcs/test/subscription_executor_test.cc b/src/ray/gcs/test/subscription_executor_test.cc deleted file mode 100644 index 44326bf97..000000000 --- a/src/ray/gcs/test/subscription_executor_test.cc +++ /dev/null @@ -1,225 +0,0 @@ -// Copyright 2017 The Ray Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "ray/gcs/subscription_executor.h" - -#include "gtest/gtest.h" -#include "ray/gcs/callback.h" -#include "ray/gcs/entry_change_notification.h" -#include "ray/gcs/redis_gcs_client.h" -#include "ray/gcs/test/accessor_test_base.h" - -namespace ray { - -namespace gcs { - -class SubscriptionExecutorTest : public AccessorTestBase { - public: - typedef SubscriptionExecutor - ActorSubExecutor; - - virtual void SetUp() { - AccessorTestBase::SetUp(); - - 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); - ASSERT_TRUE(it != id_to_data_.end()); - --sub_pending_count_; - }; - - sub_done_ = [this](Status status) { - ASSERT_TRUE(status.ok()) << status; - --do_sub_pending_count_; - }; - - unsub_done_ = [this](Status status) { - ASSERT_TRUE(status.ok()) << status; - --do_unsub_pending_count_; - }; - } - - virtual void TearDown() { - AccessorTestBase::TearDown(); - ASSERT_EQ(sub_pending_count_, 0); - ASSERT_EQ(do_sub_pending_count_, 0); - ASSERT_EQ(do_unsub_pending_count_, 0); - } - - protected: - virtual void GenTestData() { - for (size_t i = 0; i < 100; ++i) { - std::shared_ptr actor = std::make_shared(); - actor->set_max_restarts(1); - actor->set_num_restarts(0); - JobID job_id = JobID::FromInt(i); - actor->set_job_id(job_id.Binary()); - actor->set_state(ActorTableData::ALIVE); - ActorID actor_id = ActorID::Of(job_id, RandomTaskId(), /*parent_task_counter=*/i); - actor->set_actor_id(actor_id.Binary()); - id_to_data_[actor_id] = actor; - } - } - - size_t AsyncRegisterActorToGcs() { - ActorInfoAccessor &actor_accessor = gcs_client_->Actors(); - for (const auto &elem : id_to_data_) { - const auto &actor = elem.second; - auto done = [this](Status status) { - ASSERT_TRUE(status.ok()); - --pending_count_; - }; - ++pending_count_; - Status status = actor_accessor.AsyncRegister(actor, done); - RAY_CHECK_OK(status); - } - return id_to_data_.size(); - } - - protected: - std::unique_ptr actor_sub_executor_; - - std::atomic sub_pending_count_{0}; - std::atomic do_sub_pending_count_{0}; - std::atomic do_unsub_pending_count_{0}; - - SubscribeCallback subscribe_{nullptr}; - StatusCallback sub_done_{nullptr}; - StatusCallback unsub_done_{nullptr}; -}; - -TEST_F(SubscriptionExecutorTest, SubscribeAllTest) { - ++do_sub_pending_count_; - Status status = - actor_sub_executor_->AsyncSubscribeAll(NodeID::Nil(), subscribe_, sub_done_); - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - ASSERT_TRUE(status.ok()); - sub_pending_count_ = id_to_data_.size(); - AsyncRegisterActorToGcs(); - status = actor_sub_executor_->AsyncSubscribeAll(NodeID::Nil(), subscribe_, sub_done_); - ASSERT_TRUE(status.IsInvalid()); - WaitPendingDone(sub_pending_count_, wait_pending_timeout_); -} - -TEST_F(SubscriptionExecutorTest, SubscribeOneWithClientIDTest) { - const auto &item = id_to_data_.begin(); - ++do_sub_pending_count_; - ++sub_pending_count_; - Status status = actor_sub_executor_->AsyncSubscribe(NodeID::FromRandom(), item->first, - subscribe_, sub_done_); - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - ASSERT_TRUE(status.ok()); - AsyncRegisterActorToGcs(); - WaitPendingDone(sub_pending_count_, wait_pending_timeout_); - status = actor_sub_executor_->AsyncSubscribe(NodeID::FromRandom(), item->first, - subscribe_, sub_done_); - ASSERT_TRUE(status.IsInvalid()); -} - -TEST_F(SubscriptionExecutorTest, SubscribeOneAfterActorRegistrationWithClientIDTest) { - const auto &item = id_to_data_.begin(); - ++do_sub_pending_count_; - ++sub_pending_count_; - AsyncRegisterActorToGcs(); - Status status = actor_sub_executor_->AsyncSubscribe(NodeID::FromRandom(), item->first, - subscribe_, sub_done_); - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - ASSERT_TRUE(status.ok()); - WaitPendingDone(sub_pending_count_, wait_pending_timeout_); - status = actor_sub_executor_->AsyncSubscribe(NodeID::FromRandom(), item->first, - subscribe_, sub_done_); - ASSERT_TRUE(status.IsInvalid()); -} - -TEST_F(SubscriptionExecutorTest, SubscribeAllAndSubscribeOneTest) { - ++do_sub_pending_count_; - Status status = - actor_sub_executor_->AsyncSubscribeAll(NodeID::Nil(), subscribe_, sub_done_); - ASSERT_TRUE(status.ok()); - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - for (const auto &item : id_to_data_) { - status = actor_sub_executor_->AsyncSubscribe(NodeID::FromRandom(), item.first, - subscribe_, sub_done_); - ASSERT_FALSE(status.ok()); - } - sub_pending_count_ = id_to_data_.size(); - AsyncRegisterActorToGcs(); - WaitPendingDone(sub_pending_count_, wait_pending_timeout_); -} - -TEST_F(SubscriptionExecutorTest, UnsubscribeTest) { - NodeID client_id = NodeID::FromRandom(); - Status status; - for (const auto &item : id_to_data_) { - status = actor_sub_executor_->AsyncUnsubscribe(client_id, item.first, unsub_done_); - ASSERT_TRUE(status.IsInvalid()); - } - - for (const auto &item : id_to_data_) { - ++do_sub_pending_count_; - status = - actor_sub_executor_->AsyncSubscribe(client_id, item.first, subscribe_, sub_done_); - ASSERT_TRUE(status.ok()); - } - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - for (const auto &item : id_to_data_) { - ++do_unsub_pending_count_; - status = actor_sub_executor_->AsyncUnsubscribe(client_id, item.first, unsub_done_); - ASSERT_TRUE(status.ok()); - } - WaitPendingDone(do_unsub_pending_count_, wait_pending_timeout_); - for (const auto &item : id_to_data_) { - status = actor_sub_executor_->AsyncUnsubscribe(client_id, item.first, unsub_done_); - ASSERT_TRUE(!status.ok()); - } - - for (const auto &item : id_to_data_) { - ++do_sub_pending_count_; - status = - actor_sub_executor_->AsyncSubscribe(client_id, item.first, subscribe_, sub_done_); - ASSERT_TRUE(status.ok()); - } - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - for (const auto &item : id_to_data_) { - ++do_unsub_pending_count_; - status = actor_sub_executor_->AsyncUnsubscribe(client_id, item.first, unsub_done_); - ASSERT_TRUE(status.ok()); - } - WaitPendingDone(do_unsub_pending_count_, wait_pending_timeout_); - for (const auto &item : id_to_data_) { - ++do_sub_pending_count_; - status = - actor_sub_executor_->AsyncSubscribe(client_id, item.first, subscribe_, sub_done_); - ASSERT_TRUE(status.ok()); - } - WaitPendingDone(do_sub_pending_count_, wait_pending_timeout_); - sub_pending_count_ = id_to_data_.size(); - AsyncRegisterActorToGcs(); - WaitPendingDone(pending_count_, wait_pending_timeout_); - WaitPendingDone(sub_pending_count_, wait_pending_timeout_); -} - -} // namespace gcs - -} // namespace ray - -int main(int argc, char **argv) { - ::testing::InitGoogleTest(&argc, argv); - RAY_CHECK(argc == 4); - ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; - ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; - return RUN_ALL_TESTS(); -}