diff --git a/BUILD.bazel b/BUILD.bazel index e2cbdd64bf51..0bdbe5741cf8 100644 --- a/BUILD.bazel +++ b/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 > $@ """, diff --git a/src/ray/common/buffer.h b/src/ray/common/buffer.h new file mode 100644 index 000000000000..358d903799c7 --- /dev/null +++ b/src/ray/common/buffer.h @@ -0,0 +1,45 @@ +#ifndef RAY_COMMON_BUFFER_H +#define RAY_COMMON_BUFFER_H + +#include +#include + +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 diff --git a/src/ray/core_worker/common.h b/src/ray/core_worker/common.h new file mode 100644 index 000000000000..b53c35b25fa8 --- /dev/null +++ b/src/ray/core_worker/common.h @@ -0,0 +1,71 @@ +#ifndef RAY_CORE_WORKER_COMMON_H +#define RAY_CORE_WORKER_COMMON_H + +#include + +#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 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(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 &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 GetValue() { + RAY_CHECK(data_ != nullptr) << "This argument isn't passed by value."; + return data_; + } + + private: + TaskArg(const std::shared_ptr id, const std::shared_ptr data) + : id_(id), data_(data) {} + + /// Id of the argument, if passed by reference, otherwise nullptr. + const std::shared_ptr id_; + /// Data of the argument, if passed by value, otherwise nullptr. + const std::shared_ptr data_; +}; + +} // namespace ray + +#endif // RAY_CORE_WORKER_COMMON_H diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h new file mode 100644 index 000000000000..96e51dbc4532 --- /dev/null +++ b/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 diff --git a/src/ray/core_worker/core_worker_test.cc b/src/ray/core_worker/core_worker_test.cc new file mode 100644 index 000000000000..b1be58da95b8 --- /dev/null +++ b/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 buffer = + std::make_shared(static_cast(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 diff --git a/src/ray/core_worker/object_interface.cc b/src/ray/core_worker/object_interface.cc new file mode 100644 index 000000000000..d5d5d6f883f6 --- /dev/null +++ b/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 &ids, + int64_t timeout_ms, std::vector *results) { + return Status::OK(); +} + +Status CoreWorkerObjectInterface::Wait(const std::vector &object_ids, + int num_objects, int64_t timeout_ms, + std::vector *results) { + return Status::OK(); +} + +Status CoreWorkerObjectInterface::Delete(const std::vector &object_ids, + bool local_only, bool delete_creating_tasks) { + return Status::OK(); +} + +} // namespace ray diff --git a/src/ray/core_worker/object_interface.h b/src/ray/core_worker/object_interface.h new file mode 100644 index 000000000000..424c123ee543 --- /dev/null +++ b/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 &ids, int64_t timeout_ms, + std::vector *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 &object_ids, int num_objects, + int64_t timeout_ms, std::vector *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 &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 diff --git a/src/ray/core_worker/task_execution.cc b/src/ray/core_worker/task_execution.cc new file mode 100644 index 000000000000..aea48b4de34a --- /dev/null +++ b/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 diff --git a/src/ray/core_worker/task_execution.h b/src/ray/core_worker/task_execution.h new file mode 100644 index 000000000000..308b1e6868d6 --- /dev/null +++ b/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 &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 diff --git a/src/ray/core_worker/task_interface.cc b/src/ray/core_worker/task_interface.cc new file mode 100644 index 000000000000..ab8b8950c298 --- /dev/null +++ b/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 &args, + const TaskOptions &task_options, + std::vector *return_ids) { + return Status::OK(); +} + +Status CoreWorkerTaskInterface::CreateActor( + const RayFunction &function, const std::vector &args, + const ActorCreationOptions &actor_creation_options, ActorHandle *actor_handle) { + return Status::OK(); +} + +Status CoreWorkerTaskInterface::SubmitActorTask(ActorHandle &actor_handle, + const RayFunction &function, + const std::vector &args, + const TaskOptions &task_options, + std::vector *return_ids) { + return Status::OK(); +} + +} // namespace ray diff --git a/src/ray/core_worker/task_interface.h b/src/ray/core_worker/task_interface.h new file mode 100644 index 000000000000..f667d8d5a06f --- /dev/null +++ b/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 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 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 &args, + const TaskOptions &task_options, std::vector *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 &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 &args, + const TaskOptions &task_options, + std::vector *return_ids); + + private: + /// Reference to the parent CoreWorker instance. + CoreWorker &core_worker_; +}; + +} // namespace ray + +#endif // RAY_CORE_WORKER_TASK_INTERFACE_H