[NewComm] Set use_new_comm true in default. (#58073)

GhostScreaming · web-flow · commit 263d403d33ca · 2023-10-18T07:14:33.000-05:00
* [NewComm] Set Flags_dynamic_static_unified_comm `True` in default.
New communication library will be used defaultly.

* Polish code.

* Fix problems of distributed testcases using new comm lib.

* Fix problems of testcases using new comm lib in default.

* Fix failed testcase.

* Fix falied testcases.

* [NewComm] Set use_new_comm in default.

* [NewComm] Set use_new_comm in default.

* Fix some problems.

* Fix problems of testcases.

* Disable test_fleet_private_function testcase.

* Remove test_fleet_private_function testcase.
diff --git a/python/paddle/distributed/auto_parallel/static/process_group.py b/python/paddle/distributed/auto_parallel/static/process_group.py
@@ -13,7 +13,6 @@
 # limitations under the License
 
 import hashlib
-import os
 from collections import OrderedDict
 
 import paddle
@@ -158,10 +157,10 @@ def instantiate(self):
             strategy.nrings = 1
             if core.is_compiled_with_cuda():
                 place = core.CUDAPlace(genv.device_id)
-                use_new_comm = os.getenv(
-                    "FLAGS_dynamic_static_unified_comm", "0"
-                )
-                if use_new_comm in ["1", "True", "true"]:
+                use_new_comm = paddle.get_flags(
+                    "FLAGS_dynamic_static_unified_comm"
+                )["FLAGS_dynamic_static_unified_comm"]
+                if use_new_comm:
                     store = core.create_or_get_global_tcp_store()
                     endpoints_str = ""
                     for endpoint in strategy.trainer_endpoints:
diff --git a/python/paddle/distributed/fleet/base/private_helper_function.py b/python/paddle/distributed/fleet/base/private_helper_function.py
@@ -16,6 +16,8 @@
 import time
 from contextlib import closing
 
+import paddle
+
 __all__ = []
 
 
@@ -33,6 +35,15 @@ def wait_server_ready(endpoints):
 
              >>> wait_server_ready(["127.0.0.1:8080", "127.0.0.1:8081"])
     """
+    try:
+        use_new_comm = paddle.get_flags("FLAGS_dynamic_static_unified_comm")[
+            "FLAGS_dynamic_static_unified_comm"
+        ]
+    except:
+        use_new_comm = False
+
+    if use_new_comm:
+        return
     assert not isinstance(endpoints, str)
     while True:
         all_ok = True
diff --git a/python/paddle/distributed/fleet/meta_optimizers/common.py b/python/paddle/distributed/fleet/meta_optimizers/common.py
@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 
 import paddle
 from paddle.framework import core
@@ -99,8 +98,10 @@ def _init_communicator(
             other_endpoints.remove(current_endpoint)
 
         if rank == 0 and wait_port:
-            use_new_comm = os.getenv("FLAGS_dynamic_static_unified_comm", "0")
-            if use_new_comm not in [1, "1", "True", "true"]:
+            use_new_comm = paddle.get_flags(
+                "FLAGS_dynamic_static_unified_comm"
+            )["FLAGS_dynamic_static_unified_comm"]
+            if not use_new_comm:
                 wait_server_ready(other_endpoints)
 
         def _add_sync_by_allreduce(block):
diff --git a/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py b/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py
@@ -14,6 +14,7 @@
 
 import os
 
+import paddle
 from paddle.base import core
 from paddle.incubate.optimizer import PipelineOptimizer
 from paddle.static import (
@@ -714,8 +715,10 @@ def minimize_impl(
         self._recreate_not_persist_param_as_var()
 
         self._dump_program_for_debug()
-        use_new_comm = os.getenv("FLAGS_dynamic_static_unified_comm", "0")
-        if use_new_comm not in ["1", "True", "true"]:
+        use_new_comm = paddle.get_flags("FLAGS_dynamic_static_unified_comm")[
+            "FLAGS_dynamic_static_unified_comm"
+        ]
+        if not use_new_comm:
             self._wait()
         return optimize_ops, params_grads
 
diff --git a/test/collective/fleet/CMakeLists.txt b/test/collective/fleet/CMakeLists.txt
@@ -709,11 +709,6 @@ if((WITH_GPU OR WITH_XPU) AND (LINUX OR WIN32))
     test_fleet_recompute_meta_optimizer ENVS
     "http_proxy=;https_proxy=;PYTHONPATH=../..:${PADDLE_BINARY_DIR}/python")
 endif()
-if(LOCAL_ALL_ARCH AND (LINUX OR WIN32))
-  py_test_modules(
-    test_fleet_private_function MODULES test_fleet_private_function ENVS
-    "http_proxy=;https_proxy=;PYTHONPATH=../..:${PADDLE_BINARY_DIR}/python")
-endif()
 if((WITH_GPU OR WITH_XPU) AND LOCAL_ALL_PLAT)
   bash_test_modules(
     test_new_group
diff --git a/test/collective/fleet/test_fleet_private_function.py b/test/collective/fleet/test_fleet_private_function.py
diff --git a/test/collective/fleet/testslist.csv b/test/collective/fleet/testslist.csv
@@ -61,7 +61,6 @@ test_parallel_dygraph_sparse_embedding_over_height,,ROCM,350,DIST,../../legacy_t
 test_distributed_strategy,LINUX;APPLE,,,,test_runner.py,2,,http_proxy=;https_proxy=;PYTHONPATH=../..,
 test_auto_parallel_parallelizer,,,120,DIST,../../legacy_test/dist_test.sh,2,,http_proxy=;https_proxy=;PYTHONPATH=../..,
 test_fleet_recompute_meta_optimizer,LINUX;WIN32,GPU;XPU,,,test_runner.py,2,,http_proxy=;https_proxy=;PYTHONPATH=../..,
-test_fleet_private_function,LINUX;WIN32,,,,test_runner.py,2,,http_proxy=;https_proxy=;PYTHONPATH=../..,
 test_new_group,,GPU;XPU,,DIST,test_new_group.sh,2,,http_proxy=;https_proxy=,
 test_c_comm_init_op,LINUX,GPU;XPU,120,DIST,test_c_comm_init_op.sh,2,,http_proxy=;https_proxy=,
 test_fused_attention_pass_with_mp,LINUX,GPU,120,DIST,test_fused_attention_pass_with_mp.sh,2,,http_proxy=;https_proxy=,
diff --git a/tools/parallel_UT_rule.py b/tools/parallel_UT_rule.py
@@ -284,7 +284,6 @@
     'test_depthwise_conv_mkldnn_pass',
     'test_fleet_metric',
     'test_fc_fuse_pass_cc',
-    'test_fleet_private_function',
     'test_fleet',
     'test_executor_check_feed',
     'test_py_reader_lod_level_share',
@@ -2121,7 +2120,6 @@
     'test_dgc_optimizer',
     'heter_server_test',
     'test_custom_conj',
-    'test_fleet_private_function',
     'test_fake_init_op',
     'brpc_service_sparse_sgd_test',
     'test_tf32_cudnn',
