Line data Source code
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 : #include "coll_all_gather_ring_zerocopy_exchange_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherRingZerocopyExchangeExecutor::CollAllGatherRingZerocopyExchangeExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollAllGatherRingZerocopyExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = true; // 设为true，以禁用RunLoop中的本地拷贝
19 0 : desc_.isZeroCopy = true;
20 0 : }
21 :
22 0 : HcclResult CollAllGatherRingZerocopyExchangeExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
23 : {
24 : // 调用父类编排函数建链关系计算函数
25 0 : CHK_RET(CollAllGatherRingZerocopyExecutor::CalcCommInfo(opTransport));
26 : // 额外增加数据交换的建链
27 0 : CHK_RET(CalcExchangeCommInfo(opTransport));
28 0 : return HCCL_SUCCESS;
29 : }
30 :
31 0 : HcclResult CollAllGatherRingZerocopyExchangeExecutor::CalExchangeRemoteRank(u32& remoteRankSend, u32& remoteRankRecv)
32 : {
33 : // AllGather的发端与收端与ReduceScatter相反
34 0 : return CalExchangeRemoteRankForReduceScatter(remoteRankRecv, remoteRankSend);
35 : }
36 :
37 : HcclResult
38 0 : CollAllGatherRingZerocopyExchangeExecutor::CalcExchangeCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
39 : {
40 0 : std::set<u32> commTargetUserRankSet;
41 0 : u32 remoteRankSend = 0;
42 0 : u32 remoteRankRecv = 0;
43 :
44 0 : CHK_RET(CalExchangeRemoteRank(remoteRankSend, remoteRankRecv));
45 0 : HCCL_DEBUG("[%s] remoteRankSend:%d, remoteRankRecv:%d", __func__, remoteRankSend, remoteRankRecv);
46 0 : commTargetUserRankSet.insert(remoteRankSend);
47 0 : commTargetUserRankSet.insert(remoteRankRecv);
48 : CommParaInfo commParaInfo(
49 : COMM_COMBINE_ORDER, CommType::COMM_TAG_PARTIAL_MESH_COMBINED, INVALID_VALUE_RANKID, INVALID_VALUE_RANKID, false,
50 0 : false, commTargetUserRankSet);
51 :
52 0 : TransportMemType inputType = TransportMemType::CCL_INPUT;
53 0 : TransportMemType outputType = TransportMemType::CCL_OUTPUT;
54 :
55 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_COMBINE_ORDER], inputType, outputType));
56 0 : LevelNSubCommTransport& subCommTransport = opTransport[COMM_COMBINE_ORDER];
57 0 : for (u32 subCommIndex = 0; subCommIndex < subCommTransport.size(); subCommIndex++) {
58 0 : for (auto& transportRequest : subCommTransport[subCommIndex].transportRequests) {
59 : transportRequest.isUsedRdma
60 0 : = (topoAttr_.superPodNum > 1
61 0 : || (static_cast<bool>(topoMatcher_->GetExternalInputInterHccsDisable()) && topoAttr_.serverNum > 1));
62 : }
63 : }
64 0 : return HCCL_SUCCESS;
65 0 : }
66 :
67 : HcclResult
68 0 : CollAllGatherRingZerocopyExchangeExecutor::KernelRunInterServerPreProcess(const OpParam& param, const ExecMem& execMem)
69 : {
70 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[AllGatherRingZerocopyExchangeExecutor] KernelRunInterServerPreProcess");
71 : // 计算需要交换数据的通信对端
72 0 : u32 remoteRankSend = 0;
73 0 : u32 remoteRankRecv = 0;
74 0 : CHK_RET(CalExchangeRemoteRank(remoteRankSend, remoteRankRecv));
75 :
76 0 : Stream stream = param.stream;
77 0 : u64 inputMemSize = execMem.inputMem.size();
78 0 : if (remoteRankSend != topoAttr_.userRank && remoteRankRecv != topoAttr_.userRank) { // 需要交换数据
79 : // 获取通信对端的link
80 0 : LINK sendLink;
81 0 : LINK recvLink;
82 0 : CHK_RET(GetTransportForExchange(remoteRankSend, sendLink));
83 0 : CHK_RET(GetTransportForExchange(remoteRankRecv, recvLink));
84 : // 当通信对端恰好是同server的邻居时，复用Level0的建链，其注册的内存是UserMem，需要特殊处理
85 : // 否则，在CommCombineOrder上建链，其注册内存是CCL Buffer
86 0 : if (!IsLevel0Neighbor(remoteRankSend, level0RankSize_)) {
87 : // user in mem -> ccl in mem
88 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8*>(execMem.inputPtr), inputMemSize);
89 0 : DeviceMem dstMem = execMem.inputMem.range(0, inputMemSize);
90 0 : HCCL_DEBUG(
91 : "[%s] not neighbor, srcPtr:%p, dstPtr:%p, size:%llu", __func__, srcMem.ptr(), dstMem.ptr(),
92 : inputMemSize);
93 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
94 0 : }
95 : // 执行通信
96 0 : recvLink->TxAck(stream);
97 0 : sendLink->RxAck(stream);
98 0 : u32 remoteLevel1Index = remoteRankSend % (level0RankSize_ * level1RankSize_) / level0RankSize_;
99 0 : u32 remoteLevel2Index = remoteRankSend / level0RankSize_ / level1RankSize_;
100 0 : u64 txDstOffset = (remoteLevel1Index * level2RankSize_ + remoteLevel2Index) * inputMemSize;
101 0 : HCCL_DEBUG(
102 : "[%s] remoteLevel1Index:%d, remoteLevel2Index:%d, txDstOffset:%llu", __func__, remoteLevel1Index,
103 : remoteLevel2Index, txDstOffset);
104 0 : if (IsLevel0Neighbor(remoteRankSend, level0RankSize_)) {
105 0 : sendLink->TxAsync(UserMemType::OUTPUT_MEM, txDstOffset, execMem.inputPtr, inputMemSize, stream);
106 0 : HCCL_DEBUG("[%s] neighbor, send data to userMem", __func__);
107 : } else {
108 0 : sendLink->TxAsync(UserMemType::OUTPUT_MEM, txDstOffset, execMem.inputMem.ptr(), inputMemSize, stream);
109 0 : HCCL_DEBUG("[%s] not neighbor, send data to ccl buffer", __func__);
110 : }
111 0 : u64 rxDstOffset = (level1Rank_ * level2RankSize_ + level2Rank_) * inputMemSize;
112 0 : u64 rxSrcOffset = IsLevel0Neighbor(remoteRankRecv, level0RankSize_) ?
113 0 : static_cast<u8*>(execMem.inputPtr) - static_cast<u8*>(param.inputPtr) :
114 0 : 0;
115 0 : HCCL_DEBUG("[%s] rxDstOffset:%llu, rxSrcOffset:%llu", __func__, rxDstOffset, rxSrcOffset);
116 0 : recvLink->RxAsync(
117 0 : UserMemType::INPUT_MEM, rxSrcOffset, static_cast<u8*>(execMem.outputMem.ptr()) + rxDstOffset, inputMemSize,
118 : stream);
119 : // 交换数据的两端之间Barrier，确认收发完成
120 0 : CHK_RET(recvLink->TxAck(stream));
121 0 : CHK_RET(sendLink->RxAck(stream));
122 0 : CHK_RET(sendLink->TxDataSignal(stream));
123 0 : CHK_RET(recvLink->RxDataSignal(stream));
124 0 : } else { // 不需要交换数据，将数据从user in拷到ccl out
125 0 : u64 dstMemOffset = (level1Rank_ * level2RankSize_ + level2Rank_) * inputMemSize;
126 0 : DeviceMem dstMem = execMem.outputMem.range(dstMemOffset, inputMemSize);
127 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8*>(execMem.inputPtr), inputMemSize);
128 0 : HCCL_DEBUG(
129 : "[%s] not exchange, just copy data from CCLOut[%p] to UserInput[%p]", __func__, dstMem.ptr(), srcMem.ptr());
130 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
131 0 : }
132 :
133 0 : return HCCL_SUCCESS;
134 0 : }
135 :
136 : HcclResult
137 0 : CollAllGatherRingZerocopyExchangeExecutor::KernelRunInterServerPostProcess(const OpParam& param, const ExecMem& execMem)
138 : {
139 : // 将通信结果从ccl output搬到user output
140 0 : if (level1RankSize_ > 1 || level2RankSize_ > 1) {
141 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
142 0 : u64 curSize = execMem.inputMem.size();
143 0 : Stream stream = param.stream;
144 0 : for (u32 i = 0; i < level1RankSize_ * level2RankSize_; i++) {
145 : DeviceMem dstMem = DeviceMem::create(
146 0 : static_cast<u8*>(execMem.outputPtr)
147 0 : + param.DataDes.count * unitSize * (level0Rank_ * level1RankSize_ * level2RankSize_ + i),
148 0 : curSize);
149 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8*>(execMem.outputMem.ptr()) + i * curSize, curSize);
150 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream));
151 0 : HCCL_DEBUG("[%s] memcopy from CCLOut[%p] to UserOut[%p]", __func__, srcMem.ptr(), dstMem.ptr());
152 0 : }
153 0 : }
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 0 : HcclResult CollAllGatherRingZerocopyExchangeExecutor::CalcLevel0DataSlices(
158 : const OpParam& param, const ExecMem& execMem, std::vector<Slice>& dataSegsSlice)
159 : {
160 0 : return CalcIntraServerDataSlicesContinuous(
161 0 : param, execMem, level0RankSize_, level1RankSize_, level2RankSize_, dataSegsSlice);
162 : }
163 :
164 : REGISTER_EXEC(
165 : "AllGatherRingZerocopyExchangeExecutor", AllGatherRingZerocopyExchange, CollAllGatherRingZerocopyExchangeExecutor);
166 :
167 : } // namespace hccl
|
