Skip to content

[fluid_ops] clean c_allreduce_prod #72057

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Apr 7, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 0 additions & 14 deletions paddle/fluid/framework/new_executor/pir_interpreter.cc
Original file line number Diff line number Diff line change
Expand Up @@ -534,11 +534,7 @@ void PirInterpreter::UpdateNcclOpNum() {
static std::set<std::string> nccl_op_set = {
"pd_op.c_softmax_with_cross_entropy",
"pd_op.c_softmax_with_multi_label_cross_entropy",
"pd_op.c_allgather",
"pd_op.c_allreduce_avg",
"pd_op.c_allreduce_min",
"pd_op.c_allreduce_sum",
"pd_op.c_allreduce_prod",
"pd_op.c_broadcast",
"pd_op.c_scatter",
"pd_op.partial_send",
Expand Down Expand Up @@ -566,10 +562,7 @@ void PirInterpreter::UpdateNcclOpNum() {
"pd_op.reduce",
"pd_op.c_softmax_with_cross_entropy_grad",
"pd_op.c_softmax_with_multi_label_cross_entropy_grad",
"pd_op.c_allgather_grad",
"pd_op.c_allreduce_min_grad",
"pd_op.c_allreduce_sum_grad",
"pd_op.c_allreduce_prod_grad",
"pd_op.c_broadcast_grad",
"pd_op.c_scatter_grad",
"pd_op.partial_send_grad",
Expand Down Expand Up @@ -598,11 +591,7 @@ void PirInterpreter::UpdateNcclOpNum() {
"pd_op.reduce_grad",
"pd_op.c_softmax_with_cross_entropy_",
"pd_op.c_softmax_with_multi_label_cross_entropy_",
"pd_op.c_allgather_",
"pd_op.c_allreduce_avg_",
"pd_op.c_allreduce_min_",
"pd_op.c_allreduce_sum_",
"pd_op.c_allreduce_prod_",
"pd_op.c_broadcast_",
"pd_op.c_scatter_",
"pd_op.partial_send_",
Expand Down Expand Up @@ -630,10 +619,7 @@ void PirInterpreter::UpdateNcclOpNum() {
"pd_op.reduce_",
"pd_op.c_softmax_with_cross_entropy_grad_",
"pd_op.c_softmax_with_multi_label_cross_entropy_grad_",
"pd_op.c_allgather_grad_",
"pd_op.c_allreduce_min_grad_",
"pd_op.c_allreduce_sum_grad_",
"pd_op.c_allreduce_prod_grad_",
"pd_op.c_broadcast_grad_",
"pd_op.c_scatter_grad_",
"pd_op.partial_send_grad_",
Expand Down
5 changes: 1 addition & 4 deletions paddle/fluid/inference/tensorrt/convert/c_allreduce_op.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,7 @@ namespace paddle::inference::tensorrt {
using ReduceType = paddle::inference::tensorrt::plugin::ReduceType;
std::map<std::string, ReduceType> op_to_reduce_type = {
{"c_allreduce_sum", paddle::inference::tensorrt::plugin::kRedSum},
{"c_allreduce_min", paddle::inference::tensorrt::plugin::kRedMin},
{"c_allreduce_prod", paddle::inference::tensorrt::plugin::kRedProd}};
};

class CAllReduceOpConverter : public OpConverter {
public:
Expand Down Expand Up @@ -88,5 +87,3 @@ class CAllReduceOpConverter : public OpConverter {
} // namespace paddle::inference::tensorrt

REGISTER_TRT_OP_CONVERTER(c_allreduce_sum, CAllReduceOpConverter);
REGISTER_TRT_OP_CONVERTER(c_allreduce_min, CAllReduceOpConverter);
REGISTER_TRT_OP_CONVERTER(c_allreduce_prod, CAllReduceOpConverter);
4 changes: 0 additions & 4 deletions paddle/fluid/inference/tensorrt/op_teller.cc
Original file line number Diff line number Diff line change
Expand Up @@ -3192,8 +3192,6 @@ struct SimpleOpTypeSetTeller : public Teller {
"fused_preln_embedding_eltwise_layernorm",
"fused_bias_dropout_residual_layer_norm",
"c_allreduce_sum",
"c_allreduce_min",
"c_allreduce_prod",
"roll",
"cast",
"preln_skip_layernorm",
Expand Down Expand Up @@ -3368,8 +3366,6 @@ struct SimpleOpTypeSetTeller : public Teller {
"preln_skip_layernorm",
"fused_bias_dropout_residual_layer_norm",
"c_allreduce_sum",
"c_allreduce_min",
"c_allreduce_prod",
"roll",
"cast",
"transformer_input_convert",
Expand Down
6 changes: 0 additions & 6 deletions paddle/phi/backends/xpu/xpu2_op_list.cc
Original file line number Diff line number Diff line change
Expand Up @@ -117,12 +117,6 @@ XPUOpMap& get_kl2_ops() {
{"bitwise_and", XPUKernelSet({phi::DataType::BOOL})},
{"bitwise_or", XPUKernelSet({phi::DataType::BOOL})},
{"broadcast", XPUKernelSet({phi::DataType::FLOAT32})},
{"c_allgather",
XPUKernelSet({phi::DataType::FLOAT16,
phi::DataType::FLOAT32,
phi::DataType::INT32,
phi::DataType::INT64,
phi::DataType::UINT8})},
{"c_alltoall",
XPUKernelSet({phi::DataType::FLOAT16,
phi::DataType::FLOAT32,
Expand Down
8 changes: 0 additions & 8 deletions paddle/phi/backends/xpu/xpu3_op_list.cc
Original file line number Diff line number Diff line change
Expand Up @@ -199,14 +199,6 @@ XPUOpMap& get_kl3_ops() {
phi::DataType::FLOAT32,
phi::DataType::FLOAT16,
phi::DataType::BFLOAT16})},
{"c_allgather",
XPUKernelSet({phi::DataType::FLOAT16,
phi::DataType::FLOAT32,
phi::DataType::FLOAT64,
phi::DataType::INT32,
phi::DataType::INT64,
phi::DataType::UINT8,
phi::DataType::BOOL})},
{"c_alltoall",
XPUKernelSet({phi::DataType::FLOAT16,
phi::DataType::FLOAT32,
Expand Down
2 changes: 0 additions & 2 deletions python/paddle/distributed/auto_parallel/static/mapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,6 @@
def is_collective_comm_op(op):
comm_list = [
"c_allreduce_sum",
"c_allreduce_min",
"c_allreduce_prod",
"all_gather",
"all_reduce",
"broadcast",
Expand Down
10 changes: 0 additions & 10 deletions python/paddle/distributed/auto_parallel/static/operators/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -513,16 +513,6 @@ def sync_and_scale_gradients(dist_ctx, op, groups, allreduce_var_names):

allreduce_type = "c_allreduce_sum"
need_scale = dist_ctx.gradient_scale
scale_using_allreduce_avg = dist_ctx.gradient_scale_using_allreduce_avg

# With nccl_version > 2.10.00, we can use c_allreduce_avg to replace c_allreduce_sum and eliminate the scale op.
if (
need_scale
and scale_using_allreduce_avg
and int(paddle.version.nccl()) > 21000
):
allreduce_type = "c_allreduce_avg"
need_scale = False

for group in groups:
group_size = len(group.ranks)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -436,7 +436,6 @@ def _update_program(self, grad_groups):

remove_op_types = [
'scale',
'c_allreduce_avg',
'c_allreduce_sum',
'c_wait_compute',
]
Expand Down Expand Up @@ -492,9 +491,8 @@ def _update_program(self, grad_groups):

allreduce_op = block.ops[group.allreduce_op_idx]
assert allreduce_op.type in [
'c_allreduce_avg',
'c_allreduce_sum',
], f"should found c_allreduce_avg or c_allreduce_sum op but found {allreduce_op}"
], f"should found c_allreduce_sum op but found {allreduce_op}"
allreduce_op_dist_attr = (
self.dist_context.get_op_dist_attr_for_program(allreduce_op)
)
Expand Down
4 changes: 1 addition & 3 deletions python/paddle/distributed/passes/auto_parallel_sharding.py
Original file line number Diff line number Diff line change
Expand Up @@ -1035,7 +1035,6 @@ def op_depend_on_group(op, group):
):
cur_group.is_in_local_shard = True
assert ops[i + 1].type in [
"c_allreduce_avg",
"c_allreduce_sum",
], "Sharding should reduce grad first and than allreduce if Hybrid Sharding with Data-Parallel"
assert (
Expand Down Expand Up @@ -1237,7 +1236,7 @@ def _overlap_grad_comm(
grad_comm_op_to_stream_idx = {}
for idx, op in enumerate(ops):
if is_data_parallel_reduce_op(op):
if op.type in ["c_allreduce_avg", "c_allreduce_sum"]:
if op.type in ["c_allreduce_sum"]:
continue
stream_idx = reduce_op_count % self.grad_comm_stream_num
grad_comm_op_to_stream_idx[op] = stream_idx
Expand Down Expand Up @@ -1291,7 +1290,6 @@ def _overlap_grad_comm(
if self.sharding_hybrid_dp and grad_group.is_in_local_shard:
next_op = ops[idx + 1]
assert next_op.type in [
"c_allreduce_avg",
"c_allreduce_sum",
]
assert next_op.output("Out")[0] == reduce_varname
Expand Down
3 changes: 0 additions & 3 deletions python/paddle/distributed/passes/fuse_all_reduce.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,9 +149,6 @@ def filter_all_collective_op_indices(block):
# NOTE: should add more collective ops
all_collective_ops = {
"c_allreduce_sum",
"c_allreduce_prod",
"c_allreduce_min",
"c_allgather",
"c_broadcast",
"broadcast",
"all_gather",
Expand Down
15 changes: 0 additions & 15 deletions test/ir/inference/test_trt_convert_c_allreduce.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,21 +51,6 @@ def init_case(self):
self.precision = "int8"


# class TestMax(TestDistTRT):
#
# def init_case(self):
# self.op_type = "c_allreduce_max"
# self.target_value = 2.
# self.precision = "fp16"
#
#
# class TestProd(TestDistTRT):
#
# def init_case(self):
# self.op_type = "c_allreduce_prod"
# self.target_value = 2.
# self.precision = "fp16"

if __name__ == '__main__':
paddle.enable_static()
unittest.main()
Loading