rename ActorTable to LogBasedActorTable and add new ActorTable (#7643)

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<bool()> condition, int timeout_ms) {
   int wait_time = 0;
   while (true) {

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

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<rpc::ActorTableData> *actor_table_data_list) = 0;
+
   /// Get actor specification from GCS asynchronously.
   ///
   /// \param actor_id The ID of actor to look up in the GCS.

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<ActorTableData> *actor_table_data_list) {
+  return Status::Invalid("Not implemented");
+}
 
 Status ServiceBasedActorInfoAccessor::AsyncGet(
     const ActorID &actor_id, const OptionalItemCallback<rpc::ActorTableData> &callback) {

src/ray/gcs/gcs_client/service_based_accessor.h

@@ -58,6 +58,8 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor {
 
   virtual ~ServiceBasedActorInfoAccessor() = default;
 
+  Status GetAll(std::vector<ActorTableData> *actor_table_data_list) override;
+
   Status AsyncGet(const ActorID &actor_id,
                   const OptionalItemCallback<rpc::ActorTableData> &callback) override;
 
@@ -89,12 +91,13 @@ class ServiceBasedActorInfoAccessor : public ActorInfoAccessor {
       const ActorID &actor_id,
       const OptionalItemCallback<rpc::ActorCheckpointIdData> &callback) override;
 
+ protected:
+  ClientID subscribe_id_;
+
  private:
   ServiceBasedGcsClient *client_impl_;
 
-  ClientID subscribe_id_;
-
-  typedef SubscriptionExecutor<ActorID, ActorTableData, ActorTable>
+  typedef SubscriptionExecutor<ActorID, ActorTableData, LogBasedActorTable>
       ActorSubscriptionExecutor;
   ActorSubscriptionExecutor actor_sub_executor_;
 

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<rpc::JobTableData> &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;
   }
 

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;
   }

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<ActorID> 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<ActorTableData> *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<ActorTableData> &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<ActorTableData> &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<ActorTableData> &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<ActorID, ActorTableData> &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<ActorID, ActorTableData> &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<ActorCheckpointData> &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<ActorCheckpointData> &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<ActorCheckpointIdData> &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<ActorID> 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<ActorTableData> &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<ActorTableData> &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<ActorTableData> &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<ActorID, ActorTableData> &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<ActorID, ActorTableData> &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()) {}
 

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<ActorTableData> *actor_table_data_list) override;
 
   Status AsyncGet(const ActorID &actor_id,
                   const OptionalItemCallback<ActorTableData> &callback) override;
@@ -66,6 +68,10 @@ class RedisActorInfoAccessor : public ActorInfoAccessor {
       const ActorID &actor_id,
       const OptionalItemCallback<ActorCheckpointIdData> &callback) override;
 
+ protected:
+  virtual std::vector<ActorID> 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<ActorID, ActorTableData, LogBasedActorTable>
+      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<ActorTableData> &callback) override;
+
+  Status AsyncRegister(const std::shared_ptr<ActorTableData> &data_ptr,
+                       const StatusCallback &callback) override;
+
+  Status AsyncUpdate(const ActorID &actor_id,
+                     const std::shared_ptr<ActorTableData> &data_ptr,
+                     const StatusCallback &callback) override;
+
+  Status AsyncSubscribeAll(const SubscribeCallback<ActorID, ActorTableData> &subscribe,
+                           const StatusCallback &done) override;
+
+  Status AsyncSubscribe(const ActorID &actor_id,
+                        const SubscribeCallback<ActorID, ActorTableData> &subscribe,
+                        const StatusCallback &done) override;
+
+  Status AsyncUnsubscribe(const ActorID &actor_id, const StatusCallback &done) override;
+
+ protected:
+  std::vector<ActorID> GetAllActorID() const override;
+  Status Get(const ActorID &actor_id, ActorTableData *actor_table_data) const override;
+
+ private:
   typedef SubscriptionExecutor<ActorID, ActorTableData, ActorTable>
       ActorSubscriptionExecutor;
   ActorSubscriptionExecutor actor_sub_executor_;

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<std::string> &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<CallbackReply> RedisContext::RunArgvSync(
+    const std::vector<std::string> &args) {
+  RAY_CHECK(context_);
+  // Build the arguments.
+  std::vector<const char *> argv;
+  std::vector<size_t> argc;
+  for (const auto &arg : args) {
+    argv.push_back(arg.data());
+    argc.push_back(arg.size());
+  }
+  auto redis_reply = reinterpret_cast<redisReply *>(
+      ::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<CallbackReply> callback_reply(new CallbackReply(redis_reply));
+  freeReplyObject(redis_reply);
+  return callback_reply;
+}
+
 Status RedisContext::RunArgvAsync(const std::vector<std::string> &args) {
   RAY_CHECK(redis_async_context_);
   // Build the arguments.

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<std::string> &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<std::string> 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<CallbackReply> RunArgvSync(const std::vector<std::string> &args);
+
   /// Run an operation on some table key.
   ///
   /// \param command The command to run. This must match a registered Ray Redis

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() {

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<ObjectTable> object_table_;
   std::unique_ptr<raylet::TaskTable> raylet_task_table_;
+  std::unique_ptr<LogBasedActorTable> log_based_actor_table_;
   std::unique_ptr<ActorTable> actor_table_;
   std::unique_ptr<TaskReconstructionLog> task_reconstruction_log_;
   std::unique_ptr<TaskLeaseTable> task_lease_table_;

src/ray/gcs/subscription_executor.cc

@@ -198,6 +198,7 @@ Status SubscriptionExecutor<ID, Data, Table>::AsyncUnsubscribe(
   return table_.CancelNotifications(JobID::Nil(), id, client_id, on_done);
 }
 
+template class SubscriptionExecutor<ActorID, ActorTableData, LogBasedActorTable>;
 template class SubscriptionExecutor<ActorID, ActorTableData, ActorTable>;
 template class SubscriptionExecutor<JobID, JobTableData, JobTable>;
 template class SubscriptionExecutor<TaskID, TaskTableData, raylet::TaskTable>;

src/ray/gcs/tables.cc

@@ -746,6 +746,84 @@ Status TaskLeaseTable::Subscribe(const JobID &job_id, const ClientID &client_id,
   return Table<TaskID, TaskLeaseData>::Subscribe(job_id, client_id, on_subscribe, done);
 }
 
+std::vector<ActorID> SyncGetAllActorID(redisContext *redis_context,
+                                       const std::string &table_prefix) {
+  std::unordered_set<ActorID> 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<redisReply *>(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<ActorID> 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<ActorID> 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<ActorID> 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<UniqueID, ProfileTableData>;
 template class Table<ActorCheckpointID, ActorCheckpointData>;
 template class Table<ActorID, ActorCheckpointIdData>;
 template class Table<WorkerID, WorkerFailureData>;
+template class Table<ActorID, ActorTableData>;
 
 template class Log<ClientID, ResourceTableData>;
 template class Hash<ClientID, ResourceTableData>;

src/ray/gcs/tables.h

@@ -716,19 +716,44 @@ class JobTable : public Log<JobID, JobTableData> {
   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<ActorID, ActorTableData> {
+/// 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<ActorID, ActorTableData> {
  public:
-  ActorTable(const std::vector<std::shared_ptr<RedisContext>> &contexts,
-             RedisGcsClient *client)
+  LogBasedActorTable(const std::vector<std::shared_ptr<RedisContext>> &contexts,
+                     RedisGcsClient *client)
       : Log(contexts, client) {
     pubsub_channel_ = TablePubsub::ACTOR_PUBSUB;
     prefix_ = TablePrefix::ACTOR;
   }
+
+  /// Get all actor id synchronously.
+  std::vector<ActorID> 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<ActorID, ActorTableData> {
+ public:
+  ActorTable(const std::vector<std::shared_ptr<RedisContext>> &contexts,
+             RedisGcsClient *client)
+      : Table(contexts, client) {
+    pubsub_channel_ = TablePubsub::ACTOR_PUBSUB;
+    prefix_ = TablePrefix::ACTOR;
+  }
+
+  /// Get all actor id synchronously.
+  std::vector<ActorID> GetAllActorID();
+
+  /// Get actor table data by actor id synchronously.
+  Status Get(const ActorID &actor_id, ActorTableData *actor_table_data);
 };
 
 class WorkerFailureTable : public Table<WorkerID, WorkerFailureData> {

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<ActorTableData> 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) {

src/ray/gcs/test/subscription_executor_test.cc

@@ -25,12 +25,13 @@ namespace gcs {
 
 class SubscriptionExecutorTest : public AccessorTestBase<ActorID, ActorTableData> {
  public:
-  typedef SubscriptionExecutor<ActorID, ActorTableData, ActorTable> ActorSubExecutor;
+  typedef SubscriptionExecutor<ActorID, ActorTableData, LogBasedActorTable>
+      ActorSubExecutor;
 
   virtual void SetUp() {
     AccessorTestBase<ActorID, ActorTableData>::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);