Initial high-level code structure of CoreWorker. (#4875)

BUILD.bazel

@@ -105,6 +105,36 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "core_worker_lib",
+    srcs = glob(
+        [
+            "src/ray/core_worker/*.cc",
+        ],
+        exclude = [
+            "src/ray/core_worker/*_test.cc",
+        ],
+    ),
+    hdrs = glob([
+        "src/ray/core_worker/*.h",
+    ]),
+    copts = COPTS,
+    deps = [
+        ":ray_common",
+        ":ray_util",
+    ],
+)
+
+cc_test(
+    name = "core_worker_test",
+    srcs = ["src/ray/core_worker/core_worker_test.cc"],
+    copts = COPTS,
+    deps = [
+        ":core_worker_lib",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "lineage_cache_test",
     srcs = ["src/ray/raylet/lineage_cache_test.cc"],
@@ -277,6 +307,7 @@ cc_library(
         "src/ray/common/common_protocol.cc",
     ],
     hdrs = [
+        "src/ray/common/buffer.h",
         "src/ray/common/client_connection.h",
         "src/ray/common/common_protocol.h",
     ],
@@ -637,8 +668,8 @@ genrule(
         cp -f $(location //:raylet) $$WORK_DIR/python/ray/core/src/ray/raylet/ &&
         for f in $(locations //:python_gcs_fbs); do cp -f $$f $$WORK_DIR/python/ray/core/generated/; done &&
         mkdir -p $$WORK_DIR/python/ray/core/generated/ray/protocol/ &&
-        for f in $(locations //:python_node_manager_fbs); do 
-            cp -f $$f $$WORK_DIR/python/ray/core/generated/ray/protocol/; 
+        for f in $(locations //:python_node_manager_fbs); do
+            cp -f $$f $$WORK_DIR/python/ray/core/generated/ray/protocol/;
         done &&
         echo $$WORK_DIR > $@
     """,

src/ray/common/buffer.h

@@ -0,0 +1,45 @@
+#ifndef RAY_COMMON_BUFFER_H
+#define RAY_COMMON_BUFFER_H
+
+#include <cstdint>
+#include <cstdio>
+
+namespace ray {
+
+/// The interface that represents a buffer of bytes.
+class Buffer {
+ public:
+  /// Pointer to the data.
+  virtual uint8_t *Data() const = 0;
+
+  /// Size of this buffer.
+  virtual size_t Size() const = 0;
+
+  virtual ~Buffer() {}
+
+  bool operator==(const Buffer &rhs) const {
+    return this->Data() == rhs.Data() && this->Size() == rhs.Size();
+  }
+};
+
+/// Represents a byte buffer in local memory.
+class LocalMemoryBuffer : public Buffer {
+ public:
+  LocalMemoryBuffer(uint8_t *data, size_t size) : data_(data), size_(size) {}
+
+  uint8_t *Data() const override { return data_; }
+
+  size_t Size() const override { return size_; }
+
+  ~LocalMemoryBuffer() {}
+
+ private:
+  /// Pointer to the data.
+  uint8_t *data_;
+  /// Size of the buffer.
+  size_t size_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_COMMON_BUFFER_H

src/ray/core_worker/common.h

@@ -0,0 +1,71 @@
+#ifndef RAY_CORE_WORKER_COMMON_H
+#define RAY_CORE_WORKER_COMMON_H
+
+#include <string>
+
+#include "ray/common/buffer.h"
+#include "ray/id.h"
+
+namespace ray {
+
+/// Type of this worker.
+enum class WorkerType { WORKER, DRIVER };
+
+/// Language of Ray tasks and workers.
+enum class Language { PYTHON, JAVA };
+
+/// Information about a remote function.
+struct RayFunction {
+  /// Language of the remote function.
+  const Language language;
+  /// Function descriptor of the remote function.
+  const std::vector<std::string> function_descriptor;
+};
+
+/// Argument of a task.
+class TaskArg {
+ public:
+  /// Create a pass-by-reference task argument.
+  ///
+  /// \param[in] object_id Id of the argument.
+  /// \return The task argument.
+  static TaskArg PassByReference(const ObjectID &object_id) {
+    return TaskArg(std::make_shared<ObjectID>(object_id), nullptr);
+  }
+
+  /// Create a pass-by-reference task argument.
+  ///
+  /// \param[in] object_id Id of the argument.
+  /// \return The task argument.
+  static TaskArg PassByValue(const std::shared_ptr<Buffer> &data) {
+    return TaskArg(nullptr, data);
+  }
+
+  /// Return true if this argument is passed by reference, false if passed by value.
+  bool IsPassedByReference() const { return id_ != nullptr; }
+
+  /// Get the reference object ID.
+  ObjectID &GetReference() {
+    RAY_CHECK(id_ != nullptr) << "This argument isn't passed by reference.";
+    return *id_;
+  }
+
+  /// Get the value.
+  std::shared_ptr<Buffer> GetValue() {
+    RAY_CHECK(data_ != nullptr) << "This argument isn't passed by value.";
+    return data_;
+  }
+
+ private:
+  TaskArg(const std::shared_ptr<ObjectID> id, const std::shared_ptr<Buffer> data)
+      : id_(id), data_(data) {}
+
+  /// Id of the argument, if passed by reference, otherwise nullptr.
+  const std::shared_ptr<ObjectID> id_;
+  /// Data of the argument, if passed by value, otherwise nullptr.
+  const std::shared_ptr<Buffer> data_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_COMMON_H

src/ray/core_worker/core_worker.h

@@ -0,0 +1,68 @@
+#ifndef RAY_CORE_WORKER_CORE_WORKER_H
+#define RAY_CORE_WORKER_CORE_WORKER_H
+
+#include "common.h"
+#include "object_interface.h"
+#include "ray/common/buffer.h"
+#include "task_execution.h"
+#include "task_interface.h"
+
+namespace ray {
+
+/// The root class that contains all the core and language-independent functionalities
+/// of the worker. This class is supposed to be used to implement app-language (Java,
+/// Python, etc) workers.
+class CoreWorker {
+ public:
+  /// Construct a CoreWorker instance.
+  ///
+  /// \param[in] worker_type Type of this worker.
+  /// \param[in] langauge Language of this worker.
+  CoreWorker(const WorkerType worker_type, const Language language)
+      : worker_type_(worker_type),
+        language_(language),
+        task_interface_(*this),
+        object_interface_(*this),
+        task_execution_interface_(*this) {}
+
+  /// Connect this worker to Raylet.
+  Status Connect() { return Status::OK(); }
+
+  /// Type of this worker.
+  enum WorkerType WorkerType() const { return worker_type_; }
+
+  /// Language of this worker.
+  enum Language Language() const { return language_; }
+
+  /// Return the `CoreWorkerTaskInterface` that contains the methods related to task
+  /// submisson.
+  CoreWorkerTaskInterface &Tasks() { return task_interface_; }
+
+  /// Return the `CoreWorkerObjectInterface` that contains methods related to object
+  /// store.
+  CoreWorkerObjectInterface &Objects() { return object_interface_; }
+
+  /// Return the `CoreWorkerTaskExecutionInterface` that contains methods related to
+  /// task execution.
+  CoreWorkerTaskExecutionInterface &Execution() { return task_execution_interface_; }
+
+ private:
+  /// Type of this worker.
+  const enum WorkerType worker_type_;
+
+  /// Language of this worker.
+  const enum Language language_;
+
+  /// The `CoreWorkerTaskInterface` instance.
+  CoreWorkerTaskInterface task_interface_;
+
+  /// The `CoreWorkerObjectInterface` instance.
+  CoreWorkerObjectInterface object_interface_;
+
+  /// The `CoreWorkerTaskExecutionInterface` instance.
+  CoreWorkerTaskExecutionInterface task_execution_interface_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_CORE_WORKER_H

src/ray/core_worker/core_worker_test.cc

@@ -0,0 +1,38 @@
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "core_worker.h"
+#include "ray/common/buffer.h"
+
+namespace ray {
+
+class CoreWorkerTest : public ::testing::Test {
+ public:
+  CoreWorkerTest() : core_worker_(WorkerType::WORKER, Language::PYTHON) {}
+
+ protected:
+  CoreWorker core_worker_;
+};
+
+TEST_F(CoreWorkerTest, TestTaskArg) {
+  // Test by-reference argument.
+  ObjectID id = ObjectID::from_random();
+  TaskArg by_ref = TaskArg::PassByReference(id);
+  ASSERT_TRUE(by_ref.IsPassedByReference());
+  ASSERT_EQ(by_ref.GetReference(), id);
+  // Test by-value argument.
+  std::shared_ptr<LocalMemoryBuffer> buffer =
+      std::make_shared<LocalMemoryBuffer>(static_cast<uint8_t *>(0), 0);
+  TaskArg by_value = TaskArg::PassByValue(buffer);
+  ASSERT_FALSE(by_value.IsPassedByReference());
+  auto data = by_value.GetValue();
+  ASSERT_TRUE(data != nullptr);
+  ASSERT_EQ(*data, *buffer);
+}
+
+TEST_F(CoreWorkerTest, TestAttributeGetters) {
+  ASSERT_EQ(core_worker_.WorkerType(), WorkerType::WORKER);
+  ASSERT_EQ(core_worker_.Language(), Language::PYTHON);
+}
+
+}  // namespace ray

src/ray/core_worker/object_interface.cc

@@ -0,0 +1,25 @@
+#include "object_interface.h"
+
+namespace ray {
+
+Status CoreWorkerObjectInterface::Put(const Buffer &buffer, const ObjectID *object_id) {
+  return Status::OK();
+}
+
+Status CoreWorkerObjectInterface::Get(const std::vector<ObjectID> &ids,
+                                      int64_t timeout_ms, std::vector<Buffer> *results) {
+  return Status::OK();
+}
+
+Status CoreWorkerObjectInterface::Wait(const std::vector<ObjectID> &object_ids,
+                                       int num_objects, int64_t timeout_ms,
+                                       std::vector<bool> *results) {
+  return Status::OK();
+}
+
+Status CoreWorkerObjectInterface::Delete(const std::vector<ObjectID> &object_ids,
+                                         bool local_only, bool delete_creating_tasks) {
+  return Status::OK();
+}
+
+}  // namespace ray

src/ray/core_worker/object_interface.h

@@ -0,0 +1,61 @@
+#ifndef RAY_CORE_WORKER_OBJECT_INTERFACE_H
+#define RAY_CORE_WORKER_OBJECT_INTERFACE_H
+
+#include "common.h"
+#include "ray/common/buffer.h"
+#include "ray/id.h"
+#include "ray/status.h"
+
+namespace ray {
+
+class CoreWorker;
+
+/// The interface that contains all `CoreWorker` methods that are related to object store.
+class CoreWorkerObjectInterface {
+ public:
+  CoreWorkerObjectInterface(CoreWorker &core_worker) : core_worker_(core_worker) {}
+
+  /// Put an object into object store.
+  ///
+  /// \param[in] buffer Data buffer of the object.
+  /// \param[out] object_id Generated ID of the object.
+  /// \return Status.
+  Status Put(const Buffer &buffer, const ObjectID *object_id);
+
+  /// Get a list of objects from the object store.
+  ///
+  /// \param[in] ids IDs of the objects to get.
+  /// \param[in] timeout_ms Timeout in milliseconds, wait infinitely if it's negative.
+  /// \param[out] results Result list of objects data.
+  /// \return Status.
+  Status Get(const std::vector<ObjectID> &ids, int64_t timeout_ms,
+             std::vector<Buffer> *results);
+
+  /// Wait for a list of objects to appear in the object store.
+  ///
+  /// \param[in] IDs of the objects to wait for.
+  /// \param[in] num_returns Number of objects that should appear.
+  /// \param[in] timeout_ms Timeout in milliseconds, wait infinitely if it's negative.
+  /// \param[out] results A bitset that indicates each object has appeared or not.
+  /// \return Status.
+  Status Wait(const std::vector<ObjectID> &object_ids, int num_objects,
+              int64_t timeout_ms, std::vector<bool> *results);
+
+  /// Delete a list of objects from the object store.
+  ///
+  /// \param[in] object_ids IDs of the objects to delete.
+  /// \param[in] local_only Whether only delete the objects in local node, or all nodes in
+  /// the cluster.
+  /// \param[in] delete_creating_tasks Whether also delete the tasks that
+  /// created these objects. \return Status.
+  Status Delete(const std::vector<ObjectID> &object_ids, bool local_only,
+                bool delete_creating_tasks);
+
+ private:
+  /// Reference to the parent CoreWorker instance.
+  CoreWorker &core_worker_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_OBJECT_INTERFACE_H

src/ray/core_worker/task_execution.cc

@@ -0,0 +1,7 @@
+#include "task_execution.h"
+
+namespace ray {
+
+void CoreWorkerTaskExecutionInterface::Start(const TaskExecutor &executor) {}
+
+}  // namespace ray

src/ray/core_worker/task_execution.h

@@ -0,0 +1,36 @@
+#ifndef RAY_CORE_WORKER_TASK_EXECUTION_H
+#define RAY_CORE_WORKER_TASK_EXECUTION_H
+
+#include "common.h"
+#include "ray/common/buffer.h"
+#include "ray/status.h"
+
+namespace ray {
+
+class CoreWorker;
+
+/// The interface that contains all `CoreWorker` methods that are related to task
+/// execution.
+class CoreWorkerTaskExecutionInterface {
+ public:
+  CoreWorkerTaskExecutionInterface(CoreWorker &core_worker) : core_worker_(core_worker) {}
+
+  /// The callback provided app-language workers that executes tasks.
+  ///
+  /// \param ray_function[in] Information about the function to execute.
+  /// \param args[in] Arguments of the task.
+  /// \return Status.
+  using TaskExecutor = std::function<Status(const RayFunction &ray_function,
+                                            const std::vector<Buffer> &args)>;
+
+  /// Start receving and executes tasks in a infinite loop.
+  void Start(const TaskExecutor &executor);
+
+ private:
+  /// Reference to the parent CoreWorker instance.
+  CoreWorker &core_worker_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_TASK_EXECUTION_H

src/ray/core_worker/task_interface.cc

@@ -0,0 +1,26 @@
+#include "task_interface.h"
+
+namespace ray {
+
+Status CoreWorkerTaskInterface::SubmitTask(const RayFunction &function,
+                                           const std::vector<TaskArg> &args,
+                                           const TaskOptions &task_options,
+                                           std::vector<ObjectID> *return_ids) {
+  return Status::OK();
+}
+
+Status CoreWorkerTaskInterface::CreateActor(
+    const RayFunction &function, const std::vector<TaskArg> &args,
+    const ActorCreationOptions &actor_creation_options, ActorHandle *actor_handle) {
+  return Status::OK();
+}
+
+Status CoreWorkerTaskInterface::SubmitActorTask(ActorHandle &actor_handle,
+                                                const RayFunction &function,
+                                                const std::vector<TaskArg> &args,
+                                                const TaskOptions &task_options,
+                                                std::vector<ObjectID> *return_ids) {
+  return Status::OK();
+}
+
+}  // namespace ray

src/ray/core_worker/task_interface.h

@@ -0,0 +1,96 @@
+#ifndef RAY_CORE_WORKER_TASK_INTERFACE_H
+#define RAY_CORE_WORKER_TASK_INTERFACE_H
+
+#include "common.h"
+#include "ray/common/buffer.h"
+#include "ray/id.h"
+#include "ray/status.h"
+
+namespace ray {
+
+class CoreWorker;
+
+/// Options of a non-actor-creation task.
+struct TaskOptions {
+  /// Number of returns of this task.
+  const int num_returns = 1;
+  /// Resources required by this task.
+  const std::unordered_map<std::string, double> resources;
+};
+
+/// Options of an actor creation task.
+struct ActorCreationOptions {
+  /// Maximum number of times that the actor should be reconstructed when it dies
+  /// unexpectedly. It must be non-negative. If it's 0, the actor won't be reconstructed.
+  const uint64_t max_reconstructions = 0;
+  /// Resources required by the whole lifetime of this actor.
+  const std::unordered_map<std::string, double> resources;
+};
+
+/// A handle to an actor.
+class ActorHandle {
+ public:
+  ActorHandle(const ActorID &actor_id, const ActorHandleID &actor_handle_id)
+      : actor_id_(actor_id), actor_handle_id_(actor_handle_id) {}
+
+  /// ID of the actor.
+  const class ActorID &ActorID() const { return actor_id_; }
+
+  /// ID of this actor handle.
+  const class ActorHandleID &ActorHandleID() const { return actor_handle_id_; }
+
+ private:
+  /// ID of the actor.
+  const class ActorID actor_id_;
+  /// ID of this actor handle.
+  const class ActorHandleID actor_handle_id_;
+};
+
+/// The interface that contains all `CoreWorker` methods that are related to task
+/// submission.
+class CoreWorkerTaskInterface {
+ public:
+  CoreWorkerTaskInterface(CoreWorker &core_worker) : core_worker_(core_worker) {}
+
+  /// Submit a normal task.
+  ///
+  /// \param[in] function The remote function to execute.
+  /// \param[in] args Arguments of this task.
+  /// \param[in] task_options Options for this task.
+  /// \param[out] return_ids Ids of the return objects.
+  /// \return Status.
+  Status SubmitTask(const RayFunction &function, const std::vector<TaskArg> &args,
+                    const TaskOptions &task_options, std::vector<ObjectID> *return_ids);
+
+  /// Create an actor.
+  ///
+  /// \param[in] function The remote function that generates the actor object.
+  /// \param[in] args Arguments of this task.
+  /// \param[in] actor_creation_options Options for this actor creation task.
+  /// \param[out] actor_handle Handle to the actor.
+  /// \return Status.
+  Status CreateActor(const RayFunction &function, const std::vector<TaskArg> &args,
+                     const ActorCreationOptions &actor_creation_options,
+                     ActorHandle *actor_handle);
+
+  /// Submit an actor task.
+  ///
+  /// \param[in] actor_handle Handle to the actor.
+  /// \param[in] function The remote function to execute.
+  /// \param[in] args Arguments of this task.
+  /// \param[in] task_options Options for this task.
+  /// \param[out] return_ids Ids of the return objects.
+  /// \return Status.
+  Status SubmitActorTask(ActorHandle &actor_handle, const RayFunction &function,
+                         const std::vector<TaskArg> &args,
+                         const TaskOptions &task_options,
+                         std::vector<ObjectID> *return_ids);
+
+ private:
+  /// Reference to the parent CoreWorker instance.
+  CoreWorker &core_worker_;
+};
+
+}  // namespace ray
+
+#endif  // RAY_CORE_WORKER_TASK_INTERFACE_H