Revert "[gRPC] Migrate raylet client implementation to grpc (#5120)"

This reverts commit 40395ac.

Author: simon-mo

BUILD.bazel

@@ -52,7 +52,7 @@ proto_library(
 
 cc_proto_library(
     name = "node_manager_cc_proto",
-    deps = [":node_manager_proto"],
+    deps = ["node_manager_proto"],
 )
 
 proto_library(
@@ -62,21 +62,7 @@ proto_library(
 
 cc_proto_library(
     name = "object_manager_cc_proto",
-    deps = [":object_manager_proto"],
-)
-
-proto_library(
-    name = "raylet_proto",
-    srcs = ["src/ray/protobuf/raylet.proto"],
-    deps = [
-        ":common_proto",
-        ":gcs_proto",
-    ],
-)
-
-cc_proto_library(
-    name = "raylet_cc_proto",
-    deps = [":raylet_proto"],
+    deps = ["object_manager_proto"],
 )
 
 proto_library(
@@ -105,7 +91,7 @@ cc_proto_library(
 
 # === Begin of rpc definitions ===
 
-# GRPC common lib.
+# grpc common lib
 cc_library(
     name = "grpc_common_lib",
     srcs = glob([
@@ -155,7 +141,7 @@ cc_grpc_library(
     deps = [":object_manager_cc_proto"],
 )
 
-# Object manager rpc server and client.
+# Object manager server and client.
 cc_library(
     name = "object_manager_rpc",
     hdrs = glob([
@@ -171,43 +157,15 @@ cc_library(
     ],
 )
 
-# Raylet gRPC lib.
-cc_grpc_library(
-    name = "raylet_cc_grpc",
-    srcs = [":raylet_proto"],
-    grpc_only = True,
-    deps = [":raylet_cc_proto"],
-)
-
-# Raylet rpc server and client.
-cc_library(
-    name = "raylet_rpc",
-    srcs = glob([
-        "src/ray/rpc/raylet/*.cc",
-    ]),
-    hdrs = glob([
-        "src/ray/rpc/raylet/*.h",
-        "src/ray/raylet/*.h",
-    ]),
-    copts = COPTS,
-    deps = [
-        ":grpc_common_lib",
-        ":ray_common",
-        ":raylet_cc_grpc",
-        "@boost//:asio",
-        "@com_github_grpc_grpc//:grpc++",
-    ],
-)
-
-# Worker gRPC lib.
+# worker gRPC lib.
 cc_grpc_library(
     name = "worker_cc_grpc",
     srcs = [":worker_proto"],
     grpc_only = True,
     deps = [":worker_cc_proto"],
 )
 
-# Worker server and client.
+# worker server and client.
 cc_library(
     name = "worker_rpc",
     hdrs = glob([
@@ -243,6 +201,7 @@ cc_library(
     copts = COPTS,
     deps = [
         ":common_cc_proto",
+        ":node_manager_fbs",
         ":ray_util",
         "@boost//:asio",
         "@com_github_grpc_grpc//:grpc++",
@@ -341,11 +300,11 @@ cc_library(
     deps = [
         ":common_cc_proto",
         ":gcs",
+        ":node_manager_fbs",
         ":node_manager_rpc",
         ":object_manager",
         ":ray_common",
         ":ray_util",
-        ":raylet_rpc",
         ":stats_lib",
         ":worker_rpc",
         "@boost//:asio",
@@ -416,6 +375,7 @@ cc_test(
     srcs = ["src/ray/raylet/lineage_cache_test.cc"],
     copts = COPTS,
     deps = [
+        ":node_manager_fbs",
         ":raylet_lib",
         "@com_google_googletest//:gtest_main",
     ],
@@ -426,6 +386,7 @@ cc_test(
     srcs = ["src/ray/raylet/reconstruction_policy_test.cc"],
     copts = COPTS,
     deps = [
+        ":node_manager_fbs",
         ":object_manager",
         ":raylet_lib",
         "@com_google_googletest//:gtest_main",
@@ -615,6 +576,7 @@ cc_library(
     deps = [
         ":gcs_cc_proto",
         ":hiredis",
+        ":node_manager_fbs",
         ":node_manager_rpc",
         ":ray_common",
         ":ray_util",
@@ -670,6 +632,13 @@ flatbuffer_cc_library(
     out_prefix = "src/ray/common/",
 )
 
+flatbuffer_cc_library(
+    name = "node_manager_fbs",
+    srcs = ["src/ray/raylet/format/node_manager.fbs"],
+    flatc_args = FLATC_ARGS,
+    out_prefix = "src/ray/raylet/format/",
+)
+
 flatbuffer_cc_library(
     name = "object_manager_fbs",
     srcs = ["src/ray/object_manager/format/object_manager.fbs"],

java/runtime/src/main/java/org/ray/runtime/raylet/RayletClientImpl.java

@@ -37,9 +37,9 @@ public class RayletClientImpl implements RayletClient {
   private long client = 0;
 
   // TODO(qwang): JobId parameter can be removed once we embed jobId in driverId.
-  public RayletClientImpl(String schedulerSockName, UniqueId workerId,
+  public RayletClientImpl(String schedulerSockName, UniqueId clientId,
       boolean isWorker, JobId jobId) {
-    client = nativeInit(schedulerSockName, workerId.getBytes(),
+    client = nativeInit(schedulerSockName, clientId.getBytes(),
         isWorker, jobId.getBytes());
   }
 

python/ray/__init__.py

@@ -5,13 +5,6 @@
 import os
 import sys
 
-# MUST import ray._raylet before pyarrow to initialize some global variables.
-# It seems the library related to memory allocation in pyarrow will destroy the
-# initialization of grpc if we import pyarrow at first.
-# NOTE(JoeyJiang): See https://github.com/ray-project/ray/issues/5219 for more
-# details.
-import ray._raylet
-
 if "pyarrow" in sys.modules:
     raise ImportError("Ray must be imported before pyarrow because Ray "
                       "requires a specific version of pyarrow (which is "

python/ray/_raylet.pyx

@@ -220,14 +220,14 @@ cdef class RayletClient:
     cdef unique_ptr[CRayletClient] client
 
     def __cinit__(self, raylet_socket,
-                  WorkerID worker_id,
+                  ClientID client_id,
                   c_bool is_worker,
                   JobID job_id):
         # We know that we are using Python, so just skip the language
         # parameter.
         # TODO(suquark): Should we allow unicode chars in "raylet_socket"?
         self.client.reset(new CRayletClient(
-            raylet_socket.encode("ascii"), worker_id.native(), is_worker,
+            raylet_socket.encode("ascii"), client_id.native(), is_worker,
             job_id.native(), LANGUAGE_PYTHON))
 
     def disconnect(self):
@@ -374,7 +374,7 @@ cdef class RayletClient:
 
     @property
     def client_id(self):
-        return ClientID(self.client.get().GetWorkerId().Binary())
+        return ClientID(self.client.get().GetClientID().Binary())
 
     @property
     def job_id(self):

python/ray/includes/common.pxd

@@ -88,16 +88,16 @@ cdef extern from "ray/common/id.h" namespace "ray" nogil:
 
 
 cdef extern from "ray/protobuf/common.pb.h" nogil:
-    cdef cppclass CLanguage "ray::rpc::Language":
+    cdef cppclass CLanguage "Language":
         pass
 
 
 # This is a workaround for C++ enum class since Cython has no corresponding
 # representation.
-cdef extern from "ray/protobuf/common.pb.h" namespace "ray::rpc::Language" nogil:
-    cdef CLanguage LANGUAGE_PYTHON "ray::rpc::Language::PYTHON"
-    cdef CLanguage LANGUAGE_CPP "ray::rpc::Language::CPP"
-    cdef CLanguage LANGUAGE_JAVA "ray::rpc::Language::JAVA"
+cdef extern from "ray/protobuf/common.pb.h" namespace "Language" nogil:
+    cdef CLanguage LANGUAGE_PYTHON "Language::PYTHON"
+    cdef CLanguage LANGUAGE_CPP "Language::CPP"
+    cdef CLanguage LANGUAGE_JAVA "Language::JAVA"
 
 
 cdef extern from "ray/common/task/scheduling_resources.h" \

python/ray/includes/libraylet.pxd

@@ -23,14 +23,14 @@ from ray.includes.task cimport CTaskSpec
 
 
 cdef extern from "ray/protobuf/gcs.pb.h" nogil:
-    cdef cppclass GCSProfileEvent "ray::rpc::ProfileTableData::ProfileEvent":
+    cdef cppclass GCSProfileEvent "ProfileTableData::ProfileEvent":
         void set_event_type(const c_string &value)
         void set_start_time(double value)
         void set_end_time(double value)
         c_string set_extra_data(const c_string &value)
         GCSProfileEvent()
 
-    cdef cppclass GCSProfileTableData "ray::rpc::ProfileTableData":
+    cdef cppclass GCSProfileTableData "ProfileTableData":
         void set_component_type(const c_string &value)
         void set_component_id(const c_string &value)
         void set_node_ip_address(const c_string &value)
@@ -43,12 +43,13 @@ ctypedef unordered_map[c_string, c_vector[pair[int64_t, double]]] \
 ctypedef pair[c_vector[CObjectID], c_vector[CObjectID]] WaitResultPair
 
 
-cdef extern from "ray/rpc/raylet/raylet_client.h" namespace "ray::rpc" nogil:
-    cdef cppclass CRayletClient "ray::rpc::RayletClient":
+cdef extern from "ray/raylet/raylet_client.h" nogil:
+    cdef cppclass CRayletClient "RayletClient":
         CRayletClient(const c_string &raylet_socket,
-                      const CWorkerID &worker_id,
+                      const CClientID &client_id,
                       c_bool is_worker, const CJobID &job_id,
                       const CLanguage &language)
+        CRayStatus Disconnect()
         CRayStatus SubmitTask(const CTaskSpec &task_spec)
         CRayStatus GetTask(unique_ptr[CTaskSpec] *task_spec)
         CRayStatus TaskDone()
@@ -72,8 +73,7 @@ cdef extern from "ray/rpc/raylet/raylet_client.h" namespace "ray::rpc" nogil:
             const CActorID &actor_id, const CActorCheckpointID &checkpoint_id)
         CRayStatus SetResource(const c_string &resource_name, const double capacity, const CClientID &client_Id)
         CLanguage GetLanguage() const
-        CWorkerID GetWorkerId() const
+        CClientID GetClientID() const
         CJobID GetJobID() const
         c_bool IsWorker() const
-        CRayStatus Disconnect()
         const ResourceMappingType &GetResourceIDs() const

python/ray/tests/test_multi_node.py

@@ -458,9 +458,11 @@ def g():
         # Make sure the first driver ran to completion.
         assert "success" in out
 
+    nonexistent_id_bytes = _random_string()
+    nonexistent_id_hex = ray.utils.binary_to_hex(nonexistent_id_bytes)
     # Define a driver that creates one task that depends on a nonexistent
     # object. This task will be queued as waiting to execute.
-    driver_script_template = """
+    driver_script = """
 import time
 import ray
 ray.init(redis_address="{}")
@@ -470,24 +472,22 @@ def g(x):
 g.remote(ray.ObjectID(ray.utils.hex_to_binary("{}")))
 time.sleep(1)
 print("success")
-"""
+""".format(redis_address, nonexistent_id_hex)
 
     # Create some drivers and let them exit and make sure everything is
     # still alive.
     for _ in range(3):
-        nonexistent_id_bytes = _random_string()
-        nonexistent_id_hex = ray.utils.binary_to_hex(nonexistent_id_bytes)
-        driver_script = driver_script_template.format(redis_address,
-                                                      nonexistent_id_hex)
         out = run_string_as_driver(driver_script)
         # Simulate the nonexistent dependency becoming available.
         ray.worker.global_worker.put_object(
             ray.ObjectID(nonexistent_id_bytes), None)
         # Make sure the first driver ran to completion.
         assert "success" in out
 
+    nonexistent_id_bytes = _random_string()
+    nonexistent_id_hex = ray.utils.binary_to_hex(nonexistent_id_bytes)
     # Define a driver that calls `ray.wait` on a nonexistent object.
-    driver_script_template = """
+    driver_script = """
 import time
 import ray
 ray.init(redis_address="{}")
@@ -497,15 +497,11 @@ def g():
 g.remote()
 time.sleep(1)
 print("success")
-"""
+""".format(redis_address, nonexistent_id_hex)
 
     # Create some drivers and let them exit and make sure everything is
     # still alive.
     for _ in range(3):
-        nonexistent_id_bytes = _random_string()
-        nonexistent_id_hex = ray.utils.binary_to_hex(nonexistent_id_bytes)
-        driver_script = driver_script_template.format(redis_address,
-                                                      nonexistent_id_hex)
         out = run_string_as_driver(driver_script)
         # Simulate the nonexistent dependency becoming available.
         ray.worker.global_worker.put_object(

python/ray/worker.py

@@ -41,6 +41,7 @@
 from ray import (
     ActorHandleID,
     ActorID,
+    ClientID,
     WorkerID,
     JobID,
     ObjectID,
@@ -1946,7 +1947,7 @@ def connect(node,
 
     worker.raylet_client = ray._raylet.RayletClient(
         node.raylet_socket_name,
-        WorkerID(worker.worker_id),
+        ClientID(worker.worker_id),
         (mode == WORKER_MODE),
         worker.current_job_id,
     )

src/ray/common/grpc_util.h

@@ -4,6 +4,7 @@
 #include <google/protobuf/map.h>
 #include <google/protobuf/repeated_field.h>
 #include <grpcpp/grpcpp.h>
+
 #include "status.h"
 
 namespace ray {
@@ -72,17 +73,6 @@ inline std::vector<T> VectorFromProtobuf(
   return std::vector<T>(pb_repeated.begin(), pb_repeated.end());
 }
 
-template <typename Message>
-using AddFunction = void (Message::*)(const ::std::string &value);
-/// Add a vector of type ID to protobuf message.
-template <typename ID, typename Message>
-inline void IdVectorToProtobuf(const std::vector<ID> &ids, Message &message,
-                               AddFunction<Message> add_func) {
-  for (const auto &id : ids) {
-    (message.*add_func)(id.Binary());
-  }
-}
-
 /// Converts a Protobuf `RepeatedField` to a vector of IDs.
 template <class ID>
 inline std::vector<ID> IdVectorFromProtobuf(

src/ray/common/ray_config_def.h

@@ -20,12 +20,8 @@ RAY_CONFIG(int64_t, ray_cookie, 0x5241590000000000)
 /// warning is logged that the handler is taking too long.
 RAY_CONFIG(int64_t, handler_warning_timeout_ms, 100)
 
-/// The duration between heartbeats. This value is used for both worker and raylet.
+/// The duration between heartbeats. These are sent by the raylet.
 RAY_CONFIG(int64_t, heartbeat_timeout_milliseconds, 100)
-/// Worker heartbeats also use `heartbeat_timeout_milliseconds` as timer timeout period.
-/// If a worker has not sent a heartbeat in the last `num_worker_heartbeats_timeout`
-/// heartbeat intervals, raylet will mark this worker as dead.
-RAY_CONFIG(int64_t, num_worker_heartbeats_timeout, 30)
 /// If a component has not sent a heartbeat in the last num_heartbeats_timeout
 /// heartbeat intervals, the raylet monitor process will report
 /// it as dead to the db_client table.
@@ -156,9 +152,6 @@ RAY_CONFIG(uint32_t, num_actor_checkpoints_to_keep, 20)
 /// Maximum number of ids in one batch to send to GCS to delete keys.
 RAY_CONFIG(uint32_t, maximum_gcs_deletion_batch_size, 1000)
 
-/// Number of times for a raylet client to retry to register.
-RAY_CONFIG(int, num_raylet_client_retry_times, 25)
-
 /// When getting objects from object store, print a warning every this number of attempts.
 RAY_CONFIG(uint32_t, object_store_get_warn_per_num_attempts, 50)