[clean old comm] remove FLAGS_dynamic_static_unified_comm in python directory (#70727)

co63oc · web-flow · commit 74eb10a376de · 2025-01-09T14:47:58.000+08:00
diff --git a/python/paddle/distributed/auto_parallel/static/process_group.py b/python/paddle/distributed/auto_parallel/static/process_group.py
@@ -160,58 +160,42 @@ def instantiate(self):
             strategy.nrings = 1
             if core.is_compiled_with_cuda():
                 place = core.CUDAPlace(genv.device_id)
-                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:
-                        endpoints_str += endpoint
-                    endpoints_str += f"ring_id:{ring_id}"
-                    endpoints_str_hash = hashlib.md5(
-                        endpoints_str.encode(encoding='UTF-8')
-                    ).hexdigest()
-
-                    core.CommContextManager.set_device_id(genv.device_id)
-                    core.CommContextManager.create_nccl_comm_context(
-                        store,
-                        str(ring_id),
-                        strategy.local_rank,
-                        strategy.nranks,
-                        endpoints_str_hash,
-                    )
-                else:
-                    core.NCCLParallelContext(strategy, place).init_with_ring_id(
-                        ring_id
-                    )
+                store = core.create_or_get_global_tcp_store()
+                endpoints_str = ""
+                for endpoint in strategy.trainer_endpoints:
+                    endpoints_str += endpoint
+                endpoints_str += f"ring_id:{ring_id}"
+                endpoints_str_hash = hashlib.md5(
+                    endpoints_str.encode(encoding='UTF-8')
+                ).hexdigest()
+
+                core.CommContextManager.set_device_id(genv.device_id)
+                core.CommContextManager.create_nccl_comm_context(
+                    store,
+                    str(ring_id),
+                    strategy.local_rank,
+                    strategy.nranks,
+                    endpoints_str_hash,
+                )
             elif core.is_compiled_with_xpu():
                 place = core.XPUPlace(genv.device_id)
-                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:
-                        endpoints_str += endpoint
-                    endpoints_str += f"ring_id:{ring_id}"
-                    endpoints_str_hash = hashlib.md5(
-                        endpoints_str.encode(encoding='UTF-8')
-                    ).hexdigest()
-
-                    core.CommContextManager.set_device_id(genv.device_id)
-                    core.CommContextManager.create_bkcl_comm_context(
-                        store,
-                        str(ring_id),
-                        strategy.local_rank,
-                        strategy.nranks,
-                        endpoints_str_hash,
-                    )
-                else:
-                    core.BKCLParallelContext(strategy, place).init_with_ring_id(
-                        ring_id
-                    )
+                store = core.create_or_get_global_tcp_store()
+                endpoints_str = ""
+                for endpoint in strategy.trainer_endpoints:
+                    endpoints_str += endpoint
+                endpoints_str += f"ring_id:{ring_id}"
+                endpoints_str_hash = hashlib.md5(
+                    endpoints_str.encode(encoding='UTF-8')
+                ).hexdigest()
+
+                core.CommContextManager.set_device_id(genv.device_id)
+                core.CommContextManager.create_bkcl_comm_context(
+                    store,
+                    str(ring_id),
+                    strategy.local_rank,
+                    strategy.nranks,
+                    endpoints_str_hash,
+                )
             elif genv.device_type in core.get_all_custom_device_type():
                 place = core.CustomPlace(genv.device_type, genv.device_id)
                 core.XCCLParallelContext(strategy, place).init_with_ring_id(
diff --git a/python/paddle/distributed/fleet/base/private_helper_function.py b/python/paddle/distributed/fleet/base/private_helper_function.py
@@ -11,12 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import socket
-import sys
-import time
-from contextlib import closing
 
-import paddle
 
 __all__ = []
 
@@ -35,39 +30,4 @@ 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
-        not_ready_endpoints = []
-        for ep in endpoints:
-            ip_port = ep.split(":")
-            with closing(
-                socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            ) as sock:
-                sock.settimeout(2)
-                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                if hasattr(socket, 'SO_REUSEPORT'):
-                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
-
-                result = sock.connect_ex((ip_port[0], int(ip_port[1])))
-                if result != 0:
-                    all_ok = False
-                    not_ready_endpoints.append(ep)
-        if not all_ok:
-            sys.stderr.write("server not ready, wait 3 sec to retry...\n")
-            sys.stderr.write(
-                "not ready endpoints:" + str(not_ready_endpoints) + "\n"
-            )
-            sys.stderr.flush()
-            time.sleep(3)
-        else:
-            break
+    return
diff --git a/python/paddle/distributed/fleet/meta_optimizers/common.py b/python/paddle/distributed/fleet/meta_optimizers/common.py
@@ -97,13 +97,6 @@ def _init_communicator(
             other_endpoints = endpoints[:]
             other_endpoints.remove(current_endpoint)
 
-        if rank == 0 and wait_port:
-            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):
             sync_var = block.create_var(
                 name=unique_name.generate('sync_var'),
diff --git a/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py b/python/paddle/distributed/fleet/meta_optimizers/sharding_optimizer.py
@@ -14,7 +14,6 @@
 
 import os
 
-import paddle
 from paddle.base import core
 from paddle.incubate.optimizer import PipelineOptimizer
 from paddle.static import (
@@ -705,11 +704,6 @@ def minimize_impl(
         self._recreate_not_persist_param_as_var()
 
         self._dump_program_for_debug()
-        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
 
     def _init_pair_comm(self, pair, ring_id):
