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