From efbaf155b2d5333c7530a5dc91dd8aa94db49322 Mon Sep 17 00:00:00 2001 From: fangfengbin <869218239a@zju.edu.cn> Date: Wed, 15 Apr 2020 19:24:52 +0800 Subject: [PATCH] [GCS]Add publish and subscribe function of gcs table (#7909) --- BUILD.bazel | 54 +++-- src/ray/gcs/pubsub/gcs_pub_sub.cc | 100 +++++++++ src/ray/gcs/pubsub/gcs_pub_sub.h | 99 +++++++++ src/ray/gcs/pubsub/message_publisher.h | 33 --- src/ray/gcs/pubsub/redis_message_publisher.cc | 44 ---- src/ray/gcs/pubsub/redis_message_publisher.h | 32 --- src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc | 189 ++++++++++++++++++ src/ray/gcs/redis_context.cc | 42 +++- src/ray/gcs/redis_context.h | 28 +++ src/ray/protobuf/gcs.proto | 6 + 10 files changed, 499 insertions(+), 128 deletions(-) create mode 100644 src/ray/gcs/pubsub/gcs_pub_sub.cc create mode 100644 src/ray/gcs/pubsub/gcs_pub_sub.h delete mode 100644 src/ray/gcs/pubsub/message_publisher.h delete mode 100644 src/ray/gcs/pubsub/redis_message_publisher.cc delete mode 100644 src/ray/gcs/pubsub/redis_message_publisher.h create mode 100644 src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc diff --git a/BUILD.bazel b/BUILD.bazel index adf15c46c..ef5261562 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -305,6 +305,42 @@ cc_binary( ], ) +cc_library( + name = "gcs_pub_sub_lib", + srcs = glob( + [ + "src/ray/gcs/pubsub/gcs_pub_sub.cc", + ], + ), + hdrs = glob( + [ + "src/ray/gcs/pubsub/gcs_pub_sub.h", + ], + ), + copts = COPTS, + deps = [ + ":gcs", + ":ray_common", + ":redis_client", + ], +) + +cc_test( + name = "gcs_pub_sub_test", + srcs = ["src/ray/gcs/pubsub/test/gcs_pub_sub_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_pub_sub_lib", + "@com_google_googletest//:gtest_main", + ], +) + cc_library( name = "gcs_server_lib", srcs = glob( @@ -1075,24 +1111,6 @@ cc_test( ], ) -cc_library( - name = "gcs_pubsub", - srcs = [ - "src/ray/gcs/pubsub/redis_message_publisher.cc", - ], - hdrs = [ - "src/ray/gcs/callback.h", - "src/ray/gcs/pubsub/message_publisher.h", - "src/ray/gcs/pubsub/redis_message_publisher.h", - ], - copts = COPTS, - deps = [ - ":ray_common", - ":ray_util", - ":redis_client", - ], -) - cc_library( name = "gcs", srcs = glob( diff --git a/src/ray/gcs/pubsub/gcs_pub_sub.cc b/src/ray/gcs/pubsub/gcs_pub_sub.cc new file mode 100644 index 000000000..aacf34d2e --- /dev/null +++ b/src/ray/gcs/pubsub/gcs_pub_sub.cc @@ -0,0 +1,100 @@ +// 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 "gcs_pub_sub.h" + +namespace ray { +namespace gcs { + +Status GcsPubSub::Publish(const std::string &channel, const std::string &id, + const std::string &data, const StatusCallback &done) { + rpc::PubSubMessage message; + message.set_id(id); + message.set_data(data); + + auto on_done = [done](std::shared_ptr reply) { + if (done) { + done(Status::OK()); + } + }; + + return redis_client_->GetPrimaryContext()->PublishAsync( + GenChannelPattern(channel, boost::optional(id)), + message.SerializeAsString(), on_done); +} + +Status GcsPubSub::Subscribe(const std::string &channel, const std::string &id, + const Callback &subscribe, const StatusCallback &done) { + return SubscribeInternal(channel, subscribe, done, boost::optional(id)); +} + +Status GcsPubSub::SubscribeAll(const std::string &channel, const Callback &subscribe, + const StatusCallback &done) { + return SubscribeInternal(channel, subscribe, done); +} + +Status GcsPubSub::Unsubscribe(const std::string &channel, const std::string &id) { + return redis_client_->GetPrimaryContext()->PUnsubscribeAsync( + GenChannelPattern(channel, boost::optional(id))); +} + +Status GcsPubSub::SubscribeInternal(const std::string &channel, const Callback &subscribe, + const StatusCallback &done, + const boost::optional &id) { + std::string pattern = GenChannelPattern(channel, id); + auto callback = [this, pattern, subscribe](std::shared_ptr reply) { + if (!reply->IsNil()) { + if (reply->IsUnsubscribeCallback()) { + absl::MutexLock lock(&mutex_); + ray::gcs::RedisCallbackManager::instance().remove( + subscribe_callback_index_[pattern]); + subscribe_callback_index_.erase(pattern); + } else { + const auto reply_data = reply->ReadAsPubsubData(); + if (!reply_data.empty()) { + rpc::PubSubMessage message; + message.ParseFromString(reply_data); + subscribe(message.id(), message.data()); + } + } + } + }; + + int64_t out_callback_index; + auto status = redis_client_->GetPrimaryContext()->PSubscribeAsync(pattern, callback, + &out_callback_index); + if (id) { + absl::MutexLock lock(&mutex_); + subscribe_callback_index_[pattern] = out_callback_index; + } + if (done) { + done(status); + } + return status; +} + +std::string GcsPubSub::GenChannelPattern(const std::string &channel, + const boost::optional &id) { + std::stringstream pattern; + pattern << channel << ":"; + if (id) { + pattern << *id; + } else { + pattern << "*"; + } + return pattern.str(); +} + +} // namespace gcs +} // namespace ray diff --git a/src/ray/gcs/pubsub/gcs_pub_sub.h b/src/ray/gcs/pubsub/gcs_pub_sub.h new file mode 100644 index 000000000..e28141ee4 --- /dev/null +++ b/src/ray/gcs/pubsub/gcs_pub_sub.h @@ -0,0 +1,99 @@ +// 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. + +#ifndef RAY_GCS_GCS_PUB_SUB_H_ +#define RAY_GCS_GCS_PUB_SUB_H_ + +#include "absl/synchronization/mutex.h" + +#include "ray/gcs/callback.h" +#include "ray/gcs/redis_client.h" +#include "ray/gcs/redis_context.h" +#include "ray/protobuf/gcs.pb.h" + +namespace ray { +namespace gcs { + +/// \class GcsPubSub +/// +/// GcsPubSub supports publishing, subscription and unsubscribing of data. +/// This class is thread safe. +class GcsPubSub { + public: + /// The callback is called when a subscription message is received. + using Callback = std::function; + + explicit GcsPubSub(std::shared_ptr redis_client) + : redis_client_(redis_client) {} + + virtual ~GcsPubSub() = default; + + /// Posts a message to the given channel. + /// + /// \param channel The channel to publish to redis. + /// \param id The id of message to be published to redis. + /// \param data The data of message to be published to redis. + /// \param done Callback that will be called when the message is published to redis. + /// \return Status + Status Publish(const std::string &channel, const std::string &id, + const std::string &data, const StatusCallback &done); + + /// Subscribe to messages with the specified ID under the specified channel. + /// + /// \param channel The channel to subscribe from redis. + /// \param id The id of message to be subscribed from redis. + /// \param subscribe Callback that will be called when a subscription message is + /// received. + /// \param done Callback that will be called when subscription is complete. + /// \return Status + Status Subscribe(const std::string &channel, const std::string &id, + const Callback &subscribe, const StatusCallback &done); + + /// Subscribe to messages with the specified channel. + /// + /// \param channel The channel to subscribe from redis. + /// \param subscribe Callback that will be called when a subscription message is + /// received. + /// \param done Callback that will be called when subscription is complete. + /// \return Status + Status SubscribeAll(const std::string &channel, const Callback &subscribe, + const StatusCallback &done); + + /// Unsubscribe to messages with the specified ID under the specified channel. + /// + /// \param channel The channel to unsubscribe from redis. + /// \param id The id of message to be unsubscribed from redis. + /// \return Status + Status Unsubscribe(const std::string &channel, const std::string &id); + + private: + Status SubscribeInternal(const std::string &channel, const Callback &subscribe, + const StatusCallback &done, + const boost::optional &id = boost::none); + + std::string GenChannelPattern(const std::string &channel, + const boost::optional &id); + + std::shared_ptr redis_client_; + + /// Mutex to protect the subscribe_callback_index_ field. + absl::Mutex mutex_; + + std::unordered_map subscribe_callback_index_ GUARDED_BY(mutex_); +}; + +} // namespace gcs +} // namespace ray + +#endif // RAY_GCS_GCS_PUB_SUB_H_ diff --git a/src/ray/gcs/pubsub/message_publisher.h b/src/ray/gcs/pubsub/message_publisher.h deleted file mode 100644 index 4b2660d48..000000000 --- a/src/ray/gcs/pubsub/message_publisher.h +++ /dev/null @@ -1,33 +0,0 @@ -#ifndef RAY_GCS_PUBSUB_MESSAGE_PUBLISHER_H -#define RAY_GCS_PUBSUB_MESSAGE_PUBLISHER_H - -#include -#include "ray/common/status.h" -#include "ray/gcs/callback.h" -#include "ray/util/io_service_pool.h" -#include "ray/util/logging.h" - -namespace ray { - -namespace gcs { - -class MessagePublisher { - public: - virtual ~MessagePublisher() {} - - virtual Status Init() = 0; - - virtual void Shutdown() = 0; - - virtual Status PublishMessage(const std::string &channel, const std::string &message, - const StatusCallback &callback) = 0; - - protected: - MessagePublisher() {} -}; - -} // namespace gcs - -} // namespace ray - -#endif // RAY_GCS_PUBSUB_MESSAGE_PUBLISHER_H diff --git a/src/ray/gcs/pubsub/redis_message_publisher.cc b/src/ray/gcs/pubsub/redis_message_publisher.cc deleted file mode 100644 index 3131b5506..000000000 --- a/src/ray/gcs/pubsub/redis_message_publisher.cc +++ /dev/null @@ -1,44 +0,0 @@ -#include "ray/gcs/pubsub/redis_message_publisher.h" -#include "ray/gcs/redis_context.h" - -namespace ray { - -namespace gcs { - -RedisMessagePublisher::RedisMessagePublisher( - const RedisClientOptions &options, std::shared_ptr io_service_pool) - : redis_client_(new RedisClient(options)), - io_service_pool_(std::move(io_service_pool)) {} - -Status RedisMessagePublisher::Init() { - auto io_services = io_service_pool_->GetAll(); - Status status = redis_client_->Connect(io_services); - RAY_LOG(INFO) << "RedisMessagePublisher::Init finished with status " - << status.ToString(); - return status; -} - -void RedisMessagePublisher::Shutdown() { - redis_client_->Disconnect(); - RAY_LOG(INFO) << "RedisMessagePublisher::Shutdown."; -} - -Status RedisMessagePublisher::PublishMessage(const std::string &channel, - const std::string &message, - const StatusCallback &callback) { - std::vector args = {"PUBLISH", channel, message}; - - RedisCallback pub_callback = nullptr; - if (callback) { - pub_callback = [callback](std::shared_ptr reply) { - callback(Status::OK()); - }; - } - // Select shard context by channel. - auto shard_context = redis_client_->GetShardContext(channel); - return shard_context->RunArgvAsync(args, pub_callback); -} - -} // namespace gcs - -} // namespace ray diff --git a/src/ray/gcs/pubsub/redis_message_publisher.h b/src/ray/gcs/pubsub/redis_message_publisher.h deleted file mode 100644 index 30833dc88..000000000 --- a/src/ray/gcs/pubsub/redis_message_publisher.h +++ /dev/null @@ -1,32 +0,0 @@ -#ifndef RAY_GCS_PUBSUB_REDIS_MESSAGE_PUBLISHER_H -#define RAY_GCS_PUBSUB_REDIS_MESSAGE_PUBLISHER_H - -#include "ray/gcs/pubsub/message_publisher.h" -#include "ray/gcs/redis_client.h" - -namespace ray { - -namespace gcs { - -class RedisMessagePublisher : public MessagePublisher { - public: - RedisMessagePublisher(const RedisClientOptions &options, - std::shared_ptr io_service_pool); - - Status Init() override; - - void Shutdown() override; - - Status PublishMessage(const std::string &channel, const std::string &message, - const StatusCallback &callback) override; - - private: - std::shared_ptr redis_client_; - std::shared_ptr io_service_pool_; -}; - -} // namespace gcs - -} // namespace ray - -#endif // RAY_GCS_PUBSUB_REDIS_MESSAGE_PUBLISHER_H diff --git a/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc b/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc new file mode 100644 index 000000000..25f4970b5 --- /dev/null +++ b/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc @@ -0,0 +1,189 @@ +// 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 + +#include "gtest/gtest.h" +#include "ray/common/test_util.h" +#include "ray/gcs/pubsub/gcs_pub_sub.h" + +namespace ray { + +class GcsPubSubTest : public RedisServiceManagerForTest { + protected: + virtual void SetUp() override { + thread_io_service_.reset(new std::thread([this] { + std::unique_ptr work( + new boost::asio::io_service::work(io_service_)); + io_service_.run(); + })); + + gcs::RedisClientOptions redis_client_options("127.0.0.1", REDIS_SERVER_PORT, "", + true); + client_ = std::make_shared(redis_client_options); + RAY_CHECK_OK(client_->Connect(io_service_)); + pub_sub_ = std::make_shared(client_); + } + + virtual void TearDown() override { + pub_sub_.reset(); + client_->Disconnect(); + io_service_.stop(); + client_.reset(); + thread_io_service_->join(); + thread_io_service_.reset(); + } + + void Subscribe(const std::string &channel, const std::string &id, + std::vector &result) { + std::promise promise; + auto done = [&promise](Status status) { promise.set_value(status.ok()); }; + auto subscribe = [&result](const std::string &id, const std::string &data) { + result.push_back(data); + }; + RAY_CHECK_OK((pub_sub_->Subscribe(channel, id, subscribe, done))); + WaitReady(promise.get_future(), timeout_ms_); + } + + void SubscribeAll(const std::string &channel, + std::vector> &result) { + std::promise promise; + auto done = [&promise](Status status) { promise.set_value(status.ok()); }; + auto subscribe = [&result](const std::string &id, const std::string &data) { + result.push_back(std::make_pair(id, data)); + }; + RAY_CHECK_OK((pub_sub_->SubscribeAll(channel, subscribe, done))); + WaitReady(promise.get_future(), timeout_ms_); + } + + bool Unsubscribe(const std::string &channel, const std::string &id) { + return pub_sub_->Unsubscribe(channel, id).ok(); + } + + bool Publish(const std::string &channel, const std::string &id, + const std::string &data) { + std::promise promise; + auto done = [&promise](Status status) { promise.set_value(status.ok()); }; + RAY_CHECK_OK((pub_sub_->Publish(channel, id, data, done))); + return WaitReady(promise.get_future(), timeout_ms_); + } + + bool WaitReady(std::future future, const std::chrono::milliseconds &timeout_ms) { + auto status = future.wait_for(timeout_ms); + return status == std::future_status::ready && future.get(); + } + + template + void WaitPendingDone(const std::vector &data, int expected_count) { + auto condition = [&data, expected_count]() { + return (int)data.size() == expected_count; + }; + EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); + } + + std::shared_ptr client_; + const std::chrono::milliseconds timeout_ms_{60000}; + std::shared_ptr pub_sub_; + + private: + boost::asio::io_service io_service_; + std::unique_ptr thread_io_service_; +}; + +TEST_F(GcsPubSubTest, TestPubSubApi) { + std::string channel("channel"); + std::string id("id"); + std::string data("data"); + std::vector> all_result; + SubscribeAll(channel, all_result); + std::vector result; + Subscribe(channel, id, result); + Publish(channel, id, data); + + WaitPendingDone(result, 1); + WaitPendingDone(all_result, 1); + Unsubscribe(channel, id); + Publish(channel, id, data); + usleep(100 * 1000); + EXPECT_EQ(result.size(), 1); + + Subscribe(channel, id, result); + Publish(channel, id, data); + WaitPendingDone(result, 2); + WaitPendingDone(all_result, 3); +} + +TEST_F(GcsPubSubTest, TestMultithreading) { + std::string channel("channel"); + auto sub_message_count = std::make_shared>(0); + auto sub_finished_count = std::make_shared>(0); + int size = 5; + std::vector> threads; + threads.resize(size); + for (int index = 0; index < size; ++index) { + std::stringstream ss; + ss << index; + auto id = ss.str(); + threads[index].reset( + new std::thread([this, sub_message_count, sub_finished_count, id, channel] { + auto subscribe = [sub_message_count](const std::string &id, + const std::string &data) { + ++(*sub_message_count); + }; + auto on_done = [sub_finished_count](Status status) { + RAY_CHECK_OK(status); + ++(*sub_finished_count); + }; + RAY_CHECK_OK(pub_sub_->Subscribe(channel, id, subscribe, on_done)); + })); + } + auto sub_finished_condition = [sub_finished_count, size]() { + return sub_finished_count->load() == size; + }; + EXPECT_TRUE(WaitForCondition(sub_finished_condition, timeout_ms_.count())); + for (auto &thread : threads) { + thread->join(); + thread.reset(); + } + + std::string data("data"); + for (int index = 0; index < size; ++index) { + std::stringstream ss; + ss << index; + auto id = ss.str(); + threads[index].reset(new std::thread([this, channel, id, data] { + RAY_CHECK_OK(pub_sub_->Publish(channel, id, data, nullptr)); + })); + } + + auto sub_message_condition = [sub_message_count, size]() { + return sub_message_count->load() == size; + }; + EXPECT_TRUE(WaitForCondition(sub_message_condition, timeout_ms_.count())); + for (auto &thread : threads) { + thread->join(); + thread.reset(); + } +} + +} // namespace ray + +int main(int argc, char **argv) { + ::testing::InitGoogleTest(&argc, argv); + RAY_CHECK(argc == 4); + ray::REDIS_SERVER_EXEC_PATH = argv[1]; + ray::REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + return RUN_ALL_TESTS(); +} diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc index 1cddcd781..c9185154c 100644 --- a/src/ray/gcs/redis_context.cc +++ b/src/ray/gcs/redis_context.cc @@ -89,15 +89,24 @@ CallbackReply::CallbackReply(redisReply *redis_reply) : reply_type_(redis_reply- } case REDIS_REPLY_ARRAY: { redisReply *message_type = redis_reply->element[0]; - if (strcmp(message_type->str, "subscribe") == 0) { + if (strcmp(message_type->str, "subscribe") == 0 || + strcmp(message_type->str, "psubscribe") == 0) { // If the message is for the initial subscription call, return the empty // string as a response to signify that subscription was successful. + } else if (strcmp(message_type->str, "punsubscribe") == 0) { + is_unsubscribe_callback_ = true; } else if (strcmp(message_type->str, "message") == 0) { // If the message is from a PUBLISH, make sure the data is nonempty. redisReply *message = redis_reply->element[redis_reply->elements - 1]; // data is a notification message. string_reply_ = std::string(message->str, message->len); RAY_CHECK(!string_reply_.empty()) << "Empty message received on subscribe channel."; + } else if (strcmp(message_type->str, "pmessage") == 0) { + // If the message is from a PUBLISH, make sure the data is nonempty. + redisReply *message = redis_reply->element[redis_reply->elements - 1]; + // data is a notification message. + string_reply_ = std::string(message->str, message->len); + RAY_CHECK(!string_reply_.empty()) << "Empty message received on subscribe channel."; } else { // Array replies are used for scan or get. ParseAsStringArrayOrScanArray(redis_reply); @@ -391,6 +400,37 @@ Status RedisContext::SubscribeAsync(const ClientID &client_id, return status; } +Status RedisContext::PSubscribeAsync(const std::string &pattern, + const RedisCallback &redisCallback, + int64_t *out_callback_index) { + RAY_CHECK(async_redis_subscribe_context_); + + int64_t callback_index = + RedisCallbackManager::instance().add(redisCallback, true, io_service_); + *out_callback_index = callback_index; + std::string redis_command = "PSUBSCRIBE %b"; + return async_redis_subscribe_context_->RedisAsyncCommand( + reinterpret_cast(&GlobalRedisCallback), + reinterpret_cast(callback_index), redis_command.c_str(), pattern.c_str(), + pattern.size()); +} + +Status RedisContext::PUnsubscribeAsync(const std::string &pattern) { + RAY_CHECK(async_redis_subscribe_context_); + + std::string redis_command = "PUNSUBSCRIBE %b"; + return async_redis_subscribe_context_->RedisAsyncCommand( + reinterpret_cast(&GlobalRedisCallback), + reinterpret_cast(-1), redis_command.c_str(), pattern.c_str(), + pattern.size()); +} + +Status RedisContext::PublishAsync(const std::string &channel, const std::string &message, + const RedisCallback &redisCallback) { + std::vector args = {"PUBLISH", channel, message}; + return RunArgvAsync(args, redisCallback); +} + } // namespace gcs } // namespace ray diff --git a/src/ray/gcs/redis_context.h b/src/ray/gcs/redis_context.h index 0af743efe..39ea29838 100644 --- a/src/ray/gcs/redis_context.h +++ b/src/ray/gcs/redis_context.h @@ -76,6 +76,8 @@ class CallbackReply { /// \return size_t The next cursor for scan. size_t ReadAsScanArray(std::vector *array) const; + bool IsUnsubscribeCallback() const { return is_unsubscribe_callback_; } + private: /// Parse redis reply as string array or scan array. void ParseAsStringArrayOrScanArray(redisReply *redis_reply); @@ -99,6 +101,8 @@ class CallbackReply { /// Represent the reply of StringArray or ScanArray. std::vector string_array_reply_; + bool is_unsubscribe_callback_ = false; + /// Represent the reply of SCanArray, means the next scan cursor for scan request. size_t next_scan_cursor_reply_{0}; }; @@ -234,6 +238,30 @@ class RedisContext { Status SubscribeAsync(const ClientID &client_id, const TablePubsub pubsub_channel, const RedisCallback &redisCallback, int64_t *out_callback_index); + /// Subscribes the client to the given pattern. + /// + /// \param pattern The pattern of subscription channel. + /// \param redisCallback The callback function that the notification calls. + /// \param out_callback_index The output pointer to callback index. + /// \return Status. + Status PSubscribeAsync(const std::string &pattern, const RedisCallback &redisCallback, + int64_t *out_callback_index); + + /// Unsubscribes the client from the given pattern. + /// + /// \param pattern The pattern of unsubscription channel. + /// \return Status. + Status PUnsubscribeAsync(const std::string &pattern); + + /// Posts a message to the given channel. + /// + /// \param channel The channel for message publishing to redis. + /// \param message The message to be published to redis. + /// \param redisCallback The callback will be called when the message is published to + /// redis. \return Status. + Status PublishAsync(const std::string &channel, const std::string &message, + const RedisCallback &redisCallback); + redisContext *sync_context() { RAY_CHECK(context_); return context_; diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index 5b9b3e261..47184d9f1 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -310,6 +310,12 @@ message ResourceMap { message ObjectTableDataList { repeated ObjectTableData items = 1; } + +message PubSubMessage { + bytes id = 1; + bytes data = 2; +} + // This enum type is used as object's metadata to indicate the object's creating // task has failed because of a certain error. // TODO(hchen): We may want to make these errors more specific. E.g., we may want