Add a lock in the coordinator's RunStep as well.

See also: alibaba#1445

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

coordinator/gscoordinator/coordinator.py

@@ -215,6 +215,9 @@ def __init__(self, launcher, dangling_timeout_seconds, log_level="INFO"):
             )
             self._dangling_detecting_timer.start()
 
+        # a lock that protects the coordinator
+        self._lock = threading.Lock()
+
         atexit.register(self._cleanup)
 
     def __del__(self):
@@ -593,8 +596,9 @@ def run_on_coordinator(
         return response_head, response_bodies
 
     def RunStep(self, request_iterator, context):
-        for response in self.RunStepWrapped(request_iterator, context):
-            yield response
+        with self._lock:
+            for response in self.RunStepWrapped(request_iterator, context):
+                yield response
 
     def _RunStep(self, request_iterator, context):
         # split dag