Fix/shutdown lock (#204)

fed/api.py

@@ -294,7 +294,9 @@ def shutdown():
     """
     Shutdown a RayFed client.
     """
-    _shutdown(True)
+    global_context = get_global_context()
+    if global_context is not None and global_context.acquire_shutdown_flag():
+        _shutdown(True)
 
 
 def _shutdown(intended=True):

fed/tests/test_cross_silo_error.py

@@ -180,5 +180,52 @@ def test_cross_silo_not_expose_error_trace():
     assert p_bob.exitcode == 0
 
 
+@fed.remote
+def foo(e):
+    print(e)
+
+
+def run4(party):
+    compatible_utils.init_ray(address='local')
+    addresses = {
+        'alice': '127.0.0.1:11012',
+        'bob': '127.0.0.1:11011',
+    }
+
+    fed.init(
+        addresses=addresses,
+        party=party,
+        logging_level='debug',
+        config={
+            'cross_silo_comm': {
+                'timeout_ms': 20 * 1000,
+                'expose_error_trace': False,
+            },
+        },
+    )
+
+    a = error_func.party("alice").remote()
+    o = foo.party('bob').remote(a)
+    if party == 'bob':
+        # Wait a while to receive error from alice.
+        import time
+
+        time.sleep(10)
+    # Alice will shutdown once exactly.
+    fed.shutdown()
+    ray.shutdown()
+
+
+def test_cross_silo_alice_send_error_and_shutdown_once():
+    p_alice = multiprocessing.Process(target=run4, args=('alice',))
+    p_bob = multiprocessing.Process(target=run4, args=('bob',))
+    p_alice.start()
+    p_bob.start()
+    p_alice.join()
+    p_bob.join()
+    assert p_alice.exitcode == 0
+    assert p_bob.exitcode == 0
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-sv", __file__]))