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 118 : int RaHwAsyncInit(unsigned int chipId, pid_t pid)
26 : {
27 : int ret;
28 :
29 118 : ret = pthread_mutex_init(&gHdcAsyncInitPara.mutex, NULL);
30 118 : CHK_PRT_RETURN(ret != 0, hccp_err("g_hdc_async_init_para mutex_init failed ret %d", ret), -ESYSFUNC);
31 :
32 118 : gHdcAsyncInitPara.chipId = chipId;
33 118 : gHdcAsyncInitPara.hostTgid = pid;
34 :
35 118 : ret = pthread_mutex_init(&gHdcAsync[chipId].sendMutex, NULL);
36 118 : 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 118 : RaHdcInitOpSec(&gHdcAsync[chipId].opSec, BUCKET_DEPTH, true);
42 118 : 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, (unsigned long)"hccp_ra_async");
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),
112 : "ra_async_thread", 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, (unsigned long)"hccp_hw_async");
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 121 : void RaHwAsyncDeinit(void)
192 : {
193 121 : pthread_mutex_destroy(&gHdcAsync[gHdcAsyncInitPara.chipId].sendMutex);
194 121 : pthread_mutex_destroy(&gHdcAsyncInitPara.mutex);
195 121 : }
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,
206 : opResult);
207 1 : asyncData = (union OpAsyncHdcConnectData *)(inBuf + sizeof(struct MsgHead));
208 : HCCP_CHECK_PARAM_LEN_RET_HOST(asyncData->txData.queueSize, 0, MAX_POOL_QUEUE_SIZE, opResult);
209 : HCCP_CHECK_PARAM_LEN_RET_HOST(asyncData->txData.threadNum, 0, MAX_POOL_THREAD_NUM, opResult);
210 :
211 1 : phyId = gHdcAsyncInitPara.chipId;
212 1 : gHdcAsync[phyId].pool = RaHdcPoolCreate(asyncData->txData.queueSize, asyncData->txData.threadNum);
213 1 : if (gHdcAsync[phyId].pool == NULL) {
214 0 : hccp_err("ra_hdc_pool_create failed, queueSize:%u threadNum:%u phyId:%u",
215 : asyncData->txData.queueSize, asyncData->txData.threadNum, asyncData->txData.phyId);
216 0 : *opResult = -ESYSFUNC;
217 0 : return 0;
218 : }
219 :
220 1 : ret = pthread_create(&tidp, NULL, (void *)RaHwAsyncHdcInit, NULL);
221 1 : if (ret != 0) {
222 0 : hccp_err("Create pthread failed, ret(%d)", ret);
223 0 : *opResult = -ESYSFUNC;
224 0 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
225 0 : gHdcAsync[phyId].pool = NULL;
226 0 : return 0;
227 : }
228 :
229 : // will block until time out: RA_THREAD_TRY_TIME * RA_THREAD_SLEEP_TIME us
230 2 : while (gHdcAsyncInitPara.hdcFlag != 1 && timeout > 0) {
231 1 : usleep(RA_THREAD_SLEEP_TIME);
232 1 : timeout--;
233 : }
234 :
235 1 : if (gHdcAsyncInitPara.hdcFlag == 0 || timeout <= 0) {
236 0 : hccp_err("HDC server thread create timeout, flag %d, timeout %d", gHdcAsyncInitPara.hdcFlag, timeout);
237 0 : *opResult = -ESRCH;
238 0 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
239 0 : gHdcAsync[phyId].pool = NULL;
240 0 : return 0;
241 : }
242 :
243 1 : *opResult = 0;
244 1 : return 0;
245 : }
246 :
247 1 : int RaRsAsyncHdcSessionClose(char *inBuf, char *outBuf, int *outLen, int *opResult, int rcvBufLen)
248 : {
249 1 : int tryAgain = HDC_TRY_TIME;
250 1 : unsigned int phyId = 0;
251 :
252 : HCCP_CHECK_PARAM_LEN_RET_HOST(sizeof(union OpAsyncHdcCloseData), sizeof(struct MsgHead), rcvBufLen,
253 : opResult);
254 :
255 1 : RA_PTHREAD_MUTEX_LOCK(&gHdcAsyncInitPara.mutex);
256 1 : gHdcAsyncInitPara.threadStatus = THREAD_DESTROYING;
257 1 : RA_PTHREAD_MUTEX_UNLOCK(&gHdcAsyncInitPara.mutex);
258 :
259 201 : while ((gHdcAsyncInitPara.threadStatus != THREAD_HALT) && tryAgain != 0) {
260 200 : usleep(HDC_USLEEP_TIME);
261 200 : tryAgain--;
262 : }
263 :
264 1 : if (tryAgain <= 0) {
265 1 : hccp_warn("hdc async message thread quit timeout");
266 : }
267 :
268 1 : phyId = gHdcAsyncInitPara.chipId;
269 1 : RaHdcPoolDestroy(gHdcAsync[phyId].pool);
270 1 : gHdcAsync[phyId].pool = NULL;
271 1 : *opResult = 0;
272 1 : return 0;
273 : }
|
