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 <pthread.h>
13 : #include "securec.h"
14 : #include "user_log.h"
15 : #include "ra_hdc.h"
16 : #include "ra_rs_err.h"
17 : #include "ra_adp_pool.h"
18 :
19 2 : STATIC void *RaHdcWorkerThread(void *arg)
20 : {
21 2 : struct RaHdcThreadPool *pool = (struct RaHdcThreadPool *)arg;
22 2 : pthread_t tidp = pthread_self();
23 2 : struct RaHdcTask task = {0};
24 :
25 2 : (void)prctl(PR_SET_NAME, (uintptr_t) "hccp_rs_work", 0, 0, 0);
26 :
27 : while (1) {
28 2 : RA_PTHREAD_MUTEX_LOCK(&pool->poolMutex);
29 : // block thread until task received
30 4 : while (pool->taskNum == 0 && pool->shutdown != SHUTDOWN_SIGNAL) {
31 2 : pthread_cond_wait(&pool->condition, &pool->poolMutex);
32 : }
33 2 : if (pool->shutdown == SHUTDOWN_SIGNAL) {
34 2 : pool->threadNum--;
35 2 : RA_PTHREAD_MUTEX_UNLOCK(&pool->poolMutex);
36 2 : break;
37 : }
38 :
39 : // consume a task
40 0 : task.func = pool->taskQueue[pool->queueCi].func;
41 0 : task.args.chipId = pool->taskQueue[pool->queueCi].args.chipId;
42 0 : task.args.recvBuf = pool->taskQueue[pool->queueCi].args.recvBuf;
43 0 : task.args.recvLen = pool->taskQueue[pool->queueCi].args.recvLen;
44 0 : pool->queueCi = (pool->queueCi + 1) % pool->queueSize;
45 0 : pool->taskNum--;
46 :
47 : // notify manager to produce
48 0 : pthread_cond_signal(&pool->condition);
49 0 : RA_PTHREAD_MUTEX_UNLOCK(&pool->poolMutex);
50 :
51 : // do task
52 0 : task.func(task.args.chipId, task.args.recvBuf, task.args.recvLen);
53 0 : free(task.args.recvBuf);
54 0 : task.args.recvBuf = NULL;
55 : }
56 :
57 2 : hccp_run_info("tidp:%ld exit", tidp);
58 2 : return NULL;
59 : }
60 :
61 2 : STATIC int RaHdcPoolMutexCondInit(struct RaHdcThreadPool *pool)
62 : {
63 : int ret;
64 :
65 2 : ret = pthread_mutex_init(&pool->poolMutex, NULL);
66 2 : CHK_PRT_RETURN(ret != 0, hccp_err("pool_mutex mutex_init failed ret %d", ret), -ESYSFUNC);
67 :
68 2 : ret = pthread_cond_init(&pool->condition, NULL);
69 2 : if (ret != 0) {
70 0 : hccp_err("condition cond_init failed ret %d", ret);
71 0 : ret = -ESYSFUNC;
72 0 : goto deinit_pool_mutex;
73 : }
74 :
75 2 : return 0;
76 :
77 0 : deinit_pool_mutex:
78 0 : pthread_mutex_destroy(&pool->poolMutex);
79 0 : return ret;
80 : }
81 :
82 2 : STATIC void RaHdcPoolMutexCondDeinit(struct RaHdcThreadPool *pool)
83 : {
84 2 : pthread_cond_destroy(&pool->condition);
85 2 : pthread_mutex_destroy(&pool->poolMutex);
86 2 : }
87 :
88 2 : STATIC void RaHdcPoolFreeWorkers(struct RaHdcThreadPool *pool)
89 : {
90 2 : int timeout = RA_THREAD_TRY_TIME;
91 : unsigned int i;
92 :
93 2 : RA_PTHREAD_MUTEX_LOCK(&pool->poolMutex);
94 2 : pool->shutdown = SHUTDOWN_SIGNAL;
95 4 : for (i = 0; i < pool->threadNum; i++) {
96 2 : pthread_cond_signal(&pool->condition);
97 : }
98 2 : RA_PTHREAD_MUTEX_UNLOCK(&pool->poolMutex);
99 :
100 : // wait for all threads exit until time out: RA_THREAD_TRY_TIME * RA_THREAD_SLEEP_TIME us
101 3 : while (pool->threadNum > 0 && timeout > 0) {
102 1 : usleep(RA_THREAD_SLEEP_TIME);
103 1 : timeout--;
104 : }
105 2 : if (pool->threadNum > 0 && timeout <= 0) {
106 0 : hccp_warn("destroy thread pool timeout, threadNum:%u > 0 and timeout:%d <= 0", pool->threadNum, timeout);
107 : }
108 2 : }
109 :
110 2 : struct RaHdcThreadPool *RaHdcPoolCreate(unsigned int queueSize, unsigned int threadNum)
111 : {
112 2 : struct RaHdcThreadPool *pool = NULL;
113 : unsigned int i;
114 : int ret;
115 :
116 2 : pool = (struct RaHdcThreadPool *)calloc(1, sizeof(struct RaHdcThreadPool));
117 2 : CHK_PRT_RETURN(pool == NULL, hccp_err("calloc pool failed"), NULL);
118 2 : pool->taskQueue = (struct RaHdcTask *)calloc(queueSize, sizeof(struct RaHdcTask));
119 2 : if (pool->taskQueue == NULL) {
120 0 : hccp_err("calloc task_queue failed, queueSize:%u", queueSize);
121 0 : goto free_pool;
122 : }
123 2 : pool->queueSize = queueSize;
124 :
125 2 : ret = RaHdcPoolMutexCondInit(pool);
126 2 : if (ret != 0) {
127 0 : hccp_err("ra_hdc_pool_mutex_cond_init failed, ret:%d", ret);
128 0 : goto free_queue;
129 : }
130 :
131 2 : pool->workerThreads = (pthread_t *)calloc(threadNum, sizeof(pthread_t));
132 2 : if (pool->workerThreads == NULL) {
133 0 : hccp_err("calloc worker_threads failed, threadNum:%u", threadNum);
134 0 : goto free_cond;
135 : }
136 4 : for (i = 0; i < threadNum; i++) {
137 3 : ret = pthread_create(&pool->workerThreads[i], NULL, (void *)RaHdcWorkerThread, pool);
138 3 : if (ret != 0) {
139 1 : hccp_err("Create pthread i:%u failed, ret:%d", i, ret);
140 1 : pool->threadNum = i;
141 1 : goto free_thread;
142 : }
143 : }
144 1 : pool->threadNum = threadNum;
145 :
146 1 : return pool;
147 1 : free_thread:
148 1 : RaHdcPoolFreeWorkers(pool);
149 1 : free(pool->workerThreads);
150 1 : pool->workerThreads = NULL;
151 1 : free_cond:
152 1 : RaHdcPoolMutexCondDeinit(pool);
153 1 : free_queue:
154 1 : free(pool->taskQueue);
155 1 : pool->taskQueue = NULL;
156 1 : free_pool:
157 1 : free(pool);
158 1 : pool = NULL;
159 1 : return NULL;
160 : }
161 :
162 1 : void RaHdcPoolAddTask(struct RaHdcThreadPool *pool, TaskFuncT func, unsigned int chipId, void *recvBuf,
163 : unsigned int recvLen)
164 : {
165 1 : RA_PTHREAD_MUTEX_LOCK(&pool->poolMutex);
166 : // block until task can be received
167 1 : while (pool->taskNum == pool->queueSize) {
168 0 : pthread_cond_wait(&pool->condition, &pool->poolMutex);
169 : }
170 :
171 : // produce a task
172 1 : pool->taskQueue[pool->queuePi].func = func;
173 1 : pool->taskQueue[pool->queuePi].args.chipId = chipId;
174 1 : pool->taskQueue[pool->queuePi].args.recvBuf = recvBuf;
175 1 : pool->taskQueue[pool->queuePi].args.recvLen = recvLen;
176 1 : pool->queuePi = (pool->queuePi + 1) % pool->queueSize;
177 1 : pool->taskNum++;
178 :
179 : // notify worker to consume
180 1 : pthread_cond_signal(&pool->condition);
181 1 : RA_PTHREAD_MUTEX_UNLOCK(&pool->poolMutex);
182 1 : }
183 :
184 1 : int RaHdcPoolDestroy(struct RaHdcThreadPool *pool)
185 : {
186 1 : CHK_PRT_RETURN(pool == NULL, hccp_err("param invalid, pool is NULL"), -EINVAL);
187 :
188 1 : RaHdcPoolFreeWorkers(pool);
189 1 : if (pool->taskQueue != NULL) {
190 1 : free(pool->taskQueue);
191 1 : pool->taskQueue = NULL;
192 : }
193 1 : if (pool->workerThreads) {
194 1 : free(pool->workerThreads);
195 1 : pool->workerThreads = NULL;
196 : }
197 1 : RaHdcPoolMutexCondDeinit(pool);
198 1 : free(pool);
199 1 : pool = NULL;
200 1 : return 0;
201 : }
|
