part-2: remove duplicate in topology.py and dygraph_sharding_optimize…

…r.py

python/paddle/distributed/fleet/base/topology.py

@@ -212,19 +212,6 @@ def __init__(self, topology):
             group=self._pp_comm_group,
         )
 
-        # create comm group for pipe parallel
-        self._pp_group, self._pp_comm_group = self._set_comm_group("pipe")
-        # NOTE(shenliang03): In pipeline parallel, we use batch_isend_irecv.
-        # if batch_isend_irecv is the first collective operation, all ranks of
-        # the pipeline group must participate in this call. In order to avoid
-        # this situation, we perform a collective communication in advance and
-        # create a communicator.
-        paddle.distributed.all_reduce(
-            paddle.zeros([1], dtype="int32"),
-            op=paddle.distributed.ReduceOp.SUM,
-            group=self._pp_comm_group,
-        )
-
         # create comm group for data parallel
         self._dp_group, self._dp_comm_group = self._set_comm_group("data")
 

python/paddle/distributed/fleet/meta_optimizers/dygraph_optimizer/dygraph_sharding_optimizer.py

@@ -39,16 +39,6 @@
         not g_shard_use_reduce
     ), "g_shard_norm_align_dp is not supported if g_shard_use_reduce is true"
 
-g_shard_use_reduce = int(os.environ.get("FLAGS_shard_use_reduce", 0))
-logger.info(f"g_shard_use_reduce {g_shard_use_reduce}")
-g_shard_norm_align_dp = int(os.environ.get("FLAGS_shard_norm_align_dp", 1))
-logger.info(f"g_shard_norm_align_dp {g_shard_norm_align_dp}")
-
-if g_shard_norm_align_dp:
-    assert (
-        not g_shard_use_reduce
-    ), "g_shard_norm_align_dp is not support if g_shard_use_reduce is true"
-
 
 def _is_trainable(param):
     return not param.stop_gradient