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 "externalinput_pub.h"
12 : #include "device_capacity.h"
13 : #include "adapter_rts_common.h"
14 : #include "sal_pub.h"
15 : #include "workspace_resource_impl.h"
16 :
17 : namespace hccl {
18 : const std::string HCCL_KERNEL_OP_TYPE_REDUCESCATTER = "HcomReduceScatter";
19 : const std::string HCCL_KERNEL_OP_TYPE_ALLTOALL = "HcomAllToAll";
20 :
21 : constexpr s64 HCCL_SUB_STREAM_NUM_THREE = 3; // 最大subStream 数量为3
22 521 : WorkspaceResourceImpl::WorkspaceResourceImpl(u32 devicePhyId, s32 deviceLogicId, CCLBufferManager *cclBufferManagerPtr)
23 521 : : devicePhyId_(devicePhyId), deviceLogicId_(deviceLogicId), cclBufferManagerPtr_(cclBufferManagerPtr)
24 : {
25 521 : }
26 :
27 520 : WorkspaceResourceImpl::~WorkspaceResourceImpl()
28 : {
29 520 : opBaseDeviceMemMap_.clear();
30 519 : remoteOpStreamMap_.clear();
31 518 : }
32 :
33 91 : HcclResult WorkspaceResourceImpl::GetWorkspaceMemSize(const std::string &opType, u64 count,
34 : HcclDataType dataType, u32 rankSize, u64 &memSize, DevType deviceType) const
35 : {
36 : // 以一个页的大小4kb 去分配
37 91 : u64 alignSize = HCCL_ALIGN_SIZE;
38 91 : u64 tempMemSize = HCCL_WORKSPACE_MEM_32_KB;
39 :
40 91 : u32 dataTypeSize = 0;
41 91 : HcclResult ret = SalGetDataTypeSize(dataType, dataTypeSize);
42 91 : CHK_PRT_RET(ret != HCCL_SUCCESS,
43 : HCCL_ERROR("[WorkspaceResourceImpl][GetWorkspaceMemSize]op[%s] dataType[%s] is invalid. ret[%d]",
44 : opType.c_str(), GetDataTypeEnumStr(dataType).c_str(), ret), ret);
45 :
46 91 : u64 opMemSize = 0;
47 : // 判断是否需要申请额外的reduce scatter scratch mem
48 91 : if (opType == HCCL_KERNEL_OP_TYPE_REDUCESCATTER) {
49 39 : if (deviceType == DevType::DEV_TYPE_310P3) {
50 0 : opMemSize += 0;
51 : } else {
52 : // ReduceScatter 算子所需memory大小为 单个数据size * count * rank_size
53 39 : opMemSize += count * dataTypeSize * rankSize;
54 : }
55 : }
56 :
57 91 : if (opType == HCCL_KERNEL_OP_TYPE_ALLTOALL) {
58 0 : opMemSize += count * dataTypeSize * rankSize;
59 : }
60 :
61 91 : tempMemSize += opMemSize;
62 91 : memSize = (tempMemSize + alignSize - 1) / alignSize * alignSize;
63 91 : HCCL_INFO("[WorkspaceResourceImpl][GetWorkspaceMemSize]workspace memory memSize: "
64 : "op[%s], data type[%s], count[%llu], rank memSize[%u], memory memSize[%llu].",
65 : opType.c_str(), GetDataTypeEnumStr(dataType).c_str(), count, rankSize, memSize);
66 :
67 91 : return HCCL_SUCCESS;
68 : }
69 :
70 0 : HcclResult WorkspaceResourceImpl::RegisterMaster(const std::string &tag, Stream stream)
71 : {
72 0 : return offloadStreamManager_.RegisterMaster(tag, stream);
73 : }
74 :
75 250 : HcclResult WorkspaceResourceImpl::SetMemResource(const std::string &tag, void *memPtr, u64 &maxSize)
76 : {
77 250 : return workSpaceMem_.SetMemResource(tag, memPtr, maxSize);
78 : }
79 :
80 124 : HcclResult WorkspaceResourceImpl::SetStreamResource(const std::string &tag, std::vector<rtStream_t> &stream)
81 : {
82 124 : HCCL_DEBUG("[WorkspaceResourceImpl][SetStreamResource]setting stream resources, input stream size[%u], tag[%s]",
83 : stream.size(), tag.c_str());
84 : // 后继考虑是否将 OffloadStreamManager 和 WorkSpaceMem 的实现直接包在这个类中
85 124 : std::vector<Stream> offloadSlaves;
86 396 : for (u32 i = 0; i < stream.size(); i++) {
87 : // 当前GE WorkSpaceResource中Create的流都是从流，后继若有主流情况，再作扩展
88 272 : offloadSlaves.emplace_back(Stream(stream[i], false));
89 272 : if (!offloadSlaves[i].ptr()) {
90 0 : HCCL_ERROR("[WorkspaceResourceImpl][SetStreamResource]create offload stream[%u] fail.", i);
91 0 : return HCCL_E_INTERNAL;
92 : }
93 : }
94 124 : return offloadStreamManager_.RegisterSlaves(tag, offloadSlaves);
95 124 : }
96 :
97 : // 基于tag 初始设置资源，包含 Stream 资源 和 DeviceMem 资源
98 250 : HcclResult WorkspaceResourceImpl::SetWorkspaceResource(const std::string &tag, void *memPtr,
99 : u64 &maxSize, std::vector<rtStream_t> &stream)
100 : {
101 : // 设定 workspace memory 资源
102 250 : if (memPtr == nullptr) {
103 0 : HCCL_WARNING("[WorkspaceResourceImpl][SetWorkspaceResource] workspace mem ptr is null, tag[%s] maxSize[%llu]",
104 : tag.c_str(), maxSize);
105 : } else {
106 250 : CHK_RET(SetMemResource(tag, memPtr, maxSize));
107 : }
108 :
109 : /* 设定 workspace stream 资源 */
110 250 : if (stream.size() != 0) {
111 124 : CHK_RET(SetStreamResource(tag, stream));
112 : }
113 250 : return HCCL_SUCCESS;
114 : }
115 :
116 : // 基于 tag 销毁资源，包含 Stream 资源 和 DeviceMem 资源
117 16 : void WorkspaceResourceImpl::DestroyWorkspaceResource(const std::string &tag)
118 : {
119 : // 销毁 work space memory 资源
120 16 : HcclResult ret = workSpaceMem_.DestroyMemResource(tag);
121 11 : if (ret != HCCL_SUCCESS) {
122 0 : HCCL_ERROR("[WorkspaceResourceImpl][DestroyWorkspaceResource]Destroy workspace mem failed. ret[%d]", ret);
123 : }
124 :
125 : // 销毁 work space stream资源
126 11 : if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
127 11 : ret = offloadStreamManager_.ClearSlaves(tag);
128 16 : if (ret != HCCL_SUCCESS) {
129 0 : HCCL_ERROR("[WorkspaceResourceImpl][DestroyWorkspaceResource]Destroy workspace stream failed. "
130 : "ret[%d]", ret);
131 : }
132 : }
133 16 : }
134 :
135 : // 销毁 Workspace全局资源，包含 Stream 资源 和 内存 资源
136 0 : void WorkspaceResourceImpl::DestroyWorkspaceResource()
137 : {
138 : // 销毁 work space memory 资源
139 0 : workSpaceMem_.DestroyMemResource();
140 :
141 : // 销毁 work space stream资源
142 0 : if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
143 0 : offloadStreamManager_.ClearSlaves();
144 : }
145 0 : return;
146 : }
147 :
148 : // 基于tag 分配 Stream 资源
149 16 : std::vector<Stream> WorkspaceResourceImpl::AllocSlaveStreams(const std::string &tag, u32 num)
150 : {
151 16 : return offloadStreamManager_.GetSlaves(tag, num);
152 : }
153 :
154 : // 基于tag 销毁 Stream 资源
155 0 : HcclResult WorkspaceResourceImpl::DestroyStream(const std::string &tag)
156 : {
157 0 : return offloadStreamManager_.ClearSlaves(tag);
158 : }
159 :
160 : // 基于tag 分配 DeviceMem 资源
161 24 : DeviceMem WorkspaceResourceImpl::AllocDeviceMem(const std::string &tag, u64 size)
162 : {
163 24 : return DeviceMem::create(workSpaceMem_.AllocMem(tag, size), size);
164 : }
165 :
166 : // 基于tag 销毁 DeviceMem 资源
167 0 : HcclResult WorkspaceResourceImpl::DestroyDeviceMem(const std::string &tag)
168 : {
169 0 : return workSpaceMem_.DestroyMemResource(tag);
170 : }
171 :
172 :
173 95 : bool WorkspaceResourceImpl::IsExistResourceWorkSpaceMem(const std::string &tag)
174 : {
175 95 : return workSpaceMem_.IsExist(tag);
176 : }
177 :
178 76 : HcclResult WorkspaceResourceImpl::GetDevMemSize(const std::string &tag)
179 : {
180 76 : auto interIter = opBaseDeviceMemMap_.find(tag);
181 76 : if (interIter == opBaseDeviceMemMap_.end()) {
182 0 : HCCL_INFO("[WorkspaceResourceImpl][GetDevMemSize]tag[%s] is exit, don't need get memsize", tag.c_str());
183 0 : return HCCL_SUCCESS;
184 : } else {
185 76 : return (interIter->second.size() == HCCL_WORKSPACE_MEM_32_KB) ? HCCL_SUCCESS : HCCL_E_INTERNAL;
186 : }
187 : }
188 :
189 95 : HcclResult WorkspaceResourceImpl::InsertDevMem(const std::string &tag, DeviceMem &deviceMem)
190 : {
191 95 : std::unique_lock<std::mutex> lock(memResMutex_);
192 95 : opBaseDeviceMemMap_.erase(tag);
193 95 : opBaseDeviceMemMap_.insert(std::pair<std::string, DeviceMem>(tag, std::move(deviceMem)));
194 95 : lock.unlock();
195 95 : return HCCL_SUCCESS;
196 95 : }
197 :
198 0 : HcclResult WorkspaceResourceImpl::DestroyRemoteOpBasedMem(const std::string &tag)
199 : {
200 0 : std::unique_lock<std::mutex> lock(memResMutex_);
201 0 : opBaseDeviceMemMap_.erase(tag);
202 0 : remoteOpStreamMap_.erase(tag);
203 0 : lock.unlock();
204 :
205 0 : DestroyWorkspaceResource(tag);
206 :
207 0 : return HCCL_SUCCESS;
208 0 : }
209 :
210 : // 获取算子所需workspace memory大小[byte]
211 95 : HcclResult WorkspaceResourceImpl::GetOpBasedMemSize(const HcclCMDType &opType, u64 &size,
212 : const HcomCollOpInfo &opInfo)
213 : {
214 95 : u64 opMemSize = 0;
215 :
216 95 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
217 : // ReduceScatter 算子所需memory大小为 CCLBuffSize
218 : DevType devType;
219 0 : CHK_RET(hrtGetDeviceType(devType));
220 0 : if (IsSupportSDMAReduce(opInfo.inputAddr, opInfo.outputAddr, opInfo.dataType, opInfo.reduceOp) &&
221 0 : IsSupportRDMAReduce(opInfo.dataType, opInfo.reduceOp) && devType == DevType::DEV_TYPE_910B) {
222 0 : opMemSize = 0;
223 : } else {
224 0 : if (cclBufferManagerPtr_ == nullptr) {
225 0 : opMemSize = GetExternalInputCCLBuffSize();
226 : } else {
227 0 : opMemSize = cclBufferManagerPtr_->GetInCCLbufferSize();
228 : }
229 : }
230 : } else {
231 95 : opMemSize = 0;
232 : }
233 95 : size = HCCL_WORKSPACE_MEM_32_KB + opMemSize;
234 95 : HCCL_INFO("[WorkspaceResourceImpl][GetOpBasedMemSize]workspace memory size: op[%d], memory size[%llu].",
235 : opType, size);
236 95 : return HCCL_SUCCESS;
237 : }
238 :
239 95 : HcclResult WorkspaceResourceImpl::CreateOpBasedResources(const HcclCMDType &opType, const std::string &tag,
240 : const HcomCollOpInfo &opInfo)
241 : {
242 95 : if (IsExistResourceWorkSpaceMem(tag) && GetDevMemSize(tag) != HCCL_SUCCESS) {
243 0 : HCCL_INFO("[WorkspaceResourceImpl][CreateOpBasedResources]tag[%s] is exit, don't create workspace Memory",
244 : tag.c_str());
245 0 : return HCCL_SUCCESS;
246 : }
247 : // HCCL内部申请设备内存
248 95 : u64 memSize = 0;
249 95 : CHK_RET(GetOpBasedMemSize(opType, memSize, opInfo));
250 :
251 : // 创建 device memory
252 95 : DeviceMem deviceMem;
253 95 : CHK_RET(DeviceMem::alloc(deviceMem, memSize));
254 95 : std::vector<rtStream_t> stream;
255 95 : u64 maxSize = deviceMem.size();
256 95 : CHK_RET(SetWorkspaceResource(tag, deviceMem.ptr(), maxSize, stream));
257 95 : HCCL_INFO("[WorkspaceResourceImpl][CreateOpBasedResources]create workspace memory success. "
258 : "tag[%s] workspace addr[%p] workspace size[%llu].", tag.c_str(), deviceMem.ptr(), deviceMem.size());
259 95 : CHK_RET(InsertDevMem(tag, deviceMem));
260 95 : return HCCL_SUCCESS;
261 95 : }
262 :
263 0 : HcclResult WorkspaceResourceImpl::InsertRemoteOpStream(const std::string &tag, std::vector<Stream> &stream)
264 : {
265 0 : auto interIter = remoteOpStreamMap_.find(tag);
266 0 : CHK_PRT_RET(interIter != remoteOpStreamMap_.end(),
267 : HCCL_ERROR("[WorkspaceResourceImpl][InsertRemoteOpStream]tag[%s] is exit, "
268 : "don't insert remote operation stream", tag.c_str()), HCCL_E_INTERNAL);
269 0 : remoteOpStreamMap_[tag] = std::move(stream);
270 0 : return HCCL_SUCCESS;
271 : }
272 :
273 0 : HcclResult WorkspaceResourceImpl::CreateAndInsertDevMem(const std::string &tag, u64 memSize,
274 : std::vector<rtStream_t> &streamPtr)
275 : {
276 : // 创建device memory
277 0 : DeviceMem deviceMem;
278 0 : CHK_RET(DeviceMem::alloc(deviceMem, memSize));
279 :
280 0 : CHK_RET(SetWorkspaceResource(tag, deviceMem.ptr(), memSize, streamPtr));
281 0 : HCCL_INFO("[WorkspaceResourceImpl][CreateAndInsertDevMem]create workspace memory success. "
282 : "tag[%s] workspace addr[%p] workspace size[%llu]", tag.c_str(), deviceMem.ptr(), deviceMem.size());
283 :
284 : // 资源管理
285 0 : CHK_RET(InsertDevMem(tag, deviceMem));
286 0 : return HCCL_SUCCESS;
287 0 : }
288 :
289 0 : HcclResult WorkspaceResourceImpl::CreateAndInsertRemoteOpStream(const std::string &tag,
290 : std::vector<rtStream_t> &streamPtr)
291 : {
292 : // 创建三条流
293 0 : Stream stream1(StreamType::STREAM_TYPE_ONLINE);
294 0 : CHK_PRT_RET(stream1.ptr() == nullptr, HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
295 : "In create workspace stream 1,malloc failed."), HCCL_E_MEMORY);
296 0 : Stream stream2(StreamType::STREAM_TYPE_ONLINE);
297 0 : CHK_PRT_RET(stream2.ptr() == nullptr, HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
298 : "In create workspace stream 2,malloc failed."), HCCL_E_MEMORY);
299 0 : Stream stream3(StreamType::STREAM_TYPE_ONLINE);
300 0 : CHK_PRT_RET(stream3.ptr() == nullptr, HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
301 : "In create workspace stream 3,malloc failed."), HCCL_E_MEMORY);
302 :
303 : // 将流指针插入streamPtr
304 0 : streamPtr.push_back(stream1.ptr());
305 0 : streamPtr.push_back(stream2.ptr());
306 0 : streamPtr.push_back(stream3.ptr());
307 :
308 : // 管理流对象
309 0 : std::vector<Stream> streamObjs;
310 0 : streamObjs.reserve(HCCL_SUB_STREAM_NUM_THREE);
311 0 : streamObjs.push_back(std::move(stream1));
312 0 : streamObjs.push_back(std::move(stream2));
313 0 : streamObjs.push_back(std::move(stream3));
314 :
315 0 : CHK_RET(InsertRemoteOpStream(tag, streamObjs)); // 当前无实际作用
316 0 : return HCCL_SUCCESS;
317 0 : }
318 :
319 0 : HcclResult WorkspaceResourceImpl::CreateRemoteOpBasedResources(u64 memSize, const std::string &tag)
320 : {
321 0 : if (IsExistResourceWorkSpaceMem(tag)) {
322 0 : HCCL_INFO("[WorkspaceResourceImpl][CreateRemoteOpBasedResources]tag[%s] is exit, "
323 : "don't create workspace Memory", tag.c_str());
324 0 : return HCCL_SUCCESS;
325 : }
326 :
327 0 : std::vector<rtStream_t> streamPtr;
328 0 : CHK_RET(CreateAndInsertRemoteOpStream(tag, streamPtr));
329 0 : CHK_RET(CreateAndInsertDevMem(tag, memSize, streamPtr));
330 :
331 0 : return HCCL_SUCCESS;
332 0 : }
333 :
334 0 : HcclResult WorkspaceResourceImpl::CreateOrUpdateRemoteOpBasedResources(u64 memSize, const std::string &tag)
335 : {
336 0 : if (IsExistResourceWorkSpaceMem(tag)) {
337 0 : auto interMemIter = workSpaceMem_.memResMap_.find(tag);
338 0 : if (interMemIter->second.maxSize >= memSize) {
339 0 : HCCL_INFO("[WorkspaceResourceImpl][CreateOrUpdateRemoteOpBasedResources]tag[%s] is exit, "
340 : "and memSize meets the requirements, don't create workspace Memory", tag.c_str());
341 0 : return HCCL_SUCCESS;
342 : }
343 : }
344 :
345 : // workspace stream 资源已存在, 将流指针插入stream ptr
346 0 : std::vector<rtStream_t> streamPtr;
347 0 : auto interStreamIter = remoteOpStreamMap_.find(tag);
348 0 : if (interStreamIter != remoteOpStreamMap_.end()) {
349 : // 复用workspace stream 资源
350 0 : for (auto &stream : interStreamIter->second) {
351 0 : streamPtr.push_back(stream.ptr());
352 : }
353 : } else {
354 0 : CHK_RET(CreateAndInsertRemoteOpStream(tag, streamPtr));
355 : }
356 :
357 0 : CHK_RET(CreateAndInsertDevMem(tag, memSize, streamPtr));
358 0 : return HCCL_SUCCESS;
359 0 : }
360 :
361 : } // namespace hccl
|
