diff --git a/src/ray/core_worker/lib/java/org_ray_runtime_object_NativeObjectStore.cc b/src/ray/core_worker/lib/java/org_ray_runtime_object_NativeObjectStore.cc index 94d71223f..24ad763bb 100644 --- a/src/ray/core_worker/lib/java/org_ray_runtime_object_NativeObjectStore.cc +++ b/src/ray/core_worker/lib/java/org_ray_runtime_object_NativeObjectStore.cc @@ -46,11 +46,6 @@ Java_org_ray_runtime_object_NativeObjectStore_nativePut__J_3BLorg_ray_runtime_ob RAY_CHECK(ray_object != nullptr); auto status = GetObjectInterfaceFromPointer(nativeCoreWorkerPointer).Put(*ray_object, object_id); - if (status.IsIOError() && - status.message() == "object already exists in the plasma store") { - // Ignore duplicated put on the same object ID. - return; - } THROW_EXCEPTION_AND_RETURN_IF_NOT_OK(env, status, (void)0); } diff --git a/src/ray/core_worker/store_provider/local_plasma_provider.cc b/src/ray/core_worker/store_provider/local_plasma_provider.cc index 7441883c5..8295a9fd5 100644 --- a/src/ray/core_worker/store_provider/local_plasma_provider.cc +++ b/src/ray/core_worker/store_provider/local_plasma_provider.cc @@ -19,9 +19,16 @@ Status CoreWorkerLocalPlasmaStoreProvider::Put(const RayObject &object, std::shared_ptr out_buffer; { std::unique_lock guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.Create( + arrow::Status status = store_client_.Create( plasma_id, data ? data->Size() : 0, metadata ? metadata->Data() : nullptr, - metadata ? metadata->Size() : 0, &out_buffer)); + metadata ? metadata->Size() : 0, &out_buffer); + if (plasma::IsPlasmaObjectExists(status)) { + // TODO(hchen): Should we propagate this error out of `ObjectInterface::put`? + RAY_LOG(WARNING) << "Trying to put an object that already existed in plasma: " + << object_id << "."; + return Status::OK(); + } + RAY_ARROW_RETURN_NOT_OK(status); } if (data != nullptr) { diff --git a/src/ray/core_worker/transport/direct_actor_transport.cc b/src/ray/core_worker/transport/direct_actor_transport.cc index 2f5733d3c..0c59fa0b7 100644 --- a/src/ray/core_worker/transport/direct_actor_transport.cc +++ b/src/ray/core_worker/transport/direct_actor_transport.cc @@ -63,7 +63,8 @@ Status CoreWorkerDirectActorTaskSubmitter::SubmitTask( // Submit request. auto &client = rpc_clients_[actor_id]; - return PushTask(*client, *request, task_id, num_returns); + PushTask(*client, *request, task_id, num_returns); + return Status::OK(); } else { // Actor is dead, treat the task as failure. RAY_CHECK(iter->second.state_ == ActorTableData::DEAD); @@ -92,6 +93,17 @@ Status CoreWorkerDirectActorTaskSubmitter::SubscribeActorUpdates() { } else { // Remove rpc client if it's dead or being reconstructed. rpc_clients_.erase(actor_id); + // If this actor is permanently dead and there are pending requests, treat + // the pending tasks as failed. + if (actor_data.state() == ActorTableData::DEAD && + pending_requests_.count(actor_id) > 0) { + for (const auto &request : pending_requests_[actor_id]) { + TreatTaskAsFailed(TaskID::FromBinary(request->task_spec().task_id()), + request->task_spec().num_returns(), + rpc::ErrorType::ACTOR_DIED); + } + pending_requests_.erase(actor_id); + } } RAY_LOG(INFO) << "received notification on actor, state=" @@ -114,17 +126,16 @@ void CoreWorkerDirectActorTaskSubmitter::ConnectAndSendPendingTasks( auto &requests = pending_requests_[actor_id]; while (!requests.empty()) { const auto &request = *requests.front(); - auto status = - PushTask(*client, request, TaskID::FromBinary(request.task_spec().task_id()), - request.task_spec().num_returns()); + PushTask(*client, request, TaskID::FromBinary(request.task_spec().task_id()), + request.task_spec().num_returns()); requests.pop_front(); } } -Status CoreWorkerDirectActorTaskSubmitter::PushTask(rpc::DirectActorClient &client, - const rpc::PushTaskRequest &request, - const TaskID &task_id, - int num_returns) { +void CoreWorkerDirectActorTaskSubmitter::PushTask(rpc::DirectActorClient &client, + const rpc::PushTaskRequest &request, + const TaskID &task_id, + int num_returns) { auto status = client.PushTask( request, [this, task_id, num_returns](Status status, const rpc::PushTaskReply &reply) { @@ -153,7 +164,9 @@ Status CoreWorkerDirectActorTaskSubmitter::PushTask(rpc::DirectActorClient &clie store_provider_->Put(RayObject(data_buffer, metadata_buffer), object_id)); } }); - return status; + if (!status.ok()) { + TreatTaskAsFailed(task_id, num_returns, rpc::ErrorType::ACTOR_DIED); + } } void CoreWorkerDirectActorTaskSubmitter::TreatTaskAsFailed( diff --git a/src/ray/core_worker/transport/direct_actor_transport.h b/src/ray/core_worker/transport/direct_actor_transport.h index 90a256900..f9528a852 100644 --- a/src/ray/core_worker/transport/direct_actor_transport.h +++ b/src/ray/core_worker/transport/direct_actor_transport.h @@ -42,15 +42,17 @@ class CoreWorkerDirectActorTaskSubmitter : public CoreWorkerTaskSubmitter { /// Subscribe to all actor updates. Status SubscribeActorUpdates(); - /// Helper function to push a task to an actor. + /// Push a task to a remote actor via the given client. + /// Note, this function doesn't return any error status code. If an error occurs while + /// sending the request, this task will be treated as failed. /// /// \param[in] client The RPC client to send tasks to an actor. /// \param[in] request The request to send. /// \param[in] task_id The ID of a task. /// \param[in] num_returns Number of return objects. - /// \return Status. - Status PushTask(rpc::DirectActorClient &client, const rpc::PushTaskRequest &request, - const TaskID &task_id, int num_returns); + /// \return Void. + void PushTask(rpc::DirectActorClient &client, const rpc::PushTaskRequest &request, + const TaskID &task_id, int num_returns); /// Treat a task as failed. /// diff --git a/src/ray/core_worker/transport/raylet_transport.cc b/src/ray/core_worker/transport/raylet_transport.cc index c6bd9ab8e..004434f9d 100644 --- a/src/ray/core_worker/transport/raylet_transport.cc +++ b/src/ray/core_worker/transport/raylet_transport.cc @@ -49,10 +49,10 @@ void CoreWorkerRayletTaskReceiver::HandleAssignTask( /*transport_type=*/static_cast(TaskTransportType::RAYLET)); Status status = object_interface_.Put(*results[i], id); if (!status.ok()) { - // TODO (kfstorm): RAY_LOG(FATAL) except the error is about the object to put - // already exists. - RAY_LOG(WARNING) << "Task " << task_spec.TaskId() << " failed to put object " << id - << " in store: " << status.message(); + // NOTE(hchen): `PlasmaObjectExists` error is already ignored inside + // `ObjectInterface::Put`, we treat other error types as fatal here. + RAY_LOG(FATAL) << "Task " << task_spec.TaskId() << " failed to put object " << id + << " in store: " << status.message(); } else { RAY_LOG(DEBUG) << "Task " << task_spec.TaskId() << " put object " << id << " in store.";