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