A3 back2back CP enable-hcomm-AtomGit

Msrc/legacy/ascend910/algorithm/base/alg_template/temp_alltoallv/alltoallv_continuous_pipeline.cc+12-6

	@@ -178,13 +178,18 @@ HcclResult AlltoallvContinuousPipeline::SplitBuffer(const bool enablePingPong)
178	// 如果是pingpong模式，需要两倍的大小	178	// 如果是pingpong模式，需要两倍的大小
179	inBufferDataSize_.resize(blockNum);	179	inBufferDataSize_.resize(blockNum);
180		180
181	- const u64 minBufferSize = globalCountsInfoSize + HCCL_MIN_SLICE_ALIGN * blockNum;	181	+ u64 alignSize = HCCL_MIN_SLICE_ALIGN;
		182	+ const u64 minBufferSize = globalCountsInfoSize + alignSize * blockNum;
182	CHK_PRT_RET(bufferSize < minBufferSize,	183	CHK_PRT_RET(bufferSize < minBufferSize,
183	HCCL_ERROR("[AlltoallvContinuousPipeline][SplitBuffer]Insufficient buffer size [%llu Byte]; it needs to be "	184	HCCL_ERROR("[AlltoallvContinuousPipeline][SplitBuffer]Insufficient buffer size [%llu Byte]; it needs to be "
184	"greater than [%llu Byte].", bufferSize, minBufferSize), HCCL_E_MEMORY);	185	"greater than [%llu Byte].", bufferSize, minBufferSize), HCCL_E_MEMORY);
185		186
		187	+ if (bufferSize > globalCountsInfoSize + HCCL_MIN_SLICE_ALIGN_910_93 * blockNum) {
		188	+ alignSize = HCCL_MIN_SLICE_ALIGN_910_93;
		189	+ }
		190	+
186	countsPerBlock_ = (((bufferSize - globalCountsInfoSize) / blockNum) /	191	countsPerBlock_ = (((bufferSize - globalCountsInfoSize) / blockNum) /
187	- HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN) / unitSize_; // 前面已经可以保证countsPerBlock_大于0，不再检查	192	+ alignSize * alignSize) / unitSize_; // 前面已经可以保证countsPerBlock_大于0，不再检查
188	sizePerBlock_ = countsPerBlock_ * unitSize_;	193	sizePerBlock_ = countsPerBlock_ * unitSize_;
189		194
190	for (u32 rank = 0; rank < userRankSize_; ++rank) {	195	for (u32 rank = 0; rank < userRankSize_; ++rank) {
	@@ -195,8 +200,8 @@ HcclResult AlltoallvContinuousPipeline::SplitBuffer(const bool enablePingPong)
195	dataBlockOffsets_.emplace_back(sizePerBlock_ * blockIdx);	200	dataBlockOffsets_.emplace_back(sizePerBlock_ * blockIdx);
196	}	201	}
197		202
198	- HCCL_INFO("[AlltoallvContinuousPipeline][SplitBuffer] Split buffer done, sizePerBlock[%llu], countsPerBlock[%llu], "	203	+ HCCL_INFO("[AlltoallvContinuousPipeline][SplitBuffer] Split buffer done, alignSize[%llu], sizePerBlock[%llu], "
199	- "blockNum[%u]", sizePerBlock_, countsPerBlock_, blockNum);	204	+ "countsPerBlock[%llu], blockNum[%u]", alignSize, sizePerBlock_, countsPerBlock_, blockNum);
200	return HCCL_SUCCESS;	205	return HCCL_SUCCESS;
201	}	206	}
202		207
	@@ -204,7 +209,8 @@ HcclResult AlltoallvContinuousPipeline::PartitionSubStreamsAndNotifies(const std
204	const std::vector<std::shared_ptr<LocalNotify>> &signalMainToSub,	209	const std::vector<std::shared_ptr<LocalNotify>> &signalMainToSub,
205	const std::vector<std::shared_ptr<LocalNotify>> &signalSubToMain)	210	const std::vector<std::shared_ptr<LocalNotify>> &signalSubToMain)
206	{	211	{
207	- const u32 sdmaConcurrentNum = intraRankSize_ - 1;	212	+ constexpr u32 DEVICE_EIGHT = 8;
		213	+ const u32 sdmaConcurrentNum = std::min(intraRankSize_ - 1, DEVICE_EIGHT);
208	const u32 totalSubstreamSize = rdmaConcurrentNum_ + sdmaConcurrentNum;	214	const u32 totalSubstreamSize = rdmaConcurrentNum_ + sdmaConcurrentNum;
209	CHK_PRT_RET(subStreams.size() < totalSubstreamSize \|\| signalMainToSub.size() < totalSubstreamSize \|\|	215	CHK_PRT_RET(subStreams.size() < totalSubstreamSize \|\| signalMainToSub.size() < totalSubstreamSize \|\|
210	signalSubToMain.size() < totalSubstreamSize,	216	signalSubToMain.size() < totalSubstreamSize,
	@@ -251,7 +257,7 @@ HcclResult AlltoallvContinuousPipeline::PartitionSubStreamsAndNotifies(const std
251		257
252	inline u32 AlltoallvContinuousPipeline::GetSdmaSubStreamIdx(const u32 remoteRank) const	258	inline u32 AlltoallvContinuousPipeline::GetSdmaSubStreamIdx(const u32 remoteRank) const
253	{	259	{
254	- return remoteRank > intraRankId_ ? remoteRank - 1 : remoteRank;	260	+ return (remoteRank > intraRankId_ ? remoteRank - 1 : remoteRank) % sdmaSubStreams_.size();
255	}	261	}
256		262
257	inline u64 AlltoallvContinuousPipeline::GetLocalSendCountOfRank(const u32 targetRank) const	263	inline u64 AlltoallvContinuousPipeline::GetLocalSendCountOfRank(const u32 targetRank) const

Msrc/legacy/ascend910/algorithm/impl/coll_executor/coll_all_to_all/CMakeLists.txt+2-0

	@@ -23,6 +23,7 @@ set(src_list
23	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_direct_fullmesh_aiv_executor.cc	23	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_direct_fullmesh_aiv_executor.cc
24	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_aiv_direct_executor.cc	24	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_aiv_direct_executor.cc
25	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_symmetric_memory_executor.cc	25	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_symmetric_memory_executor.cc
		26	+ ${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_pipeline_for_910_93_executor.cc
26	)	27	)
27		28
28	if(TARGET hccl_alg)	29	if(TARGET hccl_alg)
	@@ -42,6 +43,7 @@ if(TARGET ccl_kernel)
42	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_2level_pipeline_excecutor.cc	43	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_2level_pipeline_excecutor.cc
43	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_continuous_pipeline_executor.cc	44	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_continuous_pipeline_executor.cc
44	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_staged_executor.cc	45	${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_staged_executor.cc
		46	+ ${CMAKE_CURRENT_SOURCE_DIR}/coll_all_to_all_v_pipeline_for_910_93_executor.cc
45	)	47	)
46	target_include_directories(ccl_kernel PRIVATE	48	target_include_directories(ccl_kernel PRIVATE
47	${CMAKE_CURRENT_SOURCE_DIR}	49	${CMAKE_CURRENT_SOURCE_DIR}

Asrc/legacy/ascend910/algorithm/impl/coll_executor/coll_all_to_all/coll_all_to_all_v_pipeline_for_910_93_executor.cc+254-0

@@ -0,0 +1,254 @@
	1	+/**
	2	+ * Copyright (c) 2025 Huawei Technologies Co., Ltd.
	3	+ * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
	4	+ * CANN Open Software License Agreement Version 2.0 (the "License").
	5	+ * Please refer to the License for details. You may not use this file except in compliance with the License.
	6	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
	7	+ * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
	8	+ * See LICENSE in the root of the software repository for the full text of the License.
	9	+ */
	10	+
	11	+
	12	+#include "coll_all_to_all_v_pipeline_for_910_93_executor.h"
	13	+
	14	+namespace hccl {
	15	+
	16	+CollAlltoAllVPipelineFor91093::CollAlltoAllVPipelineFor91093(const HcclDispatcher dispatcher,
	17	+ std::unique_ptr<TopoMatcher> &topoMatcher)
	18	+ : CollAlltoAllExecutor(dispatcher, topoMatcher)
	19	+{
	20	+}
	21	+
	22	+HcclResult CollAlltoAllVPipelineFor91093::CalcTransportMemType(
	23	+ TransportMemType &inputType, TransportMemType &outputType)
	24	+{
	25	+ inputType = TransportMemType::CCL_INPUT;
	26	+ outputType = TransportMemType::CCL_OUTPUT;
	27	+ HCCL_INFO("[CollAlltoAllVPipelineFor91093][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
	28	+ tag_.c_str(), inputType, outputType);
	29	+ return HCCL_SUCCESS;
	30	+}
	31	+
	32	+HcclResult CollAlltoAllVPipelineFor91093::CalcLevel1CommInfo(TransportMemType inputType,TransportMemType outputType,
	33	+ std::vector<LevelNSubCommTransport>& opTransport)
	34	+{
	35	+ // level0 + level1 - Mesh建链
	36	+ CommParaInfo commParaCombineL1(COMM_COMBINE_L1, CommType::COMM_TAG_MESH);
	37	+ CHK_RET(CalcCommPlaneInfo(tag_, commParaCombineL1, opTransport[COMM_COMBINE_L1], inputType, outputType));
	38	+ return HCCL_SUCCESS;
	39	+}
	40	+
	41	+HcclResult CollAlltoAllVPipelineFor91093::CalcLevel2CommInfo(TransportMemType inputType, TransportMemType outputType,
	42	+ std::vector<LevelNSubCommTransport>& opTransport)
	43	+{
	44	+ // level2 - Mesh建链
	45	+ CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MESH);
	46	+ CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
	47	+ return HCCL_SUCCESS;
	48	+}
	49	+
	50	+HcclResult CollAlltoAllVPipelineFor91093::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
	51	+{
	52	+ TransportMemType inputType = TransportMemType::RESERVED;
	53	+ TransportMemType outputType = TransportMemType::RESERVED;
	54	+ CHK_RET(CalcTransportMemType(inputType, outputType));
	55	+ CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
	56	+ CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
	57	+ return HCCL_SUCCESS;
	58	+}
	59	+
	60	+HcclResult CollAlltoAllVPipelineFor91093::CalcStreamNum(u32& streamNum)
	61	+{
	62	+ const u32 level0StreamNum = std::min(topoAttr_.deviceNumPerAggregation - 1, DEVICE_EIGHT);
	63	+ const u32 level2StreamNum = 1; // 先固定1条RDMA 流
	64	+ streamNum = level0StreamNum + level2StreamNum; // 最大流数量为9条, RDMA 1条, SDMA 最大8条
	65	+ HCCL_INFO("[CollAlltoAllVPipelineFor91093]tag[%s] level0StreamNum[%u], level2StreamNum[%u], streamNum[%u]",
	66	+ tag_.c_str(), level0StreamNum, level2StreamNum, streamNum);
	67	+ return HCCL_SUCCESS;
	68	+}
	69	+
	70	+HcclResult CollAlltoAllVPipelineFor91093::CalLocalSendRecvInfo(const OpParam &param, SendRecvInfo &info)
	71	+{
	72	+ if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALL) {
	73	+ CalcA2ASendRecvInfo(param, info);
	74	+ } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV) {
	75	+ CalcA2AvSendRecvInfo(param, info);
	76	+ } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) {
	77	+ CalcA2AvcSendRecvInfo(param, info);
	78	+ } else {
	79	+ HCCL_ERROR("[CollAlltoAllVPipelineFor91093] get invalid optype.");
	80	+ return HCCL_E_PARA;
	81	+ }
	82	+ return HCCL_SUCCESS;
	83	+}
	84	+
	85	+HcclResult CollAlltoAllVPipelineFor91093::CalcA2ASendRecvInfo(const OpParam &param, SendRecvInfo &info)
	86	+{
	87	+ const u32 userRankSize = topoAttr_.userRankSize;
	88	+ info.sendCounts.resize(userRankSize);
	89	+ info.sendDispls.resize(userRankSize);
	90	+ info.recvCounts.resize(userRankSize);
	91	+ info.recvDispls.resize(userRankSize);
	92	+ u64 sdispl = 0, rdispl = 0;
	93	+
	94	+ for(u32 i = 0; i < userRankSize; i++) {
	95	+ info.sendCounts[i] = param.All2AllDataDes.sendCount;
	96	+ info.recvCounts[i] = param.All2AllDataDes.sendCount;
	97	+ info.sendDispls[i] = sdispl;
	98	+ info.recvDispls[i] = rdispl;
	99	+ sdispl += param.All2AllDataDes.sendCount;
	100	+ rdispl += param.All2AllDataDes.sendCount;
	101	+ }
	102	+ if (UNLIKELY(HcclCheckLogLevel(DLOG_DEBUG))) {
	103	+ for(u32 i=0; i< userRankSize ;i++) {
	104	+ HCCL_DEBUG("CalcA2ASendRecvInfo sendCounts[%d]=%ld, recvCounts[%d]=%ld, sdispls[%d]=%ld, rdispls[%d]=%ld",
	105	+ i, info.sendCounts[i], i, info.recvCounts[i], i, info.sendDispls[i], i, info.recvDispls[i]);
	106	+ }
	107	+ }
	108	+ return HCCL_SUCCESS;
	109	+}
	110	+
	111	+HcclResult CollAlltoAllVPipelineFor91093::CalcA2AvSendRecvInfo(const OpParam &param, SendRecvInfo &info)
	112	+{
	113	+ const bool hasRecvInfo = param.All2AllDataDes.recvCounts != nullptr;
	114	+ const u32 userRankSize = topoAttr_.userRankSize;
	115	+
	116	+ info.sendCounts.resize(userRankSize);
	117	+ info.sendDispls.resize(userRankSize);
	118	+
	119	+ if (hasRecvInfo) {
	120	+ info.recvCounts.resize(userRankSize);
	121	+ info.recvDispls.resize(userRankSize);
	122	+ }
	123	+
	124	+ for (u32 i = 0; i < userRankSize; ++i) {
	125	+ info.sendCounts[i] = (static_cast<const u64 >(param.All2AllDataDes.sendCounts) + i);
	126	+ info.sendDispls[i] = (static_cast<const u64 >(param.All2AllDataDes.sdispls) + i);
	127	+
	128	+ if (hasRecvInfo) {
	129	+ info.recvCounts[i] = (static_cast<const u64 >(param.All2AllDataDes.recvCounts) + i);
	130	+ info.recvDispls[i] = (static_cast<const u64 >(param.All2AllDataDes.rdispls) + i);
	131	+ }
	132	+ }
	133	+
	134	+ if (UNLIKELY(HcclCheckLogLevel(DLOG_INFO))) {
	135	+ for(u32 i=0; i< userRankSize ;i++) {
	136	+ HCCL_DEBUG("CalcA2AvSendRecvInfo sendCounts[%d]=%ld, recvCounts[%d]=%ld, sdispls[%d]=%ld, rdispls[%d]=%ld",
	137	+ i, info.sendCounts[i], i, info.recvCounts[i], i, info.sendDispls[i], i, info.recvDispls[i]);
	138	+ }
	139	+ }
	140	+ return HCCL_SUCCESS;
	141	+}
	142	+
	143	+HcclResult CollAlltoAllVPipelineFor91093::CalcA2AvcSendRecvInfo(const OpParam& param, SendRecvInfo &info)
	144	+{
	145	+ const u32 userRankSize = topoAttr_.userRankSize;
	146	+ const u32 userRank = topoAttr_.userRank;
	147	+ info.sendCounts.resize(userRankSize);
	148	+ info.sendDispls.resize(userRankSize);
	149	+ info.recvCounts.resize(userRankSize);
	150	+ info.recvDispls.resize(userRankSize);
	151	+ u64 sdispl = 0, rdispl = 0;
	152	+
	153	+ u64* sendCountMatrix = static_cast<u64 *>(param.All2AllDataDes.sendCountMatrix);
	154	+ for(u32 i = 0; i < userRankSize; i++) {
	155	+ info.sendCounts[i] = (sendCountMatrix + userRank userRankSize + i);
	156	+ info.recvCounts[i] = (sendCountMatrix + userRank + userRankSize i);
	157	+ info.sendDispls[i] = sdispl;
	158	+ info.recvDispls[i] = rdispl;
	159	+ sdispl += (sendCountMatrix + userRank userRankSize + i);
	160	+ rdispl += (sendCountMatrix + userRank + userRankSize i);
	161	+ }
	162	+ if (UNLIKELY(HcclCheckLogLevel(DLOG_INFO))) {
	163	+ for(u32 i=0; i< userRankSize ;i++) {
	164	+ HCCL_DEBUG("CalcA2AvcSendRecvInfo sendCounts[%d]=%ld, recvCounts[%d]=%ld, sdispls[%d]=%ld, rdispls[%d]=%ld",
	165	+ i, info.sendCounts[i], i, info.recvCounts[i], i, info.sendDispls[i], i, info.recvDispls[i]);
	166	+ }
	167	+ }
	168	+ return HCCL_SUCCESS;
	169	+}
	170	+
	171	+HcclResult CollAlltoAllVPipelineFor91093::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
	172	+{
	173	+ HcclUs startut = TIME_NOW();
	174	+ HcclResult ret = HCCL_SUCCESS;
	175	+ tag_ = param.tag;
	176	+ algResResp_ = &algRes;
	177	+ AlltoAllVParam_ = param;
	178	+ ExecMem execMem;
	179	+ execMem.count = 0;
	180	+ execMem.inputPtr = param.inputPtr;
	181	+ execMem.outputPtr = param.outputPtr;
	182	+ execMem.inputMem = algRes.cclInputMem;
	183	+ execMem.outputMem = algRes.cclOutputMem;
	184	+ ret = KernelRun(param, execMem);
	185	+
	186	+ CHK_PRT_RET(ret != HCCL_SUCCESS,
	187	+ HCCL_ERROR("[CollAlltoAllVPipelineFor91093][Orchestrate]errNo[0x%016llx]executor run failed",
	188	+ HCCL_ERROR_CODE(ret)), ret);
	189	+
	190	+ HCCL_INFO("tag[%s], CollAlltoAllVPipelineFor91093 tempAlg orchestrate success, take time [%lld]us.",
	191	+ param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
	192	+ return HCCL_SUCCESS;
	193	+}
	194	+
	195	+HcclResult CollAlltoAllVPipelineFor91093::KernelRun(const OpParam &param, ExecMem &execMem)
	196	+{
	197	+ HCCL_CONFIG_INFO(HCCL_ALG, "[CollAlltoAllVPipelineFor91093][KernelRun] AllToAllV npu direct start.");
	198	+ // 获取通信域
	199	+ SubCommInfo level1CommInfo;
	200	+ SubCommInfo level2CommInfo;
	201	+ CHK_RET(CheckCommSize(COMM_COMBINE_L1, COMM_INDEX_0 + 1));
	202	+ CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
	203	+ level1CommInfo = GetSubCommInfo(COMM_COMBINE_L1, COMM_INDEX_0);
	204	+ level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
	205	+
	206	+ // 执行
	207	+ std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
	208	+ TemplateType::TEMPLATE_ALL_2_ALL_V_CONTINUOUS_PIPELINE, dispatcher_);
	209	+ CHK_SMART_PTR_NULL(tempAlg);
	210	+
	211	+ SendRecvInfo sendRecvInfo;
	212	+ CHK_RET(CalLocalSendRecvInfo(param, sendRecvInfo));
	213	+
	214	+ A2aPipelineMemory a2aPipelineMemory;
	215	+ a2aPipelineMemory.userInput = algResResp_->paramInputMem;
	216	+ a2aPipelineMemory.userOutput = algResResp_->paramOutputMem;
	217	+ a2aPipelineMemory.cclInBuffer = execMem.inputMem;
	218	+ a2aPipelineMemory.cclOutBuffer = execMem.outputMem;
	219	+
	220	+ HCCL_INFO("[CollAlltoAllVPipelineFor91093] Memory info[addr, size]: userInput[%p, %llu], userOutput[%p, %llu], "
	221	+ "cclInBuffer[%p, %llu] ,cclOutBuffer[%p, %llu].",
	222	+ a2aPipelineMemory.userInput.ptr(), a2aPipelineMemory.userInput.size(),
	223	+ a2aPipelineMemory.userOutput.ptr(), a2aPipelineMemory.userOutput.size(),
	224	+ a2aPipelineMemory.cclInBuffer.ptr(), a2aPipelineMemory.cclInBuffer.size(),
	225	+ a2aPipelineMemory.cclOutBuffer.ptr(), a2aPipelineMemory.cclOutBuffer.size()
	226	+ );
	227	+
	228	+#ifndef OPEN_HCCL_TEST
	229	+ std::vector<SendRecvInfo> sendRecvInfoList{sendRecvInfo};
	230	+#else
	231	+ // 适配算法检查器，传入全局的info
	232	+ std::vector<SendRecvInfo> sendRecvInfoList = allMeshAggregationSendRecvInfo_;
	233	+#endif
	234	+
	235	+ CHK_RET(tempAlg->Prepare(topoAttr_.userRank,
	236	+ a2aPipelineMemory,
	237	+ level1CommInfo,
	238	+ level2CommInfo,
	239	+ param.stream,
	240	+ algResResp_->slaveStreams,
	241	+ algResResp_->notifiesMain,
	242	+ algResResp_->notifiesAux,
	243	+ sendRecvInfoList,
	244	+ param.All2AllDataDes.sendType,
	245	+ workflowMode_));
	246	+
	247	+ CHK_RET(tempAlg->RunAsync());
	248	+
	249	+ HCCL_INFO("[CollAlltoAllVPipelineFor91093] executor run success.");
	250	+ return HCCL_SUCCESS;
	251	+}
	252	+
	253	+REGISTER_EXEC("RunAlltoAllVPipelineFor91093", AlltoAllVPipelineFor91093, CollAlltoAllVPipelineFor91093);
	254	+} // namespace hccl

Asrc/legacy/ascend910/algorithm/impl/coll_executor/coll_all_to_all/coll_all_to_all_v_pipeline_for_910_93_executor.h+38-0

@@ -0,0 +1,38 @@
	1	+/**
	2	+ * Copyright (c) 2025 Huawei Technologies Co., Ltd.
	3	+ * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
	4	+ * CANN Open Software License Agreement Version 2.0 (the "License").
	5	+ * Please refer to the License for details. You may not use this file except in compliance with the License.
	6	+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
	7	+ * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
	8	+ * See LICENSE in the root of the software repository for the full text of the License.
	9	+ */
	10	+
	11	+#ifndef COLL_ALLTOALLV_PIPELINE_FOR_910_93_EXECUTOR_H
	12	+#define COLL_ALLTOALLV_PIPELINE_FOR_910_93_EXECUTOR_H
	13	+#include "coll_all_to_all_executor.h"
	14	+namespace hccl {
	15	+class CollAlltoAllVPipelineFor91093 : public CollAlltoAllExecutor { // A3 背靠背机型算法
	16	+
	17	+public:
	18	+ CollAlltoAllVPipelineFor91093(const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher> &topoMatcher);
	19	+ ~CollAlltoAllVPipelineFor91093() override = default;
	20	+ HcclResult Orchestrate(OpParam& param, AlgResourceResponse& algRes) override;
	21	+
	22	+private:
	23	+ HcclResult CalcStreamNum(u32& streamNum) override;
	24	+ HcclResult CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType);
	25	+ HcclResult CalcLevel1CommInfo(TransportMemType inputType, TransportMemType outputType,
	26	+ std::vector<LevelNSubCommTransport>& opTransport) override;
	27	+ HcclResult CalcLevel2CommInfo(TransportMemType inputType, TransportMemType outputType,
	28	+ std::vector<LevelNSubCommTransport>& opTransport) override;
	29	+ HcclResult CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport) override;
	30	+ HcclResult KernelRun(const OpParam &param, ExecMem &execMem) override;
	31	+ HcclResult CalLocalSendRecvInfo(const OpParam &param, SendRecvInfo &info);
	32	+ HcclResult CalcA2ASendRecvInfo(const OpParam &param, SendRecvInfo &info);
	33	+ HcclResult CalcA2AvSendRecvInfo(const OpParam &param, SendRecvInfo &info);
	34	+ HcclResult CalcA2AvcSendRecvInfo(const OpParam& param, SendRecvInfo &info);
	35	+};
	36	+
	37	+} // namespace hccl
	38	+#endif

Msrc/legacy/ascend910/algorithm/impl/operator/alltoall_operator.cc+29-0

	@@ -265,6 +265,10 @@ HcclResult AlltoAllOperator::SelectAlgforAlltoAll(const OpParam& param, std::str
265	return HCCL_SUCCESS ;	265	return HCCL_SUCCESS ;
266	} else if (isCommon310P3DUO_) {	266	} else if (isCommon310P3DUO_) {
267	algName = "RunAlltoAllVFor310PExecutor";	267	algName = "RunAlltoAllVFor310PExecutor";
		268	+ } else if(IsSatisfyA2AContinuousPipelineFor91093Condition(param)) {
		269	+ algName = "RunAlltoAllVPipelineFor91093";
		270	+ HCCL_INFO("[SelectAlgforAlltoAll] AllToAll algName is [%s]", algName.c_str());
		271	+ return HCCL_SUCCESS;
268	} else if (IsSupportDirectFullmeshForAlltoallv(param, deviceType_, useSuperPodMode_, serverNum_,	272	} else if (IsSupportDirectFullmeshForAlltoallv(param, deviceType_, useSuperPodMode_, serverNum_,
269	isSingleMeshAggregation_, userRankSize_, cclBufferManager_.GetInCCLbufferSize()) \|\|	273	isSingleMeshAggregation_, userRankSize_, cclBufferManager_.GetInCCLbufferSize()) \|\|
270	(deviceType_ == DevType::DEV_TYPE_910_93 && param.aicpuUnfoldMode) \|\| deviceType_ == DevType::DEV_TYPE_310P3) {	274	(deviceType_ == DevType::DEV_TYPE_910_93 && param.aicpuUnfoldMode) \|\| deviceType_ == DevType::DEV_TYPE_310P3) {
	@@ -644,6 +648,31 @@ bool AlltoAllOperator::IsBufferSatisfyAlltoAllAivCondition(const OpParam& param)
644	return true;	648	return true;
645	}	649	}
646		650
		651	+bool AlltoAllOperator::IsSatisfyA2AContinuousPipelineFor91093Condition(const OpParam& param)
		652	+{
		653	+ constexpr u32 SERVERNUM = 2;
		654	+ constexpr u32 RANKSPERSERVER = 1;
		655	+ bool cclBigEnough = cclBufferManager_.GetInCCLbufferSize() >= ALLTOALL_PIPELINE_MIN_CCL_SIZE;
		656	+ bool multiServer = (serverNum_ == SERVERNUM) && (superPodNum_ == SERVERNUM);
		657	+ bool multiRankPerServer = meshAggregationRankSize_ > RANKSPERSERVER;
		658	+ bool isOpbse = GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
		659	+ bool isAlltoAll = param.opType == HcclCMDType::HCCL_CMD_ALLTOALLV
		660	+ \|\| param.opType == HcclCMDType::HCCL_CMD_ALLTOALL
		661	+ \|\| param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC;
		662	+ bool isDefaultAlgo = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING)
		663	+ \|\| (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD)
		664	+ \|\| (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR);
		665	+ bool isPipelineAlgo = (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_PIPELINE);
		666	+ bool satisfyAlgType = isPipelineAlgo \|\| isDefaultAlgo;
		667	+ bool res = (deviceType_ == DevType::DEV_TYPE_910_93) && multiServer && multiRankPerServer
		668	+ && !multiSuperPodDiffDeviceNumMode_ && isOpbse && isAlltoAll && cclBigEnough && satisfyAlgType;
		669	+
		670	+ HCCL_DEBUG("[AlltoAllOperator][IsSatisfyA2AContinuousPipelineFor91093Condition] isSatisfy[%d], serverNum[%d], superPodNum %u,"
		671	+ "meshAggregationRankSize_ %u, isOpbse %u, isAlltoAll(vc\|v) %u, multiSuperPodDiffDeviceNumMode_ %u, aicpuUnfoldMode[%u].",
		672	+ res, serverNum_, superPodNum_, meshAggregationRankSize_, isOpbse, isAlltoAll, multiSuperPodDiffDeviceNumMode_, param.aicpuUnfoldMode);
		673	+ return res;
		674	+}
		675	+
647	bool AlltoAllOperator::IsSatisfyAlltoallContinuousPipelineCondition(const OpParam& param)	676	bool AlltoAllOperator::IsSatisfyAlltoallContinuousPipelineCondition(const OpParam& param)
648	{	677	{
649	std::vector<HcclAlgoType> algoTypeArr = topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALLV);	678	std::vector<HcclAlgoType> algoTypeArr = topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALLV);

Msrc/legacy/ascend910/algorithm/pub_inc/alltoall_operator.h+1-0

	@@ -52,6 +52,7 @@ public:
52	bool IsSatisfyAlltoAllAivCondition(const OpParam& param);	52	bool IsSatisfyAlltoAllAivCondition(const OpParam& param);
53	bool IsSatisfy91093OffloadCondition();	53	bool IsSatisfy91093OffloadCondition();
54	bool IsSatisfyAlltoallContinuousPipelineCondition(const OpParam& param);	54	bool IsSatisfyAlltoallContinuousPipelineCondition(const OpParam& param);
		55	+ bool IsSatisfyA2AContinuousPipelineFor91093Condition(const OpParam& param);
55		56
56	private:	57	private:
57	bool IsSatisfyAlltoallPipelineCondition();	58	bool IsSatisfyAlltoallPipelineCondition();

Msrc/legacy/ascend910/framework/communicator/impl/hccl_communicator_host.cc+5-2

	@@ -4790,7 +4790,9 @@ namespace hccl
4790	AlltoAllOperator alltoAllOperator = dynamic_cast<AlltoAllOperator >(algOperator.get());	4790	AlltoAllOperator alltoAllOperator = dynamic_cast<AlltoAllOperator >(algOperator.get());
4791	CHK_PTR_NULL(alltoAllOperator);	4791	CHK_PTR_NULL(alltoAllOperator);
4792		4792
4793	- if (alltoAllOperator->IsSatisfyAlltoallContinuousPipelineCondition(opParam)) {	4793	+ bool isSatisfyA2ACPForA3Condition = alltoAllOperator->IsSatisfyA2AContinuousPipelineFor91093Condition(opParam);
		4794	+ bool IsSatisfyA2ACPForA2Condition = alltoAllOperator->IsSatisfyAlltoallContinuousPipelineCondition(opParam);
		4795	+ if (IsSatisfyA2ACPForA2Condition \|\| isSatisfyA2ACPForA3Condition) {
4794	opParam.aicpuUnfoldMode = true;	4796	opParam.aicpuUnfoldMode = true;
4795	opParam.aicpuCacheEnable = GetExternalInputAicpuCacheEnable();	4797	opParam.aicpuCacheEnable = GetExternalInputAicpuCacheEnable();
4796	}	4798	}
	@@ -4981,7 +4983,8 @@ namespace hccl
4981	"RunAlltoAllDirectFullmesh",	4983	"RunAlltoAllDirectFullmesh",
4982	"RunAlltoAllVTwoLevelPipeline",	4984	"RunAlltoAllVTwoLevelPipeline",
4983	"RunAlltoAllFullMeshSymmetricMemory",	4985	"RunAlltoAllFullMeshSymmetricMemory",
4984	- "RunAlltoAllVContinuousPipeline"	4986	+ "RunAlltoAllVContinuousPipeline",
		4987	+ "RunAlltoAllVPipelineFor91093"
4985	};	4988	};
4986	return aicpuAlgs.count(algName) > 0;	4989	return aicpuAlgs.count(algName) > 0;
4987	};	4990	};

Mtest/st/algorithm/testcase/testcase/testcase_all_to_all.cc+29-0

	@@ -741,4 +741,33 @@ TEST_F(AllToAllTest, alltoall_test_910_93_opbase_RunAlltoAllFullMeshSymmetricMem
741	Checker checker;	741	Checker checker;
742	HcclResult ret;	742	HcclResult ret;
743	ret = checker.Check(checkerOpParam, topoMeta);	743	ret = checker.Check(checkerOpParam, topoMeta);
		744	+}
		745	+
		746	+TEST_F(AllToAllTest, RunAlltoAllVPipelineFor91093)
		747	+{
		748	+ RankTable_For_LLT gen;
		749	+ TopoMeta topoMeta;
		750	+ gen.GenTopoMeta(topoMeta, 2, 1, 8);
		751	+ setenv("HCCL_BUFFSIZE", "1", 1);
		752	+
		753	+ CheckerOpParam checkerOpParam;
		754	+ checkerOpParam.opType = CheckerOpType::ALLTOALL;
		755	+ checkerOpParam.tag = "AllToAll";
		756	+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
		757	+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
		758	+
		759	+ checkerOpParam.All2AllDataDes.sendType = CheckerDataType::DATA_TYPE_INT8;
		760	+ checkerOpParam.All2AllDataDes.recvType = CheckerDataType::DATA_TYPE_INT8;
		761	+
		762	+ // 生成sendCountMatrix矩阵，alltoall的底层实现走alltoallvc
		763	+ u32 rankNum = GetRankNumFormTopoMeta(topoMeta);
		764	+ checkerOpParam.All2AllDataDes.sendCountMatrix = GenerateSendCountMatrix(100, rankNum);
		765	+ checkerOpParam.All2AllDataDes.sendType = CheckerDataType::DATA_TYPE_INT8;
		766	+ checkerOpParam.All2AllDataDes.recvType = CheckerDataType::DATA_TYPE_INT8;
		767	+ checkerOpParam.All2AllDataDes.sendCount = 100;
		768	+
		769	+ Checker checker;
		770	+ HcclResult ret;
		771	+ ret = checker.Check(checkerOpParam, topoMeta);
		772	+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
744	}	773	}

Mtest/st/algorithm/testcase/testcase/testcase_all_to_allv.cc+27-0

	@@ -848,5 +848,32 @@ TEST_F(AllToAllVTest, alltoallv_test_910B_opbase_RunAlltoAllVContinuousPipeline_
848	HcclResult ret;	848	HcclResult ret;
849	ret = checker.Check(checkerOpParam, topoMeta);	849	ret = checker.Check(checkerOpParam, topoMeta);
850		850
		851	+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
		852	+}
		853	+
		854	+TEST_F(AllToAllVTest, RunAlltoAllVPipelineFor91093)
		855	+{
		856	+ RankTable_For_LLT gen;
		857	+ TopoMeta topoMeta;
		858	+ gen.GenTopoMeta(topoMeta, 2, 1, 16);
		859	+ setenv("HCCL_BUFFSIZE", "1", 1);
		860	+
		861	+ CheckerOpParam checkerOpParam;
		862	+ checkerOpParam.opType = CheckerOpType::ALLTOALLV;
		863	+ checkerOpParam.tag = "AllToAll";
		864	+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
		865	+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
		866	+
		867	+ checkerOpParam.All2AllDataDes.sendType = CheckerDataType::DATA_TYPE_INT8;
		868	+ checkerOpParam.All2AllDataDes.recvType = CheckerDataType::DATA_TYPE_INT8;
		869	+
		870	+ u32 rankNum = GetRankNumFormTopoMeta(topoMeta);
		871	+ GenAllToAllVParams(rankNum, 10, checkerOpParam.All2AllDataDes.sendCounts,
		872	+ checkerOpParam.All2AllDataDes.sdispls, checkerOpParam.All2AllDataDes.recvCounts,
		873	+ checkerOpParam.All2AllDataDes.rdispls);
		874	+
		875	+ Checker checker;
		876	+ HcclResult ret;
		877	+ ret = checker.Check(checkerOpParam, topoMeta);
851	EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);	878	EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
852	}	879	}

Mtest/st/algorithm/testcase/testcase/testcase_all_to_allvc.cc+24-0

	@@ -428,3 +428,27 @@ TEST_F(AllToAllVCTest, alltoallvc_test_910_93_opbase_2superpod_RunAlltoAllDirect
428		428
429	EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);	429	EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
430	}	430	}
		431	+
		432	+TEST_F(AllToAllVCTest, RunAlltoAllVPipelineFor91093)
		433	+{
		434	+ RankTable_For_LLT gen;
		435	+ TopoMeta topoMeta;
		436	+ gen.GenTopoMeta(topoMeta, 2, 1, 7);
		437	+ setenv("HCCL_BUFFSIZE", "1", 1);
		438	+
		439	+ CheckerOpParam checkerOpParam;
		440	+ checkerOpParam.opType = CheckerOpType::ALLTOALLVC;
		441	+ checkerOpParam.tag = "AllToAll";
		442	+ checkerOpParam.opMode = CheckerOpMode::OPBASE;
		443	+ checkerOpParam.devtype = CheckerDevType::DEV_TYPE_910_93;
		444	+
		445	+ u32 rankNum = GetRankNumFormTopoMeta(topoMeta);
		446	+ checkerOpParam.All2AllDataDes.sendCountMatrix = GenerateSendCountMatrix(100, rankNum);
		447	+ checkerOpParam.All2AllDataDes.sendType = CheckerDataType::DATA_TYPE_INT8;
		448	+ checkerOpParam.All2AllDataDes.recvType = CheckerDataType::DATA_TYPE_INT8;
		449	+
		450	+ Checker checker;
		451	+ HcclResult ret;
		452	+ ret = checker.Check(checkerOpParam, topoMeta);
		453	+ EXPECT_EQ(ret, HcclResult::HCCL_SUCCESS);
		454	+}