]> git.proxmox.com Git - mirror_frr.git/blob - lib/workqueue.c
lib: Remove tests for ipv[4|6]_prefix_table
[mirror_frr.git] / lib / workqueue.c
1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /*
3 * Quagga Work Queue Support.
4 *
5 * Copyright (C) 2005 Sun Microsystems, Inc.
6 */
7
8 #include <zebra.h>
9 #include "thread.h"
10 #include "memory.h"
11 #include "workqueue.h"
12 #include "linklist.h"
13 #include "command.h"
14 #include "log.h"
15
16 DEFINE_MTYPE(LIB, WORK_QUEUE, "Work queue");
17 DEFINE_MTYPE_STATIC(LIB, WORK_QUEUE_ITEM, "Work queue item");
18 DEFINE_MTYPE_STATIC(LIB, WORK_QUEUE_NAME, "Work queue name string");
19
20 /* master list of work_queues */
21 static struct list _work_queues;
22 /* pointer primarily to avoid an otherwise harmless warning on
23 * ALL_LIST_ELEMENTS_RO
24 */
25 static struct list *work_queues = &_work_queues;
26
27 #define WORK_QUEUE_MIN_GRANULARITY 1
28
29 static struct work_queue_item *work_queue_item_new(struct work_queue *wq)
30 {
31 struct work_queue_item *item;
32 assert(wq);
33
34 item = XCALLOC(MTYPE_WORK_QUEUE_ITEM, sizeof(struct work_queue_item));
35
36 return item;
37 }
38
39 static void work_queue_item_free(struct work_queue_item *item)
40 {
41 XFREE(MTYPE_WORK_QUEUE_ITEM, item);
42 return;
43 }
44
45 static void work_queue_item_remove(struct work_queue *wq,
46 struct work_queue_item *item)
47 {
48 assert(item && item->data);
49
50 /* call private data deletion callback if needed */
51 if (wq->spec.del_item_data)
52 wq->spec.del_item_data(wq, item->data);
53
54 work_queue_item_dequeue(wq, item);
55
56 work_queue_item_free(item);
57
58 return;
59 }
60
61 /* create new work queue */
62 struct work_queue *work_queue_new(struct thread_master *m,
63 const char *queue_name)
64 {
65 struct work_queue *new;
66
67 new = XCALLOC(MTYPE_WORK_QUEUE, sizeof(struct work_queue));
68
69 new->name = XSTRDUP(MTYPE_WORK_QUEUE_NAME, queue_name);
70 new->master = m;
71 SET_FLAG(new->flags, WQ_UNPLUGGED);
72
73 STAILQ_INIT(&new->items);
74
75 listnode_add(work_queues, new);
76
77 new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY;
78
79 /* Default values, can be overridden by caller */
80 new->spec.hold = WORK_QUEUE_DEFAULT_HOLD;
81 new->spec.yield = THREAD_YIELD_TIME_SLOT;
82 new->spec.retry = WORK_QUEUE_DEFAULT_RETRY;
83
84 return new;
85 }
86
87 void work_queue_free_and_null(struct work_queue **wqp)
88 {
89 struct work_queue *wq = *wqp;
90
91 THREAD_OFF(wq->thread);
92
93 while (!work_queue_empty(wq)) {
94 struct work_queue_item *item = work_queue_last_item(wq);
95
96 work_queue_item_remove(wq, item);
97 }
98
99 listnode_delete(work_queues, wq);
100
101 XFREE(MTYPE_WORK_QUEUE_NAME, wq->name);
102 XFREE(MTYPE_WORK_QUEUE, wq);
103
104 *wqp = NULL;
105 }
106
107 bool work_queue_is_scheduled(struct work_queue *wq)
108 {
109 return thread_is_scheduled(wq->thread);
110 }
111
112 static int work_queue_schedule(struct work_queue *wq, unsigned int delay)
113 {
114 /* if appropriate, schedule work queue thread */
115 if (CHECK_FLAG(wq->flags, WQ_UNPLUGGED) &&
116 !thread_is_scheduled(wq->thread) && !work_queue_empty(wq)) {
117 /* Schedule timer if there's a delay, otherwise just schedule
118 * as an 'event'
119 */
120 if (delay > 0) {
121 thread_add_timer_msec(wq->master, work_queue_run, wq,
122 delay, &wq->thread);
123 thread_ignore_late_timer(wq->thread);
124 } else
125 thread_add_event(wq->master, work_queue_run, wq, 0,
126 &wq->thread);
127
128 /* set thread yield time, if needed */
129 if (thread_is_scheduled(wq->thread) &&
130 wq->spec.yield != THREAD_YIELD_TIME_SLOT)
131 thread_set_yield_time(wq->thread, wq->spec.yield);
132 return 1;
133 } else
134 return 0;
135 }
136
137 void work_queue_add(struct work_queue *wq, void *data)
138 {
139 struct work_queue_item *item;
140
141 assert(wq);
142
143 item = work_queue_item_new(wq);
144
145 item->data = data;
146 work_queue_item_enqueue(wq, item);
147
148 work_queue_schedule(wq, wq->spec.hold);
149
150 return;
151 }
152
153 static void work_queue_item_requeue(struct work_queue *wq,
154 struct work_queue_item *item)
155 {
156 work_queue_item_dequeue(wq, item);
157
158 /* attach to end of list */
159 work_queue_item_enqueue(wq, item);
160 }
161
162 DEFUN (show_work_queues,
163 show_work_queues_cmd,
164 "show work-queues",
165 SHOW_STR
166 "Work Queue information\n")
167 {
168 struct listnode *node;
169 struct work_queue *wq;
170
171 vty_out(vty, "%c %8s %5s %8s %8s %21s\n", ' ', "List", "(ms) ",
172 "Q. Runs", "Yields", "Cycle Counts ");
173 vty_out(vty, "%c %8s %5s %8s %8s %7s %6s %8s %6s %s\n", 'P', "Items",
174 "Hold", "Total", "Total", "Best", "Gran.", "Total", "Avg.",
175 "Name");
176
177 for (ALL_LIST_ELEMENTS_RO(work_queues, node, wq)) {
178 vty_out(vty, "%c %8d %5d %8ld %8ld %7d %6d %8ld %6u %s\n",
179 (CHECK_FLAG(wq->flags, WQ_UNPLUGGED) ? ' ' : 'P'),
180 work_queue_item_count(wq), wq->spec.hold, wq->runs,
181 wq->yields, wq->cycles.best, wq->cycles.granularity,
182 wq->cycles.total,
183 (wq->runs) ? (unsigned int)(wq->cycles.total / wq->runs)
184 : 0,
185 wq->name);
186 }
187
188 return CMD_SUCCESS;
189 }
190
191 void workqueue_cmd_init(void)
192 {
193 install_element(VIEW_NODE, &show_work_queues_cmd);
194 }
195
196 /* 'plug' a queue: Stop it from being scheduled,
197 * ie: prevent the queue from draining.
198 */
199 void work_queue_plug(struct work_queue *wq)
200 {
201 THREAD_OFF(wq->thread);
202
203 UNSET_FLAG(wq->flags, WQ_UNPLUGGED);
204 }
205
206 /* unplug queue, schedule it again, if appropriate
207 * Ie: Allow the queue to be drained again
208 */
209 void work_queue_unplug(struct work_queue *wq)
210 {
211 SET_FLAG(wq->flags, WQ_UNPLUGGED);
212
213 /* if thread isnt already waiting, add one */
214 work_queue_schedule(wq, wq->spec.hold);
215 }
216
217 /* timer thread to process a work queue
218 * will reschedule itself if required,
219 * otherwise work_queue_item_add
220 */
221 void work_queue_run(struct thread *thread)
222 {
223 struct work_queue *wq;
224 struct work_queue_item *item, *titem;
225 wq_item_status ret = WQ_SUCCESS;
226 unsigned int cycles = 0;
227 char yielded = 0;
228
229 wq = THREAD_ARG(thread);
230
231 assert(wq);
232
233 /* calculate cycle granularity:
234 * list iteration == 1 run
235 * listnode processing == 1 cycle
236 * granularity == # cycles between checks whether we should yield.
237 *
238 * granularity should be > 0, and can increase slowly after each run to
239 * provide some hysteris, but not past cycles.best or 2*cycles.
240 *
241 * Best: starts low, can only increase
242 *
243 * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased
244 * if we run to end of time slot, can increase otherwise
245 * by a small factor.
246 *
247 * We could use just the average and save some work, however we want to
248 * be
249 * able to adjust quickly to CPU pressure. Average wont shift much if
250 * daemon has been running a long time.
251 */
252 if (wq->cycles.granularity == 0)
253 wq->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY;
254
255 STAILQ_FOREACH_SAFE (item, &wq->items, wq, titem) {
256 assert(item->data);
257
258 /* dont run items which are past their allowed retries */
259 if (item->ran > wq->spec.max_retries) {
260 work_queue_item_remove(wq, item);
261 continue;
262 }
263
264 /* run and take care of items that want to be retried
265 * immediately */
266 do {
267 ret = wq->spec.workfunc(wq, item->data);
268 item->ran++;
269 } while ((ret == WQ_RETRY_NOW)
270 && (item->ran < wq->spec.max_retries));
271
272 switch (ret) {
273 case WQ_QUEUE_BLOCKED: {
274 /* decrement item->ran again, cause this isn't an item
275 * specific error, and fall through to WQ_RETRY_LATER
276 */
277 item->ran--;
278 }
279 case WQ_RETRY_LATER: {
280 goto stats;
281 }
282 case WQ_REQUEUE: {
283 item->ran--;
284 work_queue_item_requeue(wq, item);
285 /* If a single node is being used with a meta-queue
286 * (e.g., zebra),
287 * update the next node as we don't want to exit the
288 * thread and
289 * reschedule it after every node. By definition,
290 * WQ_REQUEUE is
291 * meant to continue the processing; the yield logic
292 * will kick in
293 * to terminate the thread when time has exceeded.
294 */
295 if (titem == NULL)
296 titem = item;
297 break;
298 }
299 case WQ_RETRY_NOW:
300 /* a RETRY_NOW that gets here has exceeded max_tries, same as
301 * ERROR */
302 /* fallthru */
303 case WQ_SUCCESS:
304 default: {
305 work_queue_item_remove(wq, item);
306 break;
307 }
308 }
309
310 /* completed cycle */
311 cycles++;
312
313 /* test if we should yield */
314 if (!(cycles % wq->cycles.granularity)
315 && thread_should_yield(thread)) {
316 yielded = 1;
317 goto stats;
318 }
319 }
320
321 stats:
322
323 #define WQ_HYSTERESIS_FACTOR 4
324
325 /* we yielded, check whether granularity should be reduced */
326 if (yielded && (cycles < wq->cycles.granularity)) {
327 wq->cycles.granularity =
328 ((cycles > 0) ? cycles : WORK_QUEUE_MIN_GRANULARITY);
329 }
330 /* otherwise, should granularity increase? */
331 else if (cycles >= (wq->cycles.granularity)) {
332 if (cycles > wq->cycles.best)
333 wq->cycles.best = cycles;
334
335 /* along with yielded check, provides hysteresis for granularity
336 */
337 if (cycles > (wq->cycles.granularity * WQ_HYSTERESIS_FACTOR
338 * WQ_HYSTERESIS_FACTOR))
339 wq->cycles.granularity *=
340 WQ_HYSTERESIS_FACTOR; /* quick ramp-up */
341 else if (cycles
342 > (wq->cycles.granularity * WQ_HYSTERESIS_FACTOR))
343 wq->cycles.granularity += WQ_HYSTERESIS_FACTOR;
344 }
345 #undef WQ_HYSTERIS_FACTOR
346
347 wq->runs++;
348 wq->cycles.total += cycles;
349 if (yielded)
350 wq->yields++;
351
352 /* Is the queue done yet? If it is, call the completion callback. */
353 if (!work_queue_empty(wq)) {
354 if (ret == WQ_RETRY_LATER ||
355 ret == WQ_QUEUE_BLOCKED)
356 work_queue_schedule(wq, wq->spec.retry);
357 else
358 work_queue_schedule(wq, 0);
359
360 } else if (wq->spec.completion_func)
361 wq->spec.completion_func(wq);
362 }