[core worker] add task submission & execution interface

BUILD.bazel

@@ -114,6 +114,7 @@ cc_library(
         ],
         exclude = [
             "src/ray/core_worker/*_test.cc",
+            "src/ray/core_worker/mock_worker.cc",
         ],
     ),
     hdrs = glob([
@@ -127,7 +128,15 @@ cc_library(
     ],
 )
 
-# This test is run by src/ray/test/run_core_worker_tests.sh
+cc_binary(
+    name = "mock_worker",
+    srcs = ["src/ray/core_worker/mock_worker.cc"],
+    copts = COPTS,
+    deps = [
+        ":core_worker_lib",
+    ],
+)
+
 cc_binary(
     name = "core_worker_test",
     srcs = ["src/ray/core_worker/core_worker_test.cc"],

src/ray/core_worker/common.h

@@ -12,12 +12,12 @@ namespace ray {
 enum class WorkerType { WORKER, DRIVER };
 
 /// Language of Ray tasks and workers.
-enum class Language { PYTHON, JAVA };
+enum class WorkerLanguage { PYTHON, JAVA };
 
 /// Information about a remote function.
 struct RayFunction {
   /// Language of the remote function.
-  const Language language;
+  const WorkerLanguage language;
   /// Function descriptor of the remote function.
   const std::vector<std::string> function_descriptor;
 };

src/ray/core_worker/context.cc

@@ -23,7 +23,6 @@ struct WorkerThreadContext {
   void SetCurrentTask(const raylet::TaskSpecification &spec) {
     SetCurrentTask(spec.TaskId());
   }
-
  private:
   /// The task ID for current task.
   TaskID current_task_id;

src/ray/core_worker/core_worker.cc

@@ -3,14 +3,16 @@
 
 namespace ray {
 
-CoreWorker::CoreWorker(const enum WorkerType worker_type, const enum Language language,
+CoreWorker::CoreWorker(const enum WorkerType worker_type,
+                       const enum WorkerLanguage language,
                        const std::string &store_socket, const std::string &raylet_socket,
                        DriverID driver_id)
     : worker_type_(worker_type),
       language_(language),
       worker_context_(worker_type, driver_id),
       store_socket_(store_socket),
       raylet_socket_(raylet_socket),
+      is_initialized_(false),
       task_interface_(*this),
       object_interface_(*this),
       task_execution_interface_(*this) {}
@@ -20,10 +22,8 @@ Status CoreWorker::Connect() {
   RAY_ARROW_RETURN_NOT_OK(store_client_.Connect(store_socket_));
 
   // connect to raylet.
-  ::Language lang = ::Language::PYTHON;
-  if (language_ == ray::Language::JAVA) {
-    lang = ::Language::JAVA;
-  }
+  ::Language language = (language_ == ray::WorkerLanguage::JAVA) ? (::Language::JAVA)
+                                                                 : (::Language::PYTHON);
 
   // TODO: currently RayletClient would crash in its constructor if it cannot
   // connect to Raylet after a number of retries, this needs to be changed
@@ -32,7 +32,8 @@ Status CoreWorker::Connect() {
   raylet_client_ = std::unique_ptr<RayletClient>(
       new RayletClient(raylet_socket_, worker_context_.GetWorkerID(),
                        (worker_type_ == ray::WorkerType::WORKER),
-                       worker_context_.GetCurrentDriverID(), lang));
+                       worker_context_.GetCurrentDriverID(), language));
+  is_initialized_ = true;
   return Status::OK();
 }
 

src/ray/core_worker/core_worker.h

@@ -20,7 +20,7 @@ class CoreWorker {
   ///
   /// \param[in] worker_type Type of this worker.
   /// \param[in] langauge Language of this worker.
-  CoreWorker(const WorkerType worker_type, const Language language,
+  CoreWorker(const WorkerType worker_type, const WorkerLanguage language,
              const std::string &store_socket, const std::string &raylet_socket,
              DriverID driver_id = DriverID::Nil());
 
@@ -31,7 +31,7 @@ class CoreWorker {
   enum WorkerType WorkerType() const { return worker_type_; }
 
   /// Language of this worker.
-  enum Language Language() const { return language_; }
+  enum WorkerLanguage Language() const { return language_; }
 
   /// Return the `CoreWorkerTaskInterface` that contains the methods related to task
   /// submisson.
@@ -50,7 +50,7 @@ class CoreWorker {
   const enum WorkerType worker_type_;
 
   /// Language of this worker.
-  const enum Language language_;
+  const enum WorkerLanguage language_;
 
   /// Worker context per thread.
   WorkerContext worker_context_;
@@ -64,9 +64,15 @@ class CoreWorker {
   /// Plasma store client.
   plasma::PlasmaClient store_client_;
 
+  /// Mutex to protect store_client_.
+  std::mutex store_client_mutex_;
+
   /// Raylet client.
   std::unique_ptr<RayletClient> raylet_client_;
 
+  /// Whether this worker has been initialized.
+  bool is_initialized_;
+
   /// The `CoreWorkerTaskInterface` instance.
   CoreWorkerTaskInterface task_interface_;