Fix global_state not disconnected after ray.shutdown (#4354)

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,

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.")

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