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 <sys/prctl.h>
12 : #include "securec.h"
13 : #include "user_log.h"
14 : #include "ra_hdc_async.h"
15 : #include "ra_rs_comm.h"
16 : #include "ra_rs_err.h"
17 : #include "ra_adp.h"
18 : #include "ra_adp_pool.h"
19 : #include "ra_adp_async.h"
20 :
21 : struct RaHdcAsyncInfo gHdcAsync[RA_MAX_PHY_ID_NUM] = {0};
22 : struct RaHdcInitPara gHdcAsyncInitPara = {0};
23 : struct RsPthreadInfo gRaAsyncThreadInfo = {0};
24 :
25 119 : int RaHwAsyncInit(unsigned int chipId, pid_t pid)
26 : {
27 : int ret;
28 :
29 119 : ret = pthread_mutex_init(&gHdcAsyncInitPara.mutex, NULL);
30 119 : CHK_PRT_RETURN(ret != 0, hccp_err("g_hdc_async_init_para mutex_init failed ret %d", ret), -ESYSFUNC);
31 :
32 119 : gHdcAsyncInitPara.chipId = chipId;
33 119 : gHdcAsyncInitPara.hostTgid = pid;
34 :
35 119 : ret = pthread_mutex_init(&gHdcAsync[chipId].sendMutex, NULL);
36 119 : if (ret != 0) {
37 0 : hccp_err("send_mutex mutex_init failed ret %d", ret);
38 0 : pthread_mutex_destroy(&gHdcAsyncInitPara.mutex);
39 0 : return -ESYSFUNC;
40 : }
41 119 : RaHdcInitOpSec(&gHdcAsync[chipId].opSec, BUCKET_DEPTH, true);
42 119 : return 0;
43 : }
44 :
45 1 : STATIC int RaHdcHandleSendPkt(unsigned int chipId, void *recvBuf, unsigned int recvLen)
46 : {
47 1 : unsigned int closeSession = 0;
48 1 : void *sendBuf = NULL;
49 1 : int sendLen = 0;
50 : int ret;
51 :
52 1 : RsSetCtx(chipId);
53 :
54 1 : ret = RaHandle(&gHdcAsync[chipId].opSec, recvBuf, recvLen, (char **)&sendBuf, &sendLen, &closeSession);
55 1 : if (ret != 0) {
56 0 : hccp_err("ra_handle failed, ret:%d", ret);
57 0 : goto out;
58 : }
59 :
60 1 : ret = RaHdcAsyncSendPkt(&gHdcAsync[chipId], chipId, sendBuf, sendLen);
61 1 : if (ret != 0) {
62 1 : hccp_err("send_pkt failed, ret:%d", ret);
63 1 : goto err;
64 : }
65 :
66 0 : err:
67 1 : free(sendBuf);
68 1 : sendBuf = NULL;
69 1 : out:
70 1 : return ret;
71 : }
72 :
73 1 : STATIC void RaAsyncHandlePkt(unsigned int chipId, void *recvBuf, unsigned int recvLen)
74 : {
75 1 : struct MsgHead *recvMsgHead = (struct MsgHead *)recvBuf;
76 1 : bool closeSession = false;
77 :
78 : // should handle RA_RS_HDC_SESSION_CLOSE on recv thread
79 1 : if (recvLen < sizeof(struct MsgHead) || recvMsgHead->opcode == RA_RS_HDC_SESSION_CLOSE) {
80 1 : closeSession = true;
81 : }
82 1 : if (closeSession) {
83 1 : (void)RaHdcHandleSendPkt(chipId, recvBuf, recvLen);
84 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
85 1 : gHdcAsyncInitPara.connectStatus = HDC_UNCONNECTED;
86 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
87 1 : return;
88 : }
89 :
90 : // handle other opcode: generate task and process the msg with work thread
91 0 : RaHdcPoolAddTask(gHdcAsync[chipId].pool, RaHdcHandleSendPkt, chipId, recvBuf, recvLen);
92 : }
93 :
94 1 : STATIC void *RaAsyncPthread(void *arg)
95 : {
96 1 : unsigned int chipId = gHdcAsyncInitPara.chipId;
97 1 : unsigned int recvLen = 0;
98 1 : void *recvBuf = NULL;
99 : int ret;
100 :
101 1 : ret = pthread_detach(pthread_self());
102 1 : CHK_PRT_RETURN(ret != 0, hccp_err("pthread detach failed ret %d", ret), NULL);
103 :
104 1 : (void)prctl(PR_SET_NAME, (uintptr_t) "hccp_ra_async", 0, 0, 0);
105 :
106 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
107 1 : gHdcAsyncInitPara.threadStatus = THREAD_RUNNING;
108 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
109 :
110 1 : RsGetCurTime(&gRaAsyncThreadInfo.lastCheckTime);
111 1 : ret = strncpy_s((char *)gRaAsyncThreadInfo.pthreadName, sizeof(gRaAsyncThreadInfo.pthreadName), "ra_async_thread",
112 : strlen("ra_async_thread"));
113 1 : CHK_PRT_RETURN(ret != 0, hccp_err("strncpy_s pthread name failed, ret[%d]", ret), NULL);
114 :
115 1 : hccp_run_info("pthread[%s] is alive!", gRaAsyncThreadInfo.pthreadName);
116 : while (1) {
117 1 : if (gHdcAsyncInitPara.threadStatus == THREAD_DESTROYING) {
118 0 : break;
119 : }
120 :
121 1 : if (gHdcAsyncInitPara.connectStatus != HDC_CONNECTED) {
122 0 : usleep(THREAD_SLEEP_TIME);
123 0 : continue;
124 : }
125 1 : RsHeartbeatAlivePrint(&gRaAsyncThreadInfo);
126 : // recv msg from hdc, alloc recv_buf in ra_async_pthread, free in work_pthread
127 1 : ret = RaHdcAsyncRecvPkt(&gHdcAsync[chipId], chipId, &recvBuf, &recvLen);
128 1 : if (ret != 0) {
129 1 : hccp_err("ra_hdc_async_recv_pkt failed, ret:%d chipId:%u", ret, chipId);
130 1 : break;
131 : }
132 :
133 0 : RaAsyncHandlePkt(chipId, recvBuf, recvLen);
134 : }
135 :
136 1 : hccp_info("thread [%d] is out, cleaning resources", getpid());
137 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
138 1 : gHdcAsyncInitPara.threadStatus = THREAD_HALT;
139 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
140 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsync[chipId].sendMutex);
141 1 : RaHdcCloseSession(&gHdcAsync[chipId].hdcSession);
142 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsync[chipId].sendMutex);
143 1 : return NULL;
144 : }
145 :
146 1 : STATIC void RaHwAsyncHdcInit(void *arg)
147 : {
148 1 : unsigned int chipId = gHdcAsyncInitPara.chipId;
149 : pthread_t tidp;
150 : int ret;
151 :
152 1 : ret = pthread_detach(pthread_self());
153 1 : if (ret != 0) {
154 0 : hccp_err("pthread detach failed chip_id(%u), ret %d", chipId, ret);
155 0 : return;
156 : }
157 :
158 1 : (void)prctl(PR_SET_NAME, (uintptr_t) "hccp_hw_async", 0, 0, 0);
159 :
160 1 : hccp_info("chip_id(%u)", chipId);
161 1 : gHdcAsyncInitPara.hdcFlag = 1;
162 :
163 1 : ret = pthread_create(&tidp, NULL, (void *)RaAsyncPthread, NULL);
164 1 : if (ret != 0) {
165 0 : hccp_err("Create pthread failed, chipId(%u), ret(%d) ", chipId, ret);
166 0 : return;
167 : }
168 :
169 : while (1) {
170 1 : if (gHdcAsyncInitPara.connectStatus != HDC_UNCONNECTED) {
171 0 : usleep(HDC_ACCEPT_SLEEP_TIME);
172 0 : continue;
173 : }
174 1 : ret = RaHdcSessionAccept(chipId, &gHdcAsync[chipId].hdcSession, (int)gHdcAsyncInitPara.hostTgid);
175 1 : if (ret != 0) {
176 0 : gHdcAsyncInitPara.hdcFlag = 0;
177 0 : return;
178 : }
179 : // should continue to accept: host_tgid != g_hdc_async_init_para.host_tgid
180 1 : if (ret == 0 && gHdcAsync[chipId].hdcSession == NULL) {
181 0 : continue;
182 : }
183 :
184 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
185 1 : gHdcAsyncInitPara.connectStatus = HDC_CONNECTED;
186 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
187 1 : return;
188 : }
189 : }
190 :
191 122 : void RaHwAsyncDeinit(void)
192 : {
193 122 : pthread_mutex_destroy(&gHdcAsync[gHdcAsyncInitPara.chipId].sendMutex);
194 122 : pthread_mutex_destroy(&gHdcAsyncInitPara.mutex);
195 122 : }
196 :
197 1 : int RaRsAsyncHdcSessionConnect(char *inBuf, char *outBuf, int *outLen, int *opResult, int rcvBufLen)
198 : {
199 1 : union OpAsyncHdcConnectData *asyncData = NULL;
200 1 : int timeout = RA_THREAD_TRY_TIME;
201 1 : unsigned int phyId = 0;
202 : pthread_t tidp;
203 : int ret;
204 :
205 : HCCP_CHECK_PARAM_LEN_RET_HOST(sizeof(union OpAsyncHdcConnectData), sizeof(struct MsgHead), rcvBufLen, opResult);
206 1 : asyncData = (union OpAsyncHdcConnectData *)(inBuf + sizeof(struct MsgHead));
207 : HCCP_CHECK_PARAM_LEN_RET_HOST(asyncData->txData.queueSize, 0, MAX_POOL_QUEUE_SIZE, opResult);
208 : HCCP_CHECK_PARAM_LEN_RET_HOST(asyncData->txData.threadNum, 0, MAX_POOL_THREAD_NUM, opResult);
209 :
210 1 : phyId = gHdcAsyncInitPara.chipId;
211 1 : gHdcAsync[phyId].pool = RaHdcPoolCreate(asyncData->txData.queueSize, asyncData->txData.threadNum);
212 1 : if (gHdcAsync[phyId].pool == NULL) {
213 0 : hccp_err("ra_hdc_pool_create failed, queueSize:%u threadNum:%u phyId:%u", asyncData->txData.queueSize,
214 : asyncData->txData.threadNum, asyncData->txData.phyId);
215 0 : *opResult = -ESYSFUNC;
216 0 : return 0;
217 : }
218 :
219 1 : ret = pthread_create(&tidp, NULL, (void *)RaHwAsyncHdcInit, NULL);
220 1 : if (ret != 0) {
221 0 : hccp_err("Create pthread failed, ret(%d)", ret);
222 0 : *opResult = -ESYSFUNC;
223 0 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
224 0 : gHdcAsync[phyId].pool = NULL;
225 0 : return 0;
226 : }
227 :
228 : // will block until time out: RA_THREAD_TRY_TIME * RA_THREAD_SLEEP_TIME us
229 2 : while (gHdcAsyncInitPara.hdcFlag != 1 && timeout > 0) {
230 1 : usleep(RA_THREAD_SLEEP_TIME);
231 1 : timeout--;
232 : }
233 :
234 1 : if (gHdcAsyncInitPara.hdcFlag == 0 || timeout <= 0) {
235 0 : hccp_err("HDC server thread create timeout, flag %d, timeout %d", gHdcAsyncInitPara.hdcFlag, timeout);
236 0 : *opResult = -ESRCH;
237 0 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
238 0 : gHdcAsync[phyId].pool = NULL;
239 0 : return 0;
240 : }
241 :
242 1 : *opResult = 0;
243 1 : return 0;
244 : }
245 :
246 1 : int RaRsAsyncHdcSessionClose(char *inBuf, char *outBuf, int *outLen, int *opResult, int rcvBufLen)
247 : {
248 1 : int tryAgain = HDC_TRY_TIME;
249 1 : unsigned int phyId = 0;
250 :
251 : HCCP_CHECK_PARAM_LEN_RET_HOST(sizeof(union OpAsyncHdcCloseData), sizeof(struct MsgHead), rcvBufLen, opResult);
252 :
253 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
254 1 : gHdcAsyncInitPara.threadStatus = THREAD_DESTROYING;
255 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
256 :
257 201 : while ((gHdcAsyncInitPara.threadStatus != THREAD_HALT) && tryAgain != 0) {
258 200 : usleep(HDC_USLEEP_TIME);
259 200 : tryAgain--;
260 : }
261 :
262 1 : if (tryAgain <= 0) {
263 1 : hccp_warn("hdc async message thread quit timeout");
264 : }
265 :
266 1 : phyId = gHdcAsyncInitPara.chipId;
267 1 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
268 1 : gHdcAsync[phyId].pool = NULL;
269 1 : *opResult = 0;
270 1 : return 0;
271 : }
|
