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 "tensor_data_transfer.h"
12 : #include <map>
13 : #include <mutex>
14 : #include <sstream>
15 : #include <unordered_map>
16 :
17 : #include "data_common.h"
18 : #include "tdt_host_interface.h"
19 :
20 : #include "log_inner.h"
21 : #include "acl/acl_tdt_queue.h"
22 : #include "acl_tdt_queue/queue.h"
23 : #include "runtime/rt_mem_queue.h"
24 : #include "runtime/mem.h"
25 : #include "runtime/context.h"
26 : #include "runtime/rts/rts_mem.h"
27 : #include "utils/file_utils.h"
28 : #include "utils/data_type_utils.h"
29 :
30 : namespace {
31 : std::mutex aclChannleMutex;
32 : std::map<std::string, acltdtChannelHandle *> aclChannleMap;
33 : std::map<std::string, aclDataType> aclDataTypeStrMap =
34 : {
35 : {"bool", ACL_BOOL},
36 : {"int8", ACL_INT8},
37 : {"uint8", ACL_UINT8},
38 : {"half", ACL_FLOAT16},
39 : {"int16", ACL_INT16},
40 : {"uint16", ACL_UINT16},
41 : {"float", ACL_FLOAT},
42 : {"int32", ACL_INT32},
43 : {"uint32", ACL_UINT32},
44 : {"int64", ACL_INT64},
45 : {"uint64", ACL_UINT64},
46 : {"double", ACL_DOUBLE},
47 : {"string", ACL_STRING}
48 : };
49 : constexpr uint32_t VERSION_NAME = 1U;
50 : constexpr size_t TDT_TENSOR_ALIGNE_UNIT = 64UL;
51 : const std::vector<size_t> GEAR_SIZE{1U * 1024U * 1024U, 10U * 1024U * 1024U, 100U * 1024U * 1024U,
52 : 500U * 1024U * 1024U};
53 6 : size_t Get64AlignedSize(const size_t size)
54 : {
55 6 : return (size + TDT_TENSOR_ALIGNE_UNIT - 1UL) / TDT_TENSOR_ALIGNE_UNIT * TDT_TENSOR_ALIGNE_UNIT;
56 : }
57 :
58 : using TdtHostInitFunc = int32_t (*)(uint32_t);
59 : using TdtHostPreparePopDataFunc = int32_t (*)();
60 : using TdtHostPopDataFunc = int32_t (*)(const std::string &, std::vector<tdt::DataItem> &);
61 : using TdtHostPushDataFunc = int32_t (*)(const std::string &, const std::vector<tdt::DataItem> &, uint32_t deviceId);
62 : using TdtHostStopFunc = int32_t (*)(const std::string &);
63 : using TdtHostDestroyFunc = int32_t (*)();
64 :
65 : #ifndef RUN_TEST
66 : void *GetHandler()
67 : {
68 : std::string soPath;
69 : if (acl::file_utils::GetSoRealPath(soPath) != ACL_SUCCESS) {
70 : ACL_LOG_ERROR("Get libacl_tdt_channel.so path failed.");
71 : return nullptr;
72 : }
73 : std::string soName = soPath + "libdatatransfer.so";
74 : // Load the "libdatatransfer.so" library until the program ends. During the process, Dlclose is not invoked
75 : // to prevent the destruction of the global state information saved in ibdatatransfer.so.
76 : void *handler = mmDlopen(soName.c_str(), RTLD_NOW | RTLD_GLOBAL);
77 : if (handler == nullptr) {
78 : ACL_LOG_ERROR("The corresponding dependent dynamic library cannot be found. "
79 : "Please confirm whether the environment supports it and if the extension package has been correctly installed. "
80 : "soName is %s.", soName.c_str());
81 : }
82 : return handler;
83 : }
84 : #endif
85 :
86 6 : void *GetFunction(const std::string &func_name)
87 : {
88 : #ifdef RUN_TEST
89 : std::unordered_map<std::string, void*> stubFunctionMap = {
90 0 : {"TdtHostInit", reinterpret_cast<void*>(&tdt::TdtHostInit)},
91 0 : {"TdtHostPushData", reinterpret_cast<void*>(&tdt::TdtHostPushData)},
92 0 : {"TdtHostDestroy", reinterpret_cast<void*>(&tdt::TdtHostDestroy)},
93 0 : {"TdtHostPreparePopData", reinterpret_cast<void*>(&tdt::TdtHostPreparePopData)},
94 0 : {"TdtHostPopData", reinterpret_cast<void*>(&tdt::TdtHostPopData)},
95 0 : {"TdtHostStop", reinterpret_cast<void*>(&tdt::TdtHostStop)}
96 48 : };
97 6 : auto it = stubFunctionMap.find(func_name);
98 6 : if (it != stubFunctionMap.end()) {
99 6 : return it->second;
100 : }
101 0 : return nullptr;
102 : #else
103 : static void *handler = GetHandler();
104 : if (handler == nullptr) {
105 : ACL_LOG_ERROR("Get handler failed when get %s function.", func_name.c_str());
106 : return nullptr;
107 : }
108 : void *func_ptr = mmDlsym(handler, func_name.c_str());
109 : if (func_ptr == nullptr) {
110 : ACL_LOG_ERROR("The corresponding symbol cannot be found. Please confirm whether the installed extension package is correct, %s.", mmDlerror());
111 : }
112 : return func_ptr;
113 : #endif
114 12 : }
115 : }
116 :
117 : namespace acl {
118 4 : bool GetTensorShape(const std::string &dimsStr, std::vector<int64_t> &dims)
119 : {
120 : // change "[32,224,224,3]" => "32,224,224,3"
121 : // tensor_shape.size() - 2 is the second to last
122 4 : if (dimsStr.size() < 2) {
123 2 : ACL_LOG_INNER_ERROR("[Check][dimsStr]Invalid shape string: %s", dimsStr.c_str());
124 2 : return false;
125 : }
126 :
127 2 : std::string str = dimsStr.substr(1, dimsStr.size() - 2);
128 2 : std::string::size_type index = 0;
129 2 : if (!str.empty()) {
130 2 : while ((index = str.find(' ', index)) != std::string::npos) {
131 0 : str.erase(index, 1);
132 : }
133 : }
134 2 : std::string split = ",";
135 2 : std::string::size_type pos2 = str.find(split);
136 2 : std::string::size_type pos1 = 0;
137 3 : while (pos2 != std::string::npos) {
138 : try {
139 1 : dims.push_back(std::stoll(str.substr(pos1, pos2 - pos1)));
140 0 : } catch (...) {
141 0 : ACL_LOG_INNER_ERROR("[Check][Shape]Invalid shape string: %s", dimsStr.c_str());
142 0 : return false;
143 0 : }
144 : // string::size_type can store the length of any string object
145 1 : pos1 = pos2 + split.size();
146 1 : pos2 = str.find(split, pos1);
147 : }
148 2 : if (pos1 != str.length()) {
149 : try {
150 3 : dims.push_back(std::stoll(str.substr(pos1)));
151 1 : } catch (...) {
152 1 : ACL_LOG_INNER_ERROR("[Check][Shape]Invalid shape string: %s", dimsStr.c_str());
153 1 : return false;
154 1 : }
155 : }
156 1 : return true;
157 2 : }
158 :
159 5 : aclError GetTdtDataTypeByAclDataType(acltdtTensorType aclType, tdt::TdtDataType &tdtDataType)
160 : {
161 5 : switch (aclType) {
162 1 : case ACL_TENSOR_DATA_END_OF_SEQUENCE: {
163 1 : tdtDataType = tdt::TDT_END_OF_SEQUENCE;
164 1 : break;
165 : }
166 2 : case ACL_TENSOR_DATA_TENSOR: {
167 2 : tdtDataType = tdt::TDT_TENSOR;
168 2 : break;
169 : }
170 1 : case ACL_TENSOR_DATA_ABNORMAL: {
171 1 : tdtDataType = tdt::TDT_ABNORMAL;
172 1 : break;
173 : }
174 1 : default: {
175 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
176 2 : std::vector<const char *>({"func", "value", "param", "expect"}),
177 1 : std::vector<const char *>({"Obtaining the tdt data type", acl::GetTensorTypeDesc(aclType),
178 2 : "aclType", "ACL_TENSOR_DATA_END_OF_SEQUENCE or ACL_TENSOR_DATA_TENSOR or ACL_TENSOR_DATA_ABNORMAL"}));
179 1 : return ACL_ERROR_INVALID_PARAM;
180 : }
181 : }
182 4 : return ACL_SUCCESS;
183 : }
184 :
185 12 : aclError GetTdtDataTypeByAclDataTypeV2(acltdtTensorType aclType, int32_t &tdtDataType)
186 : {
187 12 : switch (aclType) {
188 1 : case ACL_TENSOR_DATA_END_OF_SEQUENCE: {
189 1 : tdtDataType = 1;
190 1 : break;
191 : }
192 7 : case ACL_TENSOR_DATA_TENSOR: {
193 7 : tdtDataType = 0;
194 7 : break;
195 : }
196 1 : case ACL_TENSOR_DATA_ABNORMAL: {
197 1 : tdtDataType = 2;
198 1 : break;
199 : }
200 3 : default: {
201 3 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
202 6 : std::vector<const char *>({"func", "value", "param", "expect"}),
203 3 : std::vector<const char *>({"Obtaining the tdt data type", acl::GetTensorTypeDesc(aclType),
204 6 : "aclType", "ACL_TENSOR_DATA_END_OF_SEQUENCE or ACL_TENSOR_DATA_TENSOR or ACL_TENSOR_DATA_ABNORMAL"}));
205 3 : return ACL_ERROR_INVALID_PARAM;
206 : }
207 : }
208 9 : return ACL_SUCCESS;
209 : }
210 :
211 6 : aclError GetAclTypeByTdtDataType(tdt::TdtDataType tdtDataType, acltdtTensorType &aclType)
212 : {
213 6 : switch (tdtDataType) {
214 2 : case tdt::TDT_END_OF_SEQUENCE: {
215 2 : aclType = ACL_TENSOR_DATA_END_OF_SEQUENCE;
216 2 : break;
217 : }
218 2 : case tdt::TDT_TENSOR: {
219 2 : aclType = ACL_TENSOR_DATA_TENSOR;
220 2 : break;
221 : }
222 1 : case tdt::TDT_ABNORMAL: {
223 1 : aclType = ACL_TENSOR_DATA_ABNORMAL;
224 1 : break;
225 : }
226 1 : default: {
227 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
228 2 : std::vector<const char *>({"func", "value", "param", "expect"}),
229 1 : std::vector<const char *>({"Obtaining the tensor data type", acl::GetTdtDataTypeDesc(tdtDataType), "tdtDataType",
230 2 : "TDT_END_OF_SEQUENCE or TDT_TENSOR or TDT_ABNORMAL"}));
231 1 : return ACL_ERROR_UNSUPPORTED_DATA_TYPE;
232 : }
233 : }
234 5 : return ACL_SUCCESS;
235 : }
236 :
237 8 : aclError GetAclTypeByTdtDataTypeV2(int32_t tdtDataType, acltdtTensorType &aclType)
238 : {
239 8 : switch (tdtDataType) {
240 1 : case 1: {
241 1 : aclType = ACL_TENSOR_DATA_END_OF_SEQUENCE;
242 1 : break;
243 : }
244 3 : case 0: {
245 3 : aclType = ACL_TENSOR_DATA_TENSOR;
246 3 : break;
247 : }
248 1 : case 2: {
249 1 : aclType = ACL_TENSOR_DATA_ABNORMAL;
250 1 : break;
251 : }
252 1 : case 3: {
253 1 : aclType = ACL_TENSOR_DATA_SLICE_TENSOR;
254 1 : break;
255 : }
256 1 : case 4: {
257 1 : aclType = ACL_TENSOR_DATA_END_TENSOR;
258 1 : break;
259 : }
260 1 : default: {
261 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
262 2 : std::vector<const char *>({"func", "value", "param", "expect"}),
263 1 : std::vector<const char *>({"Obtaining the tensor data type", acl::GetTdtDataTypeDescV2(tdtDataType), "tdtDataType",
264 2 : "[TDT_TENSOR(0), TDT_END_OF_SEQUENCE(1), TDT_ABNORMAL(2), TDT_SLICE_TENSOR(3), TDT_END_TENSOR(4)]"}));
265 1 : return ACL_ERROR_UNSUPPORTED_DATA_TYPE;
266 : }
267 : }
268 7 : return ACL_SUCCESS;
269 : }
270 :
271 8 : aclError TensorDatasetSerializes(const acltdtDataset *dataset, std::vector<tdt::DataItem> &itemVec)
272 : {
273 8 : ACL_REQUIRES_NOT_NULL(dataset);
274 :
275 9 : for (size_t i = 0; i < dataset->blobs.size(); ++i) {
276 1 : tdt::DataItem item;
277 : tdt::TdtDataType tdtDataType;
278 1 : auto ret = GetTdtDataTypeByAclDataType(dataset->blobs[i]->tdtType, tdtDataType);
279 1 : if (ret != ACL_SUCCESS) {
280 0 : ACL_LOG_INNER_ERROR("[Check][Dataset]TensorDatasetSerializes failed, "
281 : "invalid tdt type %d", dataset->blobs[i]->tdtType);
282 0 : itemVec.clear();
283 0 : return ret;
284 : }
285 :
286 1 : item.dataType_ = tdtDataType;
287 1 : item.tensorShape_ = dataset->blobs[i]->dimsStr;
288 1 : item.tensorType_ = dataset->blobs[i]->dataTypeStr;
289 1 : item.dataLen_ = dataset->blobs[i]->dataLen;
290 1 : item.dataPtr_ = dataset->blobs[i]->dataPtr;
291 1 : itemVec.emplace_back(item);
292 1 : }
293 8 : return ACL_SUCCESS;
294 : }
295 :
296 9 : aclError TensorDatasetSerializesV2(const acltdtDataset *dataset, std::vector<acl::aclTdtDataItemInfo> &itemVec)
297 : {
298 9 : ACL_REQUIRES_NOT_NULL(dataset);
299 16 : for (size_t i = 0; i < dataset->blobs.size(); ++i) {
300 9 : acl::aclTdtDataItemInfo item;
301 : int32_t tdtDataType;
302 9 : auto ret = GetTdtDataTypeByAclDataTypeV2(dataset->blobs[i]->tdtType, tdtDataType);
303 9 : if (ret != ACL_SUCCESS) {
304 2 : ACL_LOG_INNER_ERROR("[Check][Dataset]TensorDatasetSerializes failed, "
305 : "invalid tdt type %d", dataset->blobs[i]->tdtType);
306 2 : return ret;
307 : }
308 :
309 7 : item.ctrlInfo.dataType = tdtDataType;
310 7 : item.ctrlInfo.tensorType = dataset->blobs[i]->dataType;
311 7 : item.ctrlInfo.dimNum = dataset->blobs[i]->dims.size();
312 7 : item.dims = dataset->blobs[i]->dims;
313 7 : item.ctrlInfo.dataLen = dataset->blobs[i]->dataLen;
314 7 : item.dataPtr = dataset->blobs[i]->dataPtr;
315 7 : itemVec.emplace_back(item);
316 7 : ACL_LOG_DEBUG("TensorDatasetSerializesWithQueue, dataType %d, tensorType %d, dimNum %u, dataLen %lu",
317 : item.ctrlInfo.dataType, item.ctrlInfo.tensorType, item.ctrlInfo.dimNum, item.ctrlInfo.dataLen);
318 9 : }
319 7 : return ACL_SUCCESS;
320 : }
321 :
322 8 : aclError TensorDatasetDeserializes(const std::vector<tdt::DataItem> &itemVec, acltdtDataset *dataset)
323 : {
324 8 : ACL_REQUIRES_NOT_NULL(dataset);
325 8 : if (dataset->blobs.size() != 0) {
326 4 : const std::string sizeVal = std::to_string(dataset->blobs.size());
327 4 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
328 8 : std::vector<const char *>({"func", "value", "param", "reason"}),
329 4 : std::vector<const char *>({"tdt data deserialization", sizeVal.c_str(), "dataset->blobs.size",
330 8 : "dataset must be empty before deserialization"}));
331 4 : return ACL_ERROR_INVALID_PARAM;
332 4 : }
333 4 : aclError ret = ACL_SUCCESS;
334 5 : for (size_t i = 0; i < itemVec.size(); ++i) {
335 : acltdtTensorType aclType;
336 3 : ret = GetAclTypeByTdtDataType(itemVec[i].dataType_, aclType);
337 3 : if (ret != ACL_SUCCESS) {
338 1 : ACL_LOG_INNER_ERROR("[Check][Dataset]TensorDatasetDeserializes failed, invalid data type %d",
339 : itemVec[i].dataType_);
340 2 : break;
341 : }
342 :
343 2 : if (aclType == ACL_TENSOR_DATA_TENSOR) {
344 1 : std::vector<int64_t> dims;
345 1 : if (!GetTensorShape(itemVec[i].tensorShape_, dims)) {
346 1 : ACL_LOG_INNER_ERROR("[Check][TensorDataset]TensorDatasetDeserializes failed, "
347 : "invalid tensor shape[%s]", itemVec[i].tensorShape_.c_str());
348 1 : ret = ACL_ERROR_INTERNAL_ERROR;
349 1 : break;
350 : }
351 :
352 : std::map<std::string, aclDataType>::const_iterator iter =
353 0 : aclDataTypeStrMap.find(itemVec[i].tensorType_);
354 0 : if (iter == aclDataTypeStrMap.cend()) {
355 0 : ACL_LOG_INNER_ERROR("[Deserialize][TensorDataset]TensorDatasetDeserializes failed, "
356 : "unknown data type[%s]", itemVec[i].tensorType_.c_str());
357 0 : ret = ACL_ERROR_INTERNAL_ERROR;
358 0 : break;
359 : }
360 0 : aclDataType dataType = iter->second;
361 : acltdtDataItem *item = new(std::nothrow) acltdtDataItem(aclType,
362 0 : &dims[0], dims.size(), itemVec[i].tensorShape_,
363 0 : dataType, itemVec[i].tensorType_,
364 0 : itemVec[i].dataPtr_, itemVec[i].dataLen_);
365 0 : if (item == nullptr) {
366 0 : ACL_LOG_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
367 0 : std::string sizeStr = std::to_string(sizeof(acltdtDataItem));
368 0 : acl::AclErrorLogManager::ReportInputError(acl::ALLOC_MEMORY_FAILED_MSG,
369 0 : std::vector<const char *>({"buf_size", "alloc_interface"}),
370 0 : std::vector<const char *>({sizeStr.c_str(), "new"}));
371 0 : ret = ACL_ERROR_BAD_ALLOC;
372 0 : break;
373 0 : }
374 0 : dataset->blobs.push_back(item);
375 1 : } else {
376 : acltdtDataItem *item = new(std::nothrow) acltdtDataItem(aclType,
377 2 : nullptr, 0, itemVec[i].tensorShape_, ACL_DT_UNDEFINED,
378 1 : itemVec[i].tensorType_, itemVec[i].dataPtr_, itemVec[i].dataLen_);
379 1 : if (item == nullptr) {
380 0 : ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
381 0 : ret = ACL_ERROR_BAD_ALLOC;
382 0 : break;
383 : }
384 1 : dataset->blobs.push_back(item);
385 : }
386 : }
387 :
388 4 : if (ret != ACL_SUCCESS) {
389 2 : for (size_t i = 0; i < dataset->blobs.size(); ++i) {
390 0 : ACL_DELETE_AND_SET_NULL(dataset->blobs[i]);
391 : }
392 2 : dataset->blobs.clear();
393 : }
394 4 : dataset->freeSelf = true;
395 4 : return ret;
396 : }
397 :
398 4 : aclError TensorDatasetDeserializesV2(const std::vector<acl::aclTdtDataItemInfo> &itemVec, acltdtDataset *dataset)
399 : {
400 4 : ACL_REQUIRES_NOT_NULL(dataset);
401 4 : if (!dataset->blobs.empty() && !dataset->freeSelf) {
402 1 : const std::string sizeVal = std::to_string(dataset->blobs.size());
403 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
404 2 : std::vector<const char *>({"func", "value", "param", "reason"}),
405 1 : std::vector<const char *>({"tdt data deserialization", sizeVal.c_str(), "dataset->blobs.size",
406 2 : "dataset must be empty or freeSelf must be true before deserialization"}));
407 1 : return ACL_ERROR_INVALID_PARAM;
408 1 : }
409 3 : for (auto it = dataset->blobs.begin(); it != dataset->blobs.end(); ++it) {
410 0 : ACL_DELETE_AND_SET_NULL(*it);
411 : }
412 3 : dataset->blobs.clear();
413 3 : aclError ret = ACL_SUCCESS;
414 6 : for (size_t i = 0; i < itemVec.size(); ++i) {
415 : acltdtTensorType aclType;
416 3 : ret = GetAclTypeByTdtDataTypeV2(itemVec[i].ctrlInfo.dataType, aclType);
417 3 : if (ret != ACL_SUCCESS) {
418 0 : ACL_LOG_INNER_ERROR("[Check][Dataset]Failed to convert TdtDataType %d to acltdtTensorType.",
419 : itemVec[i].ctrlInfo.dataType);
420 0 : break;
421 : }
422 3 : if ((aclType == ACL_TENSOR_DATA_TENSOR) || (aclType == ACL_TENSOR_DATA_SLICE_TENSOR)
423 1 : || (aclType == ACL_TENSOR_DATA_END_TENSOR)) {
424 2 : if (itemVec[i].ctrlInfo.version == VERSION_NAME) {
425 1 : void *dataReal = (itemVec[i].priorityDataPtr_ != nullptr) ?
426 1 : itemVec[i].priorityDataPtr_ : itemVec[i].dataPtr.get();
427 1 : dataset->name.assign(static_cast<char *>(dataReal), itemVec[i].ctrlInfo.dataLen);
428 1 : ACL_LOG_INFO("get dataset name is %s", dataset->name.c_str());
429 1 : continue;
430 1 : }
431 1 : std::vector<int64_t> dims = itemVec[i].dims;
432 1 : aclDataType dataType = static_cast<aclDataType>(itemVec[i].ctrlInfo.tensorType);
433 : acltdtDataItem *item = new(std::nothrow) acltdtDataItem(aclType,
434 1 : &dims[0], dims.size(), "",
435 : dataType, "",
436 6 : itemVec[i].dataPtr, itemVec[i].ctrlInfo.dataLen);
437 1 : if (item == nullptr) {
438 0 : ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
439 0 : ret = ACL_ERROR_BAD_ALLOC;
440 0 : break;
441 : }
442 1 : item->sliceNum = itemVec[i].ctrlInfo.sliceNum;
443 1 : item->sliceId = itemVec[i].ctrlInfo.sliceId;
444 1 : item->priorityData_ = itemVec[i].priorityDataPtr_;
445 1 : dataset->blobs.push_back(item);
446 2 : } else {
447 : acltdtDataItem *item = new(std::nothrow) acltdtDataItem(aclType,
448 : nullptr, 0, "", ACL_DT_UNDEFINED,
449 5 : "", itemVec[i].dataPtr, itemVec[i].ctrlInfo.dataLen);
450 1 : if (item == nullptr) {
451 0 : ACL_LOG_INNER_ERROR("[Check][Item]TensorDatasetDeserializes alloc failed");
452 0 : ret = ACL_ERROR_BAD_ALLOC;
453 0 : break;
454 : }
455 1 : item->priorityData_ = itemVec[i].priorityDataPtr_;
456 1 : dataset->blobs.push_back(item);
457 : }
458 : }
459 :
460 3 : if (ret != ACL_SUCCESS) {
461 0 : for (size_t i = 0; i < dataset->blobs.size(); ++i) {
462 0 : ACL_DELETE_AND_SET_NULL(dataset->blobs[i]);
463 : }
464 0 : dataset->blobs.clear();
465 : }
466 3 : dataset->freeSelf = true;
467 3 : return ret;
468 : }
469 :
470 28 : void GetTensorDimsString(const int64_t *dims, size_t dimNum, std::string &dimsStr)
471 : {
472 91 : for (size_t i = 0; i < dimNum; ++i) {
473 84 : dimsStr += std::to_string(dims[i]);
474 84 : if (i + 1 == dimNum) {
475 21 : break;
476 : }
477 63 : dimsStr.push_back(',');
478 : }
479 28 : dimsStr += "]";
480 28 : }
481 :
482 6 : aclError SaveCtrlSharedPtrToVec(const datasetMemType memType, rtMemQueueBuffInfo &qItem,
483 : const std::shared_ptr<uint8_t> &ctrlSharedPtr, std::vector<std::shared_ptr<uint8_t>> &ctrlSharedPtrVec)
484 : {
485 6 : void *ctrlPtr = ctrlSharedPtr.get();
486 6 : if (memType == MEM_DEVICE) {
487 3 : uint8_t *devPtr = nullptr;
488 3 : std::shared_ptr<uint8_t> ctrlSharedDevPtr;
489 3 : ctrlSharedDevPtr.reset(devPtr, [](void *p) {
490 3 : if (p != nullptr) {
491 0 : (void)rtFree(p);
492 : }
493 3 : });
494 3 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
495 : rtMalloc(reinterpret_cast<void **>(&devPtr), qItem.len, RT_MEMORY_DEFAULT, acl::ACL_MODE_ID_U16),
496 : rtMalloc);
497 3 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(
498 : rtMemcpy(devPtr, qItem.len, ctrlPtr, qItem.len, RT_MEMCPY_HOST_TO_DEVICE), rtMemcpy);
499 3 : qItem.addr = devPtr;
500 3 : ctrlSharedPtrVec.push_back(ctrlSharedDevPtr);
501 3 : } else {
502 3 : qItem.addr = ctrlPtr;
503 3 : ctrlSharedPtrVec.push_back(ctrlSharedPtr);
504 : }
505 6 : return ACL_SUCCESS;
506 : }
507 :
508 4 : aclError UnpackageRecvDataInfo(uint8_t *outputHostAddr, size_t size, std::vector<acl::aclTdtDataItemInfo> &itemVec)
509 : {
510 4 : ItemInfo *head = reinterpret_cast<ItemInfo *>(outputHostAddr);
511 4 : uint32_t cnt = head->cnt;
512 4 : ACL_LOG_INFO("get tensor cnt is %u", cnt);
513 4 : size_t offset = 0;
514 5 : for (uint32_t i = 0; i < cnt; ++i) {
515 3 : if (offset + sizeof(ItemInfo) > size) {
516 1 : ACL_LOG_ERROR("offset is %zu, size is %zu", offset, size);
517 2 : return ACL_ERROR_FAILURE;
518 : }
519 2 : acl::aclTdtDataItemInfo item;
520 2 : ItemInfo *tmp = reinterpret_cast<ItemInfo *>(outputHostAddr + offset);
521 2 : item.ctrlInfo = *tmp;
522 2 : ACL_LOG_INFO("UnpackInfo version %d, dataType %d, curCnt %u, cnt %u, tensorType %d, dimNum %u, "
523 : "dynamicBitSize %u, sliceNum %u, sliceId %u, dataLen %lu", tmp->version, tmp->dataType, tmp->curCnt,
524 : tmp->cnt, tmp->tensorType, tmp->dimNum, tmp->dynamicBitSize, static_cast<uint32_t>(tmp->sliceNum),
525 : static_cast<uint32_t>(tmp->sliceId), tmp->dataLen);
526 2 : offset += sizeof(ItemInfo);
527 :
528 3 : for (uint32_t j = 0; j < tmp->dimNum; ++j) {
529 2 : if (offset + sizeof(int64_t) > size) {
530 1 : ACL_LOG_ERROR("offset is %zu, size is %zu", offset, size);
531 1 : return ACL_ERROR_FAILURE;
532 : }
533 1 : int64_t dimTmp = *(reinterpret_cast<int64_t *>(outputHostAddr + offset));
534 1 : item.dims.push_back(dimTmp);
535 1 : ACL_LOG_INFO("current dims[%u] is %ld", j, dimTmp);
536 1 : offset += sizeof(int64_t);
537 : }
538 :
539 1 : if (offset + tmp->dataLen > size) {
540 0 : ACL_LOG_ERROR("offset is %zu, data len is %lu, size is %zu", offset, tmp->dataLen, size);
541 0 : return ACL_ERROR_FAILURE;
542 : }
543 1 : if (tmp->dataLen > 0U) {
544 1 : item.priorityDataPtr_ = outputHostAddr + offset;
545 1 : offset += tmp->dataLen;
546 : } else {
547 0 : ACL_LOG_INFO("data length is 0");
548 : }
549 1 : ACL_LOG_INFO("after %u tensor, offset is %zu", i + 1, offset);
550 1 : itemVec.push_back(item);
551 2 : }
552 2 : return ACL_SUCCESS;
553 : }
554 :
555 6 : aclError TensorDataitemSerialize(std::vector<acl::aclTdtDataItemInfo> &itemVec, const datasetMemType memType,
556 : std::vector<rtMemQueueBuffInfo> &qBufVec, std::vector<std::shared_ptr<uint8_t>> &ctrlSharedPtrVec)
557 : {
558 6 : uint32_t currentCnt = 0;
559 6 : size_t lastDataSize = 0U;
560 12 : for (size_t i = 0; i < itemVec.size(); ++i) {
561 6 : itemVec[i].ctrlInfo.curCnt = currentCnt;
562 6 : itemVec[i].ctrlInfo.cnt = itemVec.size();
563 6 : size_t ctrlSize = sizeof(ItemInfo) + itemVec[i].dims.size() * sizeof(int64_t);
564 : // 64n + lastDataSize + 64n - lastDataSize
565 6 : size_t alignedSize = Get64AlignedSize(ctrlSize + lastDataSize) - lastDataSize;
566 6 : itemVec[i].ctrlInfo.dynamicBitSize = alignedSize - sizeof(ItemInfo);
567 : std::shared_ptr<uint8_t> ctrlSharedPtr(
568 6 : new (std::nothrow) uint8_t[alignedSize], std::default_delete<uint8_t[]>());
569 6 : ACL_CHECK_MALLOC_RESULT_REPORT_RET(ctrlSharedPtr.get(), alignedSize, "new", ACL_ERROR_BAD_ALLOC);
570 6 : void *ctrlPtr = ctrlSharedPtr.get();
571 6 : ACL_LOG_DEBUG("TensorDataitemSerialize alignedSize is %zu, ctrlSize is %zu, dynamicBitSize is %u, i is %zu,"
572 : " lastDataSize is %zu, shape size is %zu", alignedSize, ctrlSize, itemVec[i].ctrlInfo.dynamicBitSize,
573 : i, lastDataSize, itemVec[i].dims.size());
574 6 : auto memcpyRet = memcpy_s(ctrlPtr, alignedSize, &itemVec[i].ctrlInfo, sizeof(ItemInfo));
575 6 : if (memcpyRet != EN_OK) {
576 0 : ACL_LOG_INNER_ERROR("[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%zu",
577 : memcpyRet, sizeof(ItemInfo), alignedSize);
578 : }
579 6 : size_t offset = sizeof(ItemInfo);
580 30 : for (size_t j = 0; j < itemVec[i].dims.size(); ++j) {
581 24 : memcpyRet = memcpy_s(reinterpret_cast<uint8_t *>(ctrlPtr) + offset,
582 24 : alignedSize - offset, &itemVec[i].dims[j], sizeof(int64_t));
583 24 : if (memcpyRet != EN_OK) {
584 0 : ACL_LOG_INNER_ERROR("[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%zu",
585 : memcpyRet, sizeof(int64_t), alignedSize - offset);
586 : }
587 24 : offset += sizeof(int64_t);
588 : }
589 6 : rtMemQueueBuffInfo qItem = {};
590 6 : qItem.len = alignedSize;
591 6 : ACL_REQUIRES_OK(SaveCtrlSharedPtrToVec(memType, qItem, ctrlSharedPtr, ctrlSharedPtrVec));
592 6 : qBufVec.push_back(qItem);
593 :
594 6 : if (itemVec[i].ctrlInfo.dataLen > 0U) {
595 6 : rtMemQueueBuffInfo tmpQItem = {itemVec[i].dataPtr.get(), itemVec[i].ctrlInfo.dataLen};
596 6 : qBufVec.push_back(tmpQItem);
597 : } else {
598 0 : ACL_LOG_DEBUG("no need to insert data buf");
599 : }
600 : // current total size is (64n + lastDataSize)
601 6 : lastDataSize = itemVec[i].ctrlInfo.dataLen;
602 6 : ++currentCnt;
603 6 : }
604 6 : return ACL_SUCCESS;
605 : }
606 :
607 8 : aclError acltdtSendTensorV2(const acltdtChannelHandle *handle, const acltdtDataset *dataset, int32_t timeout)
608 : {
609 8 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
610 8 : std::vector<acl::aclTdtDataItemInfo> itemVec;
611 8 : auto ret = acl::TensorDatasetSerializesV2(dataset, itemVec);
612 8 : if (ret != ACL_SUCCESS) {
613 2 : ACL_LOG_INNER_ERROR("[Serialize][Dataset]Failed to TensorDatasetSerializesV2, device is %u, name is %s.",
614 : handle->devId, handle->name.c_str());
615 2 : itemVec.clear();
616 2 : return ret;
617 : }
618 6 : std::vector<std::shared_ptr<uint8_t>> ctrlSharedPtrVec;
619 6 : std::vector<rtMemQueueBuffInfo> queueBufInfoVec;
620 6 : ret = acl::TensorDataitemSerialize(itemVec, dataset->memType, queueBufInfoVec, ctrlSharedPtrVec);
621 6 : if (ret != ACL_SUCCESS) {
622 0 : ACL_LOG_INNER_ERROR("[Serialize][Dataset]Failed to TensorDataitemSerialize, device is %u, name is %s.",
623 : handle->devId, handle->name.c_str());
624 0 : return ret;
625 : }
626 :
627 6 : rtMemQueueBuff_t queueBuf = {nullptr, 0U, nullptr, 0U};
628 6 : queueBuf.buffCount = queueBufInfoVec.size();
629 6 : queueBuf.buffInfo = queueBufInfoVec.data();
630 6 : ret = rtMemQueueEnQueueBuff(handle->devId, handle->qid, &queueBuf, timeout);
631 6 : if (ret == ACL_ERROR_RT_QUEUE_FULL) {
632 2 : ACL_LOG_DEBUG("queue is full, device is %u, name is %s", handle->devId, handle->name.c_str());
633 2 : return ret;
634 : }
635 4 : if (ret != RT_ERROR_NONE) {
636 2 : return ret;
637 : }
638 2 : ACL_LOG_DEBUG("success to execute acltdtSendTensor, device is %u, name is %s",
639 : handle->devId, handle->name.c_str());
640 2 : return ACL_SUCCESS;
641 8 : }
642 :
643 11 : aclError EnsureCurrentThreadHasContext(const acltdtChannelHandle *handle)
644 : {
645 11 : rtContext_t rtCtx = nullptr;
646 11 : const rtError_t rtRet = rtCtxGetCurrent(&rtCtx);
647 11 : if ((rtRet != ACL_RT_SUCCESS) && (rtRet != ACL_ERROR_RT_CONTEXT_NULL)) {
648 1 : return rtRet;
649 : }
650 10 : if (rtCtx == nullptr) {
651 10 : if (handle->ctx_ == nullptr) {
652 10 : ACL_LOG_INFO("current thread need to create new context");
653 10 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtCtxCreateEx(&rtCtx, static_cast<uint32_t>(RT_CTX_NORMAL_MODE),
654 : static_cast<int32_t>(handle->devId)), rtCtxCreateEx);
655 10 : const_cast<acltdtChannelHandle *>(handle)->ctx_.reset(rtCtx,
656 10 : [](void *p) {if (p != nullptr) {(void)rtCtxDestroyEx(p);}});
657 : }
658 10 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtCtxSetCurrent(handle->ctx_.get()), rtCtxSetCurrent);
659 : }
660 10 : return ACL_SUCCESS;
661 : }
662 :
663 7 : size_t GetMallocSize(const size_t bufLen)
664 : {
665 : // 超出当前档位就是bufLen, 在档位内就是上限值，并保存当前申请的值
666 21 : for (const size_t& size : GEAR_SIZE) {
667 19 : if (bufLen <= size) {
668 5 : return size;
669 : }
670 : }
671 2 : return bufLen;
672 : }
673 :
674 11 : aclError GetOrMallocHostMem(const acltdtChannelHandle *handle, acltdtDataset *dataset,
675 : size_t bufLen, void *&hostPtr)
676 : {
677 11 : ACL_LOG_INFO("current need size is %zu, current mem size is %zu", bufLen, dataset->sharedMemSize_);
678 11 : ACL_REQUIRES_OK(EnsureCurrentThreadHasContext(handle));
679 10 : if (bufLen > dataset->sharedMemSize_) {
680 7 : const size_t mallocSize = GetMallocSize(bufLen);
681 7 : ACL_LOG_INFO("need mallochost size %zu, bufLen is %zu", mallocSize, bufLen);
682 7 : void *outHostAddr = nullptr;
683 7 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMallocHost(&outHostAddr, mallocSize, acl::ACL_MODE_ID_U16), rtMallocHost);
684 7 : ACL_CHECK_MALLOC_RESULT(outHostAddr);
685 14 : dataset->sharedMem_.reset(outHostAddr, [](void *p) {if (p != nullptr) {(void)rtFreeHost(p);}});
686 7 : dataset->sharedMemSize_ = mallocSize;
687 : }
688 10 : hostPtr = dataset->sharedMem_.get();
689 10 : return ACL_SUCCESS;
690 : }
691 :
692 5 : aclError acltdtReceiveTensorV2(const acltdtChannelHandle *handle, acltdtDataset *dataset, int32_t timeout)
693 : {
694 5 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
695 5 : size_t bufLen = 0;
696 5 : auto ret = rtMemQueuePeek(handle->devId, handle->qid, &bufLen, timeout);
697 5 : if (ret == ACL_ERROR_RT_QUEUE_EMPTY) {
698 1 : ACL_LOG_INFO("queue is empty, device is %u, name is %s", handle->devId, handle->name.c_str());
699 1 : return ret;
700 : }
701 4 : if (ret != RT_ERROR_NONE) {
702 1 : ACL_LOG_ERROR("peek queue [%u] failed", handle->qid);
703 1 : return ret;
704 : }
705 3 : ACL_LOG_INFO("peek queue [%u] success, bufLen is %zu", handle->qid, bufLen);
706 3 : if (bufLen == 0) {
707 0 : ACL_LOG_INNER_ERROR("[Check][bufLen]peek queue len cannot be zero");
708 0 : return ACL_ERROR_FAILURE;
709 : }
710 3 : void *hostPtr = nullptr;
711 3 : ACL_REQUIRES_OK(GetOrMallocHostMem(handle, dataset, bufLen, hostPtr));
712 :
713 3 : rtMemQueueBuff_t queueBuf = {nullptr, 0U, nullptr, 0U};
714 3 : rtMemQueueBuffInfo queueBufInfo = {hostPtr, bufLen};
715 3 : queueBuf.buffCount = 1;
716 3 : queueBuf.buffInfo = &queueBufInfo;
717 3 : ret = rtMemQueueDeQueueBuff(handle->devId, handle->qid, &queueBuf, 0);
718 3 : if (ret == ACL_ERROR_RT_QUEUE_EMPTY) {
719 1 : ACL_LOG_INFO("queue is empty, device is %u, name is %s", handle->devId, handle->name.c_str());
720 1 : return ret;
721 : }
722 2 : if (ret != RT_ERROR_NONE) {
723 1 : ACL_LOG_ERROR("Failed to rtMemQueueDeQueueBuf, device is %u, name is %s.",
724 : handle->devId, handle->name.c_str());
725 1 : return ret;
726 : }
727 :
728 1 : std::vector<acl::aclTdtDataItemInfo> itemVec;
729 1 : ret = acl::UnpackageRecvDataInfo(static_cast<uint8_t *>(hostPtr), bufLen, itemVec);
730 1 : if (ret != ACL_SUCCESS) {
731 0 : ACL_LOG_ERROR("Failed to unpackage received data, device is %u, name is %s.",
732 : handle->devId, handle->name.c_str());
733 0 : return ret;
734 : }
735 1 : ret = acl::TensorDatasetDeserializesV2(itemVec, dataset);
736 1 : if (ret != ACL_SUCCESS) {
737 0 : ACL_LOG_INNER_ERROR("[Deserialize][Dataset]Failed to deserialize tensor dataset, device is %u, name is %s.",
738 : handle->devId, handle->name.c_str());
739 0 : return ret;
740 : }
741 1 : ACL_LOG_INFO("success to execute acltdtReceiveTensorV2, device is %u, name is %s.",
742 : handle->devId, handle->name.c_str());
743 1 : return ACL_SUCCESS;
744 1 : }
745 : } // namespace acl
746 :
747 2 : acltdtTensorType acltdtGetTensorTypeFromItem(const acltdtDataItem *dataItem)
748 : {
749 6 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, ACL_TENSOR_DATA_UNDEFINED);
750 1 : return dataItem->tdtType;
751 : }
752 :
753 2 : aclDataType acltdtGetDataTypeFromItem(const acltdtDataItem *dataItem)
754 : {
755 6 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, ACL_DT_UNDEFINED);
756 1 : return dataItem->dataType;
757 : }
758 :
759 1 : void *acltdtGetDataAddrFromItem(const acltdtDataItem *dataItem)
760 : {
761 1 : ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(dataItem);
762 1 : if (dataItem->priorityData_ != nullptr) {
763 1 : return dataItem->priorityData_;
764 : }
765 0 : return dataItem->dataPtr.get();
766 : }
767 :
768 2 : size_t acltdtGetDataSizeFromItem(const acltdtDataItem *dataItem)
769 : {
770 6 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, 0);
771 1 : return dataItem->dataLen;
772 : }
773 :
774 2 : size_t acltdtGetDimNumFromItem(const acltdtDataItem *dataItem)
775 : {
776 6 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataItem, 0);
777 1 : return dataItem->dims.size();
778 : }
779 :
780 1 : aclError acltdtGetSliceInfoFromItem(const acltdtDataItem *dataItem, size_t *sliceNum, size_t *sliceId)
781 : {
782 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
783 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(sliceNum);
784 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(sliceId);
785 1 : *sliceNum = dataItem->sliceNum;
786 1 : *sliceId = dataItem->sliceId;
787 1 : return ACL_SUCCESS;
788 : }
789 :
790 5 : aclError acltdtGetDimsFromItem(const acltdtDataItem *dataItem, int64_t *dims, size_t dimNum)
791 : {
792 5 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
793 : // check dims and dimNum
794 5 : if (dims == nullptr && dimNum != 0) {
795 2 : ACL_LOG_ERROR("[Check][Params]acltdtGetDimsFromItem failed, invalid dims and dimNum[%zu]", dimNum);
796 2 : std::string value = "nullptr/" + std::to_string(dimNum);
797 2 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
798 4 : std::vector<const char *>({"func", "value", "param", "reason"}),
799 4 : std::vector<const char *>({__func__, value.c_str(), "dims/dimNum", "If dims is nullptr, dimNum should be 0"}));
800 2 : return ACL_ERROR_INVALID_PARAM;
801 2 : }
802 :
803 3 : if (dims != nullptr && dimNum == 0) {
804 1 : ACL_LOG_ERROR("[Check][Params]acltdtGetDimsFromItem failed, invalid dims and dimNum[%zu]", dimNum);
805 1 : std::string value = std::to_string(*dims) + "/" + std::to_string(dimNum);
806 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
807 2 : std::vector<const char *>({"func", "value", "param", "reason"}),
808 2 : std::vector<const char *>({__func__, value.c_str(), "dims/dimNum", "If dims is not nullptr, dimNum should be greater than 0"}));
809 1 : return ACL_ERROR_INVALID_PARAM;
810 1 : }
811 :
812 2 : if (dimNum < dataItem->dims.size()) {
813 1 : ACL_LOG_ERROR("[Check][dimNum]output dimNum[%zu] cannot be less than dims number[%zu]",
814 : dimNum, dataItem->dims.size());
815 1 : const std::string dimNumVal = std::to_string(dimNum);
816 : std::string errMsg = acl::AclErrorLogManager::FormatStr("dimNum %zu cannot be less than the size of dataItem's dims %zu",
817 1 : dimNum, dataItem->dims.size());
818 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
819 2 : std::vector<const char *>({"func", "value", "param", "reason"}),
820 2 : std::vector<const char *>({__func__, dimNumVal.c_str(), "dimNum", errMsg.c_str()}));
821 1 : return ACL_ERROR_INVALID_PARAM;
822 1 : }
823 :
824 5 : for (size_t i = 0; i < dataItem->dims.size(); ++i) {
825 4 : dims[i] = dataItem->dims[i];
826 : }
827 :
828 1 : return ACL_SUCCESS;
829 : }
830 :
831 3 : const char *acltdtGetDatasetName(const acltdtDataset *dataset)
832 : {
833 7 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataset, nullptr);
834 2 : return dataset->name.c_str();
835 : }
836 :
837 32 : acltdtDataItem *acltdtCreateDataItem(acltdtTensorType tdtType,
838 : const int64_t *dims, size_t dimNum, aclDataType dataType, void *data, size_t size)
839 : {
840 32 : if (dims == nullptr && dimNum != 0) {
841 0 : ACL_LOG_ERROR("[Check][Params]acltdtCreateDataItem failed, invalid dims and dimNum[%zu]", dimNum);
842 0 : std::string value = "nullptr/" + std::to_string(dimNum);
843 0 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
844 0 : std::vector<const char *>({"func", "value", "param", "reason"}),
845 0 : std::vector<const char *>({__func__, value.c_str(), "dims/dimNum", "If dims is nullptr, dimNum should be 0"}));
846 0 : return nullptr;
847 0 : }
848 :
849 32 : if (dims != nullptr && dimNum == 0) {
850 2 : ACL_LOG_ERROR("[Check][Params]acltdtCreateDataItem failed, invalid dims and dimNum[%zu]", dimNum);
851 2 : std::string value = std::to_string(*dims) + "/" + std::to_string(dimNum);
852 2 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
853 4 : std::vector<const char *>({"func", "value", "param", "reason"}),
854 2 : std::vector<const char *>({__func__, value.c_str(),
855 4 : "dims/dimNum", "If dims is not nullptr, dimNum should be greater than 0"}));
856 2 : return nullptr;
857 2 : }
858 :
859 30 : constexpr size_t MAX_DIM_CNT = 128UL;
860 30 : if (dimNum > MAX_DIM_CNT) {
861 1 : ACL_LOG_ERROR("[Check][Dimnum]acltdtCreateDataItem failed, dimNum[%zu] can't be larger than "
862 : "MAX_DIM_CNT[%zu]", dimNum, MAX_DIM_CNT);
863 1 : std::string expect = "less than or equal to " + std::to_string(MAX_DIM_CNT);
864 1 : const std::string dimNumVal = std::to_string(dimNum);
865 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
866 2 : std::vector<const char *>({"func", "value", "param", "expect"}),
867 2 : std::vector<const char *>({__func__, dimNumVal.c_str(), "dimNum", expect.c_str()}));
868 1 : return nullptr;
869 1 : }
870 :
871 29 : if (tdtType != ACL_TENSOR_DATA_TENSOR) {
872 1 : if (dims != nullptr) {
873 1 : ACL_LOG_ERROR("[Check][Dims]acltdtCreateDataItem failed, "
874 : "dims must be nullptr. tdtType is %d", tdtType);
875 1 : const std::string dimsVal = std::to_string(reinterpret_cast<uintptr_t>(dims));
876 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_VALUE_MSG,
877 2 : std::vector<const char *>({"func", "value", "param", "expect"}),
878 2 : std::vector<const char *>({__func__, dimsVal.c_str(), "dims", "nullptr"}));
879 1 : return nullptr;
880 1 : }
881 0 : return new(std::nothrow) acltdtDataItem(tdtType, dims, dimNum, "[]", ACL_DT_UNDEFINED, "", nullptr, 0);
882 : }
883 :
884 : // tdtType: ACL_TENSOR_DATA_TENSOR
885 28 : std::string dimsStr = "[";
886 28 : acl::GetTensorDimsString(dims, dimNum, dimsStr);
887 :
888 28 : std::string typeStr;
889 184 : for (const auto &item: aclDataTypeStrMap) {
890 184 : if (item.second == dataType) {
891 28 : typeStr = item.first;
892 28 : break;
893 : }
894 : }
895 28 : std::shared_ptr<void> dataPtr;
896 28 : dataPtr.reset(data, [](const void *) {});
897 28 : return new(std::nothrow) acltdtDataItem(tdtType, dims, dimNum, dimsStr, dataType, typeStr, dataPtr, size);
898 28 : }
899 :
900 32 : aclError acltdtDestroyDataItem(acltdtDataItem *dataItem)
901 : {
902 32 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
903 31 : ACL_DELETE_AND_SET_NULL(dataItem);
904 31 : return ACL_SUCCESS;
905 : }
906 :
907 36 : acltdtDataset *acltdtCreateDataset()
908 : {
909 36 : return new(std::nothrow) acltdtDataset();
910 : }
911 :
912 36 : aclError acltdtDestroyDataset(acltdtDataset *dataset)
913 : {
914 36 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
915 36 : ACL_DELETE_AND_SET_NULL(dataset);
916 36 : return ACL_SUCCESS;
917 : }
918 :
919 33 : aclError acltdtAddDataItem(acltdtDataset *dataset, acltdtDataItem *dataItem)
920 : {
921 33 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
922 33 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataItem);
923 32 : if (dataset->freeSelf) {
924 1 : acl::AclErrorLogManager::ReportInputError(acl::UNSUPPORTED_FEATURE_MSG,
925 2 : std::vector<const char *>({"feature", "reason"}),
926 2 : std::vector<const char *>({__func__, "item cannot be added because internal item already exists"}));
927 1 : return ACL_ERROR_FEATURE_UNSUPPORTED;
928 : }
929 31 : datasetMemType currentMemType = MEM_UNKNOWN;
930 31 : if (dataItem->dataPtr != nullptr) {
931 23 : rtPtrAttributes_t attr = {};
932 23 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtsPointerGetAttributes(dataItem->dataPtr.get(), &attr), rtsPointerGetAttributes);
933 23 : if ((attr.location.type == RT_MEMORY_LOC_HOST) || (attr.location.type == RT_MEMORY_LOC_UNREGISTERED)) {
934 13 : currentMemType = MEM_HOST;
935 : } else {
936 10 : currentMemType = MEM_DEVICE;
937 : }
938 : }
939 31 : if (dataset->memType == MEM_UNKNOWN) {
940 : // only MEM_UNKNOWN status can be refreshed
941 19 : dataset->memType = currentMemType;
942 : }
943 :
944 31 : if ((dataset->memType != MEM_UNKNOWN) && (currentMemType != MEM_UNKNOWN)) {
945 23 : if (dataset->memType != currentMemType) {
946 6 : ACL_LOG_ERROR("The memTypes in the dataset must be all host-side or device-side address");
947 6 : const std::string memTypeVal = acl::DatasetMemTypeToString(dataset->memType);
948 6 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
949 12 : std::vector<const char *>({"func", "value", "param", "reason"}),
950 6 : std::vector<const char *>({__func__, memTypeVal.c_str(),
951 12 : "dataset->memType", "The memTypes in the dataset must be all host-side or device-side address"}));
952 6 : return ACL_ERROR_INVALID_PARAM;
953 6 : }
954 : }
955 25 : dataset->blobs.push_back(dataItem);
956 25 : return ACL_SUCCESS;
957 : }
958 :
959 6 : acltdtDataItem *acltdtGetDataItem(const acltdtDataset *dataset, size_t index)
960 : {
961 10 : ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(dataset);
962 5 : if (index >= dataset->blobs.size()) {
963 : std::string errMsg =
964 4 : acl::AclErrorLogManager::FormatStr("index %zu is greater than or equal to dataset size %zu", index, dataset->blobs.size());
965 4 : const std::string indexVal = std::to_string(index);
966 4 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
967 8 : std::vector<const char *>({"func", "value", "param", "reason"}),
968 8 : std::vector<const char *>({__func__, indexVal.c_str(), "index", errMsg.c_str()}));
969 4 : return nullptr;
970 4 : }
971 :
972 1 : return dataset->blobs[index];
973 : }
974 :
975 3 : size_t acltdtGetDatasetSize(const acltdtDataset *dataset)
976 : {
977 7 : ACL_REQUIRES_NOT_NULL_RET_INPUT_REPORT(dataset, 0);
978 2 : return dataset->blobs.size();
979 : }
980 :
981 19 : acltdtChannelHandle *acltdtCreateChannel(uint32_t deviceId, const char *name)
982 : {
983 19 : ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(name);
984 21 : static TdtHostInitFunc tdtHostInit = (TdtHostInitFunc)GetFunction("TdtHostInit");
985 19 : if (tdtHostInit == nullptr) {
986 0 : return nullptr;
987 : }
988 19 : auto ret = tdtHostInit(deviceId);
989 19 : if (ret != 0) {
990 1 : ACL_LOG_INNER_ERROR("[Init][Tdt]tdt host init failed, tdt result = %d", ret);
991 1 : return nullptr;
992 : }
993 18 : acltdtChannelHandle *handle = new(std::nothrow) acltdtChannelHandle(deviceId, name);
994 18 : if (handle != nullptr) {
995 18 : if (!handle->recvName.empty()) {
996 3 : static TdtHostPreparePopDataFunc tdtHostPreparePopData = (TdtHostPreparePopDataFunc)GetFunction("TdtHostPreparePopData");
997 1 : if (tdtHostPreparePopData == nullptr) {
998 0 : return nullptr;
999 : }
1000 1 : (void)tdtHostPreparePopData();
1001 : }
1002 : {
1003 18 : std::unique_lock<std::mutex> lk(aclChannleMutex);
1004 36 : aclChannleMap[name] = handle;
1005 18 : }
1006 : }
1007 18 : return handle;
1008 : }
1009 :
1010 9 : acltdtChannelHandle *acltdtCreateChannelWithCapacity(uint32_t deviceId, const char *name, size_t capacity)
1011 : {
1012 9 : ACL_REQUIRES_NOT_NULL_RET_NULL_INPUT_REPORT(name);
1013 9 : ACL_LOG_INFO("acltdtCreateChannelWithCapacity devId is %u, name is %s, capacity is %zu", deviceId, name, capacity);
1014 9 : if (strlen(name) + 1 > RT_MQ_MAX_NAME_LEN) {
1015 1 : ACL_LOG_ERROR("name [%s] length %zu cannot be larger than %d", name, (strlen(name) + 1U), RT_MQ_MAX_NAME_LEN);
1016 : std::string errMsg =
1017 1 : acl::AclErrorLogManager::FormatStr("name [%s] length %zu cannot be larger than %d", name, (strlen(name) + 1U), RT_MQ_MAX_NAME_LEN);
1018 1 : acl::AclErrorLogManager::ReportInputError(acl::INVALID_PARAM_REASON_MSG,
1019 2 : std::vector<const char *>({"func", "value", "param", "reason"}),
1020 2 : std::vector<const char *>({__func__, name, "name", errMsg.c_str()}));
1021 1 : return nullptr;
1022 1 : }
1023 8 : acltdtChannelHandle *handle = new(std::nothrow) acltdtChannelHandle(deviceId, name);
1024 8 : ACL_CHECK_MALLOC_RESULT_REPORT_RET(handle, sizeof(acltdtChannelHandle), "new", nullptr);
1025 8 : handle->isTdtProcess = false;
1026 8 : acltdtQueueAttr attr{};
1027 8 : const size_t count = strlen(name) + 1U;
1028 8 : const auto ret = memcpy_s(attr.name, RT_MQ_MAX_NAME_LEN, name, count);
1029 8 : if (ret != EN_OK) {
1030 0 : const std::string retCode = std::to_string(ret);
1031 0 : std::stringstream ss;
1032 0 : ss << std::hex << "name=0x" << reinterpret_cast<uintptr_t>(name)
1033 0 : << ", dest=0x" << reinterpret_cast<uintptr_t>(attr.name)
1034 0 : << std::dec << ", dest_max=" << RT_MQ_MAX_NAME_LEN << ", count=" << count << ".";
1035 0 : const std::string extendInfo = ss.str();
1036 0 : acl::AclErrorLogManager::ReportInputError(acl::STANDARD_FUNC_FAILED_MSG,
1037 0 : std::vector<const char *>({"func1", "func2", "ret_code", "reason", "extend_info"}),
1038 0 : std::vector<const char *>({__func__, "memcpy_s", retCode.c_str(),
1039 0 : strerror(ret), extendInfo.c_str()}));
1040 0 : ACL_LOG_ERROR("[Call][MemCpy]call memcpy failed, result=%d, srcLen=%zu, dstLen=%d",
1041 : ret, count, RT_MQ_MAX_NAME_LEN);
1042 0 : ACL_DELETE_AND_SET_NULL(handle);
1043 0 : return nullptr;
1044 0 : }
1045 8 : attr.depth = static_cast<uint32_t>(capacity);
1046 8 : attr.workMode = RT_MQ_MODE_DEFAULT;
1047 8 : attr.flowCtrlFlag = false;
1048 8 : attr.flowCtrlDropTime = 0;
1049 8 : attr.overWriteFlag = false;
1050 : // queue init should be invoked when device is open
1051 8 : auto rtError = rtMemQueueInit(deviceId);
1052 8 : if (rtError == ACL_ERROR_RT_FEATURE_NOT_SUPPORT) {
1053 1 : ACL_LOG_INFO("queue init failed due to runtime does not support.");
1054 1 : ACL_DELETE_AND_SET_NULL(handle);
1055 1 : return nullptr;
1056 : }
1057 7 : if ((rtError != RT_ERROR_NONE) && (rtError != ACL_ERROR_RT_REPEATED_INIT)) {
1058 1 : ACL_LOG_INNER_ERROR("queue init failed, rtError is %d", rtError);
1059 1 : ACL_DELETE_AND_SET_NULL(handle);
1060 1 : return nullptr;
1061 : }
1062 6 : const rtError_t rtErr = rtMemQueueCreate(static_cast<int32_t>(deviceId), &attr, &handle->qid);
1063 6 : if (rtErr != RT_ERROR_NONE) {
1064 1 : ACL_DELETE_AND_SET_NULL(handle);
1065 1 : return nullptr;
1066 : }
1067 5 : ACL_LOG_INFO("acltdtCreateChannelWithCapacity devId is %u, name is %s, real name is %s, qid is %u",
1068 : deviceId, handle->name.c_str(), name, handle->qid);
1069 5 : return handle;
1070 : }
1071 :
1072 2 : aclError acltdtStopChannel(acltdtChannelHandle *handle)
1073 : {
1074 2 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1075 2 : ACL_LOG_INFO("start to acltdtStopChannel, device is %u, name is %s",
1076 : handle->devId, handle->name.c_str());
1077 2 : if (!handle->isTdtProcess) {
1078 0 : ACL_LOG_INFO("new process , stop channel is no use");
1079 0 : return ACL_SUCCESS;
1080 : }
1081 2 : if (!handle->recvName.empty()) {
1082 3 : static TdtHostStopFunc tdtHostStop = (TdtHostStopFunc)GetFunction("TdtHostStop");
1083 1 : if (tdtHostStop == nullptr) {
1084 0 : return ACL_ERROR_FAILURE;
1085 : }
1086 1 : auto ret = tdtHostStop(handle->recvName);
1087 1 : if (ret != 0) {
1088 1 : ACL_LOG_INNER_ERROR("[Init][Tdt]tdt host stop failed for channel %s, tdt result = %d",
1089 : handle->name.c_str(), ret);
1090 1 : return ACL_ERROR_FAILURE;
1091 : }
1092 : }
1093 1 : ACL_LOG_INFO("acltdtStopChannel success, device is %u, name is %s",
1094 : handle->devId, handle->name.c_str());
1095 1 : return ACL_SUCCESS;
1096 : }
1097 :
1098 23 : aclError acltdtDestroyChannel(acltdtChannelHandle *handle)
1099 : {
1100 23 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1101 23 : ACL_LOG_INFO("start to acltdtDestroyChannel, device is %u, name is %s",
1102 : handle->devId, handle->name.c_str());
1103 23 : if (!handle->isTdtProcess) {
1104 7 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemQueueDestroy(static_cast<int32_t>(handle->devId), handle->qid),
1105 : rtMemQueueDestroy);
1106 7 : ACL_LOG_INFO("acltdtDestroyChannel success, device is %u, name is %s",
1107 : handle->devId, handle->name.c_str());
1108 7 : ACL_DELETE_AND_SET_NULL(handle);
1109 7 : return ACL_SUCCESS;
1110 : }
1111 16 : std::unique_lock<std::mutex> lk(aclChannleMutex);
1112 16 : aclChannleMap.erase(handle->name);
1113 16 : if (aclChannleMap.size() == 0) {
1114 17 : static TdtHostDestroyFunc tdtHostDestroy = (TdtHostDestroyFunc)GetFunction("TdtHostDestroy");
1115 15 : if (tdtHostDestroy == nullptr) {
1116 0 : return ACL_ERROR_FAILURE;
1117 : }
1118 15 : auto ret = tdtHostDestroy();
1119 15 : if (ret != 0) {
1120 1 : ACL_LOG_INNER_ERROR("[Destroy][Tdt]TdtHostDestroy failed, tdt result = %d", ret);
1121 : }
1122 : }
1123 :
1124 16 : ACL_DELETE_AND_SET_NULL(handle);
1125 16 : return ACL_SUCCESS;
1126 16 : }
1127 :
1128 5 : aclError acltdtCleanChannel(acltdtChannelHandle *handle)
1129 : {
1130 5 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1131 4 : ACL_LOG_INFO("start to acltdtCleanChannel, device is %u, name is %s",
1132 : handle->devId, handle->name.c_str());
1133 4 : if (!handle->isTdtProcess) {
1134 2 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemQueueReset(handle->devId, handle->qid), rtMemQueueReset);
1135 1 : ACL_LOG_INFO("acltdtCleanChannel success, device is %u, name is %s",
1136 : handle->devId, handle->name.c_str());
1137 1 : return ACL_SUCCESS;
1138 : }
1139 2 : return ACL_ERROR_FEATURE_UNSUPPORTED;
1140 : }
1141 :
1142 8 : aclError acltdtSendTensor(const acltdtChannelHandle *handle, const acltdtDataset *dataset, int32_t timeout)
1143 : {
1144 8 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1145 7 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
1146 6 : ACL_LOG_DEBUG("start to execute acltdtSendTensor, device is %u, name is %s",
1147 : handle->devId, handle->name.c_str());
1148 6 : if (!handle->isTdtProcess) {
1149 2 : ACL_LOG_DEBUG("new process, use queue process");
1150 2 : return acl::acltdtSendTensorV2(handle, dataset, timeout);
1151 : }
1152 : // -1 represents infinite wait, it is must be -1 now
1153 7 : ACL_CHECK_INVALID_PARAM_WITH_REASON(timeout != -1, timeout,
1154 : "Only never timeout is supported, timeout can only be set to -1");
1155 :
1156 3 : std::vector<tdt::DataItem> itemVec;
1157 3 : auto ret = acl::TensorDatasetSerializes(dataset, itemVec);
1158 3 : if (ret != ACL_SUCCESS) {
1159 0 : ACL_LOG_INNER_ERROR("[Serialize][Dataset]Failed to TensorDatasetSerializes, device is %u, name is %s.",
1160 : handle->devId, handle->name.c_str());
1161 0 : itemVec.clear();
1162 0 : return ret;
1163 : }
1164 :
1165 5 : static TdtHostPushDataFunc tdtHostPushData = (TdtHostPushDataFunc)GetFunction("TdtHostPushData");
1166 3 : if (tdtHostPushData == nullptr) {
1167 0 : return ACL_ERROR_FAILURE;
1168 : }
1169 3 : int32_t sendRet = tdtHostPushData(handle->name, itemVec, 0);
1170 3 : if (sendRet != 0) {
1171 2 : ACL_LOG_INNER_ERROR("[Push][Data]Failed to push data, tdt result = %d, device is %u, name is %s.",
1172 : sendRet, handle->devId, handle->name.c_str());
1173 2 : return ACL_ERROR_FAILURE;
1174 : }
1175 :
1176 1 : ACL_LOG_DEBUG("success to execute acltdtSendTensor, device is %u, name is %s",
1177 : handle->devId, handle->name.c_str());
1178 1 : return ACL_SUCCESS;
1179 3 : }
1180 :
1181 7 : aclError acltdtReceiveTensor(const acltdtChannelHandle *handle, acltdtDataset *dataset, int32_t timeout)
1182 : {
1183 7 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1184 6 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(dataset);
1185 5 : ACL_LOG_INFO("start to execute acltdtReceiveTensor, device is %u, name is %s",
1186 : handle->devId, handle->name.c_str());
1187 5 : if (!handle->isTdtProcess) {
1188 0 : ACL_LOG_INFO("new process, use queue process");
1189 0 : return acl::acltdtReceiveTensorV2(handle, dataset, timeout);
1190 : }
1191 : // -1 represents infinite wait, it is must be -1 now
1192 8 : ACL_CHECK_INVALID_PARAM_WITH_REASON(timeout != -1, timeout,
1193 : "Only never timeout is supported, timeout can only be set to -1");
1194 :
1195 4 : if (handle->recvName.empty()) {
1196 2 : ACL_LOG_ERROR("[Check][Recvname]it is not a receive channel, failed to receive, device is %u, name is %s",
1197 : handle->devId, handle->name.c_str());
1198 2 : return ACL_ERROR_INVALID_PARAM;
1199 : }
1200 :
1201 2 : std::vector<tdt::DataItem> itemVec;
1202 4 : static TdtHostPopDataFunc tdtHostPopData = (TdtHostPopDataFunc)GetFunction("TdtHostPopData");
1203 2 : if (tdtHostPopData == nullptr) {
1204 0 : return ACL_ERROR_FAILURE;
1205 : }
1206 2 : int32_t recvRet = tdtHostPopData(handle->recvName, itemVec);
1207 2 : if (recvRet != 0) {
1208 1 : ACL_LOG_INNER_ERROR("[Pop][Data]Failed to receive, tdt result = %d, device is %u, name is %s.",
1209 : recvRet, handle->devId, handle->name.c_str());
1210 1 : return ACL_ERROR_FAILURE;
1211 : }
1212 :
1213 1 : auto ret = acl::TensorDatasetDeserializes(itemVec, dataset);
1214 1 : if (ret != ACL_SUCCESS) {
1215 0 : ACL_LOG_INNER_ERROR("[Deserialize][Dataset]Failed to TensorDatasetDeserializes, device is %u, name is %s.",
1216 : handle->devId, handle->name.c_str());
1217 0 : return ret;
1218 : }
1219 :
1220 1 : ACL_LOG_INFO("success to execute acltdtReceiveTensor, device is %u, name is %s",
1221 : handle->devId, handle->name.c_str());
1222 1 : return ACL_SUCCESS;
1223 2 : }
1224 :
1225 5 : aclError acltdtQueryChannelSize(const acltdtChannelHandle *handle, size_t *size)
1226 : {
1227 5 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(handle);
1228 4 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(size);
1229 3 : if (handle->isTdtProcess) {
1230 1 : ACL_LOG_DEBUG("acltdtQueryChannelSize is not supported");
1231 1 : return ACL_ERROR_FEATURE_UNSUPPORTED;
1232 : }
1233 2 : ACL_LOG_DEBUG("start to execute acltdtQueryChannelSize, device is %u, qid is %u", handle->devId, handle->qid);
1234 : rtMemQueueInfo_t info;
1235 2 : ACL_REQUIRES_RTS_OK(
1236 : rtMemQueueQueryInfo(static_cast<int32_t>(handle->devId), handle->qid, &info));
1237 1 : *size = static_cast<size_t>(info.size);
1238 1 : ACL_LOG_DEBUG("success to execute acltdtQueryChannelSize, size is %zu, device is %u, qid is %u",
1239 : *size, handle->devId, handle->qid);
1240 1 : return ACL_SUCCESS;
1241 : }
|
