Add a lock to protect the coordinator to avoid data race condtions.

analytical_engine/.gitignore

@@ -0,0 +1,2 @@
+.cache/
+

python/graphscope/client/session.py

@@ -715,6 +715,9 @@ def __init__(
         # networkx module
         self._nx = None
 
+        # a lock that protects the coordinator
+        self._lock = threading.Lock()
+
     def __repr__(self):
         return str(self.info)
 
@@ -937,7 +940,7 @@ def _wrapper(self, dag_node):
         return dag_node
 
     def run(self, fetches, debug=False):
-        """Run operations of `fetch`.
+        """Run operations of `fetches`.
         Args:
             fetch: :class:`Operation`
 
@@ -955,6 +958,11 @@ def run(self, fetches, debug=False):
         Returns:
             Different values for different output types of :class:`Operation`
         """
+        with self._lock:
+            return self.run_fetches(fetches, debug)
+
+    def run_fetches(self, fetches, debug=False):
+        """Run operations of `fetches` without the session lock."""
         if self._closed:
             raise RuntimeError("Attempted to use a closed Session.")
         if not self._grpc_client: