[xray] Basic actor support

python/ray/actor.py

@@ -802,6 +802,8 @@ def remote(cls, *args, **kwargs):
                                             actor_creation_resources,
                                             actor_method_cpus,
                                             ray.worker.global_worker)
+            # Increment the actor counter to account for the creation task.
+            actor_counter += 1
 
             # Instantiate the actor handle.
             actor_object = cls.__new__(cls)

src/common/redis_module/ray_redis_module.cc

@@ -49,14 +49,11 @@
     return RedisModule_ReplyWithError(ctx, (MESSAGE)); \
   }
 
+// NOTE(swang): The order of prefixes here must match the TablePrefix enum
+// defined in src/ray/gcs/format/gcs.fbs.
 static const char *table_prefixes[] = {
-    NULL,
-    "TASK:",
-    "TASK:",
-    "CLIENT:",
-    "OBJECT:",
-    "FUNCTION:",
-    "TASK_RECONSTRUCTION:",
+    NULL,         "TASK:",  "TASK:",     "CLIENT:",
+    "OBJECT:",    "ACTOR:", "FUNCTION:", "TASK_RECONSTRUCTION:",
     "HEARTBEAT:",
 };
 

src/local_scheduler/local_scheduler_algorithm.cc

@@ -358,6 +358,12 @@ void handle_convert_worker_to_actor(
      * filled out, so fill out the correct worker field now. */
     algorithm_state->local_actor_infos[actor_id].worker = worker;
   }
+  /* Increment the task counter for the creator's handle to account for the
+   * actor creation task. */
+  auto &task_counters =
+      algorithm_state->local_actor_infos[actor_id].task_counters;
+  RAY_CHECK(task_counters[ActorHandleID::nil()] == 0);
+  task_counters[ActorHandleID::nil()]++;
 }
 
 /**

src/ray/CMakeLists.txt

@@ -51,7 +51,7 @@ set(RAY_SRCS
   raylet/worker.cc
   raylet/worker_pool.cc
   raylet/scheduling_resources.cc
-  raylet/actor.cc
+  raylet/actor_registration.cc
   raylet/scheduling_queue.cc
   raylet/scheduling_policy.cc
   raylet/task_dependency_manager.cc

src/ray/gcs/client.cc

@@ -10,6 +10,7 @@ AsyncGcsClient::AsyncGcsClient(const ClientID &client_id) {
   context_.reset(new RedisContext());
   client_table_.reset(new ClientTable(context_, this, client_id));
   object_table_.reset(new ObjectTable(context_, this));
+  actor_table_.reset(new ActorTable(context_, this));
   task_table_.reset(new TaskTable(context_, this));
   raylet_task_table_.reset(new raylet::TaskTable(context_, this));
   task_reconstruction_log_.reset(new TaskReconstructionLog(context_, this));
@@ -48,6 +49,8 @@ TaskTable &AsyncGcsClient::task_table() { return *task_table_; }
 
 raylet::TaskTable &AsyncGcsClient::raylet_task_table() { return *raylet_task_table_; }
 
+ActorTable &AsyncGcsClient::actor_table() { return *actor_table_; }
+
 TaskReconstructionLog &AsyncGcsClient::task_reconstruction_log() {
   return *task_reconstruction_log_;
 }

src/ray/gcs/client.h

@@ -45,12 +45,12 @@ class RAY_EXPORT AsyncGcsClient {
   inline FunctionTable &function_table();
   // TODO: Some API for getting the error on the driver
   inline ClassTable &class_table();
-  inline ActorTable &actor_table();
   inline CustomSerializerTable &custom_serializer_table();
   inline ConfigTable &config_table();
   ObjectTable &object_table();
   TaskTable &task_table();
   raylet::TaskTable &raylet_task_table();
+  ActorTable &actor_table();
   TaskReconstructionLog &task_reconstruction_log();
   ClientTable &client_table();
   HeartbeatTable &heartbeat_table();
@@ -72,6 +72,7 @@ class RAY_EXPORT AsyncGcsClient {
   std::unique_ptr<ObjectTable> object_table_;
   std::unique_ptr<TaskTable> task_table_;
   std::unique_ptr<raylet::TaskTable> raylet_task_table_;
+  std::unique_ptr<ActorTable> actor_table_;
   std::unique_ptr<TaskReconstructionLog> task_reconstruction_log_;
   std::unique_ptr<HeartbeatTable> heartbeat_table_;
   std::unique_ptr<ClientTable> client_table_;

src/ray/gcs/format/gcs.fbs

@@ -10,9 +10,10 @@ enum TablePrefix:int {
   RAYLET_TASK,
   CLIENT,
   OBJECT,
+  ACTOR,
   FUNCTION,
   TASK_RECONSTRUCTION,
-  HEARTBEAT
+  HEARTBEAT,
 }
 
 // The channel that Add operations to the Table should be published on, if any.
@@ -89,6 +90,16 @@ table ClassTableData {
 }
 
 table ActorTableData {
+  // The ID of the actor that was created.
+  actor_id: string;
+  // The dummy object ID returned by the actor creation task. If the actor
+  // dies, then this is the object that should be reconstructed for the actor
+  // to be recreated.
+  actor_creation_dummy_object_id: string;
+  // The ID of the driver that created the actor.
+  driver_id: string;
+  // The ID of the node manager that created the actor.
+  node_manager_id: string;
 }
 
 table ErrorTableData {

src/ray/gcs/tables.cc

@@ -338,6 +338,7 @@ template class Log<ObjectID, ObjectTableData>;
 template class Log<TaskID, ray::protocol::Task>;
 template class Table<TaskID, ray::protocol::Task>;
 template class Table<TaskID, TaskTableData>;
+template class Log<ActorID, ActorTableData>;
 template class Log<TaskID, TaskReconstructionData>;
 template class Table<ClientID, HeartbeatTableData>;
 

src/ray/gcs/tables.h

@@ -277,14 +277,22 @@ class FunctionTable : public Table<ObjectID, FunctionTableData> {
 using ClassTable = Table<ClassID, ClassTableData>;
 
 // TODO(swang): Set the pubsub channel for the actor table.
-using ActorTable = Table<ActorID, ActorTableData>;
+class ActorTable : public Log<ActorID, ActorTableData> {
+ public:
+  ActorTable(const std::shared_ptr<RedisContext> &context, AsyncGcsClient *client)
+      : Log(context, client) {
+    pubsub_channel_ = TablePubsub_ACTOR;
+    prefix_ = TablePrefix_TASK_RECONSTRUCTION;
+  }
+};
 
 class TaskReconstructionLog : public Log<TaskID, TaskReconstructionData> {
  public:
   TaskReconstructionLog(const std::shared_ptr<RedisContext> &context,
                         AsyncGcsClient *client)
       : Log(context, client) {
-    prefix_ = TablePrefix_TASK_RECONSTRUCTION;
+    pubsub_channel_ = TablePubsub_ACTOR;
+    prefix_ = TablePrefix_ACTOR;
   }
 };
 

src/ray/raylet/actor_registration.cc

@@ -0,0 +1,41 @@
+#include "ray/raylet/actor_registration.h"
+
+#include "ray/util/logging.h"
+
+namespace ray {
+
+namespace raylet {
+
+ActorRegistration::ActorRegistration(const ActorTableDataT &actor_table_data)
+    : actor_table_data_(actor_table_data),
+      execution_dependency_(ObjectID::nil()),
+      frontier_() {}
+
+const ClientID ActorRegistration::GetNodeManagerId() const {
+  return ClientID::from_binary(actor_table_data_.node_manager_id);
+}
+
+const ObjectID ActorRegistration::GetActorCreationDependency() const {
+  return ObjectID::from_binary(actor_table_data_.actor_creation_dummy_object_id);
+}
+
+const ObjectID ActorRegistration::GetExecutionDependency() const {
+  return execution_dependency_;
+}
+
+const std::unordered_map<ActorHandleID, ActorRegistration::FrontierLeaf, UniqueIDHasher>
+    &ActorRegistration::GetFrontier() const {
+  return frontier_;
+}
+
+void ActorRegistration::ExtendFrontier(const ActorHandleID &handle_id,
+                                       const ObjectID &execution_dependency) {
+  auto &frontier_entry = frontier_[handle_id];
+  frontier_entry.task_counter++;
+  frontier_entry.execution_dependency = execution_dependency;
+  execution_dependency_ = execution_dependency;
+}
+
+}  // namespace raylet
+
+}  // namespace ray

src/ray/raylet/actor_registration.h

@@ -0,0 +1,96 @@
+#ifndef RAY_RAYLET_ACTOR_REGISTRATION_H
+#define RAY_RAYLET_ACTOR_REGISTRATION_H
+
+#include <unordered_map>
+
+#include "ray/gcs/format/gcs_generated.h"
+#include "ray/id.h"
+
+namespace ray {
+
+namespace raylet {
+
+/// \class ActorRegistration
+///
+/// Information about an actor registered in the system. This includes the
+/// actor's current node manager location, and if local, information about its
+/// current execution state, used for reconstruction purposes.
+class ActorRegistration {
+ public:
+  /// Create an actor registration.
+  ///
+  /// \param actor_table_data Information from the global actor table about
+  /// this actor. This includes the actor's node manager location.
+  ActorRegistration(const ActorTableDataT &actor_table_data);
+
+  /// Each actor may have multiple callers, or "handles". A frontier leaf
+  /// represents the execution state of the actor with respect to a single
+  /// handle.
+  struct FrontierLeaf {
+    /// The number of tasks submitted by this handle that have executed on the
+    /// actor so far.
+    int64_t task_counter;
+    /// The execution dependency returned by the task submitted by this handle
+    /// that most recently executed on the actor.
+    ObjectID execution_dependency;
+  };
+
+  /// Get the actor's node manager location.
+  ///
+  /// \return The actor's node manager location. All tasks for the actor should
+  /// be forwarded to this node.
+  const ClientID GetNodeManagerId() const;
+
+  /// Get the object that represents the actor's initial state. This is the
+  /// execution dependency returned by this actor's creation task. If
+  /// reconstructed, this will recreate the actor.
+  ///
+  /// \return The execution dependency returned by the actor's creation task.
+  const ObjectID GetActorCreationDependency() const;
+
+  /// Get the object that represents the actor's current state. This is the
+  /// execution dependency returned by the task most recently executed on the
+  /// actor. The next task to execute on the actor should be marked as
+  /// execution-dependent on this object.
+  ///
+  /// \return The execution dependency returned by the most recently executed
+  /// task.
+  const ObjectID GetExecutionDependency() const;
+
+  /// Get the execution frontier of the actor, indexed by handle. This captures
+  /// the execution state of the actor, a summary of which tasks have executed
+  /// so far.
+  ///
+  /// \return The actor frontier, a map from handle ID to execution state for
+  /// that handle.
+  const std::unordered_map<ActorHandleID, FrontierLeaf, UniqueIDHasher> &GetFrontier()
+      const;
+
+  /// Extend the frontier of the actor by a single task. This should be called
+  /// whenever the actor executes a task.
+  ///
+  /// \param handle_id The ID of the handle that submitted the task.
+  /// \param execution_dependency The object representing the actor's new
+  /// state. This is the execution dependency returned by the task.
+  void ExtendFrontier(const ActorHandleID &handle_id,
+                      const ObjectID &execution_dependency);
+
+ private:
+  /// Information from the global actor table about this actor, including the
+  /// node manager location.
+  ActorTableDataT actor_table_data_;
+  /// The object representing the state following the actor's most recently
+  /// executed task. The next task to execute on the actor should be marked as
+  /// execution-dependent on this object.
+  ObjectID execution_dependency_;
+  /// The execution frontier of the actor, which represents which tasks have
+  /// executed so far and which tasks may execute next, based on execution
+  /// dependencies. This is indexed by handle.
+  std::unordered_map<ActorHandleID, FrontierLeaf, UniqueIDHasher> frontier_;
+};
+
+}  // namespace raylet
+
+}  // namespace ray
+
+#endif  // RAY_RAYLET_ACTOR_REGISTRATION_H