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