[NewComm] Compatiable upgrade for Fluid communication operators.

[GhostScreaming]

本issue用于跟踪记录该项目下各个算子的兼容升级进度。注：

1. 同一个通信算子需要完成兼容升级和单测添加，同学们可直接在本issue下方回复认领。
2. 该任务时间：PR 截止提交时间是9月17日，截止合入时间是9月20日。
3. PR 标题：[NewComm] No. xxx
4. PR 通过 CI 后，在评论里 @hitywt @GhostScreaming 进行code review。
5. 可以参考PR：[Auto Parallel] Upgrade fluid comm operators to be compatible with new comm library #56088 进行修改。
6. 本任务挑战性适中，但需要有单机2卡的环境，欢迎刚接触Paddle分布式开发的同学认领，同学们可以通过通过完成这个 ISSUE/PR 来踏出贡献分布式代码的第一步

待迁移算子列表（整体进度 13/13）

按 merge 的时间顺序，排名不分先后： @GreatV (1) @BeingGod (5) @Difers (1) @jinyouzhi (3) @GhostScreaming (2) @gouzil (1)

序号	算子名称	对应单测	认领人	PR(算子/单测)
1 ✅(23/9/13)	paddle/fluid/operators/class_center_sample_op.cu	test/legacy_test/test_class_center_sample_op.py	@GreatV	#57153
2✅(23/9/21)	paddle/fluid/operators/collective/partial_recv_op.cu.cc	None	@jinyouzhi	#57548
3 ✅(23/9/13)	paddle/fluid/operators/collective/global_scatter_op.cu.cc	test/collective/test_collective_global_scatter.py	@BeingGod	#57161
4 ✅(23/9/13)	paddle/fluid/operators/collective/partial_allgather_op.cu.cc	None	@BeingGod	#57252
5✅(23/9/14)	paddle/fluid/operators/collective/c_softmax_with_cross_entropy_op.cu	test/collective/test_collective_global_scatter.py	@BeingGod	#57208
6✅(23/9/14)	paddle/fluid/operators/collective/partial_send_op.cu.cc	None	@BeingGod	#57251
7✅(23/9/15)	paddle/fluid/operators/collective/global_gather_op.cu.cc	test/collective/test_collective_global_gather.py	@Difers	#57305
8✅(23/9/22)	paddle/fluid/operators/fused/fused_multi_transformer_op.cu.h	test/legacy_test/test_fused_multi_transformer_op.py	@jinyouzhi	#57564
9✅(23/9/21)	paddle/fluid/operators/fused/fused_attention_utils.h	test/legacy_test/test_fused_attention_op.py	@jinyouzhi	#57560
10✅(23/9/21)	paddle/fluid/operators/optimizers/distributed_fused_lamb_op.cu	test/legacy_test/test_distributed_fused_lamb_op_with_gradient_merge.py	@BeingGod	#57424
11 ✅(23/9/21)	paddle/fluid/operators/margin_cross_entropy_op.cu	test/legacy_test/test_margin_cross_entropy_op.py	@GhostScreaming	#57519
12✅(23/9/21)	paddle/fluid/operators/data_norm_op.cu	test/legacy_test/test_data_norm_op.py	@GhostScreaming	#57519
13✅(23/9/22)	paddle/fluid/inference/tensorrt/plugin/c_allreduce_op_plugin.cu	test/ir/inference/test_trt_convert_c_allreduce.py	@gouzil	#57233

[GhostScreaming]

问题描述 Please describe your issue

大家好，当前飞桨通信库管理整体分为三个部分：通信库初始化、分布式算子、通信异步控制。目前在这个三个模块上，动态图/静态图上不统一，存在如下问题：

1. 通信库初始化：用户调用接口统一，但是底层使用的数据结构，初始化方式完全不一样，导致debug信息不一致，代码冗余。
2. 分布式算子：用户调用接口统一，但是底层使用的通信原语存在两处，动静各自一份，导致代码冗余，难以维护。
3. 通信异步控制：由于动态图没有执行器的统一调度，接口不统一。

通信库动静统一是用统一的通信和通信管理语言描述支撑上层的需求，统一通信的多个实现为一套，避免通信部分实现的比较散乱，共享优化，并方便进行通信的动/静态检查和debug。具体架构如下：

目前动态图通信库升级工作已经完成，但静态图通信库的升级，存在新执行器管理Stream导致的精度、性能问题，以及推理框架、参数服务器相应的升级工作。目前采用兼容升级方案，通过环境变量FLAGS_dynamic_static_unified_comm控制使用新/旧通信库。为此，Fluid下的通信算子也需要进行兼容升级，支持新/旧通信库的使用。我们一共收集了14个需要迁移的通信算子，欢迎大家提交PR一起对这些算子做兼容升级。

以c_allgather_op.cu.cc为例。新通信库和环境变量的定义在 comm_context_manager.h, flags.h, nccl_comm_context.h中，首先添加对应的头文件。

使用老通信库，对应的数据结构为platform::NCCLComm，新通信库则是phi::distributed::NCCLCommContext。通信算子需要在对应的gpuStream_t上执行，分别从NCCLComm或NCCLCommContext中获取。

通过设置环境变量export FLAGS_dynamic_static_unified_comm=1来选择用新通信库，否则默认用老通信库。 NCCLCommContext通过全局单例phi::distributed::CommContextManager管理。stream可以通过GetStream()方法获取。更多类方法可以参考paddle/phi/core/distributed/nccl_comm_context.h。

由于使用新通信库，旧通信库管理NCCLComm的全局单例 platform::NCCLCommContext未初始化，通信算子原本的 platform::NCCLComm* comm仍为nullptr，使用到comm的方法，如comm->nranks()的方法需要替换成comm_ctx的对应方法，如comm_ctx->GetRank()。

stream可能切换成计算流，注意这段代码要放在if(FLAGS_dynamic_static_unified_comm)的判断外。
新通信库的通信操作在NCCLCommContext中，将NCCL的通信操作封装成类方法。而旧通信库是直接调用NCCL的通信方法。进行相应的修改。

算子修改完成后，添加对应的单测验证修改的正确性。c_allgather_op.cu.cc对应的单测是 test/collective/test_collective_allgather_api.py，它会执行collective_allgather_api.py。我们只需要参考之前的单测，添加一个以test_开头的测试函数，并设置环境变量export FLAGS_dynamic_static_unified_comm=1。

继承自test_collective_api_base.TestDistBase的单测类，只需要在环境变量中添加need_envs={"FLAGS_dynamic_static_unified_comm": "1"}即可。不同的Op的单测根据情况，完成对应的环境变量设置。

[BeingGod]

stream可能切换成计算流，注意这段代码要放在if(FLAGS_dynamic_static_unified_comm)的判断外。

对于该描述不太理解，能否解释一下，谢谢~ @GhostScreaming

[GhostScreaming]

@BeingGod 不管使用新通信库（phi::distributed::NCCLCommContext ）或者旧通信库（platform::NCCLComm），ctx.Attr<bool>("use_calc_stream") 为 True 时，都应该设置 stream = ctx.cuda_device_context().stream();

[BeingGod]

@BeingGod 不管使用新通信库（phi::distributed::NCCLCommContext ）或者旧通信库（platform::NCCLComm），ctx.Attr<bool>("use_calc_stream") 为 True 时，都应该设置 stream = ctx.cuda_device_context().stream();

好的，谢谢~

[BeingGod]

认领10

[luotao1]

通信库算子兼容性升级已全部完成，感谢参与的小伙伴们！

按 merge 的时间顺序，排名不分先后： @GreatV (1) @BeingGod (5) @Difers (1) @jinyouzhi (3) @GhostScreaming (2) @gouzil (1)

欢迎继续参与 #56689 ！