From 3b141b26cd4af491b3c1fb8ce4dbb00265246b1e Mon Sep 17 00:00:00 2001
From: Wang Qing <kingchin1218@126.com>
Date: Tue, 19 Mar 2019 07:44:49 +0800
Subject: [PATCH]  Fix global_state not disconnected after ray.shutdown (#4354)

---
 python/ray/experimental/state.py | 5 +++++
 python/ray/tests/test_basic.py   | 9 +++++++++
 python/ray/worker.py             | 5 ++++-
 3 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/python/ray/experimental/state.py b/python/ray/experimental/state.py
index eea005874..68d48a6df 100644
--- a/python/ray/experimental/state.py
+++ b/python/ray/experimental/state.py
@@ -120,6 +120,11 @@ class GlobalState(object):
             raise Exception("The ray.global_state API cannot be used before "
                             "ray.init has been called.")
 
+    def disconnect(self):
+        """Disconnect global state from GCS."""
+        self.redis_client = None
+        self.redis_clients = None
+
     def _initialize_global_state(self,
                                  redis_ip_address,
                                  redis_port,
diff --git a/python/ray/tests/test_basic.py b/python/ray/tests/test_basic.py
index 88c64d3e3..4c05e7466 100644
--- a/python/ray/tests/test_basic.py
+++ b/python/ray/tests/test_basic.py
@@ -2887,3 +2887,12 @@ def test_load_code_from_local(shutdown_only):
     base_actor_class = ray.remote(num_cpus=1)(BaseClass)
     base_actor = base_actor_class.remote(message)
     assert ray.get(base_actor.get_data.remote()) == message
+
+
+def test_shutdown_disconnect_global_state():
+    ray.init(num_cpus=0)
+    ray.shutdown()
+
+    with pytest.raises(Exception) as e:
+        ray.global_state.object_table()
+    assert str(e.value).endswith("ray.init has been called.")
diff --git a/python/ray/worker.py b/python/ray/worker.py
index 3938d9256..2e3583029 100644
--- a/python/ray/worker.py
+++ b/python/ray/worker.py
@@ -1566,6 +1566,9 @@ def shutdown(exiting_interpreter=False):
 
     disconnect()
 
+    # Disconnect global state from GCS.
+    global_state.disconnect()
+
     # Shut down the Ray processes.
     global _global_node
     if _global_node is not None:
@@ -2054,7 +2057,7 @@ def connect(info,
 
 
 def disconnect():
-    """Disconnect this worker from the scheduler and object store."""
+    """Disconnect this worker from the raylet and object store."""
     # Reset the list of cached remote functions and actors so that if more
     # remote functions or actors are defined and then connect is called again,
     # the remote functions will be exported. This is mostly relevant for the