Rename CPU barrier APIs, and change gloo release behavior to not assert its init status.

Author: bliu3650

paddle/fluid/platform/gloo_context.cc

@@ -30,14 +30,17 @@ void GlooParallelContext::Init() {
 
 void GlooParallelContext::Barrier() {
   auto gloo_ptr = paddle::framework::GlooWrapper::GetInstance();
-  CHECK_EQ(gloo_ptr->IsInitialized(), true);
+  PADDLE_ENFORCE_EQ(gloo_ptr->IsInitialized(), true,
+                    paddle::platform::errors::Unavailable(
+                        "Gloo context is not initialized."));
   gloo_ptr->Barrier();
 }
 
 void GlooParallelContext::ReleaseContext() {
   auto gloo_ptr = paddle::framework::GlooWrapper::GetInstance();
-  CHECK_EQ(gloo_ptr->IsInitialized(), true);
-  gloo_ptr.reset();
+  if (gloo_ptr->IsInitialized() == true) {
+    gloo_ptr.reset();
+  }
 }
 #endif
 

python/paddle/distributed/__init__.py

@@ -18,9 +18,9 @@
 from .parallel import get_rank  # noqa: F401
 from .parallel import get_world_size  # noqa: F401
 
-from .parallel_with_gloo import init_gloo_parallel_env
-from .parallel_with_gloo import barrier_func
-from .parallel_with_gloo import release_gloo
+from .parallel_with_gloo import gloo_init_parallel_env
+from .parallel_with_gloo import gloo_barrier
+from .parallel_with_gloo import gloo_release
 
 from paddle.distributed.fleet.dataset import InMemoryDataset  # noqa: F401
 from paddle.distributed.fleet.dataset import QueueDataset  # noqa: F401
@@ -64,9 +64,9 @@
       "ParallelEnv",
       "new_group",
       "init_parallel_env",
-      "init_gloo_parallel_env",
-      "barrier_func",
-      "release_gloo",
+      "gloo_init_parallel_env",
+      "gloo_barrier",
+      "gloo_release",
       "QueueDataset",
       "split",
       "CountFilterEntry",

python/paddle/distributed/parallel_with_gloo.py

@@ -37,7 +37,7 @@ def _start_kv_server(port, http_server_d, size):
     http_server.stop()
 
 
-def init_gloo_parallel_env(rank_id, rank_num, server_endpoint):
+def gloo_init_parallel_env(rank_id, rank_num, server_endpoint):
     """
     Initialize parallel environment with gloo for cpu only.
 
@@ -132,7 +132,7 @@ def test_init_gloo_with_multiprocess(num_of_ranks):
         http_server_proc.join()
 
 
-def barrier_func():
+def gloo_barrier():
     """
     Call barrier function with initialized gloo context.
 
@@ -186,10 +186,11 @@ def test_barrier_with_multiprocess(num_of_ranks):
                 test_barrier_with_multiprocess(2)
     """
 
+    assert _global_gloo_ctx is not None, "gloo context is not initialzed."
     _global_gloo_ctx.barrier()
 
 
-def release_gloo(rank_id):
+def gloo_release():
     """
     Release the parallel environment initialized by gloo
 
@@ -244,4 +245,5 @@ def test_release_gloo_with_multiprocess(num_of_ranks):
                 test_release_gloo_with_multiprocess(2)
     """
 
-    _global_gloo_ctx.release()
+    if _global_gloo_ctx is not None:
+        _global_gloo_ctx.release()

python/paddle/fluid/tests/unittests/test_collective_cpu_barrier_with_gloo.py

@@ -27,7 +27,7 @@
 paddle.enable_static()
 
 
-class CollectiveCPUBarrierText(unittest.TestCase):
+class CollectiveCPUBarrierWithGlooTest(unittest.TestCase):
     def find_free_port(self):
         def _free_port():
             with closing(socket.socket(socket.AF_INET,
@@ -42,42 +42,50 @@ def _free_port():
                 return port
 
     def barrier_func(self, id, rank_num, server_endpoint, out_dict, sleep_time):
-        paddle.distributed.init_gloo_parallel_env(id, rank_num, server_endpoint)
-        # 1st barrier
-        # Run barrier to synchronize processes after starting
-        paddle.distributed.barrier_func()
-        # 2nd barrier
-        # Let rank 0 sleep for one second and check that all processes
-        # saw that artificial delay through the barrier
-        start = time.time()
-        if (id == 0):
-            time.sleep(sleep_time)
-        paddle.distributed.barrier_func()
-        end = time.time()
-        out_dict[id] = end - start
-        # Release
-        paddle.distributed.release_gloo(id)
+        try:
+            paddle.distributed.gloo_init_parallel_env(id, rank_num,
+                                                      server_endpoint)
+            # 1st barrier
+            # Run barrier to synchronize processes after starting
+            paddle.distributed.gloo_barrier()
+            # 2nd barrier
+            # Let rank 0 sleep for one second and check that all processes
+            # saw that artificial delay through the barrier
+            start = time.time()
+            if (id == 0):
+                time.sleep(sleep_time)
+            paddle.distributed.gloo_barrier()
+            end = time.time()
+            out_dict[id] = end - start
+            # Release
+            paddle.distributed.gloo_release()
+        except:
+            out_dict[id] = 0
 
     def barrier_op(self, id, rank_num, server_endpoint, out_dict, sleep_time):
-        main_prog = fluid.Program()
-        startup_prog = fluid.Program()
-        paddle.distributed.init_gloo_parallel_env(id, rank_num, server_endpoint)
-        place = fluid.CPUPlace()
-        with fluid.program_guard(main_prog, startup_prog):
-            paddle.distributed.barrier()
-        exe = fluid.Executor(place)
-        # Run barrier to synchronize processes after starting
-        exe.run(main_prog)
-        # Let rank 0 sleep for one second and check that all processes
-        # saw that artificial delay through the barrier
-        start = time.time()
-        if (id == 0):
-            time.sleep(sleep_time)
-        exe.run(main_prog)
-        end = time.time()
-        out_dict[id] = end - start
-        # Release
-        paddle.distributed.release_gloo(id)
+        try:
+            main_prog = fluid.Program()
+            startup_prog = fluid.Program()
+            paddle.distributed.gloo_init_parallel_env(id, rank_num,
+                                                      server_endpoint)
+            place = fluid.CPUPlace()
+            with fluid.program_guard(main_prog, startup_prog):
+                paddle.distributed.barrier()
+            exe = fluid.Executor(place)
+            # Run barrier to synchronize processes after starting
+            exe.run(main_prog)
+            # Let rank 0 sleep for one second and check that all processes
+            # saw that artificial delay through the barrier
+            start = time.time()
+            if (id == 0):
+                time.sleep(sleep_time)
+            exe.run(main_prog)
+            end = time.time()
+            out_dict[id] = end - start
+            # Release
+            paddle.distributed.gloo_release()
+        except:
+            out_dict[id] = 0
 
     def test_barrier_func_with_multiprocess(self):
         num_of_ranks = 4
@@ -90,7 +98,7 @@ def test_barrier_func_with_multiprocess(self):
         jobs = []
         for id in range(num_of_ranks):
             p = multiprocessing.Process(
-                target=self.barrier_op,
+                target=self.barrier_func,
                 args=(id, num_of_ranks, ep_str, procs_out_dict, sleep_time))
             jobs.append(p)
             p.start()