Fixes the long-standing "Event loop is closed" issue. (#1247)

The `DriverRemoteConnection` is leaked.

The failure can be reproduced when `traverse_source` involved.

Signed-off-by: Tao He <linzhu.ht@alibaba-inc.com>

python/graphscope/client/connection.py

@@ -92,11 +92,20 @@ class Connection:
     def __init__(self, addr, gremlin_endpoint=None) -> None:
         self._addr = addr
         self._gremlin_endpoint = gremlin_endpoint
+        self._conn = None
         channel = grpc.insecure_channel(addr)
         self._ddl_service_stub = ddl_service_pb2_grpc.ClientDdlStub(channel)
         self._write_service_stub = write_service_pb2_grpc.ClientWriteStub(channel)
         self._client_id = None
 
+    def close(self):
+        if self._conn is not None:
+            try:
+                self._conn.close()
+            except Exception:
+                pass  # be silent when closing
+            self._conn = None
+
     def submit(self, requests):
         return self._ddl_service_stub.batchSubmit(requests)
 
@@ -111,7 +120,9 @@ def g(self):
 
     def gremlin(self):
         graph_url = "ws://%s/gremlin" % self._gremlin_endpoint
-        return traversal().withRemote(DriverRemoteConnection(graph_url, "g"))
+        if self._conn is None:
+            self._conn = DriverRemoteConnection(graph_url, "g")
+        return traversal().withRemote(self._conn)
 
     def _get_client_id(self):
         if self._client_id is None:

python/graphscope/interactive/query.py

@@ -210,6 +210,7 @@ def __init__(
 
         self._status = InteractiveQueryStatus.Initializing
         self._graph_url = None
+        self._conn = None
         # graph object id stored in vineyard
         self._object_id = object_id
         # interactive_query_node is None used for create a interative query
@@ -300,15 +301,24 @@ def traversal_source(self):
             raise RuntimeError(
                 "Interactive query is unavailable with %s status.", str(self._status)
             )
-        ret = traversal().withRemote(DriverRemoteConnection(self._graph_url[0], "g"))
-        return ret
+        if self._conn is None:
+            self._conn = DriverRemoteConnection(self._graph_url[0], "g")
+        return traversal().withRemote(self._conn)
 
     def close(self):
         """Close interactive instance and release resources"""
-        if not self.closed() and not self._session.closed:
-            self._session._wrapper(self._interactive_query_node.close())
-            self._session._close_interactive_instance(self)
-            self._status = InteractiveQueryStatus.Closed
+        if not self.closed():
+            if self._conn is not None:
+                try:
+                    self._conn.close()
+                except Exception:
+                    pass  # be silent when closing
+                self._conn = None
+
+            if not self._session.closed:
+                self._session._wrapper(self._interactive_query_node.close())
+                self._session._close_interactive_instance(self)
+                self._status = InteractiveQueryStatus.Closed
 
 
 class ClosedInteractiveQuery(DAGNode):