From 354d119a6569b2c6335ae9309e4606e5cccedb6a Mon Sep 17 00:00:00 2001 From: paul Date: Mon, 25 Apr 2005 16:26:42 +0000 Subject: 2005-04-25 Paul Jakma * workqueue.{c,h}: Helper API for setting up and running queues via background threads. * command.c: install the 'show workqueues' command * memtypes.c: Add work queue mtypes, and a rib-queue type for a zebra rib work queue. * memtypes.h: Updated to match memtypes.c * Makefile.am: Add new workqueue files to build. --- lib/workqueue.c | 329 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 329 insertions(+) create mode 100644 lib/workqueue.c (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c new file mode 100644 index 00000000..0c9592d2 --- /dev/null +++ b/lib/workqueue.c @@ -0,0 +1,329 @@ +/* + * Quagga Work Queue Support. + * + * Copyright (C) 2005 Sun Microsystems, Inc. + * + * This file is part of GNU Zebra. + * + * Quagga is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2, or (at your option) any + * later version. + * + * Quagga is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with Quagga; see the file COPYING. If not, write to the Free + * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA + * 02111-1307, USA. + */ + +#include +#include "thread.h" +#include "memory.h" +#include "workqueue.h" +#include "linklist.h" +#include "command.h" +#include "log.h" + +/* master list of work_queues */ +static struct list work_queues; + +#define WORK_QUEUE_MIN_GRANULARITY 1 + +static struct work_queue_item * +work_queue_item_new (struct work_queue *wq) +{ + struct work_queue_item *item; + assert (wq); + + item = XCALLOC (MTYPE_WORK_QUEUE_ITEM, + sizeof (struct work_queue_item)); + + return item; +} + +static void +work_queue_item_free (struct work_queue_item *item) +{ + XFREE (MTYPE_WORK_QUEUE_ITEM, item); + return; +} + +/* create new work queue */ +struct work_queue * +work_queue_new (struct thread_master *m, const char *queue_name) +{ + struct work_queue *new; + + new = XCALLOC (MTYPE_WORK_QUEUE, sizeof (struct work_queue)); + + if (new == NULL) + return new; + + new->name = XSTRDUP (MTYPE_WORK_QUEUE_NAME, queue_name); + new->master = m; + + if ( (new->items = list_new ()) == NULL) + { + if (new->items) + list_free (new->items); + + XFREE (MTYPE_WORK_QUEUE_NAME, new->name); + XFREE (MTYPE_WORK_QUEUE, new); + + return NULL; + } + + new->items->del = (void (*)(void *)) work_queue_item_free; + + listnode_add (&work_queues, new); + + new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY; + + return new; +} + +void +work_queue_free (struct work_queue *wq) +{ + /* list_delete frees items via callback */ + list_delete (wq->items); + listnode_delete (&work_queues, wq); + + XFREE (MTYPE_WORK_QUEUE_NAME, wq->name); + XFREE (MTYPE_WORK_QUEUE, wq); + return; +} + +void +work_queue_add (struct work_queue *wq, void *data) +{ + struct work_queue_item *item; + + assert (wq); + + if (!(item = work_queue_item_new (wq))) + { + zlog_err ("%s: unable to get new queue item", __func__); + return; + } + + item->data = data; + listnode_add (wq->items, item); + + /* if thread isnt already waiting, add one */ + if (wq->thread == NULL) + wq->thread = thread_add_background (wq->master, work_queue_run, + wq, wq->spec.hold); + + /* XXX: what if we didnt get a thread? try again? */ + + return; +} + +static void +work_queue_item_remove (struct work_queue *wq, struct listnode *ln) +{ + struct work_queue_item *item = listgetdata (ln); + + assert (item && item->data); + + /* call private data deletion callback if needed */ + if (wq->spec.del_item_data) + wq->spec.del_item_data (item->data); + + list_delete_node (wq->items, ln); + work_queue_item_free (item); + + return; +} + +static void +work_queue_item_requeue (struct work_queue *wq, struct listnode *ln) +{ + LISTNODE_DETACH (wq->items, ln); + LISTNODE_ATTACH (wq->items, ln); /* attach to end of list */ +} + +DEFUN(show_work_queues, + show_work_queues_cmd, + "show work-queues", + SHOW_STR + "Work Queue information\n") +{ + struct listnode *node; + struct work_queue *wq; + struct timeval tvnow; + + gettimeofday (&tvnow, NULL); + + vty_out (vty, + "%8s %11s %8s %21s%s", + "List","(ms) ","Q. Runs","Cycle Counts ", + VTY_NEWLINE); + vty_out (vty, + "%8s %5s %5s %8s %7s %6s %6s %s%s", + "Items", + "Delay","Hold", + "Total", + "Best","Gran.","Avg.", + "Name", + VTY_NEWLINE); + + for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) + { + vty_out (vty,"%8d %5d %5d %8ld %7d %6d %6u %s%s", + listcount (wq->items), + wq->spec.delay, wq->spec.hold, + wq->runs, + wq->cycles.best, wq->cycles.granularity, + (unsigned int)(wq->cycles.total / wq->runs), + wq->name, + VTY_NEWLINE); + } + + return CMD_SUCCESS; +} + +/* timer thread to process a work queue + * will reschedule itself if required, + * otherwise work_queue_item_add + */ +int +work_queue_run (struct thread *thread) +{ + struct work_queue *wq; + struct work_queue_item *item; + wq_item_status ret; + unsigned int cycles = 0; + struct listnode *node, *nnode; + char yielded = 0; + + wq = THREAD_ARG (thread); + wq->thread = NULL; + + assert (wq && wq->items); + + /* calculate cycle granularity: + * list iteration == 1 cycle + * granularity == # cycles between checks whether we should yield. + * + * granularity should be > 0, and can increase slowly after each run to + * provide some hysteris, but not past cycles.best or 2*cycles. + * + * Best: starts low, can only increase + * + * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased if we run to end of time + * slot, can increase otherwise by a small factor. + * + * We could use just the average and save some work, however we want to be + * able to adjust quickly to CPU pressure. Average wont shift much if + * daemon has been running a long time. + */ + if (wq->cycles.granularity == 0) + wq->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY; + + for (ALL_LIST_ELEMENTS (wq->items, node, nnode, item)) + { + assert (item && item->data); + + /* dont run items which are past their allowed retries */ + if (item->retry_count >= wq->spec.max_retries) + { + /* run error handler, if any */ + if (wq->spec.errorfunc) + wq->spec.errorfunc (wq, item->data); + work_queue_item_remove (wq, node); + continue; + } + + /* run and take care of items that want to be retried immediately */ + do + { + ret = wq->spec.workfunc (item->data); + item->retry_count++; + } + while ((ret == WQ_RETRY_NOW) + && (item->retry_count < wq->spec.max_retries)); + + switch (ret) + { + case WQ_RETRY_LATER: + { + item->retry_count++; + goto stats; + } + case WQ_REQUEUE: + { + item->retry_count++; + work_queue_item_requeue (wq, node); + break; + } + case WQ_RETRY_NOW: + case WQ_ERROR: + { + if (wq->spec.errorfunc) + wq->spec.errorfunc (wq, item); + } + /* fall through here is deliberate */ + case WQ_SUCCESS: + default: + { + work_queue_item_remove (wq, node); + break; + } + } + + /* completed cycle */ + cycles++; + + /* test if we should yield */ + if ( !(cycles % wq->cycles.granularity) + && thread_should_yield (thread)) + { + yielded = 1; + goto stats; + } + } + +stats: + +#define WQ_HYSTERIS_FACTOR 2 + + /* we yielded, check whether granularity should be reduced */ + if (yielded && (cycles < wq->cycles.granularity)) + { + wq->cycles.granularity = ((cycles > 0) ? cycles + : WORK_QUEUE_MIN_GRANULARITY); + } + + if (cycles > (wq->cycles.granularity)) + { + if (cycles > wq->cycles.best) + wq->cycles.best = cycles; + + /* along with yielded check, provides hysteris for granularity */ + if (cycles > (wq->cycles.granularity * WQ_HYSTERIS_FACTOR)) + wq->cycles.granularity += WQ_HYSTERIS_FACTOR; + } +#undef WQ_HYSTERIS_FACTOR + + wq->runs++; + wq->cycles.total += cycles; + +#if 0 + printf ("%s: cycles %d, new: best %d, worst %d\n", + __func__, cycles, wq->cycles.best, wq->cycles.granularity); +#endif + + /* Is the queue done yet? */ + if (listcount (wq->items) > 0) + wq->thread = thread_add_background (wq->master, work_queue_run, wq, + wq->spec.delay); + + return 0; +} -- cgit v1.2.3 From 843696841b55343e45d6cfdad19035882a67b99f Mon Sep 17 00:00:00 2001 From: paul Date: Wed, 27 Apr 2005 12:39:27 +0000 Subject: 2005-04-27 Paul Jakma * workqueue.h: (struct work_queue_item) change retry_count to ran, its a count of number item has been run. * workqueue.c: (show_work_queues) Fix formating of slightly bugfix: fix SIGFPE if wq->runs is 0. (work_queue_run) retry logic was slightly wrong. cycles.best is 0 initialy, granularity is 1, so update best if cycles >= granularity, not just >. --- lib/workqueue.c | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index 0c9592d2..7931d19c 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -163,7 +163,7 @@ DEFUN(show_work_queues, vty_out (vty, "%8s %11s %8s %21s%s", - "List","(ms) ","Q. Runs","Cycle Counts ", + "List","(ms) ","Q. Runs","Cycle Counts ", VTY_NEWLINE); vty_out (vty, "%8s %5s %5s %8s %7s %6s %6s %s%s", @@ -176,12 +176,13 @@ DEFUN(show_work_queues, for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) { - vty_out (vty,"%8d %5d %5d %8ld %7d %6d %6u %s%s", + vty_out (vty,"%8d %5d %5d %8ld %7d %6d %6u %s%s", listcount (wq->items), wq->spec.delay, wq->spec.hold, wq->runs, - wq->cycles.best, wq->cycles.granularity, - (unsigned int)(wq->cycles.total / wq->runs), + wq->cycles.best, wq->cycles.granularity, + (wq->runs) ? + (unsigned int) (wq->cycles.total / wq->runs) : 0, wq->name, VTY_NEWLINE); } @@ -232,7 +233,7 @@ work_queue_run (struct thread *thread) assert (item && item->data); /* dont run items which are past their allowed retries */ - if (item->retry_count >= wq->spec.max_retries) + if (item->ran > wq->spec.max_retries) { /* run error handler, if any */ if (wq->spec.errorfunc) @@ -245,21 +246,19 @@ work_queue_run (struct thread *thread) do { ret = wq->spec.workfunc (item->data); - item->retry_count++; + item->ran++; } while ((ret == WQ_RETRY_NOW) - && (item->retry_count < wq->spec.max_retries)); + && (item->ran < wq->spec.max_retries)); switch (ret) { case WQ_RETRY_LATER: { - item->retry_count++; goto stats; } case WQ_REQUEUE: { - item->retry_count++; work_queue_item_requeue (wq, node); break; } @@ -301,7 +300,7 @@ stats: : WORK_QUEUE_MIN_GRANULARITY); } - if (cycles > (wq->cycles.granularity)) + if (cycles >= (wq->cycles.granularity)) { if (cycles > wq->cycles.best) wq->cycles.best = cycles; -- cgit v1.2.3 From 3df537822f594ffefe4d5e121c0b2430c9c12806 Mon Sep 17 00:00:00 2001 From: ajs Date: Wed, 27 Apr 2005 16:29:54 +0000 Subject: 2005-04-27 Andrew J. Schorr * workqueue.c (show_work_queues): Remove unused gettimeofday call. --- lib/workqueue.c | 3 --- 1 file changed, 3 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index 7931d19c..fc61d680 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -157,9 +157,6 @@ DEFUN(show_work_queues, { struct listnode *node; struct work_queue *wq; - struct timeval tvnow; - - gettimeofday (&tvnow, NULL); vty_out (vty, "%8s %11s %8s %21s%s", -- cgit v1.2.3 From 269d74fdc39a612da8f627bf97628c68d25a16ab Mon Sep 17 00:00:00 2001 From: paul Date: Mon, 23 May 2005 13:42:46 +0000 Subject: 2005-05-23 Paul Jakma * workqueue.h: Add a WQ_QUEUE_BLOCKED item_status return code, to allow a queue function to indicate the queue is not ready/blocked - rather than any problem with the item at hand. Add a notion of being able to 'plug' and 'unplug' a queue. Add helpers to plug/unplug a queue. Add a completion callback, to be called when a queue is emptied. * workqueue.c: (work_queue_new) remove useless list_free. (work_queue_schedule) new internal helper function to schedule queue, if appropriate. (work_queue_add) use work_queue_schedule (show_work_queues) Print 'P' if queue is plugged. (work_queue_plug) new API function, plug a queue - ie prevent it from 'drained' / processed / scheduled. (work_queue_unplug) unplug a queue, allowing it to be drained / scheduled / processed again. (work_queue_run) Add support for WQ_QUEUE_BLOCKED. Add comment for RETRY_NOW case. Make hysteris more aggresive in ramping up granularity, improves performance significantly. Add support for calling completion callback when queue is emptied, possibly useful for knowing when to unplug a queue. --- lib/workqueue.c | 83 ++++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 65 insertions(+), 18 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index fc61d680..bac41302 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -69,9 +69,6 @@ work_queue_new (struct thread_master *m, const char *queue_name) if ( (new->items = list_new ()) == NULL) { - if (new->items) - list_free (new->items); - XFREE (MTYPE_WORK_QUEUE_NAME, new->name); XFREE (MTYPE_WORK_QUEUE, new); @@ -99,6 +96,22 @@ work_queue_free (struct work_queue *wq) return; } +static inline int +work_queue_schedule (struct work_queue *wq, unsigned int delay) +{ + /* if appropriate, schedule work queue thread */ + if ( (wq->flags == WQ_UNPLUGGED) + && (wq->thread == NULL) + && (listcount (wq->items) > 0) ) + { + wq->thread = thread_add_background (wq->master, work_queue_run, + wq, delay); + return 1; + } + else + return 0; +} + void work_queue_add (struct work_queue *wq, void *data) { @@ -115,12 +128,7 @@ work_queue_add (struct work_queue *wq, void *data) item->data = data; listnode_add (wq->items, item); - /* if thread isnt already waiting, add one */ - if (wq->thread == NULL) - wq->thread = thread_add_background (wq->master, work_queue_run, - wq, wq->spec.hold); - - /* XXX: what if we didnt get a thread? try again? */ + work_queue_schedule (wq, wq->spec.hold); return; } @@ -159,11 +167,12 @@ DEFUN(show_work_queues, struct work_queue *wq; vty_out (vty, - "%8s %11s %8s %21s%s", - "List","(ms) ","Q. Runs","Cycle Counts ", + "%c %8s %11s %8s %21s%s", + ' ', "List","(ms) ","Q. Runs","Cycle Counts ", VTY_NEWLINE); vty_out (vty, - "%8s %5s %5s %8s %7s %6s %6s %s%s", + "%c %8s %5s %5s %8s %7s %6s %6s %s%s", + ' ', "Items", "Delay","Hold", "Total", @@ -173,7 +182,8 @@ DEFUN(show_work_queues, for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) { - vty_out (vty,"%8d %5d %5d %8ld %7d %6d %6u %s%s", + vty_out (vty,"%c %8d %5d %5d %8ld %7d %6d %6u %s%s", + (wq->flags == WQ_PLUGGED ? 'P' : ' '), listcount (wq->items), wq->spec.delay, wq->spec.hold, wq->runs, @@ -187,6 +197,32 @@ DEFUN(show_work_queues, return CMD_SUCCESS; } +/* 'plug' a queue: Stop it from being scheduled, + * ie: prevent the queue from draining. + */ +void +work_queue_plug (struct work_queue *wq) +{ + if (wq->thread) + thread_cancel (wq->thread); + + wq->thread = NULL; + + wq->flags = WQ_PLUGGED; +} + +/* unplug queue, schedule it again, if appropriate + * Ie: Allow the queue to be drained again + */ +void +work_queue_unplug (struct work_queue *wq) +{ + wq->flags = WQ_UNPLUGGED; + + /* if thread isnt already waiting, add one */ + work_queue_schedule (wq, wq->spec.hold); +} + /* timer thread to process a work queue * will reschedule itself if required, * otherwise work_queue_item_add @@ -250,6 +286,13 @@ work_queue_run (struct thread *thread) switch (ret) { + case WQ_QUEUE_BLOCKED: + { + /* decrement item->ran again, cause this isn't an item + * specific error, and fall through to WQ_RETRY_LATER + */ + item->ran--; + } case WQ_RETRY_LATER: { goto stats; @@ -260,6 +303,7 @@ work_queue_run (struct thread *thread) break; } case WQ_RETRY_NOW: + /* a RETRY_NOW that gets here has exceeded max_tries, same as ERROR */ case WQ_ERROR: { if (wq->spec.errorfunc) @@ -303,7 +347,9 @@ stats: wq->cycles.best = cycles; /* along with yielded check, provides hysteris for granularity */ - if (cycles > (wq->cycles.granularity * WQ_HYSTERIS_FACTOR)) + if (cycles > (wq->cycles.granularity * WQ_HYSTERIS_FACTOR * 2)) + wq->cycles.granularity *= WQ_HYSTERIS_FACTOR; /* quick ramp-up */ + else if (cycles > (wq->cycles.granularity * WQ_HYSTERIS_FACTOR)) wq->cycles.granularity += WQ_HYSTERIS_FACTOR; } #undef WQ_HYSTERIS_FACTOR @@ -316,10 +362,11 @@ stats: __func__, cycles, wq->cycles.best, wq->cycles.granularity); #endif - /* Is the queue done yet? */ + /* Is the queue done yet? If it is, call the completion callback. */ if (listcount (wq->items) > 0) - wq->thread = thread_add_background (wq->master, work_queue_run, wq, - wq->spec.delay); - + work_queue_schedule (wq, wq->spec.delay); + else if (wq->spec.completion_func) + wq->spec.completion_func (wq); + return 0; } -- cgit v1.2.3 From 190880dc790007a14911ef8c170af33a50a7a674 Mon Sep 17 00:00:00 2001 From: paul Date: Mon, 14 Nov 2005 12:07:47 +0000 Subject: 2005-11-14 Paul Jakma * (general) Add state to detect queue floods. There's no sense trying to be sparing of CPU resources, if the queue is flooding and using ever more memory resources. we should just get on with clearing the queue. The sense of delay and hold were wrong way around, fix. * workqueue.h: (struct work_queue) Add status bitfield. Add 'flood' integer to workqueue spec. Add runs_since_clear counter to workqueue. * workqueue.c: (work_queue_new) set defaults for delay, hold and flood. (work_queue_add) initial schedule should use delay, not hold. (show_work_queues) Print flood field, conserve whitespace. (work_queue_unplug) use delay, not hold. (work_queue_run) consecutive runs should be seperated by hold time, not delay. Keep track of number of consecutive runs, go into 'overdrive' if queue is being flooded, we can't avoid making heavy use of resources, better to use CPU than ever more RAM. --- lib/workqueue.c | 38 +++++++++++++++++++++++++++----------- 1 file changed, 27 insertions(+), 11 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index bac41302..c2ff10db 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -80,7 +80,12 @@ work_queue_new (struct thread_master *m, const char *queue_name) listnode_add (&work_queues, new); new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY; - + + /* Default values, can be overriden by caller */ + new->spec.delay = WORK_QUEUE_DEFAULT_DELAY; + new->spec.hold = WORK_QUEUE_DEFAULT_HOLD; + new->spec.flood = WORK_QUEUE_DEFAULT_FLOOD; + return new; } @@ -128,7 +133,7 @@ work_queue_add (struct work_queue *wq, void *data) item->data = data; listnode_add (wq->items, item); - work_queue_schedule (wq, wq->spec.hold); + work_queue_schedule (wq, wq->spec.delay); return; } @@ -167,12 +172,12 @@ DEFUN(show_work_queues, struct work_queue *wq; vty_out (vty, - "%c %8s %11s %8s %21s%s", - ' ', "List","(ms) ","Q. Runs","Cycle Counts ", + "%c%c %8s %11s %8s %21s%s", + ' ', ' ', "List","(ms) ","Q. Runs","Cycle Counts ", VTY_NEWLINE); vty_out (vty, - "%c %8s %5s %5s %8s %7s %6s %6s %s%s", - ' ', + "%c%c %8s %5s %5s %8s %7s %6s %6s %s%s", + 'P', 'F', "Items", "Delay","Hold", "Total", @@ -182,8 +187,9 @@ DEFUN(show_work_queues, for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) { - vty_out (vty,"%c %8d %5d %5d %8ld %7d %6d %6u %s%s", + vty_out (vty,"%c%c %8d %5d %5d %8ld %7d %6d %6u %s%s", (wq->flags == WQ_PLUGGED ? 'P' : ' '), + (wq->runs_since_clear >= wq->spec.flood ? 'F' : ' '), listcount (wq->items), wq->spec.delay, wq->spec.hold, wq->runs, @@ -220,7 +226,7 @@ work_queue_unplug (struct work_queue *wq) wq->flags = WQ_UNPLUGGED; /* if thread isnt already waiting, add one */ - work_queue_schedule (wq, wq->spec.hold); + work_queue_schedule (wq, wq->spec.delay); } /* timer thread to process a work queue @@ -364,9 +370,19 @@ stats: /* Is the queue done yet? If it is, call the completion callback. */ if (listcount (wq->items) > 0) - work_queue_schedule (wq, wq->spec.delay); - else if (wq->spec.completion_func) - wq->spec.completion_func (wq); + { + if (++(wq->runs_since_clear) < wq->spec.flood) + work_queue_schedule (wq, wq->spec.hold); + else + work_queue_schedule (wq, 0); /* queue flooded, go into overdrive */ + } + else + { + wq->runs_since_clear = 0; + + if (wq->spec.completion_func) + wq->spec.completion_func (wq); + } return 0; } -- cgit v1.2.3 From 889e9311e5c900ce24eecf00fcb5b8b9d51bb020 Mon Sep 17 00:00:00 2001 From: paul Date: Mon, 14 Nov 2005 14:46:35 +0000 Subject: [workqueue] Update workqueue users callbacks to additional arguments 2005-11-14 Paul Jakma * (general) pass struct work-queue to callback functions. * workqueue.h: (struct work_queue) move the state flag variables to end. Add an opaque pointer to spec, for user-data global to the queue. Pass reference to work_queue to all callbacks. * workqueue.c: (work_queue_item_remove) pass ref to workqueue to user callbacks. (work_queue_run) ditto. --- lib/workqueue.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index c2ff10db..0c61f5c4 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -147,7 +147,7 @@ work_queue_item_remove (struct work_queue *wq, struct listnode *ln) /* call private data deletion callback if needed */ if (wq->spec.del_item_data) - wq->spec.del_item_data (item->data); + wq->spec.del_item_data (wq, item->data); list_delete_node (wq->items, ln); work_queue_item_free (item); @@ -284,7 +284,7 @@ work_queue_run (struct thread *thread) /* run and take care of items that want to be retried immediately */ do { - ret = wq->spec.workfunc (item->data); + ret = wq->spec.workfunc (wq, item->data); item->ran++; } while ((ret == WQ_RETRY_NOW) -- cgit v1.2.3 From 306d8890439cdb9128d063ee2f77700a11e6843c Mon Sep 17 00:00:00 2001 From: paul Date: Thu, 2 Feb 2006 17:50:19 +0000 Subject: [lib/workqueue] remove the useless 'delay' factor. 2006-02-02 Paul Jakma * workqueue.h: (struct work_queue) Remove the delay field. It served no purpose and just introduced bad behaviour. Should be excised before its allowed to escape into 1.0. This removes need for the 'flood' and runs_since_clear fields. * workqueue.c: (general) excise delay factor between queue runs, hence the 'flood' crap too.. See above. * bgp_route.c: (bgp_{clear_node,process}_queue_init) delay field is removed from workqueue spec. --- lib/workqueue.c | 37 ++++++++++++------------------------- 1 file changed, 12 insertions(+), 25 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index 0c61f5c4..1fa16ded 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -82,9 +82,7 @@ work_queue_new (struct thread_master *m, const char *queue_name) new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY; /* Default values, can be overriden by caller */ - new->spec.delay = WORK_QUEUE_DEFAULT_DELAY; new->spec.hold = WORK_QUEUE_DEFAULT_HOLD; - new->spec.flood = WORK_QUEUE_DEFAULT_FLOOD; return new; } @@ -133,7 +131,7 @@ work_queue_add (struct work_queue *wq, void *data) item->data = data; listnode_add (wq->items, item); - work_queue_schedule (wq, wq->spec.delay); + work_queue_schedule (wq, wq->spec.hold); return; } @@ -172,14 +170,14 @@ DEFUN(show_work_queues, struct work_queue *wq; vty_out (vty, - "%c%c %8s %11s %8s %21s%s", - ' ', ' ', "List","(ms) ","Q. Runs","Cycle Counts ", + "%c %8s %5s %8s %21s%s", + ' ', "List","(ms) ","Q. Runs","Cycle Counts ", VTY_NEWLINE); vty_out (vty, - "%c%c %8s %5s %5s %8s %7s %6s %6s %s%s", - 'P', 'F', + "%c %8s %5s %8s %7s %6s %6s %s%s", + 'P', "Items", - "Delay","Hold", + "Hold", "Total", "Best","Gran.","Avg.", "Name", @@ -187,11 +185,10 @@ DEFUN(show_work_queues, for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) { - vty_out (vty,"%c%c %8d %5d %5d %8ld %7d %6d %6u %s%s", + vty_out (vty,"%c %8d %5d %8ld %7d %6d %6u %s%s", (wq->flags == WQ_PLUGGED ? 'P' : ' '), - (wq->runs_since_clear >= wq->spec.flood ? 'F' : ' '), listcount (wq->items), - wq->spec.delay, wq->spec.hold, + wq->spec.hold, wq->runs, wq->cycles.best, wq->cycles.granularity, (wq->runs) ? @@ -226,7 +223,7 @@ work_queue_unplug (struct work_queue *wq) wq->flags = WQ_UNPLUGGED; /* if thread isnt already waiting, add one */ - work_queue_schedule (wq, wq->spec.delay); + work_queue_schedule (wq, wq->spec.hold); } /* timer thread to process a work queue @@ -370,19 +367,9 @@ stats: /* Is the queue done yet? If it is, call the completion callback. */ if (listcount (wq->items) > 0) - { - if (++(wq->runs_since_clear) < wq->spec.flood) - work_queue_schedule (wq, wq->spec.hold); - else - work_queue_schedule (wq, 0); /* queue flooded, go into overdrive */ - } - else - { - wq->runs_since_clear = 0; - - if (wq->spec.completion_func) - wq->spec.completion_func (wq); - } + work_queue_schedule (wq, 0); + else if (wq->spec.completion_func) + wq->spec.completion_func (wq); return 0; } -- cgit v1.2.3 From 31061288a6a8dc1a9d471a9a590e969b91c595ba Mon Sep 17 00:00:00 2001 From: paul Date: Thu, 30 Mar 2006 14:45:47 +0000 Subject: [lib] Trivial: fix line lengths of a comment in workqueu.c 2006-03-30 Paul Jakma * workqueue.c: (work_queue_run) fix line length of comment --- lib/workqueue.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index 1fa16ded..a0f48bc8 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -254,8 +254,9 @@ work_queue_run (struct thread *thread) * * Best: starts low, can only increase * - * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased if we run to end of time - * slot, can increase otherwise by a small factor. + * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased + * if we run to end of time slot, can increase otherwise + * by a small factor. * * We could use just the average and save some work, however we want to be * able to adjust quickly to CPU pressure. Average wont shift much if -- cgit v1.2.3 From b359e35f78c929071ed04d7f4d0b41f23c48e992 Mon Sep 17 00:00:00 2001 From: pilot Date: Mon, 12 Nov 2007 14:55:01 +0000 Subject: + fixed bug #418 (changing address on an existing interface doesn't cause existing static routes to be revalidated) --- lib/workqueue.c | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index a0f48bc8..8880b9e2 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -66,6 +66,8 @@ work_queue_new (struct thread_master *m, const char *queue_name) new->name = XSTRDUP (MTYPE_WORK_QUEUE_NAME, queue_name); new->master = m; + SET_FLAG (new->flags, WQ_UNPLUGGED); + UNSET_FLAG (new->flags, WQ_AIM_HEAD); if ( (new->items = list_new ()) == NULL) { @@ -103,7 +105,7 @@ static inline int work_queue_schedule (struct work_queue *wq, unsigned int delay) { /* if appropriate, schedule work queue thread */ - if ( (wq->flags == WQ_UNPLUGGED) + if ( CHECK_FLAG (wq->flags, WQ_UNPLUGGED) && (wq->thread == NULL) && (listcount (wq->items) > 0) ) { @@ -129,7 +131,10 @@ work_queue_add (struct work_queue *wq, void *data) } item->data = data; - listnode_add (wq->items, item); + if (CHECK_FLAG (wq->flags, WQ_AIM_HEAD)) + listnode_add_after (wq->items, NULL, item); + else + listnode_add (wq->items, item); work_queue_schedule (wq, wq->spec.hold); @@ -186,7 +191,7 @@ DEFUN(show_work_queues, for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq)) { vty_out (vty,"%c %8d %5d %8ld %7d %6d %6u %s%s", - (wq->flags == WQ_PLUGGED ? 'P' : ' '), + (CHECK_FLAG (wq->flags, WQ_UNPLUGGED) ? ' ' : 'P'), listcount (wq->items), wq->spec.hold, wq->runs, @@ -211,7 +216,7 @@ work_queue_plug (struct work_queue *wq) wq->thread = NULL; - wq->flags = WQ_PLUGGED; + UNSET_FLAG (wq->flags, WQ_UNPLUGGED); } /* unplug queue, schedule it again, if appropriate @@ -220,12 +225,21 @@ work_queue_plug (struct work_queue *wq) void work_queue_unplug (struct work_queue *wq) { - wq->flags = WQ_UNPLUGGED; + SET_FLAG (wq->flags, WQ_UNPLUGGED); /* if thread isnt already waiting, add one */ work_queue_schedule (wq, wq->spec.hold); } +void +work_queue_aim_head (struct work_queue *wq, const unsigned aim_head) +{ + if (aim_head) + SET_FLAG (wq->flags, WQ_AIM_HEAD); + else + UNSET_FLAG (wq->flags, WQ_AIM_HEAD); +} + /* timer thread to process a work queue * will reschedule itself if required, * otherwise work_queue_item_add -- cgit v1.2.3 From 90443036ecda07a5f8d5c3501faef465301365bd Mon Sep 17 00:00:00 2001 From: pilot Date: Mon, 2 Jun 2008 12:03:22 +0000 Subject: + initial edition of meta-queue for RIB updates processing (bug #431) --- lib/workqueue.c | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) (limited to 'lib/workqueue.c') diff --git a/lib/workqueue.c b/lib/workqueue.c index 8880b9e2..1d32d241 100644 --- a/lib/workqueue.c +++ b/lib/workqueue.c @@ -67,7 +67,6 @@ work_queue_new (struct thread_master *m, const char *queue_name) new->name = XSTRDUP (MTYPE_WORK_QUEUE_NAME, queue_name); new->master = m; SET_FLAG (new->flags, WQ_UNPLUGGED); - UNSET_FLAG (new->flags, WQ_AIM_HEAD); if ( (new->items = list_new ()) == NULL) { @@ -131,10 +130,7 @@ work_queue_add (struct work_queue *wq, void *data) } item->data = data; - if (CHECK_FLAG (wq->flags, WQ_AIM_HEAD)) - listnode_add_after (wq->items, NULL, item); - else - listnode_add (wq->items, item); + listnode_add (wq->items, item); work_queue_schedule (wq, wq->spec.hold); @@ -231,15 +227,6 @@ work_queue_unplug (struct work_queue *wq) work_queue_schedule (wq, wq->spec.hold); } -void -work_queue_aim_head (struct work_queue *wq, const unsigned aim_head) -{ - if (aim_head) - SET_FLAG (wq->flags, WQ_AIM_HEAD); - else - UNSET_FLAG (wq->flags, WQ_AIM_HEAD); -} - /* timer thread to process a work queue * will reschedule itself if required, * otherwise work_queue_item_add @@ -317,6 +304,7 @@ work_queue_run (struct thread *thread) } case WQ_REQUEUE: { + item->ran--; work_queue_item_requeue (wq, node); break; } -- cgit v1.2.3