Remove raylet client from Python worker (#6018)

src/ray/core_worker/core_worker.cc

@@ -1,7 +1,5 @@
 #include "ray/core_worker/core_worker.h"
 
-#include <cstdlib>
-
 #include "boost/fiber/all.hpp"
 #include "ray/common/ray_config.h"
 #include "ray/common/task/task_util.h"
@@ -646,6 +644,26 @@ TaskID CoreWorker::GetCallerId() const {
   return caller_id;
 }
 
+Status CoreWorker::PushError(const JobID &job_id, const std::string &type,
+                             const std::string &error_message, double timestamp) {
+  return local_raylet_client_->PushError(job_id, type, error_message, timestamp);
+}
+
+Status CoreWorker::PrepareActorCheckpoint(const ActorID &actor_id,
+                                          ActorCheckpointID *checkpoint_id) {
+  return local_raylet_client_->PrepareActorCheckpoint(actor_id, checkpoint_id);
+}
+
+Status CoreWorker::NotifyActorResumedFromCheckpoint(
+    const ActorID &actor_id, const ActorCheckpointID &checkpoint_id) {
+  return local_raylet_client_->NotifyActorResumedFromCheckpoint(actor_id, checkpoint_id);
+}
+
+Status CoreWorker::SetResource(const std::string &resource_name, const double capacity,
+                               const ClientID &client_id) {
+  return local_raylet_client_->SetResource(resource_name, capacity, client_id);
+}
+
 Status CoreWorker::SubmitTask(const RayFunction &function,
                               const std::vector<TaskArg> &args,
                               const TaskOptions &task_options,

src/ray/core_worker/core_worker.h

@@ -286,6 +286,40 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
   /// of the bytes zeroed out.
   TaskID GetCallerId() const;
 
+  /// Push an error to the relevant driver.
+  ///
+  /// \param[in] The ID of the job_id that the error is for.
+  /// \param[in] The type of the error.
+  /// \param[in] The error message.
+  /// \param[in] The timestamp of the error.
+  /// \return Status.
+  Status PushError(const JobID &job_id, const std::string &type,
+                   const std::string &error_message, double timestamp);
+
+  /// Request raylet backend to prepare a checkpoint for an actor.
+  ///
+  /// \param[in] actor_id ID of the actor.
+  /// \param[out] checkpoint_id ID of the new checkpoint (output parameter).
+  /// \return Status.
+  Status PrepareActorCheckpoint(const ActorID &actor_id,
+                                ActorCheckpointID *checkpoint_id);
+
+  /// Notify raylet backend that an actor was resumed from a checkpoint.
+  ///
+  /// \param[in] actor_id ID of the actor.
+  /// \param[in] checkpoint_id ID of the checkpoint from which the actor was resumed.
+  /// \return Status.
+  Status NotifyActorResumedFromCheckpoint(const ActorID &actor_id,
+                                          const ActorCheckpointID &checkpoint_id);
+
+  /// Sets a resource with the specified capacity and client id
+  /// \param[in] resource_name Name of the resource to be set.
+  /// \param[in] capacity Capacity of the resource.
+  /// \param[in] client_Id ClientID where the resource is to be set.
+  /// \return Status
+  Status SetResource(const std::string &resource_name, const double capacity,
+                     const ClientID &client_id);
+
   /// Submit a normal task.
   ///
   /// \param[in] function The remote function to execute.

src/ray/core_worker/lib/java/org_ray_runtime_RayNativeRuntime.cc

@@ -125,10 +125,9 @@ JNIEXPORT void JNICALL Java_org_ray_runtime_RayNativeRuntime_nativeSetResource(
   const auto node_id = JavaByteArrayToId<ClientID>(env, nodeId);
   const char *native_resource_name = env->GetStringUTFChars(resourceName, JNI_FALSE);
 
-  auto &raylet_client =
-      reinterpret_cast<ray::CoreWorker *>(nativeCoreWorkerPointer)->GetRayletClient();
-  auto status = raylet_client.SetResource(native_resource_name,
-                                          static_cast<double>(capacity), node_id);
+  auto status =
+      reinterpret_cast<ray::CoreWorker *>(nativeCoreWorkerPointer)
+          ->SetResource(native_resource_name, static_cast<double>(capacity), node_id);
   env->ReleaseStringUTFChars(resourceName, native_resource_name);
   THROW_EXCEPTION_AND_RETURN_IF_NOT_OK(env, status, (void)0);
 }

src/ray/core_worker/lib/java/org_ray_runtime_task_NativeTaskExecutor.cc

@@ -20,8 +20,7 @@ Java_org_ray_runtime_task_NativeTaskExecutor_nativePrepareCheckpoint(
   const auto &task_spec = core_worker.GetWorkerContext().GetCurrentTask();
   RAY_CHECK(task_spec->IsActorTask());
   ActorCheckpointID checkpoint_id;
-  auto status =
-      core_worker.GetRayletClient().PrepareActorCheckpoint(actor_id, checkpoint_id);
+  auto status = core_worker.PrepareActorCheckpoint(actor_id, &checkpoint_id);
   THROW_EXCEPTION_AND_RETURN_IF_NOT_OK(env, status, nullptr);
   jbyteArray result = env->NewByteArray(checkpoint_id.Size());
   env->SetByteArrayRegion(result, 0, checkpoint_id.Size(),
@@ -35,8 +34,7 @@ Java_org_ray_runtime_task_NativeTaskExecutor_nativeNotifyActorResumedFromCheckpo
   auto &core_worker = *reinterpret_cast<ray::CoreWorker *>(nativeCoreWorkerPointer);
   const auto &actor_id = core_worker.GetWorkerContext().GetCurrentActorID();
   const auto checkpoint_id = JavaByteArrayToId<ActorCheckpointID>(env, checkpointId);
-  auto status = core_worker.GetRayletClient().NotifyActorResumedFromCheckpoint(
-      actor_id, checkpoint_id);
+  auto status = core_worker.NotifyActorResumedFromCheckpoint(actor_id, checkpoint_id);
   THROW_EXCEPTION_AND_RETURN_IF_NOT_OK(env, status, (void)0);
 }
 

src/ray/raylet/raylet_client.cc

@@ -307,7 +307,7 @@ Status raylet::RayletClient::FreeObjects(const std::vector<ObjectID> &object_ids
 }
 
 Status raylet::RayletClient::PrepareActorCheckpoint(const ActorID &actor_id,
-                                                    ActorCheckpointID &checkpoint_id) {
+                                                    ActorCheckpointID *checkpoint_id) {
   flatbuffers::FlatBufferBuilder fbb;
   auto message =
       protocol::CreatePrepareActorCheckpointRequest(fbb, to_flatbuf(fbb, actor_id));
@@ -320,7 +320,7 @@ Status raylet::RayletClient::PrepareActorCheckpoint(const ActorID &actor_id,
   if (!status.ok()) return status;
   auto reply_message =
       flatbuffers::GetRoot<protocol::PrepareActorCheckpointReply>(reply.get());
-  checkpoint_id = ActorCheckpointID::FromBinary(reply_message->checkpoint_id()->str());
+  *checkpoint_id = ActorCheckpointID::FromBinary(reply_message->checkpoint_id()->str());
   return Status::OK();
 }
 

src/ray/raylet/raylet_client.h

@@ -221,11 +221,11 @@ class RayletClient : public WorkerLeaseInterface {
 
   /// Request raylet backend to prepare a checkpoint for an actor.
   ///
-  /// \param actor_id ID of the actor.
-  /// \param checkpoint_id ID of the new checkpoint (output parameter).
+  /// \param[in] actor_id ID of the actor.
+  /// \param[out] checkpoint_id ID of the new checkpoint (output parameter).
   /// \return ray::Status.
   ray::Status PrepareActorCheckpoint(const ActorID &actor_id,
-                                     ActorCheckpointID &checkpoint_id);
+                                     ActorCheckpointID *checkpoint_id);
 
   /// Notify raylet backend that an actor was resumed from a checkpoint.
   ///