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 "aicpu_ts_sync_data_c_adpt.h"
12 :
13 : #include <chrono>
14 : #include <thread>
15 : #include <atomic>
16 : #include <cmath>
17 :
18 : #include "log.h"
19 : #include "thread.h"
20 : #include "aicpu_ts_thread.h"
21 : #include "externalinput_pub.h"
22 :
23 : constexpr size_t MSG_TAG_SIZE_BYTE = 256;
24 : constexpr uint32_t TIMEOUT_SIZE_BYTE = 4; // timeout字段长度为4字节，表示超时时间，单位为秒
25 : constexpr uint32_t CTRL_HDR_DATA_SIZE_LEN = 8; // size_t 在不同平台上长度不同，取最大值
26 :
27 : // 同步等待超时(秒)，复用 HCCL_EXEC_TIMEOUT 环境变量配置的算子级执行超时，向上取整为秒
28 3 : static uint32_t GetSyncWaitTimeoutSeconds()
29 : {
30 3 : return static_cast<uint32_t>(std::ceil(GetExternalInputHcclExecTimeOut()));
31 : }
32 :
33 : // Msg 数据格式如下（单位：字节）：
34 : // +----------+--------------+-----------+-------------+---------------+-----------------+
35 : // | flag [1] | msgTag [256] | msgId [4] | timeout [4] | data size [8] | data [sizeByte] |
36 : // +----------+--------------+-----------+-------------+---------------+-----------------+
37 : // ^
38 : // handle
39 :
40 3 : static HcclResult WaitFlagReady(uint8_t* srcFlagPtr)
41 : {
42 3 : HCCL_INFO("[%s] Polling flag START.", __func__);
43 3 : const auto timeStart = std::chrono::steady_clock::now();
44 3 : const uint32_t timeoutVal = GetSyncWaitTimeoutSeconds();
45 3 : HCCL_INFO("[%s] Using timeout = %u seconds.", __func__, timeoutVal);
46 3 : auto timeoutSec = std::chrono::seconds(timeoutVal);
47 3 : uint8_t flagReadValue{0};
48 3 : errno_t ret = EOK;
49 : while (true) {
50 63732382 : ret = memcpy_s(&flagReadValue, sizeof(flagReadValue), srcFlagPtr, sizeof(flagReadValue));
51 63732383 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Polling flag ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
52 63732382 : if (flagReadValue == 1) {
53 2 : break;
54 : }
55 : const auto elapsed
56 63732380 : = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - timeStart);
57 63732380 : if (timeoutVal != 0 && elapsed > timeoutSec) {
58 1 : HCCL_ERROR("[%s] Polling flag TIMEOUT, timeout[%u]s.", __func__, timeoutVal);
59 1 : return HCCL_E_TIMEOUT;
60 : }
61 63732379 : }
62 2 : HCCL_INFO("[%s] Polling flag SUCCESS.", __func__);
63 2 : return HCCL_SUCCESS;
64 : }
65 :
66 : #ifdef __cplusplus
67 : extern "C" {
68 : #endif // __cplusplus
69 :
70 2 : int32_t HcommSendRequest(MsgHandle handle, const char* msgTag, const void* src, size_t sizeByte, uint32_t* msgId)
71 : {
72 2 : uint8_t* const dstOnDevShmem = reinterpret_cast<uint8_t*>(handle);
73 2 : CHK_PTR_NULL(dstOnDevShmem);
74 1 : CHK_PTR_NULL(msgTag);
75 1 : CHK_PTR_NULL(src);
76 :
77 1 : HCCL_INFO(
78 : "[%s] START. msgHandle[0x%llx], msgTag[%s], src[0x%llx], sizeByte[%zu].", __func__, handle, msgTag, src,
79 : sizeByte);
80 :
81 : static std::atomic<uint32_t> s_msgId{0};
82 1 : *msgId = s_msgId.fetch_add(1, std::memory_order_relaxed); // msgId 达 UINT32_MAX 后回绕到 0，仅保证唯一性
83 :
84 1 : const uint8_t flagWriteValue{1};
85 1 : uint8_t* const dstFlagPtr = dstOnDevShmem;
86 1 : uint8_t* const dstMsgTagPtr = dstFlagPtr + sizeof(flagWriteValue);
87 1 : uint8_t* const dstMsgIdPtr = dstMsgTagPtr + MSG_TAG_SIZE_BYTE;
88 1 : uint8_t* const dstDataPtr = dstMsgIdPtr + sizeof(*msgId) + TIMEOUT_SIZE_BYTE;
89 1 : errno_t ret = EOK;
90 :
91 1 : HCCL_INFO("[%s] Writing %zu bytes data from src to shared mem START.", __func__, sizeByte);
92 1 : ret = memcpy_s(dstDataPtr, sizeof(sizeByte), &sizeByte, sizeof(sizeByte));
93 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Writing data ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
94 1 : ret = memcpy_s(dstDataPtr + CTRL_HDR_DATA_SIZE_LEN, sizeByte, src, sizeByte);
95 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Writing data ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
96 1 : HCCL_INFO("[%s] Writing %zu bytes data from src to shared mem SUCCESS.", __func__, sizeByte);
97 :
98 1 : HCCL_INFO("[%s] Writing %zu bytes msgId to shared mem START. msgId = %u.", __func__, sizeof(*msgId), *msgId);
99 1 : ret = memcpy_s(dstMsgIdPtr, sizeof(*msgId), msgId, sizeof(*msgId));
100 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Writing msgId ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
101 1 : HCCL_INFO("[%s] Writing %zu bytes msgId to shared mem SUCCESS. msgId = %u.", __func__, sizeof(*msgId), *msgId);
102 :
103 1 : HCCL_INFO("[%s] Writing %zu bytes msgTag to shared mem START.", __func__, MSG_TAG_SIZE_BYTE);
104 1 : ret = memcpy_s(dstMsgTagPtr, MSG_TAG_SIZE_BYTE, msgTag, MSG_TAG_SIZE_BYTE);
105 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Writing msgTag ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
106 1 : HCCL_INFO("[%s] Writing %zu bytes msgTag to shared mem SUCCESS.", __func__, MSG_TAG_SIZE_BYTE);
107 :
108 : #if defined(__aarch64__) || defined(__arm__)
109 : asm volatile("dmb sy" ::: "memory"); // 确保之前的内存写入对其他线程可见
110 : #else
111 1 : asm volatile("" ::: "memory"); // 非 ARM 架构(x86)仅用编译屏障防止重排
112 : #endif
113 :
114 1 : HCCL_INFO("[%s] Setting flag = 1 on shared mem START.", __func__);
115 1 : ret = memcpy_s(dstFlagPtr, sizeof(flagWriteValue), &flagWriteValue, sizeof(flagWriteValue));
116 1 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Setting flag ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
117 1 : HCCL_INFO("[%s] Setting flag = 1 on shared mem SUCCESS.", __func__);
118 :
119 1 : HCCL_INFO("[%s] SUCCESS. msgId[%u].", __func__, *msgId);
120 1 : return HCCL_SUCCESS;
121 : }
122 :
123 3 : int32_t HcommWaitResponse(MsgHandle handle, void* dst, size_t sizeByte, uint32_t* msgId)
124 : {
125 3 : uint8_t* const srcOnDevShmem = reinterpret_cast<uint8_t*>(handle);
126 3 : CHK_PTR_NULL(srcOnDevShmem);
127 3 : if (sizeByte > 0) {
128 2 : CHK_PTR_NULL(dst);
129 : }
130 3 : CHK_PTR_NULL(msgId);
131 :
132 3 : HCCL_INFO("[%s] START. msgHandle[0x%llx], dst[0x%llx], sizeByte[%zu].", __func__, handle, dst, sizeByte);
133 :
134 3 : constexpr size_t sizeByteMsgId = sizeof(uint32_t);
135 3 : uint8_t flagReadValue{0};
136 3 : uint8_t* const srcFlagPtr = srcOnDevShmem;
137 3 : uint8_t* const srcMsgIdPtr = srcFlagPtr + sizeof(flagReadValue) + MSG_TAG_SIZE_BYTE;
138 3 : uint8_t* const srcTimeoutPtr = srcMsgIdPtr + sizeByteMsgId;
139 3 : uint8_t* const srcDataPtr = srcTimeoutPtr + TIMEOUT_SIZE_BYTE;
140 3 : errno_t ret = EOK;
141 :
142 3 : CHK_RET(WaitFlagReady(srcFlagPtr));
143 :
144 2 : if (sizeByte > 0) {
145 2 : HCCL_INFO("[%s] Reading %zu bytes data from shared mem START.", __func__, sizeByte);
146 2 : ret = memcpy_s(dst, sizeByte, srcDataPtr, sizeByte);
147 2 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Reading data ERROR[%d]", __func__, ret), HCCL_E_INTERNAL);
148 2 : HCCL_INFO("[%s] Reading %zu bytes data from shared mem SUCCESS.", __func__, sizeByte);
149 : }
150 :
151 2 : HCCL_INFO("[%s] Reading %zu bytes msgId from shared mem START.", __func__, sizeByteMsgId);
152 2 : ret = memcpy_s(msgId, sizeByteMsgId, srcMsgIdPtr, sizeByteMsgId);
153 2 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memcpy_s] Reading msgId ERROR[%d].", __func__, ret), HCCL_E_INTERNAL);
154 2 : HCCL_INFO("[%s] Reading %zu bytes msgId from shared mem SUCCESS. msgId = %u.", __func__, sizeByteMsgId, *msgId);
155 :
156 2 : HCCL_INFO("[%s] Setting flag = 0 on shared mem START.", __func__);
157 2 : ret = memset_s(srcFlagPtr, sizeof(flagReadValue), 0, sizeof(flagReadValue));
158 2 : CHK_PRT_RET(ret != EOK, HCCL_ERROR("[%s][memset_s] Resetting flag ERROR[%d]", __func__, ret), HCCL_E_INTERNAL);
159 2 : HCCL_INFO("[%s] Setting flag = 0 on shared mem SUCCESS.", __func__);
160 :
161 2 : HCCL_INFO("[%s] SUCCESS. msgId[%u].", __func__, *msgId);
162 2 : return HCCL_SUCCESS;
163 : }
164 :
165 2 : int32_t HcommThreadSynchronize(ThreadHandle thread)
166 : {
167 2 : hccl::Thread* threadPtr = reinterpret_cast<hccl::Thread*>(thread);
168 2 : CHK_PTR_NULL(threadPtr);
169 :
170 1 : HCCL_INFO("[%s] START. thread[0x%llx].", __func__, thread);
171 :
172 1 : if (threadPtr->IsDeviceA5()) {
173 0 : HCCL_INFO("[%s] Running on A5.", __func__);
174 0 : hccl::AicpuTsThread* aicpuTsThreadPtr = dynamic_cast<hccl::AicpuTsThread*>(threadPtr);
175 0 : uint32_t sqHead{0};
176 0 : uint32_t sqTail{0};
177 0 : HCCL_INFO("[%s] Start waiting for RTSQ's head == tail.", __func__);
178 : do {
179 0 : CHK_RET(aicpuTsThreadPtr->GetSqHeadAndTail(sqHead, sqTail));
180 0 : } while (sqHead != sqTail);
181 0 : HCCL_INFO("[%s] SUCCESS. RTSQ's head == tail.", __func__);
182 0 : return HCCL_SUCCESS;
183 : }
184 :
185 1 : HCCL_INFO("[%s] NOT Running on A5. No implementation, return SUCCESS.", __func__);
186 1 : return HCCL_SUCCESS;
187 : }
188 : #ifdef __cplusplus
189 : }
190 : #endif // __cplusplus
|
