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