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_operator.h"
12 : #include <string>
13 : #include <unordered_map>
14 : #include <algorithm>
15 : #include <functional>
16 : #include "op_type.h"
17 : #include "string_util.h"
18 : #include "binary_stream.h"
19 : namespace Hccl {
20 : constexpr u32 MAX_OP_TAG_LEN = 191; // 最大的tag 长度
21 : constexpr u32 MAX_HANDSHAKEMSGPACK_LEN = 1024; // 最大握手消息长度
22 :
23 42 : std::string MemBufferDesc(const BaseCollOperator &collOp)
24 : {
25 42 : std::string memDesc = "";
26 84 : memDesc += "inputMem=" + (collOp.inputMem ? collOp.inputMem->Describe() : "nullptr") + ", ";
27 84 : memDesc += "outputMem=" + (collOp.outputMem ? collOp.outputMem->Describe() : "nullptr") + ", ";
28 84 : memDesc += "scratchMem=" + (collOp.scratchMem ? collOp.scratchMem->Describe() : "nullptr");
29 42 : return memDesc;
30 0 : }
31 :
32 36 : std::string OpDesc(const BaseCollOperator &collOp)
33 : {
34 36 : std::string opDesc = "";
35 36 : opDesc += "opType=" + collOp.opType.Describe() + ", ";
36 36 : opDesc += "opMode=" + collOp.opMode.Describe() + ", ";
37 36 : opDesc += "dataType=" + collOp.dataType.Describe() + ", ";
38 36 : opDesc += "sendRecvRemoteRank=" + std::to_string(collOp.sendRecvRemoteRank) + ", ";
39 36 : opDesc += "Buffers=[" + MemBufferDesc(collOp) + "]";
40 36 : return opDesc;
41 0 : }
42 :
43 2 : std::string DescReduceScatter(const BaseCollOperator &collOp)
44 : {
45 : return StringFormat(
46 : "BaseCollOperator[%s, reduceOp=%s, dataCount=%llu]",
47 6 : OpDesc(collOp).c_str(), collOp.reduceOp.Describe().c_str(), collOp.dataCount
48 6 : );
49 : }
50 :
51 24 : std::string DescAllreduce(const BaseCollOperator &collOp)
52 : {
53 : return StringFormat(
54 : "BaseCollOperator[%s, reduceOp=%s, dataCount=%llu]",
55 72 : OpDesc(collOp).c_str(), collOp.reduceOp.Describe().c_str(), collOp.dataCount
56 72 : );
57 : }
58 :
59 2 : std::string DescAllgather(const BaseCollOperator &collOp)
60 : {
61 : return StringFormat(
62 : "BaseCollOperator[%s, dataCount=%llu]",
63 4 : OpDesc(collOp).c_str(), collOp.dataCount
64 4 : );
65 : }
66 :
67 2 : std::string DescScatter(const BaseCollOperator &collOp)
68 : {
69 : return StringFormat(
70 : "BaseCollOperator[%s, dataCount=%llu, root=%u]",
71 4 : OpDesc(collOp).c_str(), collOp.dataCount, collOp.root
72 4 : );
73 : }
74 :
75 2 : std::string DescAlltoall(const BaseCollOperator &collOp)
76 : {
77 : return StringFormat(
78 : "BaseCollOperator[opType=%s, opMode=%s, sendCount=%llu, recvCount=%llu, sendType=%s, recvType=%s, "
79 : "sendRecvRemoteRank=%u, Buffers=[%s]]",
80 4 : collOp.opType.Describe().c_str(), collOp.opMode.Describe().c_str(),
81 2 : collOp.all2AllDataDes.sendCount, collOp.all2AllDataDes.recvCount,
82 4 : collOp.all2AllDataDes.sendType.Describe().c_str(),
83 4 : collOp.all2AllDataDes.recvType.Describe().c_str(), collOp.sendRecvRemoteRank,
84 4 : MemBufferDesc(collOp).c_str()
85 12 : );
86 : }
87 :
88 2 : std::string DescAlltoallV(const BaseCollOperator &collOp)
89 : {
90 : return StringFormat(
91 : "BaseCollOperator[opType=%s, opMode=%s, sendType=%s, recvType=%s, sendRecvRemoteRank=%u, Buffers=[%s]]",
92 6 : collOp.opType.Describe().c_str(), collOp.opMode.Describe().c_str(),
93 4 : collOp.all2AllVDataDes.sendType.Describe().c_str(),
94 4 : collOp.all2AllVDataDes.recvType.Describe().c_str(), collOp.sendRecvRemoteRank,
95 4 : MemBufferDesc(collOp).c_str()
96 12 : );
97 : }
98 :
99 2 : std::string DescAlltoallVC(const BaseCollOperator &collOp)
100 : {
101 : return StringFormat(
102 : "BaseCollOperator[opType=%s, opMode=%s, sendType=%s, recvType=%s, sendRecvRemoteRank=%u, Buffers=[%s]]",
103 6 : collOp.opType.Describe().c_str(), collOp.opMode.Describe().c_str(),
104 4 : collOp.all2AllVCDataDes.sendType.Describe().c_str(),
105 4 : collOp.all2AllVCDataDes.recvType.Describe().c_str(), collOp.sendRecvRemoteRank,
106 4 : MemBufferDesc(collOp).c_str()
107 12 : );
108 : }
109 :
110 2 : std::string DescSend(const BaseCollOperator &collOp)
111 : {
112 : return StringFormat(
113 4 : "BaseCollOperator[%s]", OpDesc(collOp).c_str()
114 4 : );
115 : }
116 :
117 2 : std::string DescRecv(const BaseCollOperator &collOp)
118 : {
119 : return StringFormat(
120 4 : "BaseCollOperator[%s]", OpDesc(collOp).c_str()
121 4 : );
122 : }
123 :
124 2 : std::string DescReduce(const BaseCollOperator &collOp)
125 : {
126 : return StringFormat(
127 : "BaseCollOperator[%s, reduceOp=%s, dataCount=%llu, root=%u]",
128 6 : OpDesc(collOp).c_str(), collOp.reduceOp.Describe().c_str(), collOp.dataCount, collOp.root
129 6 : );
130 : }
131 :
132 0 : std::string DescBroadcast(const BaseCollOperator &collOp)
133 : {
134 : return StringFormat(
135 : "BaseCollOperator[%s, dataCount=%llu, root=%u]",
136 0 : OpDesc(collOp).c_str(), collOp.dataCount, collOp.root
137 0 : );
138 : }
139 :
140 0 : std::string DescBatchSendRecv(const BaseCollOperator &collOp)
141 : {
142 : return StringFormat(
143 : "BaseCollOperator[%s, dataCount=%llu, root=%u]",
144 0 : OpDesc(collOp).c_str(), collOp.dataCount, collOp.root
145 0 : );
146 : }
147 :
148 0 : std::string DescHalfAlltoAllV(const BaseCollOperator &collOp)
149 : {
150 : return StringFormat(
151 0 : "BaseCollOperator[%s]", OpDesc(collOp).c_str()
152 0 : );
153 : }
154 :
155 0 : std::string DescReduceScatterV(const BaseCollOperator &collOp)
156 : {
157 : return StringFormat(
158 0 : "BaseCollOperator[%s]", OpDesc(collOp).c_str()
159 0 : );
160 : }
161 :
162 0 : std::string DescAllGatherV(const BaseCollOperator &collOp)
163 : {
164 : return StringFormat(
165 0 : "BaseCollOperator[%s]", OpDesc(collOp).c_str()
166 0 : );
167 : }
168 :
169 : std::unordered_map<OpType, std::function<std::string(const BaseCollOperator &)>, std::EnumClassHash> descOpMap{
170 : {OpType::REDUCESCATTER, std::bind(&DescReduceScatter, std::placeholders::_1)},
171 : {OpType::ALLREDUCE, std::bind(&DescAllreduce, std::placeholders::_1)},
172 : {OpType::ALLGATHER, std::bind(&DescAllgather, std::placeholders::_1)},
173 : {OpType::SCATTER, std::bind(&DescScatter, std::placeholders::_1)},
174 : {OpType::ALLTOALL, std::bind(&DescAlltoall, std::placeholders::_1)},
175 : {OpType::ALLTOALLV, std::bind(&DescAlltoallV, std::placeholders::_1)},
176 : {OpType::ALLTOALLVC, std::bind(&DescAlltoallVC, std::placeholders::_1)},
177 : {OpType::SEND, std::bind(&DescSend, std::placeholders::_1)},
178 : {OpType::RECV, std::bind(&DescRecv, std::placeholders::_1)},
179 : {OpType::REDUCE, std::bind(&DescReduce, std::placeholders::_1)},
180 : {OpType::BROADCAST, std::bind(&DescBroadcast, std::placeholders::_1)},
181 : {OpType::BATCHSENDRECV, std::bind(&DescBatchSendRecv, std::placeholders::_1)},
182 : {OpType::HALFALLTOALLV, std::bind(&DescHalfAlltoAllV, std::placeholders::_1)},
183 : {OpType::REDUCESCATTERV, std::bind(&DescReduceScatterV, std::placeholders::_1)},
184 : {OpType::ALLGATHERV, std::bind(&DescAllGatherV, std::placeholders::_1)},
185 : };
186 :
187 43 : std::string CollOpToString(const BaseCollOperator &collOp)
188 : {
189 43 : auto it = descOpMap.find(collOp.opType);
190 43 : if (it != descOpMap.end()) {
191 42 : return it->second.operator()(collOp);
192 : } else {
193 2 : return "unknown";
194 : }
195 : }
196 :
197 26 : inline std::vector<char> DumpByteVector(BinaryStream &binaryStream)
198 : {
199 26 : std::vector<char> byteVector;
200 26 : binaryStream.Dump(byteVector);
201 :
202 26 : auto remainLen = MAX_HANDSHAKEMSGPACK_LEN - byteVector.size();
203 26 : byteVector.insert(byteVector.end(), remainLen, '\0');
204 :
205 26 : return byteVector;
206 0 : }
207 :
208 26 : std::vector<char> opTagToVector(const std::string &opTag)
209 : {
210 26 : std::vector<char> result(MAX_OP_TAG_LEN, '\0');
211 26 : auto copyLen = opTag.size() < MAX_OP_TAG_LEN ? opTag.size() :MAX_OP_TAG_LEN;
212 26 : std::copy_n(opTag.begin(), copyLen, result.begin());
213 :
214 26 : return result;
215 0 : }
216 :
217 7 : std::string vectorToOpTag(const std::vector<char> &opTagvector)
218 : {
219 7 : auto validSize = opTagvector.size() < MAX_OP_TAG_LEN ? opTagvector.size() : MAX_OP_TAG_LEN;
220 7 : auto firstNul = std::find(opTagvector.begin(), opTagvector.begin() + validSize, '\0');
221 :
222 14 : return std::string(opTagvector.begin(), firstNul);
223 : }
224 :
225 26 : std::vector<char> CollOperator::GetUniqueId() const
226 : {
227 78 : HCCL_INFO("[CollOperator::%s] opMode[%s], opType[%s], reduceOp[%s], dataType[%s], dataCount[%llu], root[%u], "
228 : "myRank[%d], sendRecvRemoteRank[%d], opTag[%s], staticAddr[%d], staticShape[%d], outputDataType[%s], ",
229 : __func__, opMode.Describe().c_str(), opType.Describe().c_str(), reduceOp.Describe().c_str(), dataType.Describe().c_str(),
230 : dataCount, root, myRank, sendRecvRemoteRank, opTag.c_str(), staticAddr, staticShape, outputDataType.Describe().c_str());
231 26 : BinaryStream binaryStream;
232 26 : binaryStream << opMode;
233 26 : binaryStream << opType;
234 26 : binaryStream << reduceOp;
235 26 : binaryStream << dataType;
236 26 : binaryStream << dataCount;
237 26 : binaryStream << root;
238 26 : binaryStream << myRank;
239 26 : binaryStream << sendRecvRemoteRank;
240 26 : binaryStream << opTagToVector(opTag);
241 26 : binaryStream << staticAddr;
242 26 : binaryStream << staticShape;
243 26 : binaryStream << outputDataType;
244 :
245 26 : if (opType == OpType::BATCHSENDRECV) {
246 0 : return DumpByteVector(binaryStream);
247 : }
248 :
249 26 : if (opType == OpType::ALLTOALL) {
250 2 : binaryStream << all2AllDataDes.sendType;
251 2 : binaryStream << all2AllDataDes.recvType;
252 2 : binaryStream << all2AllDataDes.sendCount;
253 2 : binaryStream << all2AllDataDes.recvCount;
254 2 : return DumpByteVector(binaryStream);
255 : }
256 :
257 24 : if (opType == OpType::ALLTOALLV) {
258 2 : binaryStream << all2AllVDataDes.sendType;
259 2 : binaryStream << all2AllVDataDes.recvType;
260 2 : return DumpByteVector(binaryStream);
261 : }
262 :
263 22 : if (opType == OpType::ALLTOALLVC) {
264 1 : binaryStream << all2AllVCDataDes.sendType;
265 1 : binaryStream << all2AllVCDataDes.recvType;
266 1 : return DumpByteVector(binaryStream);
267 : }
268 :
269 21 : if (opType == OpType::ALLGATHERV || opType == OpType::REDUCESCATTERV) {
270 2 : binaryStream << vDataDes.dataType;
271 2 : return DumpByteVector(binaryStream);
272 : }
273 :
274 19 : binaryStream << dataDes.dataCount;
275 19 : binaryStream << dataDes.dataType;
276 19 : binaryStream << dataDes.strideCount;
277 :
278 19 : return DumpByteVector(binaryStream);
279 26 : }
280 :
281 7 : CollOperatorDef CollOperator::GetPackedData(std::vector<char> &byteVector)
282 : {
283 7 : CollOperator op;
284 7 : BinaryStream binaryStream(byteVector);
285 7 : std::vector<char> vectorOpTag;
286 7 : binaryStream >> op.opMode;
287 7 : binaryStream >> op.opType;
288 7 : binaryStream >> op.reduceOp;
289 7 : binaryStream >> op.dataType;
290 7 : binaryStream >> op.dataCount;
291 7 : binaryStream >> op.root;
292 7 : binaryStream >> op.myRank;
293 7 : binaryStream >> op.sendRecvRemoteRank;
294 7 : binaryStream >> vectorOpTag;
295 7 : binaryStream >> op.staticAddr;
296 7 : binaryStream >> op.staticShape;
297 7 : binaryStream >> op.outputDataType;
298 :
299 7 : op.opTag = vectorToOpTag(vectorOpTag);
300 :
301 7 : if (op.opType == OpType::BATCHSENDRECV) {
302 0 : return op;
303 : }
304 :
305 7 : if (op.opType == OpType::ALLTOALL) {
306 1 : binaryStream >> op.all2AllDataDes.sendType;
307 1 : binaryStream >> op.all2AllDataDes.recvType;
308 1 : binaryStream >> op.all2AllDataDes.sendCount;
309 1 : binaryStream >> op.all2AllDataDes.recvCount;
310 1 : return op;
311 : }
312 :
313 6 : if (op.opType == OpType::ALLTOALLV) {
314 1 : binaryStream >> op.all2AllVDataDes.sendType;
315 1 : binaryStream >> op.all2AllVDataDes.recvType;
316 1 : return op;
317 : }
318 :
319 5 : if (op.opType == OpType::ALLTOALLVC) {
320 1 : binaryStream >> op.all2AllVCDataDes.sendType;
321 1 : binaryStream >> op.all2AllVCDataDes.recvType;
322 1 : return op;
323 : }
324 :
325 4 : if (op.opType == OpType::ALLGATHERV || op.opType == OpType::REDUCESCATTERV) {
326 1 : binaryStream >> op.vDataDes.dataType;
327 1 : return op;
328 : }
329 :
330 3 : binaryStream >> op.dataDes.dataCount;
331 3 : binaryStream >> op.dataDes.dataType;
332 3 : binaryStream >> op.dataDes.strideCount;
333 :
334 3 : return op;
335 7 : }
336 : }
|
