xref: /dpdk/examples/ip_pipeline/thread.c (revision 8b9bd0efe0b6920a08e28eebacf2bb916bdf5653)
1 /* SPDX-License-Identifier: BSD-3-Clause
2  * Copyright(c) 2010-2015 Intel Corporation
3  */
4 
5 #include <rte_common.h>
6 #include <rte_cycles.h>
7 #include <rte_pipeline.h>
8 
9 #include "pipeline_common_be.h"
10 #include "app.h"
11 #include "thread.h"
12 
13 #if APP_THREAD_HEADROOM_STATS_COLLECT
14 
15 #define PIPELINE_RUN_REGULAR(thread, pipeline)		\
16 do {							\
17 	uint64_t t0 = rte_rdtsc_precise();		\
18 	int n_pkts = rte_pipeline_run(pipeline->p);	\
19 							\
20 	if (n_pkts == 0) {				\
21 		uint64_t t1 = rte_rdtsc_precise();	\
22 							\
23 		thread->headroom_cycles += t1 - t0;	\
24 	}						\
25 } while (0)
26 
27 
28 #define PIPELINE_RUN_CUSTOM(thread, data)		\
29 do {							\
30 	uint64_t t0 = rte_rdtsc_precise();		\
31 	int n_pkts = data->f_run(data->be);		\
32 							\
33 	if (n_pkts == 0) {				\
34 		uint64_t t1 = rte_rdtsc_precise();	\
35 							\
36 		thread->headroom_cycles += t1 - t0;	\
37 	}						\
38 } while (0)
39 
40 #else
41 
42 #define PIPELINE_RUN_REGULAR(thread, pipeline)		\
43 	rte_pipeline_run(pipeline->p)
44 
45 #define PIPELINE_RUN_CUSTOM(thread, data)		\
46 	data->f_run(data->be)
47 
48 #endif
49 
50 static inline void *
51 thread_msg_recv(struct rte_ring *r)
52 {
53 	void *msg;
54 	int status = rte_ring_sc_dequeue(r, &msg);
55 
56 	if (status != 0)
57 		return NULL;
58 
59 	return msg;
60 }
61 
62 static inline void
63 thread_msg_send(struct rte_ring *r,
64 	void *msg)
65 {
66 	int status;
67 
68 	do {
69 		status = rte_ring_sp_enqueue(r, msg);
70 	} while (status == -ENOBUFS);
71 }
72 
73 static int
74 thread_pipeline_enable(struct app_thread_data *t,
75 		struct thread_pipeline_enable_msg_req *req)
76 {
77 	struct app_thread_pipeline_data *p;
78 
79 	if (req->f_run == NULL) {
80 		if (t->n_regular >= APP_MAX_THREAD_PIPELINES)
81 			return -1;
82 	} else {
83 		if (t->n_custom >= APP_MAX_THREAD_PIPELINES)
84 			return -1;
85 	}
86 
87 	p = (req->f_run == NULL) ?
88 		&t->regular[t->n_regular] :
89 		&t->custom[t->n_custom];
90 
91 	p->pipeline_id = req->pipeline_id;
92 	p->be = req->be;
93 	p->f_run = req->f_run;
94 	p->f_timer = req->f_timer;
95 	p->timer_period = req->timer_period;
96 	p->deadline = 0;
97 
98 	if (req->f_run == NULL)
99 		t->n_regular++;
100 	else
101 		t->n_custom++;
102 
103 	return 0;
104 }
105 
106 static int
107 thread_pipeline_disable(struct app_thread_data *t,
108 		struct thread_pipeline_disable_msg_req *req)
109 {
110 	uint32_t n_regular = RTE_MIN(t->n_regular, RTE_DIM(t->regular));
111 	uint32_t n_custom = RTE_MIN(t->n_custom, RTE_DIM(t->custom));
112 	uint32_t i;
113 
114 	/* search regular pipelines of current thread */
115 	for (i = 0; i < n_regular; i++) {
116 		if (t->regular[i].pipeline_id != req->pipeline_id)
117 			continue;
118 
119 		if (i < n_regular - 1)
120 			memcpy(&t->regular[i],
121 			  &t->regular[i+1],
122 			  (n_regular - 1 - i) * sizeof(struct app_thread_pipeline_data));
123 
124 		n_regular--;
125 		t->n_regular = n_regular;
126 
127 		return 0;
128 	}
129 
130 	/* search custom pipelines of current thread */
131 	for (i = 0; i < n_custom; i++) {
132 		if (t->custom[i].pipeline_id != req->pipeline_id)
133 			continue;
134 
135 		if (i < n_custom - 1)
136 			memcpy(&t->custom[i],
137 			  &t->custom[i+1],
138 			  (n_custom - 1 - i) * sizeof(struct app_thread_pipeline_data));
139 
140 		n_custom--;
141 		t->n_custom = n_custom;
142 
143 		return 0;
144 	}
145 
146 	/* return if pipeline not found */
147 	return -1;
148 }
149 
150 static int
151 thread_msg_req_handle(struct app_thread_data *t)
152 {
153 	void *msg_ptr;
154 	struct thread_msg_req *req;
155 	struct thread_msg_rsp *rsp;
156 
157 	msg_ptr = thread_msg_recv(t->msgq_in);
158 	req = msg_ptr;
159 	rsp = msg_ptr;
160 
161 	if (req != NULL)
162 		switch (req->type) {
163 		case THREAD_MSG_REQ_PIPELINE_ENABLE: {
164 			rsp->status = thread_pipeline_enable(t,
165 					(struct thread_pipeline_enable_msg_req *) req);
166 			thread_msg_send(t->msgq_out, rsp);
167 			break;
168 		}
169 
170 		case THREAD_MSG_REQ_PIPELINE_DISABLE: {
171 			rsp->status = thread_pipeline_disable(t,
172 					(struct thread_pipeline_disable_msg_req *) req);
173 			thread_msg_send(t->msgq_out, rsp);
174 			break;
175 		}
176 
177 		case THREAD_MSG_REQ_HEADROOM_READ: {
178 			struct thread_headroom_read_msg_rsp *rsp =
179 				(struct thread_headroom_read_msg_rsp *)
180 				req;
181 
182 			rsp->headroom_ratio = t->headroom_ratio;
183 			rsp->status = 0;
184 			thread_msg_send(t->msgq_out, rsp);
185 			break;
186 		}
187 		default:
188 			break;
189 		}
190 
191 	return 0;
192 }
193 
194 static void
195 thread_headroom_update(struct app_thread_data *t, uint64_t time)
196 {
197 	uint64_t time_diff = time - t->headroom_time;
198 
199 	t->headroom_ratio =
200 		((double) t->headroom_cycles) / ((double) time_diff);
201 
202 	t->headroom_cycles = 0;
203 	t->headroom_time = rte_rdtsc_precise();
204 }
205 
206 int
207 app_thread(void *arg)
208 {
209 	struct app_params *app = (struct app_params *) arg;
210 	uint32_t core_id = rte_lcore_id(), i, j;
211 	struct app_thread_data *t = &app->thread_data[core_id];
212 
213 	for (i = 0; ; i++) {
214 		uint32_t n_regular = RTE_MIN(t->n_regular, RTE_DIM(t->regular));
215 		uint32_t n_custom = RTE_MIN(t->n_custom, RTE_DIM(t->custom));
216 
217 		/* Run regular pipelines */
218 		for (j = 0; j < n_regular; j++) {
219 			struct app_thread_pipeline_data *data = &t->regular[j];
220 			struct pipeline *p = data->be;
221 
222 			PIPELINE_RUN_REGULAR(t, p);
223 		}
224 
225 		/* Run custom pipelines */
226 		for (j = 0; j < n_custom; j++) {
227 			struct app_thread_pipeline_data *data = &t->custom[j];
228 
229 			PIPELINE_RUN_CUSTOM(t, data);
230 		}
231 
232 		/* Timer */
233 		if ((i & 0xF) == 0) {
234 			uint64_t time = rte_get_tsc_cycles();
235 			uint64_t t_deadline = UINT64_MAX;
236 
237 			if (time < t->deadline)
238 				continue;
239 
240 			/* Timer for regular pipelines */
241 			for (j = 0; j < n_regular; j++) {
242 				struct app_thread_pipeline_data *data =
243 					&t->regular[j];
244 				uint64_t p_deadline = data->deadline;
245 
246 				if (p_deadline <= time) {
247 					data->f_timer(data->be);
248 					p_deadline = time + data->timer_period;
249 					data->deadline = p_deadline;
250 				}
251 
252 				if (p_deadline < t_deadline)
253 					t_deadline = p_deadline;
254 			}
255 
256 			/* Timer for custom pipelines */
257 			for (j = 0; j < n_custom; j++) {
258 				struct app_thread_pipeline_data *data =
259 					&t->custom[j];
260 				uint64_t p_deadline = data->deadline;
261 
262 				if (p_deadline <= time) {
263 					data->f_timer(data->be);
264 					p_deadline = time + data->timer_period;
265 					data->deadline = p_deadline;
266 				}
267 
268 				if (p_deadline < t_deadline)
269 					t_deadline = p_deadline;
270 			}
271 
272 			/* Timer for thread message request */
273 			{
274 				uint64_t deadline = t->thread_req_deadline;
275 
276 				if (deadline <= time) {
277 					thread_msg_req_handle(t);
278 					thread_headroom_update(t, time);
279 					deadline = time + t->timer_period;
280 					t->thread_req_deadline = deadline;
281 				}
282 
283 				if (deadline < t_deadline)
284 					t_deadline = deadline;
285 			}
286 
287 
288 			t->deadline = t_deadline;
289 		}
290 	}
291 
292 	return 0;
293 }
294