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