From ead30ca6557762e22595798f53b3ef68a41281c6 Mon Sep 17 00:00:00 2001
From: Edward Oakes <ed.nmi.oakes@gmail.com>
Date: Thu, 3 Sep 2020 19:48:31 -0500
Subject: [PATCH] [Core] fix named actor bug (#10550)

---
 python/ray/tests/test_actor_advanced.py     | 24 ++++++++++++++++++++-
 src/ray/gcs/gcs_server/gcs_actor_manager.cc |  9 ++++++++
 src/ray/gcs/gcs_server/gcs_actor_manager.h  |  2 +-
 3 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/python/ray/tests/test_actor_advanced.py b/python/ray/tests/test_actor_advanced.py
index 3d7446b02..0e018e3ab 100644
--- a/python/ray/tests/test_actor_advanced.py
+++ b/python/ray/tests/test_actor_advanced.py
@@ -11,7 +11,8 @@ import time
 import ray
 import ray.test_utils
 import ray.cluster_utils
-from ray.test_utils import run_string_as_driver, get_non_head_nodes
+from ray.test_utils import (run_string_as_driver, get_non_head_nodes,
+                            wait_for_condition)
 from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
 
 
@@ -646,6 +647,27 @@ assert ray.get(handle.ping.remote()) == "pong"
         detached_actor = ray.get_actor("actor")
         ray.get(detached_actor.ping.remote())
 
+    # Check that the names are reclaimed after actors die.
+
+    def check_name_available(name):
+        try:
+            ray.get_actor(name)
+            return False
+        except ValueError:
+            return True
+
+    @ray.remote
+    class A:
+        pass
+
+    a = A.options(name="my_actor_1").remote()
+    ray.kill(a, no_restart=True)
+    wait_for_condition(lambda: check_name_available("my_actor_1"))
+
+    b = A.options(name="my_actor_2").remote()
+    del b
+    wait_for_condition(lambda: check_name_available("my_actor_2"))
+
 
 def test_detached_actor(ray_start_regular):
     @ray.remote
diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.cc b/src/ray/gcs/gcs_server/gcs_actor_manager.cc
index e97eb2d1d..ed0f53faf 100644
--- a/src/ray/gcs/gcs_server/gcs_actor_manager.cc
+++ b/src/ray/gcs/gcs_server/gcs_actor_manager.cc
@@ -569,6 +569,15 @@ void GcsActorManager::DestroyActor(const ActorID &actor_id) {
     RemoveActorFromOwner(actor);
   }
 
+  // Remove actor from `named_actors_` if its name is not empty.
+  if (!actor->GetName().empty()) {
+    auto it = named_actors_.find(actor->GetName());
+    if (it != named_actors_.end()) {
+      RAY_CHECK(it->second == actor->GetActorID());
+      named_actors_.erase(it);
+    }
+  }
+
   // The actor is already dead, most likely due to process or node failure.
   if (actor->GetState() == rpc::ActorTableData::DEAD) {
     return;
diff --git a/src/ray/gcs/gcs_server/gcs_actor_manager.h b/src/ray/gcs/gcs_server/gcs_actor_manager.h
index 1f84cc2f5..c74a8d814 100644
--- a/src/ray/gcs/gcs_server/gcs_actor_manager.h
+++ b/src/ray/gcs/gcs_server/gcs_actor_manager.h
@@ -92,7 +92,7 @@ class GcsActor {
   ActorID GetActorID() const;
   /// Returns whether or not this is a detached actor.
   bool IsDetached() const;
-  /// Get the name of this actor (only set if it's a detached actor).
+  /// Get the name of this actor.
   std::string GetName() const;
   /// Get the task specification of this actor.
   TaskSpecification GetCreationTaskSpecification() const;