2005-11-14 Paul Jakma <paul.jakma@sun.com>

	* (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.

3 files changed, 71 insertions, 15 deletions

diff --git a/lib/ChangeLog b/lib/ChangeLog
index 106de477..b6407c4a 100644
--- a/lib/ChangeLog
+++ b/lib/ChangeLog
@@ -1,4 +1,25 @@
-2005-11-12 Paul Jakma <paul.jakma@sun.com>
+2005-11-14 Paul Jakma <paul.jakma@sun.com>
+
+	* (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.
+
+2005-11-05 Paul Jakma <paul.jakma@sun.com>
 
 	* routemap.c: (vty_show_route_map_entry) call action is
 	  seperate from exit action, latter should still be printed
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;
 }
diff --git a/lib/workqueue.h b/lib/workqueue.h
index 626d8e6c..15c72f62 100644
--- a/lib/workqueue.h
+++ b/lib/workqueue.h
@@ -25,8 +25,9 @@
 #define _QUAGGA_WORK_QUEUE_H
 
 /* Work queue default hold and cycle times - millisec */
-#define WORK_QUEUE_DEFAULT_HOLD  50  /* hold time for initial run of a queue */
-#define WORK_QUEUE_DEFAULT_DELAY 10  /* minimum delay between queue runs */
+#define WORK_QUEUE_DEFAULT_HOLD  50  /* hold-time between runs of a queue */
+#define WORK_QUEUE_DEFAULT_DELAY 10  /* minimum delay for queue runs */
+#define WORK_QUEUE_DEFAULT_FLOOD 40   /* flood factor, ~2s with prev values */
 
 /* action value, for use by item processor and item error handlers */
 typedef enum
@@ -56,12 +57,17 @@ enum work_queue_flags
 
 struct work_queue
 {
+  /* Everything but the specification struct is private */
   struct thread_master *master;       /* thread master */
   struct thread *thread;              /* thread, if one is active */
   char *name;                         /* work queue name */
+  char status;                        /* status */
+#define WQ_STATE_FLOODED	(1 << 0)
   enum work_queue_flags flags;		/* flags */
   
-  /* specification for this work queue */
+  /* Specification for this work queue.
+   * Public, must be set before use by caller. May be modified at will.
+   */
   struct {
     /* work function to process items with */
     wq_item_status (*workfunc) (void *);
@@ -80,11 +86,24 @@ struct work_queue
 
     unsigned int hold;	/* hold time for first run, in ms */
     unsigned int delay; /* min delay between queue runs, in ms */
+    
+    unsigned int flood; /* number of queue runs after which we consider
+                         * queue to be flooded, where the runs are
+                         * consecutive and each has used its full slot,
+                         * and the queue has still not been cleared. If
+                         * the queue is flooded, then we try harder to
+                         * clear it by ignoring the hold and delay
+                         * times. No point sparing CPU resources just
+                         * to use ever more memory resources.
+                         */
   } spec;
   
   /* remaining fields should be opaque to users */
   struct list *items;                 /* queue item list */
   unsigned long runs;                 /* runs count */
+  unsigned int runs_since_clear;      /* number of runs since queue was
+                                       * last cleared
+                                       */
   
   struct {
     unsigned int best;