lib/workqueue.c

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