2 * Quagga Work Queue Support.
4 * Copyright (C) 2005 Sun Microsystems, Inc.
6 * This file is part of GNU Zebra.
8 * Quagga is free software; you can redistribute it and/or modify it
9 * under the terms of the GNU General Public License as published by the
10 * Free Software Foundation; either version 2, or (at your option) any
13 * Quagga is distributed in the hope that it will be useful, but
14 * WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * General Public License for more details.
18 * You should have received a copy of the GNU General Public License along
19 * with this program; see the file COPYING; if not, write to the Free Software
20 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
26 #include "workqueue.h"
31 DEFINE_MTYPE(LIB
, WORK_QUEUE
, "Work queue")
32 DEFINE_MTYPE_STATIC(LIB
, WORK_QUEUE_ITEM
, "Work queue item")
33 DEFINE_MTYPE_STATIC(LIB
, WORK_QUEUE_NAME
, "Work queue name string")
35 /* master list of work_queues */
36 static struct list _work_queues
;
37 /* pointer primarily to avoid an otherwise harmless warning on
38 * ALL_LIST_ELEMENTS_RO
40 static struct list
*work_queues
= &_work_queues
;
42 #define WORK_QUEUE_MIN_GRANULARITY 1
44 static struct work_queue_item
*
45 work_queue_item_new (struct work_queue
*wq
)
47 struct work_queue_item
*item
;
50 item
= XCALLOC (MTYPE_WORK_QUEUE_ITEM
,
51 sizeof (struct work_queue_item
));
57 work_queue_item_free (struct work_queue_item
*item
)
59 XFREE (MTYPE_WORK_QUEUE_ITEM
, item
);
63 /* create new work queue */
65 work_queue_new (struct thread_master
*m
, const char *queue_name
)
67 struct work_queue
*new;
69 new = XCALLOC (MTYPE_WORK_QUEUE
, sizeof (struct work_queue
));
74 new->name
= XSTRDUP (MTYPE_WORK_QUEUE_NAME
, queue_name
);
76 SET_FLAG (new->flags
, WQ_UNPLUGGED
);
78 if ( (new->items
= list_new ()) == NULL
)
80 XFREE (MTYPE_WORK_QUEUE_NAME
, new->name
);
81 XFREE (MTYPE_WORK_QUEUE
, new);
86 new->items
->del
= (void (*)(void *)) work_queue_item_free
;
88 listnode_add (work_queues
, new);
90 new->cycles
.granularity
= WORK_QUEUE_MIN_GRANULARITY
;
92 /* Default values, can be overriden by caller */
93 new->spec
.hold
= WORK_QUEUE_DEFAULT_HOLD
;
94 new->spec
.yield
= THREAD_YIELD_TIME_SLOT
;
100 work_queue_free (struct work_queue
*wq
)
102 if (wq
->thread
!= NULL
)
103 thread_cancel(wq
->thread
);
105 /* list_delete frees items via callback */
106 list_delete (wq
->items
);
107 listnode_delete (work_queues
, wq
);
109 XFREE (MTYPE_WORK_QUEUE_NAME
, wq
->name
);
110 XFREE (MTYPE_WORK_QUEUE
, wq
);
115 work_queue_is_scheduled (struct work_queue
*wq
)
117 return (wq
->thread
!= NULL
);
121 work_queue_schedule (struct work_queue
*wq
, unsigned int delay
)
123 /* if appropriate, schedule work queue thread */
124 if ( CHECK_FLAG (wq
->flags
, WQ_UNPLUGGED
)
125 && (wq
->thread
== NULL
)
126 && (listcount (wq
->items
) > 0) )
129 thread_add_timer_msec (wq
->master
, work_queue_run
, wq
, delay
,
131 /* set thread yield time, if needed */
132 if (wq
->thread
&& wq
->spec
.yield
!= THREAD_YIELD_TIME_SLOT
)
133 thread_set_yield_time (wq
->thread
, wq
->spec
.yield
);
141 work_queue_add (struct work_queue
*wq
, void *data
)
143 struct work_queue_item
*item
;
147 if (!(item
= work_queue_item_new (wq
)))
149 zlog_err ("%s: unable to get new queue item", __func__
);
154 listnode_add (wq
->items
, item
);
156 work_queue_schedule (wq
, wq
->spec
.hold
);
162 work_queue_item_remove (struct work_queue
*wq
, struct listnode
*ln
)
164 struct work_queue_item
*item
= listgetdata (ln
);
166 assert (item
&& item
->data
);
168 /* call private data deletion callback if needed */
169 if (wq
->spec
.del_item_data
)
170 wq
->spec
.del_item_data (wq
, item
->data
);
172 list_delete_node (wq
->items
, ln
);
173 work_queue_item_free (item
);
179 work_queue_item_requeue (struct work_queue
*wq
, struct listnode
*ln
)
181 LISTNODE_DETACH (wq
->items
, ln
);
182 LISTNODE_ATTACH (wq
->items
, ln
); /* attach to end of list */
185 DEFUN (show_work_queues
,
186 show_work_queues_cmd
,
189 "Work Queue information\n")
191 struct listnode
*node
;
192 struct work_queue
*wq
;
195 "%c %8s %5s %8s %8s %21s",
196 ' ', "List","(ms) ","Q. Runs","Yields","Cycle Counts ");
198 "%c %8s %5s %8s %8s %7s %6s %8s %6s %s",
203 "Best","Gran.","Total","Avg.",
206 for (ALL_LIST_ELEMENTS_RO (work_queues
, node
, wq
))
208 vty_outln (vty
,"%c %8d %5d %8ld %8ld %7d %6d %8ld %6u %s",
209 (CHECK_FLAG (wq
->flags
, WQ_UNPLUGGED
) ? ' ' : 'P'),
210 listcount (wq
->items
),
212 wq
->runs
, wq
->yields
,
213 wq
->cycles
.best
, wq
->cycles
.granularity
, wq
->cycles
.total
,
215 (unsigned int) (wq
->cycles
.total
/ wq
->runs
) : 0,
223 workqueue_cmd_init (void)
225 install_element (VIEW_NODE
, &show_work_queues_cmd
);
228 /* 'plug' a queue: Stop it from being scheduled,
229 * ie: prevent the queue from draining.
232 work_queue_plug (struct work_queue
*wq
)
235 thread_cancel (wq
->thread
);
239 UNSET_FLAG (wq
->flags
, WQ_UNPLUGGED
);
242 /* unplug queue, schedule it again, if appropriate
243 * Ie: Allow the queue to be drained again
246 work_queue_unplug (struct work_queue
*wq
)
248 SET_FLAG (wq
->flags
, WQ_UNPLUGGED
);
250 /* if thread isnt already waiting, add one */
251 work_queue_schedule (wq
, wq
->spec
.hold
);
254 /* timer thread to process a work queue
255 * will reschedule itself if required,
256 * otherwise work_queue_item_add
259 work_queue_run (struct thread
*thread
)
261 struct work_queue
*wq
;
262 struct work_queue_item
*item
;
264 unsigned int cycles
= 0;
265 struct listnode
*node
, *nnode
;
268 wq
= THREAD_ARG (thread
);
271 assert (wq
&& wq
->items
);
273 /* calculate cycle granularity:
274 * list iteration == 1 run
275 * listnode processing == 1 cycle
276 * granularity == # cycles between checks whether we should yield.
278 * granularity should be > 0, and can increase slowly after each run to
279 * provide some hysteris, but not past cycles.best or 2*cycles.
281 * Best: starts low, can only increase
283 * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased
284 * if we run to end of time slot, can increase otherwise
287 * We could use just the average and save some work, however we want to be
288 * able to adjust quickly to CPU pressure. Average wont shift much if
289 * daemon has been running a long time.
291 if (wq
->cycles
.granularity
== 0)
292 wq
->cycles
.granularity
= WORK_QUEUE_MIN_GRANULARITY
;
294 for (ALL_LIST_ELEMENTS (wq
->items
, node
, nnode
, item
))
296 assert (item
&& item
->data
);
298 /* dont run items which are past their allowed retries */
299 if (item
->ran
> wq
->spec
.max_retries
)
301 /* run error handler, if any */
302 if (wq
->spec
.errorfunc
)
303 wq
->spec
.errorfunc (wq
, item
->data
);
304 work_queue_item_remove (wq
, node
);
308 /* run and take care of items that want to be retried immediately */
311 ret
= wq
->spec
.workfunc (wq
, item
->data
);
314 while ((ret
== WQ_RETRY_NOW
)
315 && (item
->ran
< wq
->spec
.max_retries
));
319 case WQ_QUEUE_BLOCKED
:
321 /* decrement item->ran again, cause this isn't an item
322 * specific error, and fall through to WQ_RETRY_LATER
333 work_queue_item_requeue (wq
, node
);
334 /* If a single node is being used with a meta-queue (e.g., zebra),
335 * update the next node as we don't want to exit the thread and
336 * reschedule it after every node. By definition, WQ_REQUEUE is
337 * meant to continue the processing; the yield logic will kick in
338 * to terminate the thread when time has exceeded.
345 /* a RETRY_NOW that gets here has exceeded max_tries, same as ERROR */
348 if (wq
->spec
.errorfunc
)
349 wq
->spec
.errorfunc (wq
, item
);
351 /* fall through here is deliberate */
355 work_queue_item_remove (wq
, node
);
360 /* completed cycle */
363 /* test if we should yield */
364 if ( !(cycles
% wq
->cycles
.granularity
)
365 && thread_should_yield (thread
))
374 #define WQ_HYSTERESIS_FACTOR 4
376 /* we yielded, check whether granularity should be reduced */
377 if (yielded
&& (cycles
< wq
->cycles
.granularity
))
379 wq
->cycles
.granularity
= ((cycles
> 0) ? cycles
380 : WORK_QUEUE_MIN_GRANULARITY
);
382 /* otherwise, should granularity increase? */
383 else if (cycles
>= (wq
->cycles
.granularity
))
385 if (cycles
> wq
->cycles
.best
)
386 wq
->cycles
.best
= cycles
;
388 /* along with yielded check, provides hysteresis for granularity */
389 if (cycles
> (wq
->cycles
.granularity
* WQ_HYSTERESIS_FACTOR
390 * WQ_HYSTERESIS_FACTOR
))
391 wq
->cycles
.granularity
*= WQ_HYSTERESIS_FACTOR
; /* quick ramp-up */
392 else if (cycles
> (wq
->cycles
.granularity
* WQ_HYSTERESIS_FACTOR
))
393 wq
->cycles
.granularity
+= WQ_HYSTERESIS_FACTOR
;
395 #undef WQ_HYSTERIS_FACTOR
398 wq
->cycles
.total
+= cycles
;
403 printf ("%s: cycles %d, new: best %d, worst %d\n",
404 __func__
, cycles
, wq
->cycles
.best
, wq
->cycles
.granularity
);
407 /* Is the queue done yet? If it is, call the completion callback. */
408 if (listcount (wq
->items
) > 0)
409 work_queue_schedule (wq
, 0);
410 else if (wq
->spec
.completion_func
)
411 wq
->spec
.completion_func (wq
);