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