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 "cpu_ts_thread.h"
12 : #include "hccl_common.h"
13 : #include "adapter_rts.h"
14 :
15 : namespace hccl {
16 : const std::unordered_map<HcclDataType, aclDataType> hccl2rtDataTypeMap = {
17 : {HCCL_DATA_TYPE_INT8, ACL_INT8},
18 : {HCCL_DATA_TYPE_INT16, ACL_INT16},
19 : {HCCL_DATA_TYPE_INT32, ACL_INT32},
20 : {HCCL_DATA_TYPE_FP16, ACL_FLOAT16},
21 : {HCCL_DATA_TYPE_FP32, ACL_FLOAT},
22 : {HCCL_DATA_TYPE_BFP16, ACL_BF16},
23 : };
24 :
25 :
26 : const std::unordered_map<HcclReduceOp, aclrtReduceKind> hccl2rtReduceOpMap = {
27 : {HCCL_REDUCE_SUM, ACL_RT_MEMCPY_SDMA_AUTOMATIC_SUM},
28 : {HCCL_REDUCE_MAX, ACL_RT_MEMCPY_SDMA_AUTOMATIC_MAX},
29 : {HCCL_REDUCE_MIN, ACL_RT_MEMCPY_SDMA_AUTOMATIC_MIN},
30 : };
31 :
32 8 : CpuTsThread::CpuTsThread(rtStream_t rtStream, uint32_t notifyNum, const NotifyLoadType notifyLoadType)
33 8 : : rtStream_(rtStream), notifyNum_(notifyNum), notifyLoadType_(notifyLoadType)
34 8 : {}
35 :
36 85 : CpuTsThread::CpuTsThread(StreamType streamType, uint32_t notifyNum, const NotifyLoadType notifyLoadType)
37 85 : : streamType_(streamType), notifyNum_(notifyNum), notifyLoadType_(notifyLoadType)
38 85 : {}
39 :
40 93 : CpuTsThread::~CpuTsThread()
41 : {
42 93 : DeInit();
43 93 : }
44 :
45 73 : HcclResult CpuTsThread::Init()
46 : {
47 : // Host 侧初始化
48 73 : CHK_RET(GetRunSideIsDevice(isDeviceSide_));
49 72 : CHK_RET(hrtGetDeviceType(devType_));
50 72 : if (!isDeviceSide_) {
51 : s32 deviceLogicId;
52 71 : CHK_RET(hrtGetDevice(&deviceLogicId));
53 71 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devId_));
54 71 : if (streamType_ == StreamType::STREAM_TYPE_DEVICE || notifyLoadType_ == NotifyLoadType::DEVICE_NOTIFY) {
55 3 : return HCCL_E_NOT_SUPPORT;
56 : }
57 68 : if (rtStream_ == nullptr) {
58 62 : stream_.reset(new (std::nothrow) Stream(streamType_));
59 62 : CHK_SMART_PTR_NULL(stream_);
60 62 : rtStream_ = stream_->ptr();
61 : } else {
62 6 : stream_.reset(new (std::nothrow) Stream(rtStream_));
63 6 : CHK_SMART_PTR_NULL(stream_);
64 : }
65 68 : notifys_.reserve(notifyNum_);
66 214 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
67 148 : notifys_.emplace_back(nullptr);
68 148 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
69 148 : CHK_SMART_PTR_NULL(notifys_[idx]);
70 148 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
71 146 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
72 44 : CHK_RET(notifys_[idx]->SetIpc());
73 : }
74 : }
75 66 : return HCCL_SUCCESS;
76 : } else {
77 1 : return HCCL_E_NOT_SUPPORT;
78 : }
79 : }
80 :
81 93 : HcclResult CpuTsThread::DeInit()
82 : {
83 93 : streamType_ = StreamType::STREAM_TYPE_RESERVED;
84 93 : notifyNum_ = 0;
85 93 : stream_ = nullptr;
86 93 : notifys_.clear();
87 93 : return HCCL_SUCCESS;
88 : }
89 :
90 2 : std::string &CpuTsThread::GetUniqueId()
91 : {
92 2 : if (!uniqueIdStr_.empty()) {
93 0 : return uniqueIdStr_;
94 : }
95 2 : return UpdateUniqueId();
96 : }
97 :
98 12 : std::string &CpuTsThread::UpdateUniqueId()
99 : {
100 : // 序列化信息
101 12 : uniqueIdStr_ = std::string();
102 12 : std::ostringstream oss;
103 12 : StreamType streamType = StreamType::STREAM_TYPE_DEVICE;
104 12 : oss.write(reinterpret_cast<const char_t *>(&streamType), sizeof(streamType));
105 12 : oss.write(reinterpret_cast<const char_t *>(¬ifyLoadType_), sizeof(notifyLoadType_));
106 12 : oss.write(reinterpret_cast<const char_t *>(&devId_), sizeof(devId_));
107 12 : oss.write(reinterpret_cast<const char_t *>(¬ifyNum_), sizeof(notifyNum_));
108 :
109 : // 临时申请一条流，用于在device侧资源展开时initStream
110 12 : if (streamDevice_ == nullptr) {
111 8 : streamDevice_.reset(new (std::nothrow) Stream(streamType));
112 : }
113 12 : if (streamDevice_ == nullptr) {
114 0 : HCCL_ERROR("[CpuTsThread][%s]reset stream failed, stream type[%d]",__func__, streamType);
115 0 : return uniqueIdStr_;
116 : }
117 :
118 12 : uint64_t size = sizeof(SqCqeContext);
119 12 : if (sqCqeContext_.ptr() == nullptr) {
120 8 : sqCqeContext_ = DeviceMem::alloc(size);
121 : }
122 12 : if (sqCqeContext_.ptr() == nullptr) {
123 0 : HCCL_ERROR("[CpuTsThread][%s]alloc mem failed, mem size[%llu]",__func__, size);
124 0 : return uniqueIdStr_;
125 : }
126 12 : HcclResult ret = hrtMemSet(sqCqeContext_.ptr(), size, size);
127 12 : if (ret != HCCL_SUCCESS) {
128 0 : HCCL_ERROR("[CpuTsThread][%s]mem set failed, mem size[%llu], ptr[%p]",__func__, size, sqCqeContext_.ptr());
129 0 : return uniqueIdStr_;
130 : }
131 :
132 12 : HcclStreamParam streamParam;
133 12 : streamParam.streamInfo.streamIds = streamDevice_->id();
134 12 : streamParam.streamInfo.sqIds = streamDevice_->sqId();
135 12 : streamParam.streamInfo.cqIds = streamDevice_->cqId();
136 12 : streamParam.streamInfo.logicCqids = streamDevice_->logicCqId();
137 12 : streamParam.sqCqContextAddr = reinterpret_cast<uint64_t>(sqCqeContext_.ptr());
138 12 : streamParam.sqCqContextSize = sqCqeContext_.size();
139 12 : oss.write(reinterpret_cast<const char_t *>(&streamParam), sizeof(streamParam));
140 :
141 12 : ret = HCCL_SUCCESS;
142 60 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
143 : HcclSignalInfo notifyInfo;
144 48 : ret = notifys_[idx]->GetNotifyData(notifyInfo);
145 48 : if (ret != HCCL_SUCCESS) {
146 0 : HCCL_ERROR("[AicpuTsThread][UpdateUniqueId]GetNotifyData failed, ret[%d]", ret);
147 0 : uniqueIdStr_ = std::string();
148 0 : return uniqueIdStr_;
149 : }
150 48 : HCCL_INFO("[AicpuTsThread][UpdateUniqueId]get local notify data success, resId[%u], tsId:%d, devId[%u]",
151 : notifyInfo.resId,
152 : notifyInfo.tsId,
153 : notifyInfo.devId);
154 48 : oss.write(reinterpret_cast<const char_t *>(¬ifyInfo), sizeof(notifyInfo));
155 : }
156 12 : HCCL_DEBUG("[AicpuTsThread][UpdateUniqueId] stream[%p], notifyNum[%u]", stream_->ptr(), notifyNum_);
157 :
158 12 : uniqueIdStr_ = oss.str();
159 12 : return uniqueIdStr_;
160 12 : }
161 :
162 51 : uint32_t CpuTsThread::GetNotifyNum() const
163 : {
164 51 : return notifyNum_;
165 : }
166 :
167 5 : LocalNotify *CpuTsThread::GetNotify(uint32_t index) const
168 : {
169 5 : if (index >= notifyNum_) {
170 0 : HCCL_ERROR(
171 : "[CpuTsThread][GetNotify] notifyNum[%u], index[%u] out of range[0, %u]", notifyNum_, index, notifyNum_ - 1);
172 0 : return nullptr;
173 : }
174 5 : return notifys_[index].get();
175 : }
176 :
177 15 : bool CpuTsThread::IsDeviceA5() const
178 : {
179 15 : return devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960;
180 : }
181 :
182 : // A3 Stream
183 51 : Stream *CpuTsThread::GetStream() const
184 : {
185 51 : return stream_.get();
186 : }
187 :
188 : // A5 Stream
189 1 : void *CpuTsThread::GetStreamLitePtr() const
190 : {
191 1 : return nullptr; // Not implemented
192 : }
193 :
194 1 : void CpuTsThread::LaunchTask() const
195 : {
196 1 : return;
197 : }
198 :
199 0 : void CpuTsThread::TryLaunchTask() const
200 : {
201 0 : HCCL_DEBUG("[%s] CpuTsThread does not support TryLaunchTask, skip", __func__);
202 0 : return;
203 : }
204 :
205 : // Local Data Plane Functions
206 1 : HcclResult CpuTsThread::LocalNotifyRecord(uint32_t notifyId) const
207 : {
208 1 : HCCL_ERROR("[CpuTsThread][%s]not support", __func__);
209 1 : return HCCL_E_NOT_SUPPORT;
210 : }
211 :
212 1 : HcclResult CpuTsThread::LocalNotifyWait(uint32_t notifyId) const
213 : {
214 1 : HCCL_ERROR("[CpuTsThread][%s]not support", __func__);
215 1 : return HCCL_E_NOT_SUPPORT;
216 : }
217 :
218 2 : HcclResult CpuTsThread::LocalNotifyRecord(ThreadHandle dstThread, uint32_t dstNotifyIdx) const
219 : {
220 : #ifndef CCL_KERNEL_AICPU
221 2 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
222 2 : HCCL_INFO("[%s]dstThread[0x%llu], dstNotifyIdx[%u].", __func__, dstThread, dstNotifyIdx);
223 2 : CHK_PRT_RET(!IsDeviceA5(), HCCL_ERROR("[CpuTsThread][%s]only support A5", __func__), HCCL_E_NOT_SUPPORT); // 只支持A5, 其他场景调用HcclLocalNotifyRecord
224 :
225 1 : Stream *stream = GetStream();
226 1 : CHK_PTR_NULL(stream);
227 1 : Thread *const dstThreadPtr = reinterpret_cast<Thread *>(dstThread);
228 1 : CHK_PTR_NULL(dstThreadPtr);
229 1 : LocalNotify *dstNotify = dstThreadPtr->GetNotify(dstNotifyIdx);
230 1 : CHK_PTR_NULL(dstNotify);
231 :
232 1 : HcclResult ret = dstNotify->Post(*stream);
233 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, dstThread[0x%llx], dstNotifyIdx[%u].",
234 : __func__, dstThread, dstNotifyIdx), ret);
235 :
236 : HcclSignalInfo signalInfo;
237 1 : CHK_RET(dstNotify->GetNotifyData(signalInfo));
238 1 : CHK_RET(ReportHostNotifyRecordTask(signalInfo.resId, beginTime, isMaster_));
239 : #endif
240 1 : return HCCL_SUCCESS;
241 : }
242 :
243 2 : HcclResult CpuTsThread::LocalNotifyWait(uint32_t notifyIdx, uint32_t timeOut) const
244 : {
245 : #ifndef CCL_KERNEL_AICPU
246 2 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
247 2 : HCCL_INFO("[%s]notifyIdx[%u], timeOut[%u].", __func__, notifyIdx, timeOut);
248 2 : CHK_PRT_RET(!IsDeviceA5(), HCCL_ERROR("[CpuTsThread][%s]only support A5", __func__), HCCL_E_NOT_SUPPORT); // 只支持A5, 其他场景调用HcclLocalNotifyWait
249 :
250 1 : Stream *stream = GetStream();
251 1 : CHK_PTR_NULL(stream);
252 1 : LocalNotify *notify = GetNotify(notifyIdx);
253 1 : CHK_PTR_NULL(notify);
254 :
255 1 : HcclResult ret = notify->Wait(*stream, timeOut);
256 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, notifyIdx[%u], timeOut[%u].",
257 : __func__, notifyIdx, timeOut), ret);
258 :
259 : HcclSignalInfo signalInfo;
260 1 : CHK_RET(notify->GetNotifyData(signalInfo));
261 1 : CHK_RET(ReportHostNotifyWaitTask(signalInfo.resId, beginTime, isMaster_));
262 : #endif
263 1 : return HCCL_SUCCESS;
264 : }
265 :
266 5 : HcclResult CpuTsThread::LocalCopy(void *dst, const void *src, uint64_t sizeByte) const
267 : {
268 : #ifndef CCL_KERNEL_AICPU
269 5 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
270 5 : HCCL_INFO("[%s]dst[%p], src[%p], sizeByte[%llu].", __func__, dst, src, sizeByte);
271 5 : CHK_PRT_RET(!IsDeviceA5(), HCCL_ERROR("[CpuTsThread][%s]only support A5", __func__), HCCL_E_NOT_SUPPORT); // 只支持A5, 其他场景调用HcclLocalCopy
272 :
273 4 : if (sizeByte == 0 || src == dst) {
274 2 : HCCL_INFO("[CpuTsThread][%s]skip, dst[%p] equals src[%p] or len[%llu] equals 0", __func__, dst, src, sizeByte);
275 2 : return HCCL_SUCCESS;
276 : }
277 :
278 2 : Stream *stream = GetStream();
279 2 : CHK_PTR_NULL(stream);
280 2 : CHK_RET(hrtMemAsyncCopy(dst, sizeByte, src, sizeByte,
281 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream->ptr()));
282 :
283 1 : CHK_RET(ReportHostLocalCopyTask(dst, src, sizeByte, beginTime, isMaster_));
284 : #endif
285 1 : return HCCL_SUCCESS;
286 : }
287 :
288 5 : HcclResult CpuTsThread::LocalReduce(
289 : void *dst, const void *src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp) const
290 : {
291 : #ifndef CCL_KERNEL_AICPU
292 5 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
293 5 : HCCL_INFO("[%s]dst[%p], src[%p], sizeByte[%llu], dataType[%d], reduceOp[%d].",
294 : __func__, dst, src, sizeByte, dataType, reduceOp);
295 5 : CHK_PRT_RET(!IsDeviceA5(), HCCL_ERROR("[CpuTsThread][%s]only support A5", __func__), HCCL_E_NOT_SUPPORT); // 只支持A5, 其他场景调用HcclLocalCopyReduce
296 :
297 4 : auto dataTypeIt = hccl2rtDataTypeMap.find(static_cast<HcclDataType>(dataType));
298 4 : if (dataTypeIt == hccl2rtDataTypeMap.end()) {
299 1 : HCCL_ERROR("[%s]data type[%s] is not supported", __func__,
300 : GetDataTypeEnumStr(static_cast<HcclDataType>(dataType)).c_str());
301 1 : return HCCL_E_PARA;
302 : }
303 :
304 3 : auto reduceOpIt = hccl2rtReduceOpMap.find(static_cast<HcclReduceOp>(reduceOp));
305 3 : if (reduceOpIt == hccl2rtReduceOpMap.end()) {
306 1 : HCCL_ERROR("[%s]reduceOp[%s] is not supported", __func__,
307 : GetReduceOpEnumStr(static_cast<HcclReduceOp>(reduceOp)).c_str());
308 1 : return HCCL_E_PARA;
309 : }
310 :
311 2 : Stream *stream = GetStream();
312 2 : CHK_PTR_NULL(stream);
313 2 : CHK_RET(hrtReduceAsync(dst, sizeByte, src, sizeByte, reduceOpIt->second, dataTypeIt->second, stream->ptr()));
314 1 : CHK_RET(ReportHostLocalReduceTask(dst, src, sizeByte, dataType, reduceOp, beginTime, isMaster_));
315 : #endif
316 1 : return HCCL_SUCCESS;
317 : }
318 7 : bool CpuTsThread::GetMaster() const {
319 7 : return isMaster_;
320 : }
321 :
322 4 : void CpuTsThread::SetIsMaster(bool isMaster) {
323 4 : isMaster_ = isMaster;
324 4 : }
325 :
326 12 : HcclResult CpuTsThread::SupplementNotify(uint32_t notifyNum)
327 : {
328 12 : if (streamType_ == StreamType::STREAM_TYPE_DEVICE || notifyLoadType_ == NotifyLoadType::DEVICE_NOTIFY) {
329 2 : HCCL_ERROR("[%s]Does not support this interface.", __func__);
330 2 : return HCCL_E_NOT_SUPPORT;
331 : }
332 10 : HCCL_INFO("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
333 :
334 10 : u32 currentNotifyNum = notifyNum_;
335 10 : notifyNum_ += notifyNum;
336 10 : notifys_.reserve(notifyNum_);
337 24 : for (uint32_t idx = currentNotifyNum; idx < notifyNum_; idx++) {
338 14 : notifys_.emplace_back(nullptr);
339 14 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
340 14 : CHK_SMART_PTR_NULL(notifys_[idx]);
341 14 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
342 14 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
343 3 : CHK_RET(notifys_[idx]->SetIpc());
344 : }
345 : }
346 :
347 10 : uniqueIdStr_.clear();
348 10 : UpdateUniqueId();
349 10 : return HCCL_SUCCESS;
350 : }
351 : } // namespace hccl
|
