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 524 : WorkspaceResourceImpl::~WorkspaceResourceImpl()
29 : {
30 524 : opBaseDeviceMemMap_.clear();
31 524 : remoteOpStreamMap_.clear();
32 522 : }
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 15 : 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 15 : std::vector<Stream> WorkspaceResourceImpl::AllocSlaveStreams(const std::string& tag, u32 num)
159 : {
160 15 : 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 23 : DeviceMem WorkspaceResourceImpl::AllocDeviceMem(const std::string& tag, u64 size)
171 : {
172 23 : 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 95 : HcclResult WorkspaceResourceImpl::GetOpBasedMemSize(const HcclCMDType& opType, u64& size, const HcomCollOpInfo& opInfo)
217 : {
218 95 : u64 opMemSize = 0;
219 :
220 95 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
221 : // ReduceScatter 算子所需memory大小为 CCLBuffSize
222 : DevType devType;
223 0 : CHK_RET(hrtGetDeviceType(devType));
224 0 : if (IsSupportSDMAReduce(opInfo.inputAddr, opInfo.outputAddr, opInfo.dataType, opInfo.reduceOp)
225 0 : && IsSupportRDMAReduce(opInfo.dataType, opInfo.reduceOp) && devType == DevType::DEV_TYPE_910B) {
226 0 : opMemSize = 0;
227 : } else {
228 0 : if (cclBufferManagerPtr_ == nullptr) {
229 0 : opMemSize = GetExternalInputCCLBuffSize();
230 : } else {
231 0 : opMemSize = cclBufferManagerPtr_->GetInCCLbufferSize();
232 : }
233 : }
234 : } else {
235 95 : opMemSize = 0;
236 : }
237 95 : size = HCCL_WORKSPACE_MEM_32_KB + opMemSize;
238 95 : HCCL_INFO(
239 : "[WorkspaceResourceImpl][GetOpBasedMemSize]workspace memory size: op[%d], memory size[%llu].", opType, size);
240 95 : return HCCL_SUCCESS;
241 : }
242 :
243 95 : HcclResult WorkspaceResourceImpl::CreateOpBasedResources(
244 : const HcclCMDType& opType, const std::string& tag, const HcomCollOpInfo& opInfo)
245 : {
246 95 : if (IsExistResourceWorkSpaceMem(tag) && GetDevMemSize(tag) != HCCL_SUCCESS) {
247 0 : HCCL_INFO(
248 : "[WorkspaceResourceImpl][CreateOpBasedResources]tag[%s] is exit, don't create workspace Memory",
249 : tag.c_str());
250 0 : return HCCL_SUCCESS;
251 : }
252 : // HCCL内部申请设备内存
253 95 : u64 memSize = 0;
254 95 : CHK_RET(GetOpBasedMemSize(opType, memSize, opInfo));
255 :
256 : // 创建 device memory
257 95 : DeviceMem deviceMem;
258 95 : CHK_RET(DeviceMem::alloc(deviceMem, memSize));
259 95 : std::vector<rtStream_t> stream;
260 95 : u64 maxSize = deviceMem.size();
261 95 : CHK_RET(SetWorkspaceResource(tag, deviceMem.ptr(), maxSize, stream));
262 95 : HCCL_INFO(
263 : "[WorkspaceResourceImpl][CreateOpBasedResources]create workspace memory success. "
264 : "tag[%s] workspace addr[%p] workspace size[%llu].",
265 : tag.c_str(), deviceMem.ptr(), deviceMem.size());
266 95 : CHK_RET(InsertDevMem(tag, deviceMem));
267 95 : return HCCL_SUCCESS;
268 95 : }
269 :
270 0 : HcclResult WorkspaceResourceImpl::InsertRemoteOpStream(const std::string& tag, std::vector<Stream>& stream)
271 : {
272 0 : auto interIter = remoteOpStreamMap_.find(tag);
273 0 : CHK_PRT_RET(
274 : interIter != remoteOpStreamMap_.end(),
275 : HCCL_ERROR(
276 : "[WorkspaceResourceImpl][InsertRemoteOpStream]tag[%s] is exit, "
277 : "don't insert remote operation stream",
278 : tag.c_str()),
279 : HCCL_E_INTERNAL);
280 0 : remoteOpStreamMap_[tag] = std::move(stream);
281 0 : return HCCL_SUCCESS;
282 : }
283 :
284 : HcclResult
285 0 : WorkspaceResourceImpl::CreateAndInsertDevMem(const std::string& tag, u64 memSize, std::vector<rtStream_t>& streamPtr)
286 : {
287 : // 创建device memory
288 0 : DeviceMem deviceMem;
289 0 : CHK_RET(DeviceMem::alloc(deviceMem, memSize));
290 :
291 0 : CHK_RET(SetWorkspaceResource(tag, deviceMem.ptr(), memSize, streamPtr));
292 0 : HCCL_INFO(
293 : "[WorkspaceResourceImpl][CreateAndInsertDevMem]create workspace memory success. "
294 : "tag[%s] workspace addr[%p] workspace size[%llu]",
295 : tag.c_str(), deviceMem.ptr(), deviceMem.size());
296 :
297 : // 资源管理
298 0 : CHK_RET(InsertDevMem(tag, deviceMem));
299 0 : return HCCL_SUCCESS;
300 0 : }
301 :
302 : HcclResult
303 0 : WorkspaceResourceImpl::CreateAndInsertRemoteOpStream(const std::string& tag, std::vector<rtStream_t>& streamPtr)
304 : {
305 : // 创建三条流
306 0 : Stream stream1(StreamType::STREAM_TYPE_ONLINE);
307 0 : CHK_PRT_RET(
308 : stream1.ptr() == nullptr,
309 : HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
310 : "In create workspace stream 1,malloc failed."),
311 : HCCL_E_MEMORY);
312 0 : Stream stream2(StreamType::STREAM_TYPE_ONLINE);
313 0 : CHK_PRT_RET(
314 : stream2.ptr() == nullptr,
315 : HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
316 : "In create workspace stream 2,malloc failed."),
317 : HCCL_E_MEMORY);
318 0 : Stream stream3(StreamType::STREAM_TYPE_ONLINE);
319 0 : CHK_PRT_RET(
320 : stream3.ptr() == nullptr,
321 : HCCL_ERROR("[WorkspaceResourceImpl][CreateAndInsertRemoteOpStream]"
322 : "In create workspace stream 3,malloc failed."),
323 : HCCL_E_MEMORY);
324 :
325 : // 将流指针插入streamPtr
326 0 : streamPtr.push_back(stream1.ptr());
327 0 : streamPtr.push_back(stream2.ptr());
328 0 : streamPtr.push_back(stream3.ptr());
329 :
330 : // 管理流对象
331 0 : std::vector<Stream> streamObjs;
332 0 : streamObjs.reserve(HCCL_SUB_STREAM_NUM_THREE);
333 0 : streamObjs.push_back(std::move(stream1));
334 0 : streamObjs.push_back(std::move(stream2));
335 0 : streamObjs.push_back(std::move(stream3));
336 :
337 0 : CHK_RET(InsertRemoteOpStream(tag, streamObjs)); // 当前无实际作用
338 0 : return HCCL_SUCCESS;
339 0 : }
340 :
341 0 : HcclResult WorkspaceResourceImpl::CreateRemoteOpBasedResources(u64 memSize, const std::string& tag)
342 : {
343 0 : if (IsExistResourceWorkSpaceMem(tag)) {
344 0 : HCCL_INFO(
345 : "[WorkspaceResourceImpl][CreateRemoteOpBasedResources]tag[%s] is exit, "
346 : "don't create workspace Memory",
347 : tag.c_str());
348 0 : return HCCL_SUCCESS;
349 : }
350 :
351 0 : std::vector<rtStream_t> streamPtr;
352 0 : CHK_RET(CreateAndInsertRemoteOpStream(tag, streamPtr));
353 0 : CHK_RET(CreateAndInsertDevMem(tag, memSize, streamPtr));
354 :
355 0 : return HCCL_SUCCESS;
356 0 : }
357 :
358 0 : HcclResult WorkspaceResourceImpl::CreateOrUpdateRemoteOpBasedResources(u64 memSize, const std::string& tag)
359 : {
360 0 : if (IsExistResourceWorkSpaceMem(tag)) {
361 0 : auto interMemIter = workSpaceMem_.memResMap_.find(tag);
362 0 : if (interMemIter->second.maxSize >= memSize) {
363 0 : HCCL_INFO(
364 : "[WorkspaceResourceImpl][CreateOrUpdateRemoteOpBasedResources]tag[%s] is exit, "
365 : "and memSize meets the requirements, don't create workspace Memory",
366 : tag.c_str());
367 0 : return HCCL_SUCCESS;
368 : }
369 : }
370 :
371 : // workspace stream 资源已存在, 将流指针插入stream ptr
372 0 : std::vector<rtStream_t> streamPtr;
373 0 : auto interStreamIter = remoteOpStreamMap_.find(tag);
374 0 : if (interStreamIter != remoteOpStreamMap_.end()) {
375 : // 复用workspace stream 资源
376 0 : for (auto& stream : interStreamIter->second) {
377 0 : streamPtr.push_back(stream.ptr());
378 : }
379 : } else {
380 0 : CHK_RET(CreateAndInsertRemoteOpStream(tag, streamPtr));
381 : }
382 :
383 0 : CHK_RET(CreateAndInsertDevMem(tag, memSize, streamPtr));
384 0 : return HCCL_SUCCESS;
385 0 : }
386 :
387 : } // namespace hccl
|
