[GCS]Add publish and subscribe function of gcs table (#7909)

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(

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<CallbackReply> reply) {
+    if (done) {
+      done(Status::OK());
+    }
+  };
+
+  return redis_client_->GetPrimaryContext()->PublishAsync(
+      GenChannelPattern(channel, boost::optional<std::string>(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<std::string>(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<std::string>(id)));
+}
+
+Status GcsPubSub::SubscribeInternal(const std::string &channel, const Callback &subscribe,
+                                    const StatusCallback &done,
+                                    const boost::optional<std::string> &id) {
+  std::string pattern = GenChannelPattern(channel, id);
+  auto callback = [this, pattern, subscribe](std::shared_ptr<CallbackReply> 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<std::string> &id) {
+  std::stringstream pattern;
+  pattern << channel << ":";
+  if (id) {
+    pattern << *id;
+  } else {
+    pattern << "*";
+  }
+  return pattern.str();
+}
+
+}  // namespace gcs
+}  // namespace ray

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<void(const std::string &id, const std::string &data)>;
+
+  explicit GcsPubSub(std::shared_ptr<RedisClient> 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<std::string> &id = boost::none);
+
+  std::string GenChannelPattern(const std::string &channel,
+                                const boost::optional<std::string> &id);
+
+  std::shared_ptr<RedisClient> redis_client_;
+
+  /// Mutex to protect the subscribe_callback_index_ field.
+  absl::Mutex mutex_;
+
+  std::unordered_map<std::string, int64_t> subscribe_callback_index_ GUARDED_BY(mutex_);
+};
+
+}  // namespace gcs
+}  // namespace ray
+
+#endif  // RAY_GCS_GCS_PUB_SUB_H_

src/ray/gcs/pubsub/message_publisher.h

@@ -1,33 +0,0 @@
-#ifndef RAY_GCS_PUBSUB_MESSAGE_PUBLISHER_H
-#define RAY_GCS_PUBSUB_MESSAGE_PUBLISHER_H
-
-#include <string>
-#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

src/ray/gcs/pubsub/redis_message_publisher.cc

@@ -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<IOServicePool> 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<std::string> args = {"PUBLISH", channel, message};
-
-  RedisCallback pub_callback = nullptr;
-  if (callback) {
-    pub_callback = [callback](std::shared_ptr<CallbackReply> 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

src/ray/gcs/pubsub/redis_message_publisher.h

@@ -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<IOServicePool> 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<RedisClient> redis_client_;
-  std::shared_ptr<IOServicePool> io_service_pool_;
-};
-
-}  // namespace gcs
-
-}  // namespace ray
-
-#endif  // RAY_GCS_PUBSUB_REDIS_MESSAGE_PUBLISHER_H

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 <memory>
+
+#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<boost::asio::io_service::work> 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<gcs::RedisClient>(redis_client_options);
+    RAY_CHECK_OK(client_->Connect(io_service_));
+    pub_sub_ = std::make_shared<gcs::GcsPubSub>(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<std::string> &result) {
+    std::promise<bool> 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<std::pair<std::string, std::string>> &result) {
+    std::promise<bool> 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<bool> 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<bool> future, const std::chrono::milliseconds &timeout_ms) {
+    auto status = future.wait_for(timeout_ms);
+    return status == std::future_status::ready && future.get();
+  }
+
+  template <typename Data>
+  void WaitPendingDone(const std::vector<Data> &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<gcs::RedisClient> client_;
+  const std::chrono::milliseconds timeout_ms_{60000};
+  std::shared_ptr<gcs::GcsPubSub> pub_sub_;
+
+ private:
+  boost::asio::io_service io_service_;
+  std::unique_ptr<std::thread> thread_io_service_;
+};
+
+TEST_F(GcsPubSubTest, TestPubSubApi) {
+  std::string channel("channel");
+  std::string id("id");
+  std::string data("data");
+  std::vector<std::pair<std::string, std::string>> all_result;
+  SubscribeAll(channel, all_result);
+  std::vector<std::string> 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<std::atomic<int>>(0);
+  auto sub_finished_count = std::make_shared<std::atomic<int>>(0);
+  int size = 5;
+  std::vector<std::unique_ptr<std::thread>> 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();
+}

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<redisCallbackFn *>(&GlobalRedisCallback),
+      reinterpret_cast<void *>(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<redisCallbackFn *>(&GlobalRedisCallback),
+      reinterpret_cast<void *>(-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<std::string> args = {"PUBLISH", channel, message};
+  return RunArgvAsync(args, redisCallback);
+}
+
 }  // namespace gcs
 
 }  // namespace ray

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<std::string> *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<std::string> 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_;

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