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