Basic C++ worker implementation (#6125)

cpp/src/example/example.cc

@@ -0,0 +1,76 @@
+
+/// This is a complete example of writing a distributed program using the C ++ worker API.
+
+/// including the header
+#include <ray/api.h>
+
+/// using namespace
+using namespace ray::api;
+
+/// general function of user code
+int Return1() { return 1; }
+int Plus1(int x) { return x + 1; }
+int Plus(int x, int y) { return x + y; }
+
+/// a class of user code
+class Counter {
+ public:
+  int count;
+
+  Counter() { count = 0; }
+
+  static Counter *FactoryCreate() { return new Counter(); }
+  /// non static function
+  int Add(int x) {
+    count += x;
+    return count;
+  }
+};
+
+int main() {
+  /// initialization
+  Ray::Init();
+
+  /// put and get object
+  auto obj = Ray::Put(123);
+  auto getRsult = obj.Get();
+
+  /// general function remote call（args passed by value）
+  auto r0 = Ray::Call(Return1);
+  auto r1 = Ray::Call(Plus1, 1);
+  auto r2 = Ray::Call(Plus, 1, 2);
+
+  int result0 = *(r0.Get());
+  int result1 = *(r1.Get());
+  int result2 = *(r2.Get());
+
+  std::cout << "Ray::call with value results: " << result0 << " " << result1 << " "
+            << result2 << std::endl;
+
+  /// general function remote call（args passed by reference）
+  auto r3 = Ray::Call(Return1);
+  auto r4 = Ray::Call(Plus1, r3);
+  auto r5 = Ray::Call(Plus, r4, 1);
+
+  int result3 = *(r3.Get());
+  int result4 = *(r4.Get());
+  int result5 = *(r5.Get());
+
+  std::cout << "Ray::call with reference results: " << result3 << " " << result4 << " "
+            << result5 << std::endl;
+
+  /// create actor and actor function remote call
+  RayActor<Counter> actor = Ray::CreateActor(Counter::FactoryCreate);
+  auto r6 = actor.Call(&Counter::Add, 5);
+  auto r7 = actor.Call(&Counter::Add, 1);
+  auto r8 = actor.Call(&Counter::Add, 1);
+  auto r9 = actor.Call(&Counter::Add, r8);
+
+  int result6 = *(r6.Get());
+  int result7 = *(r7.Get());
+  int result8 = *(r8.Get());
+  int result9 = *(r9.Get());
+
+  std::cout << "Ray::call with actor results: " << result6 << " " << result7 << " "
+            << result8 << " " << result9 << std::endl;
+}

cpp/src/ray/api.cc

@@ -0,0 +1,20 @@
+
+#include <ray/api.h>
+
+#include <ray/api/ray_config.h>
+#include "runtime/abstract_ray_runtime.h"
+
+namespace ray {
+namespace api {
+
+RayRuntime *Ray::runtime_ = nullptr;
+
+std::once_flag Ray::is_inited_;
+void Ray::Init() {
+  std::call_once(is_inited_, [] {
+    runtime_ = AbstractRayRuntime::DoInit(std::make_shared<RayConfig>());
+  });
+}
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/abstract_ray_runtime.cc

@@ -0,0 +1,105 @@
+
+#include "abstract_ray_runtime.h"
+
+#include <cassert>
+
+#include <ray/api.h>
+#include <ray/api/ray_config.h>
+#include <ray/api/ray_exception.h>
+#include "../util/address_helper.h"
+#include "local_mode_ray_runtime.h"
+
+namespace ray {
+namespace api {
+AbstractRayRuntime *AbstractRayRuntime::DoInit(std::shared_ptr<RayConfig> config) {
+  AbstractRayRuntime *runtime;
+  if (config->runMode == RunMode::SINGLE_PROCESS) {
+    GenerateBaseAddressOfCurrentLibrary();
+    runtime = new LocalModeRayRuntime(config);
+  } else {
+    throw RayException("Only single process mode supported now");
+  }
+  RAY_CHECK(runtime);
+  return runtime;
+}
+
+void AbstractRayRuntime::Put(std::shared_ptr<msgpack::sbuffer> data,
+                             const ObjectID &object_id) {
+  object_store_->Put(object_id, data);
+}
+
+ObjectID AbstractRayRuntime::Put(std::shared_ptr<msgpack::sbuffer> data) {
+  ObjectID object_id =
+      ObjectID::ForPut(worker_->GetCurrentTaskID(), worker_->GetNextPutIndex(),
+                       static_cast<uint8_t>(TaskTransportType::RAYLET));
+  Put(data, object_id);
+  return object_id;
+}
+
+std::shared_ptr<msgpack::sbuffer> AbstractRayRuntime::Get(const ObjectID &object_id) {
+  return object_store_->Get(object_id, -1);
+}
+
+std::vector<std::shared_ptr<msgpack::sbuffer>> AbstractRayRuntime::Get(
+    const std::vector<ObjectID> &ids) {
+  return object_store_->Get(ids, -1);
+}
+
+WaitResult AbstractRayRuntime::Wait(const std::vector<ObjectID> &ids, int num_objects,
+                                    int timeout_ms) {
+  return object_store_->Wait(ids, num_objects, timeout_ms);
+}
+
+ObjectID AbstractRayRuntime::Call(RemoteFunctionPtrHolder &fptr,
+                                  std::shared_ptr<msgpack::sbuffer> args) {
+  InvocationSpec invocationSpec;
+  invocationSpec.task_id =
+      TaskID::ForFakeTask();  // TODO(Guyang Song): make it from different task
+  invocationSpec.actor_id = ActorID::Nil();
+  invocationSpec.args = args;
+  invocationSpec.func_offset =
+      (size_t)(fptr.function_pointer - dynamic_library_base_addr);
+  invocationSpec.exec_func_offset =
+      (size_t)(fptr.exec_function_pointer - dynamic_library_base_addr);
+  return task_submitter_->SubmitTask(invocationSpec);
+}
+
+ActorID AbstractRayRuntime::CreateActor(RemoteFunctionPtrHolder &fptr,
+                                        std::shared_ptr<msgpack::sbuffer> args) {
+  return task_submitter_->CreateActor(fptr, args);
+}
+
+ObjectID AbstractRayRuntime::CallActor(const RemoteFunctionPtrHolder &fptr,
+                                       const ActorID &actor,
+                                       std::shared_ptr<msgpack::sbuffer> args) {
+  InvocationSpec invocationSpec;
+  invocationSpec.task_id =
+      TaskID::ForFakeTask();  // TODO(Guyang Song): make it from different task
+  invocationSpec.actor_id = actor;
+  invocationSpec.args = args;
+  invocationSpec.func_offset =
+      (size_t)(fptr.function_pointer - dynamic_library_base_addr);
+  invocationSpec.exec_func_offset =
+      (size_t)(fptr.exec_function_pointer - dynamic_library_base_addr);
+  return task_submitter_->SubmitActorTask(invocationSpec);
+}
+
+const TaskID &AbstractRayRuntime::GetCurrentTaskId() {
+  return worker_->GetCurrentTaskID();
+}
+
+const JobID &AbstractRayRuntime::GetCurrentJobID() { return worker_->GetCurrentJobID(); }
+
+ActorID AbstractRayRuntime::GetNextActorID() {
+  const int next_task_index = worker_->GetNextTaskIndex();
+  const ActorID actor_id = ActorID::Of(worker_->GetCurrentJobID(),
+                                       worker_->GetCurrentTaskID(), next_task_index);
+  return actor_id;
+}
+
+const std::unique_ptr<WorkerContext> &AbstractRayRuntime::GetWorkerContext() {
+  return worker_;
+}
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/abstract_ray_runtime.h

@@ -0,0 +1,62 @@
+
+#pragma once
+
+#include <mutex>
+
+#include <ray/api/ray_config.h>
+#include <ray/api/ray_runtime.h>
+#include <msgpack.hpp>
+#include "./object/object_store.h"
+#include "./task/task_executor.h"
+#include "./task/task_submitter.h"
+#include "ray/core.h"
+
+namespace ray {
+namespace api {
+
+class AbstractRayRuntime : public RayRuntime {
+ public:
+  virtual ~AbstractRayRuntime(){};
+
+  void Put(std::shared_ptr<msgpack::sbuffer> data, const ObjectID &object_id);
+
+  ObjectID Put(std::shared_ptr<msgpack::sbuffer> data);
+
+  std::shared_ptr<msgpack::sbuffer> Get(const ObjectID &id);
+
+  std::vector<std::shared_ptr<msgpack::sbuffer>> Get(const std::vector<ObjectID> &ids);
+
+  WaitResult Wait(const std::vector<ObjectID> &ids, int num_objects, int timeout_ms);
+
+  ObjectID Call(RemoteFunctionPtrHolder &fptr, std::shared_ptr<msgpack::sbuffer> args);
+
+  ActorID CreateActor(RemoteFunctionPtrHolder &fptr,
+                      std::shared_ptr<msgpack::sbuffer> args);
+
+  ObjectID CallActor(const RemoteFunctionPtrHolder &fptr, const ActorID &actor,
+                     std::shared_ptr<msgpack::sbuffer> args);
+
+  ActorID GetNextActorID();
+
+  const TaskID &GetCurrentTaskId();
+
+  const JobID &GetCurrentJobID();
+
+  const std::unique_ptr<WorkerContext> &GetWorkerContext();
+
+ protected:
+  std::shared_ptr<RayConfig> config_;
+  std::unique_ptr<WorkerContext> worker_;
+  std::unique_ptr<TaskSubmitter> task_submitter_;
+  std::unique_ptr<TaskExecutor> task_executor_;
+  std::unique_ptr<ObjectStore> object_store_;
+
+ private:
+  static AbstractRayRuntime *DoInit(std::shared_ptr<RayConfig> config);
+
+  void Execute(const TaskSpecification &task_spec);
+
+  friend class Ray;
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/local_mode_ray_runtime.cc

@@ -0,0 +1,22 @@
+
+#include "local_mode_ray_runtime.h"
+
+#include <ray/api.h>
+#include "../util/address_helper.h"
+#include "./object/local_mode_object_store.h"
+#include "./object/object_store.h"
+#include "./task/local_mode_task_submitter.h"
+
+namespace ray {
+namespace api {
+
+LocalModeRayRuntime::LocalModeRayRuntime(std::shared_ptr<RayConfig> config) {
+  config_ = config;
+  worker_ =
+      std::unique_ptr<WorkerContext>(new WorkerContext(WorkerType::DRIVER, JobID::Nil()));
+  object_store_ = std::unique_ptr<ObjectStore>(new LocalModeObjectStore(*this));
+  task_submitter_ = std::unique_ptr<TaskSubmitter>(new LocalModeTaskSubmitter(*this));
+}
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/local_mode_ray_runtime.h

@@ -0,0 +1,17 @@
+
+#pragma once
+
+#include <unordered_map>
+#include "abstract_ray_runtime.h"
+#include "ray/core.h"
+
+namespace ray {
+namespace api {
+
+class LocalModeRayRuntime : public AbstractRayRuntime {
+ public:
+  LocalModeRayRuntime(std::shared_ptr<RayConfig> config);
+};
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/object/local_mode_object_store.cc

@@ -0,0 +1,89 @@
+
+#include <algorithm>
+#include <chrono>
+#include <list>
+#include <thread>
+
+#include <ray/api/ray_exception.h>
+#include "../abstract_ray_runtime.h"
+#include "local_mode_object_store.h"
+
+namespace ray {
+namespace api {
+LocalModeObjectStore::LocalModeObjectStore(LocalModeRayRuntime &local_mode_ray_tuntime)
+    : local_mode_ray_tuntime_(local_mode_ray_tuntime) {
+  memory_store_ =
+      std::unique_ptr<::ray::CoreWorkerMemoryStore>(new ::ray::CoreWorkerMemoryStore());
+}
+
+void LocalModeObjectStore::PutRaw(const ObjectID &object_id,
+                                  std::shared_ptr<msgpack::sbuffer> data) {
+  auto buffer = std::make_shared<::ray::LocalMemoryBuffer>(
+      reinterpret_cast<uint8_t *>(data->data()), data->size(), true);
+  auto status = memory_store_->Put(
+      ::ray::RayObject(buffer, nullptr, std::vector<ObjectID>()), object_id);
+  if (!status) {
+    throw RayException("Put object error");
+  }
+}
+
+std::shared_ptr<msgpack::sbuffer> LocalModeObjectStore::GetRaw(const ObjectID &object_id,
+                                                               int timeout_ms) {
+  std::vector<ObjectID> object_ids;
+  object_ids.push_back(object_id);
+  auto buffers = GetRaw(object_ids, timeout_ms);
+  RAY_CHECK(buffers.size() == 1);
+  return buffers[0];
+}
+
+std::vector<std::shared_ptr<msgpack::sbuffer>> LocalModeObjectStore::GetRaw(
+    const std::vector<ObjectID> &ids, int timeout_ms) {
+  std::vector<std::shared_ptr<::ray::RayObject>> results;
+  ::ray::Status status =
+      memory_store_->Get(ids, (int)ids.size(), timeout_ms,
+                         *local_mode_ray_tuntime_.GetWorkerContext(), false, &results);
+  if (!status.ok()) {
+    throw RayException("Get object error: " + status.ToString());
+  }
+  RAY_CHECK(results.size() == ids.size());
+  std::vector<std::shared_ptr<msgpack::sbuffer>> result_sbuffers;
+  result_sbuffers.reserve(results.size());
+  for (size_t i = 0; i < results.size(); i++) {
+    auto data_buffer = results[i]->GetData();
+    auto sbuffer = std::make_shared<msgpack::sbuffer>(data_buffer->Size());
+    sbuffer->write(reinterpret_cast<const char *>(data_buffer->Data()),
+                   data_buffer->Size());
+    result_sbuffers.push_back(sbuffer);
+  }
+  return result_sbuffers;
+}
+
+WaitResult LocalModeObjectStore::Wait(const std::vector<ObjectID> &ids, int num_objects,
+                                      int timeout_ms) {
+  absl::flat_hash_set<ObjectID> memory_object_ids;
+  for (const auto &object_id : ids) {
+    memory_object_ids.insert(object_id);
+  }
+  absl::flat_hash_set<ObjectID> ready;
+  ::ray::Status status =
+      memory_store_->Wait(memory_object_ids, num_objects, timeout_ms,
+                          *local_mode_ray_tuntime_.GetWorkerContext(), &ready);
+  if (!status.ok()) {
+    throw RayException("Wait object error: " + status.ToString());
+  }
+  std::vector<ObjectID> ready_vector;
+  ready_vector.reserve(ready.size());
+  std::vector<ObjectID> unready_vector;
+  unready_vector.reserve(ids.size() - ready.size());
+  for (size_t i = 0; i < ids.size(); i++) {
+    if (ready.find(ids[i]) != ready.end()) {
+      ready_vector.push_back(ids[i]);
+    } else {
+      unready_vector.push_back(ids[i]);
+    }
+  }
+  WaitResult result(std::move(ready_vector), std::move(unready_vector));
+  return result;
+}
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/object/local_mode_object_store.h

@@ -0,0 +1,33 @@
+
+#pragma once
+
+#include <unordered_map>
+#include "ray/core.h"
+
+#include "../local_mode_ray_runtime.h"
+#include "object_store.h"
+
+namespace ray {
+namespace api {
+
+class LocalModeObjectStore : public ObjectStore {
+ public:
+  LocalModeObjectStore(LocalModeRayRuntime &local_mode_ray_tuntime);
+
+  WaitResult Wait(const std::vector<ObjectID> &ids, int num_objects, int timeout_ms);
+
+ private:
+  void PutRaw(const ObjectID &object_id, std::shared_ptr<msgpack::sbuffer> data);
+
+  std::shared_ptr<msgpack::sbuffer> GetRaw(const ObjectID &object_id, int timeout_ms);
+
+  std::vector<std::shared_ptr<msgpack::sbuffer>> GetRaw(const std::vector<ObjectID> &ids,
+                                                        int timeout_ms);
+
+  std::unique_ptr<::ray::CoreWorkerMemoryStore> memory_store_;
+
+  LocalModeRayRuntime &local_mode_ray_tuntime_;
+};
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/object/object_store.cc

@@ -0,0 +1,24 @@
+
+#include "object_store.h"
+
+#include <memory>
+#include <utility>
+
+namespace ray {
+namespace api {
+
+void ObjectStore::Put(const ObjectID &object_id, std::shared_ptr<msgpack::sbuffer> data) {
+  PutRaw(object_id, data);
+}
+
+std::shared_ptr<msgpack::sbuffer> ObjectStore::Get(const ObjectID &object_id,
+                                                   int timeout_ms) {
+  return GetRaw(object_id, timeout_ms);
+}
+
+std::vector<std::shared_ptr<msgpack::sbuffer>> ObjectStore::Get(
+    const std::vector<ObjectID> &ids, int timeout_ms) {
+  return GetRaw(ids, timeout_ms);
+}
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/object/object_store.h

@@ -0,0 +1,65 @@
+
+#pragma once
+
+#include <memory>
+
+#include <ray/api/wait_result.h>
+#include <msgpack.hpp>
+
+namespace ray {
+namespace api {
+
+class ObjectStore {
+ public:
+  /// The default timeout to get object.
+  static const int default_get_timeout_ms = 1000;
+
+  virtual ~ObjectStore(){};
+
+  /// Store an object in the object store.
+  ///
+  /// \param[in] object_id The object which should be stored.
+  /// \param[in] data The Serialized object buffer which should be stored.
+  void Put(const ObjectID &object_id, std::shared_ptr<msgpack::sbuffer> data);
+
+  /// Get a single object from the object store.
+  /// This method will be blocked until the object are ready or wait for timeout.
+  ///
+  /// \param[in] object_id The object id which should be got.
+  /// \param[in] timeout_ms The maximum wait time in milliseconds.
+  /// \return shared pointer of the result buffer.
+  std::shared_ptr<msgpack::sbuffer> Get(const ObjectID &object_id,
+                                        int timeout_ms = default_get_timeout_ms);
+
+  /// Get a list of objects from the object store.
+  /// This method will be blocked until all the objects are ready or wait for timeout.
+  ///
+  /// \param[in] ids The object id array which should be got.
+  /// \param[in] timeout_ms The maximum wait time in milliseconds.
+  /// \return shared pointer array of the result buffer.
+  std::vector<std::shared_ptr<msgpack::sbuffer>> Get(
+      const std::vector<ObjectID> &ids, int timeout_ms = default_get_timeout_ms);
+
+  /// Wait for a list of RayObjects to be locally available,
+  /// until specified number of objects are ready, or specified timeout has passed.
+  ///
+  /// \param[in] ids The object id array which should be waited.
+  /// \param[in] num_objects The minimum number of objects to wait.
+  /// \param[in] timeout_ms The maximum wait time in milliseconds.
+  /// \return WaitResult Two arrays, one containing locally available objects, one
+  /// containing the rest.
+  virtual WaitResult Wait(const std::vector<ObjectID> &ids, int num_objects,
+                          int timeout_ms) = 0;
+
+ private:
+  virtual void PutRaw(const ObjectID &object_id,
+                      std::shared_ptr<msgpack::sbuffer> data) = 0;
+
+  virtual std::shared_ptr<msgpack::sbuffer> GetRaw(const ObjectID &object_id,
+                                                   int timeout_ms) = 0;
+
+  virtual std::vector<std::shared_ptr<msgpack::sbuffer>> GetRaw(
+      const std::vector<ObjectID> &ids, int timeout_ms) = 0;
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/invocation_spec.h

@@ -0,0 +1,22 @@
+
+#pragma once
+
+#include <msgpack.hpp>
+#include "ray/core.h"
+
+namespace ray {
+namespace api {
+
+class InvocationSpec {
+ public:
+  TaskID task_id;
+  ActorID actor_id;
+  int actor_counter;
+  /// Remote function offset from base address.
+  size_t func_offset;
+  /// Executable function offset from base address.
+  size_t exec_func_offset;
+  std::shared_ptr<msgpack::sbuffer> args;
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/local_mode_task_submitter.cc

@@ -0,0 +1,110 @@
+
+#include <boost/asio/post.hpp>
+#include <memory>
+
+#include <ray/api/ray_exception.h>
+#include "../../util/address_helper.h"
+#include "../abstract_ray_runtime.h"
+#include "local_mode_task_submitter.h"
+
+namespace ray {
+namespace api {
+
+LocalModeTaskSubmitter::LocalModeTaskSubmitter(
+    LocalModeRayRuntime &local_mode_ray_tuntime)
+    : local_mode_ray_tuntime_(local_mode_ray_tuntime) {
+  thread_pool_.reset(new boost::asio::thread_pool(10));
+}
+
+ObjectID LocalModeTaskSubmitter::Submit(const InvocationSpec &invocation, TaskType type) {
+  /// TODO(Guyang Song): Make the infomation of TaskSpecification more reasonable
+  /// We just reuse the TaskSpecification class and make the single process mode work.
+  /// Maybe some infomation of TaskSpecification are not reasonable or invalid.
+  /// We will enhance this after implement the cluster mode.
+  auto functionDescriptor = FunctionDescriptorBuilder::BuildCpp(
+      "SingleProcess", std::to_string(invocation.func_offset),
+      std::to_string(invocation.exec_func_offset));
+  rpc::Address address;
+  std::unordered_map<std::string, double> required_resources;
+  std::unordered_map<std::string, double> required_placement_resources;
+  TaskSpecBuilder builder;
+  builder.SetCommonTaskSpec(invocation.task_id, rpc::Language::CPP, functionDescriptor,
+                            local_mode_ray_tuntime_.GetCurrentJobID(),
+                            local_mode_ray_tuntime_.GetCurrentTaskId(), 0,
+                            local_mode_ray_tuntime_.GetCurrentTaskId(), address, 1,
+                            required_resources, required_placement_resources);
+  if (type == TaskType::NORMAL_TASK) {
+  } else if (type == TaskType::ACTOR_CREATION_TASK) {
+    builder.SetActorCreationTaskSpec(invocation.actor_id);
+  } else if (type == TaskType::ACTOR_TASK) {
+    const TaskID actor_creation_task_id =
+        TaskID::ForActorCreationTask(invocation.actor_id);
+    const ObjectID actor_creation_dummy_object_id = ObjectID::ForTaskReturn(
+        actor_creation_task_id, 1, static_cast<int>(ray::TaskTransportType::RAYLET));
+    builder.SetActorTaskSpec(invocation.actor_id, actor_creation_dummy_object_id,
+                             ObjectID(), invocation.actor_counter);
+  } else {
+    throw RayException("unknown task type");
+  }
+  auto buffer = std::make_shared<::ray::LocalMemoryBuffer>(
+      reinterpret_cast<uint8_t *>(invocation.args->data()), invocation.args->size(),
+      true);
+  /// TODO(Guyang Song): Use both 'AddByRefArg' and 'AddByValueArg' to distinguish
+  builder.AddByValueArg(::ray::RayObject(buffer, nullptr, std::vector<ObjectID>()));
+  auto task_specification = builder.Build();
+  ObjectID return_object_id =
+      task_specification.ReturnId(0, ray::TaskTransportType::RAYLET);
+
+  std::shared_ptr<msgpack::sbuffer> actor;
+  std::shared_ptr<absl::Mutex> mutex;
+  if (type == TaskType::ACTOR_TASK) {
+    absl::MutexLock lock(&actor_contexts_mutex_);
+    actor = actor_contexts_.at(invocation.actor_id).get()->current_actor;
+    mutex = actor_contexts_.at(invocation.actor_id).get()->actor_mutex;
+  }
+  AbstractRayRuntime *runtime = &local_mode_ray_tuntime_;
+  if (type == TaskType::ACTOR_CREATION_TASK || type == TaskType::ACTOR_TASK) {
+    /// TODO(Guyang Song): Handle task dependencies.
+    /// Execute actor task directly in the main thread because we must guarantee the actor
+    /// task executed by calling order.
+    TaskExecutor::Invoke(task_specification, actor, runtime);
+  } else {
+    boost::asio::post(*thread_pool_.get(),
+                      std::bind(
+                          [actor, mutex, runtime](TaskSpecification &ts) {
+                            if (mutex) {
+                              absl::MutexLock lock(mutex.get());
+                            }
+                            TaskExecutor::Invoke(ts, actor, runtime);
+                          },
+                          std::move(task_specification)));
+  }
+  return return_object_id;
+}
+
+ObjectID LocalModeTaskSubmitter::SubmitTask(const InvocationSpec &invocation) {
+  return Submit(invocation, TaskType::NORMAL_TASK);
+}
+
+ActorID LocalModeTaskSubmitter::CreateActor(RemoteFunctionPtrHolder &fptr,
+                                            std::shared_ptr<msgpack::sbuffer> args) {
+  ActorID id = local_mode_ray_tuntime_.GetNextActorID();
+  typedef std::shared_ptr<msgpack::sbuffer> (*ExecFunction)(
+      uintptr_t base_addr, size_t func_offset, std::shared_ptr<msgpack::sbuffer> args);
+  ExecFunction exec_function = (ExecFunction)(fptr.exec_function_pointer);
+  auto data =
+      (*exec_function)(dynamic_library_base_addr,
+                       (size_t)(fptr.function_pointer - dynamic_library_base_addr), args);
+  std::unique_ptr<ActorContext> actorContext(new ActorContext());
+  actorContext->current_actor = data;
+  absl::MutexLock lock(&actor_contexts_mutex_);
+  actor_contexts_.emplace(id, std::move(actorContext));
+  return id;
+}
+
+ObjectID LocalModeTaskSubmitter::SubmitActorTask(const InvocationSpec &invocation) {
+  return Submit(invocation, TaskType::ACTOR_TASK);
+}
+
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/local_mode_task_submitter.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <boost/asio/thread_pool.hpp>
+#include <memory>
+#include <queue>
+#include "../local_mode_ray_runtime.h"
+#include "absl/synchronization/mutex.h"
+#include "invocation_spec.h"
+#include "ray/core.h"
+#include "task_executor.h"
+#include "task_submitter.h"
+
+namespace ray {
+namespace api {
+
+class LocalModeTaskSubmitter : public TaskSubmitter {
+ public:
+  LocalModeTaskSubmitter(LocalModeRayRuntime &local_mode_ray_tuntime);
+
+  ObjectID SubmitTask(const InvocationSpec &invocation);
+
+  ActorID CreateActor(RemoteFunctionPtrHolder &fptr,
+                      std::shared_ptr<msgpack::sbuffer> args);
+
+  ObjectID SubmitActorTask(const InvocationSpec &invocation);
+
+ private:
+  std::unordered_map<ActorID, std::unique_ptr<ActorContext>> actor_contexts_;
+
+  absl::Mutex actor_contexts_mutex_;
+
+  std::unique_ptr<boost::asio::thread_pool> thread_pool_;
+
+  LocalModeRayRuntime &local_mode_ray_tuntime_;
+
+  ObjectID Submit(const InvocationSpec &invocation, TaskType type);
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/task_executor.cc

@@ -0,0 +1,46 @@
+
+#include <memory>
+
+#include "../../util/address_helper.h"
+#include "../abstract_ray_runtime.h"
+#include "task_executor.h"
+
+namespace ray {
+namespace api {
+
+// TODO(Guyang Song): Make a common task execution function used for both local mode and
+// cluster mode.
+std::unique_ptr<ObjectID> TaskExecutor::Execute(const InvocationSpec &invocation) {
+  return std::unique_ptr<ObjectID>(new ObjectID());
+};
+
+void TaskExecutor::Invoke(const TaskSpecification &task_spec,
+                          std::shared_ptr<msgpack::sbuffer> actor,
+                          AbstractRayRuntime *runtime) {
+  auto args = std::make_shared<msgpack::sbuffer>(task_spec.ArgDataSize(0));
+  /// TODO(Guyang Song): Avoid the memory copy.
+  args->write(reinterpret_cast<const char *>(task_spec.ArgData(0)),
+              task_spec.ArgDataSize(0));
+  auto function_descriptor = task_spec.FunctionDescriptor();
+  auto typed_descriptor = function_descriptor->As<ray::CppFunctionDescriptor>();
+  std::shared_ptr<msgpack::sbuffer> data;
+  if (actor) {
+    typedef std::shared_ptr<msgpack::sbuffer> (*ExecFunction)(
+        uintptr_t base_addr, size_t func_offset, std::shared_ptr<msgpack::sbuffer> args,
+        std::shared_ptr<msgpack::sbuffer> object);
+    ExecFunction exec_function = (ExecFunction)(
+        dynamic_library_base_addr + std::stoul(typed_descriptor->ExecFunctionOffset()));
+    data = (*exec_function)(dynamic_library_base_addr,
+                            std::stoul(typed_descriptor->FunctionOffset()), args, actor);
+  } else {
+    typedef std::shared_ptr<msgpack::sbuffer> (*ExecFunction)(
+        uintptr_t base_addr, size_t func_offset, std::shared_ptr<msgpack::sbuffer> args);
+    ExecFunction exec_function = (ExecFunction)(
+        dynamic_library_base_addr + std::stoul(typed_descriptor->ExecFunctionOffset()));
+    data = (*exec_function)(dynamic_library_base_addr,
+                            std::stoul(typed_descriptor->FunctionOffset()), args);
+  }
+  runtime->Put(std::move(data), task_spec.ReturnId(0, ray::TaskTransportType::RAYLET));
+}
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/task_executor.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#include <memory>
+#include "absl/synchronization/mutex.h"
+#include "invocation_spec.h"
+#include "ray/core.h"
+
+namespace ray {
+namespace api {
+
+class AbstractRayRuntime;
+
+class ActorContext {
+ public:
+  std::shared_ptr<msgpack::sbuffer> current_actor = nullptr;
+
+  std::shared_ptr<absl::Mutex> actor_mutex;
+
+  ActorContext() { actor_mutex = std::shared_ptr<absl::Mutex>(new absl::Mutex); }
+};
+
+class TaskExecutor {
+ public:
+  /// TODO(Guyang Song): support multiple tasks execution
+  std::unique_ptr<ObjectID> Execute(const InvocationSpec &invocation);
+
+  static void Invoke(const TaskSpecification &task_spec,
+                     std::shared_ptr<msgpack::sbuffer> actor,
+                     AbstractRayRuntime *runtime);
+
+  virtual ~TaskExecutor(){};
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/runtime/task/task_submitter.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <memory>
+
+#include <ray/api/ray_runtime.h>
+#include "invocation_spec.h"
+
+namespace ray {
+namespace api {
+
+class TaskSubmitter {
+ public:
+  TaskSubmitter(){};
+
+  virtual ~TaskSubmitter(){};
+
+  virtual ObjectID SubmitTask(const InvocationSpec &invocation) = 0;
+
+  virtual ActorID CreateActor(RemoteFunctionPtrHolder &fptr,
+                              std::shared_ptr<msgpack::sbuffer> args) = 0;
+
+  virtual ObjectID SubmitActorTask(const InvocationSpec &invocation) = 0;
+};
+}  // namespace api
+}  // namespace ray

cpp/src/ray/test/api_test.cc

@@ -0,0 +1,134 @@
+
+#include <gtest/gtest.h>
+#include <ray/api.h>
+#include <future>
+#include <thread>
+
+using namespace ray::api;
+
+int Return1() { return 1; }
+int Plus1(int x) { return x + 1; }
+
+int Plus(int x, int y) { return x + y; }
+
+class Counter {
+ public:
+  int count;
+
+  MSGPACK_DEFINE(count);
+
+  Counter() { count = 0; }
+
+  static Counter *FactoryCreate() {
+    Counter *counter = new Counter();
+    return counter;
+  }
+
+  int Plus1(int x) { return x + 1; }
+
+  int Plus(int x, int y) { return x + y; }
+
+  int Add(int x) {
+    count += x;
+    return count;
+  }
+};
+
+TEST(RayApiTest, PutTest) {
+  Ray::Init();
+
+  auto obj1 = Ray::Put(1);
+  auto i1 = obj1.Get();
+  EXPECT_EQ(1, *i1);
+}
+
+TEST(RayApiTest, WaitTest) {
+  Ray::Init();
+  auto r0 = Ray::Call(Return1);
+  auto r1 = Ray::Call(Plus1, 3);
+  auto r2 = Ray::Call(Plus, 2, 3);
+  std::vector<ObjectID> objects = {r0.ID(), r1.ID(), r2.ID()};
+  WaitResult result = Ray::Wait(objects, 3, 1000);
+  EXPECT_EQ(result.ready.size(), 3);
+  EXPECT_EQ(result.unready.size(), 0);
+  std::vector<std::shared_ptr<int>> getResult = Ray::Get<int>(objects);
+  EXPECT_EQ(getResult.size(), 3);
+  EXPECT_EQ(*getResult[0], 1);
+  EXPECT_EQ(*getResult[1], 4);
+  EXPECT_EQ(*getResult[2], 5);
+}
+
+TEST(RayApiTest, CallWithValueTest) {
+  auto r0 = Ray::Call(Return1);
+  auto r1 = Ray::Call(Plus1, 3);
+  auto r2 = Ray::Call(Plus, 2, 3);
+
+  int result0 = *(r0.Get());
+  int result1 = *(r1.Get());
+  int result2 = *(r2.Get());
+
+  EXPECT_EQ(result0, 1);
+  EXPECT_EQ(result1, 4);
+  EXPECT_EQ(result2, 5);
+}
+
+TEST(RayApiTest, CallWithObjectTest) {
+  auto rt0 = Ray::Call(Return1);
+  auto rt1 = Ray::Call(Plus1, rt0);
+  auto rt2 = Ray::Call(Plus, rt1, 3);
+  auto rt3 = Ray::Call(Plus1, 3);
+  auto rt4 = Ray::Call(Plus, rt2, rt3);
+
+  int return0 = *(rt0.Get());
+  int return1 = *(rt1.Get());
+  int return2 = *(rt2.Get());
+  int return3 = *(rt3.Get());
+  int return4 = *(rt4.Get());
+
+  EXPECT_EQ(return0, 1);
+  EXPECT_EQ(return1, 2);
+  EXPECT_EQ(return2, 5);
+  EXPECT_EQ(return3, 4);
+  EXPECT_EQ(return4, 9);
+}
+
+TEST(RayApiTest, ActorTest) {
+  Ray::Init();
+  RayActor<Counter> actor = Ray::CreateActor(Counter::FactoryCreate);
+  auto rt1 = actor.Call(&Counter::Add, 1);
+  auto rt2 = actor.Call(&Counter::Add, 2);
+  auto rt3 = actor.Call(&Counter::Add, 3);
+  auto rt4 = actor.Call(&Counter::Add, rt3);
+
+  int return1 = *(rt1.Get());
+  int return2 = *(rt2.Get());
+  int return3 = *(rt3.Get());
+  int return4 = *(rt4.Get());
+
+  EXPECT_EQ(return1, 1);
+  EXPECT_EQ(return2, 3);
+  EXPECT_EQ(return3, 6);
+  EXPECT_EQ(return4, 12);
+}
+
+TEST(RayApiTest, CompareWithFuture) {
+  // future from a packaged_task
+  std::packaged_task<int(int)> task(Plus1);
+  std::future<int> f1 = task.get_future();
+  std::thread t(std::move(task), 1);
+  int rt1 = f1.get();
+
+  // future from an async()
+  std::future<int> f2 = std::async(std::launch::async, Plus1, 1);
+  int rt2 = f2.get();
+
+  // Ray API
+  Ray::Init();
+  auto f3 = Ray::Call(Plus1, 1);
+  int rt3 = *f3.Get();
+
+  EXPECT_EQ(rt1, 2);
+  EXPECT_EQ(rt2, 2);
+  EXPECT_EQ(rt3, 2);
+  t.join();
+}

cpp/src/ray/test/serialization_test.cc

@@ -0,0 +1,43 @@
+
+#include <gtest/gtest.h>
+#include <ray/api.h>
+
+using namespace ray::api;
+
+TEST(SerializationTest, TypeHybridTest) {
+  uint32_t in_arg1 = 123456789, out_arg1;
+  std::string in_arg2 = "123567ABC", out_arg2;
+
+  // 1 arg
+  // marshall
+  msgpack::sbuffer buffer1;
+  msgpack::packer<msgpack::sbuffer> pk1(&buffer1);
+  Serializer::Serialize(pk1, in_arg1);
+  // unmarshall
+  msgpack::unpacker upk1;
+  upk1.reserve_buffer(buffer1.size());
+  memcpy(upk1.buffer(), buffer1.data(), buffer1.size());
+  upk1.buffer_consumed(buffer1.size());
+
+  Serializer::Deserialize(upk1, &out_arg1);
+
+  EXPECT_EQ(in_arg1, out_arg1);
+
+  // 2 args
+  // marshall
+  msgpack::sbuffer buffer2;
+  msgpack::packer<msgpack::sbuffer> pk2(&buffer2);
+  Serializer::Serialize(pk2, in_arg1);
+  Serializer::Serialize(pk2, in_arg2);
+
+  // unmarshall
+  msgpack::unpacker upk2;
+  upk2.reserve_buffer(buffer2.size());
+  memcpy(upk2.buffer(), buffer2.data(), buffer2.size());
+  upk2.buffer_consumed(buffer2.size());
+  Serializer::Deserialize(upk2, &out_arg1);
+  Serializer::Deserialize(upk2, &out_arg2);
+
+  EXPECT_EQ(in_arg1, out_arg1);
+  EXPECT_EQ(in_arg2, out_arg2);
+}

cpp/src/ray/test/slow_function_test.cc

@@ -0,0 +1,36 @@
+
+#include <gtest/gtest.h>
+#include <ray/api.h>
+#include <chrono>
+#include <thread>
+
+using namespace ray::api;
+
+int slow_function(int i) {
+  std::this_thread::sleep_for(std::chrono::seconds(i));
+  return i;
+}
+
+TEST(RaySlowFunctionTest, BaseTest) {
+  Ray::Init();
+  auto time1 = std::chrono::duration_cast<std::chrono::milliseconds>(
+      std::chrono::system_clock::now().time_since_epoch());
+  auto r0 = Ray::Call(slow_function, 1);
+  auto r1 = Ray::Call(slow_function, 2);
+  auto r2 = Ray::Call(slow_function, 3);
+  auto r3 = Ray::Call(slow_function, 4);
+
+  int result0 = *(r0.Get());
+  int result1 = *(r1.Get());
+  int result2 = *(r2.Get());
+  int result3 = *(r3.Get());
+  auto time2 = std::chrono::duration_cast<std::chrono::milliseconds>(
+      std::chrono::system_clock::now().time_since_epoch());
+
+  EXPECT_EQ(result0, 1);
+  EXPECT_EQ(result1, 2);
+  EXPECT_EQ(result2, 3);
+  EXPECT_EQ(result3, 4);
+
+  EXPECT_LT(time2.count() - time1.count(), 4200);
+}

cpp/src/ray/util/address_helper.cc

@@ -0,0 +1,16 @@
+#include <dlfcn.h>
+#include <stdint.h>
+
+namespace ray {
+namespace api {
+
+uintptr_t dynamic_library_base_addr;
+
+extern "C" void GenerateBaseAddressOfCurrentLibrary() {
+  Dl_info info;
+  dladdr((void *)GenerateBaseAddressOfCurrentLibrary, &info);
+  dynamic_library_base_addr = (uintptr_t)info.dli_fbase;
+  return;
+}
+}  // namespace api
+}  // namespace ray

cpp/src/ray/util/address_helper.h

@@ -0,0 +1,14 @@
+#pragma once
+#include <dlfcn.h>
+#include <stdint.h>
+
+namespace ray {
+namespace api {
+
+/// A base address which is used to calculate function offset
+extern uintptr_t dynamic_library_base_addr;
+
+/// A fixed C language function which help to get infomation from dladdr
+extern "C" void GenerateBaseAddressOfCurrentLibrary();
+}  // namespace api
+}  // namespace ray