[core] Don't log actor restart warning if arg is detached actor (#57931)

Signed-off-by: dayshah <dhyey2019@gmail.com>
Signed-off-by: elliot-barn <elliot.barnwell@anyscale.com>

Author: dayshah

src/ray/core_worker/actor_handle.cc

@@ -37,7 +37,8 @@ rpc::ActorHandle CreateInnerActorHandle(
     bool allow_out_of_order_execution,
     bool enable_tensor_transport,
     std::optional<bool> enable_task_events,
-    const std::unordered_map<std::string, std::string> &labels) {
+    const std::unordered_map<std::string, std::string> &labels,
+    bool is_detached) {
   rpc::ActorHandle inner;
   inner.set_actor_id(actor_id.Data(), actor_id.Size());
   inner.set_owner_id(owner_id.Binary());
@@ -56,6 +57,7 @@ rpc::ActorHandle CreateInnerActorHandle(
   inner.set_enable_tensor_transport(enable_tensor_transport);
   inner.set_enable_task_events(enable_task_events.value_or(kDefaultTaskEventEnabled));
   inner.mutable_labels()->insert(labels.begin(), labels.end());
+  inner.set_is_detached(is_detached);
   return inner;
 }
 
@@ -89,6 +91,7 @@ rpc::ActorHandle CreateInnerActorHandleFromActorData(
       task_spec.actor_creation_task_spec().allow_out_of_order_execution());
   inner.set_max_pending_calls(task_spec.actor_creation_task_spec().max_pending_calls());
   inner.mutable_labels()->insert(task_spec.labels().begin(), task_spec.labels().end());
+  inner.set_is_detached(task_spec.actor_creation_task_spec().is_detached());
   return inner;
 }
 }  // namespace
@@ -109,7 +112,8 @@ ActorHandle::ActorHandle(
     bool allow_out_of_order_execution,
     bool enable_tensor_transport,
     std::optional<bool> enable_task_events,
-    const std::unordered_map<std::string, std::string> &labels)
+    const std::unordered_map<std::string, std::string> &labels,
+    bool is_detached)
     : ActorHandle(CreateInnerActorHandle(actor_id,
                                          owner_id,
                                          owner_address,
@@ -125,7 +129,8 @@ ActorHandle::ActorHandle(
                                          allow_out_of_order_execution,
                                          enable_tensor_transport,
                                          enable_task_events,
-                                         labels)) {}
+                                         labels,
+                                         is_detached)) {}
 
 ActorHandle::ActorHandle(const std::string &serialized)
     : ActorHandle(CreateInnerActorHandleFromString(serialized)) {}

src/ray/core_worker/actor_handle.h

@@ -51,7 +51,8 @@ class ActorHandle {
               bool allow_out_of_order_execution = false,
               bool enable_tensor_transport = false,
               std::optional<bool> enable_task_events = absl::nullopt,
-              const std::unordered_map<std::string, std::string> &labels = {});
+              const std::unordered_map<std::string, std::string> &labels = {},
+              bool is_detached = false);
 
   /// Constructs an ActorHandle from a serialized string.
   explicit ActorHandle(const std::string &serialized);
@@ -113,6 +114,8 @@ class ActorHandle {
 
   bool EnableTensorTransport() const { return inner_.enable_tensor_transport(); }
 
+  bool IsDetached() const { return inner_.is_detached(); }
+
   const ::google::protobuf::Map<std::string, std::string> &GetLabels() const {
     return inner_.labels();
   }

src/ray/core_worker/actor_manager.cc

@@ -119,6 +119,16 @@ bool ActorManager::CheckActorHandleExists(const ActorID &actor_id) {
   return actor_handles_.find(actor_id) != actor_handles_.end();
 }
 
+std::shared_ptr<ActorHandle> ActorManager::GetActorHandleIfExists(
+    const ActorID &actor_id) {
+  absl::MutexLock lock(&mutex_);
+  auto it = actor_handles_.find(actor_id);
+  if (it != actor_handles_.end()) {
+    return it->second;
+  }
+  return nullptr;
+}
+
 bool ActorManager::AddNewActorHandle(std::unique_ptr<ActorHandle> actor_handle,
                                      const std::string &call_site,
                                      const rpc::Address &caller_address,

src/ray/core_worker/actor_manager.h

@@ -143,6 +143,9 @@ class ActorManager {
   /// \param actor_id ID of the actor to be subscribed.
   void SubscribeActorState(const ActorID &actor_id);
 
+  /// Returns the actor handle if it exists, nullptr otherwise.
+  std::shared_ptr<ActorHandle> GetActorHandleIfExists(const ActorID &actor_id);
+
  private:
   /// Give this worker a handle to an actor.
   ///

src/ray/core_worker/core_worker.cc

@@ -2115,7 +2115,8 @@ Status CoreWorker::CreateActor(const RayFunction &function,
       actor_creation_options.allow_out_of_order_execution,
       actor_creation_options.enable_tensor_transport,
       actor_creation_options.enable_task_events,
-      actor_creation_options.labels);
+      actor_creation_options.labels,
+      is_detached);
   std::string serialized_actor_handle;
   actor_handle->Serialize(&serialized_actor_handle);
   ActorID root_detached_actor_id;
@@ -2162,13 +2163,38 @@ Status CoreWorker::CreateActor(const RayFunction &function,
     return Status::OK();
   }
 
+  auto ref_is_detached_actor = [this](const std::string &object_id) {
+    auto ref_object_id = ObjectID::FromBinary(object_id);
+    if (ObjectID::IsActorID(ref_object_id)) {
+      auto ref_actor_id = ObjectID::ToActorID(ref_object_id);
+      if (auto ref_actor_handle = actor_manager_->GetActorHandleIfExists(ref_actor_id)) {
+        if (ref_actor_handle->IsDetached()) {
+          return true;
+        }
+      }
+    }
+    return false;
+  };
   if (task_spec.MaxActorRestarts() != 0) {
     bool actor_restart_warning = false;
     for (size_t i = 0; i < task_spec.NumArgs(); i++) {
-      if (task_spec.ArgByRef(i) || !task_spec.ArgInlinedRefs(i).empty()) {
+      if (task_spec.ArgByRef(i)) {
         actor_restart_warning = true;
         break;
       }
+      if (!task_spec.ArgInlinedRefs(i).empty()) {
+        for (const auto &ref : task_spec.ArgInlinedRefs(i)) {
+          if (!ref_is_detached_actor(ref.object_id())) {
+            // There's an inlined ref that's not a detached actor, so we want to
+            // show the warning.
+            actor_restart_warning = true;
+            break;
+          }
+        }
+      }
+      if (actor_restart_warning) {
+        break;
+      }
     }
     if (actor_restart_warning) {
       RAY_LOG_ONCE_PER_PROCESS(ERROR)

src/ray/protobuf/core_worker.proto

@@ -74,6 +74,8 @@ message ActorHandle {
   map<string, string> labels = 15;
 
   bool enable_tensor_transport = 16;
+
+  bool is_detached = 17;
 }
 
 message PushTaskRequest {