[mirror_frr.git] / lib / workqueue.c

/* 
 * Quagga Work Queue Support.
 *
 * Copyright (C) 2005 Sun Microsystems, Inc.
 *
 * This file is part of GNU Zebra.
 *
 * Quagga is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2, or (at your option) any
 * later version.
 *
 * Quagga is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Quagga; see the file COPYING.  If not, write to the Free
 * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 * 02111-1307, USA.  
 */

#include <lib/zebra.h>
#include "thread.h"
#include "memory.h"
#include "workqueue.h"
#include "linklist.h"
#include "command.h"
#include "log.h"

/* master list of work_queues */
static struct list work_queues;

#define WORK_QUEUE_MIN_GRANULARITY 1

static struct work_queue_item *
work_queue_item_new (struct work_queue *wq)
{
  struct work_queue_item *item;
  assert (wq);

  item = XCALLOC (MTYPE_WORK_QUEUE_ITEM, 
                  sizeof (struct work_queue_item));
  
  return item;
}

static void
work_queue_item_free (struct work_queue_item *item)
{
  XFREE (MTYPE_WORK_QUEUE_ITEM, item);
  return;
}

/* create new work queue */
struct work_queue *
work_queue_new (struct thread_master *m, const char *queue_name)
{
  struct work_queue *new;
  
  new = XCALLOC (MTYPE_WORK_QUEUE, sizeof (struct work_queue));

  if (new == NULL)
    return new;
  
  new->name = XSTRDUP (MTYPE_WORK_QUEUE_NAME, queue_name);
  new->master = m;
  SET_FLAG (new->flags, WQ_UNPLUGGED);
  
  if ( (new->items = list_new ()) == NULL)
    {
      XFREE (MTYPE_WORK_QUEUE_NAME, new->name);
      XFREE (MTYPE_WORK_QUEUE, new);
      
      return NULL;
    }
  
  new->items->del = (void (*)(void *)) work_queue_item_free;  
  
  listnode_add (&work_queues, new);
  
  new->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY;

  /* Default values, can be overriden by caller */
  new->spec.hold = WORK_QUEUE_DEFAULT_HOLD;
    
  return new;
}

void
work_queue_free (struct work_queue *wq)
{
  if (wq->thread != NULL)
    thread_cancel(wq->thread);
  
  /* list_delete frees items via callback */
  list_delete (wq->items);
  listnode_delete (&work_queues, wq);
  
  XFREE (MTYPE_WORK_QUEUE_NAME, wq->name);
  XFREE (MTYPE_WORK_QUEUE, wq);
  return;
}

static int
work_queue_schedule (struct work_queue *wq, unsigned int delay)
{
  /* if appropriate, schedule work queue thread */
  if ( CHECK_FLAG (wq->flags, WQ_UNPLUGGED)
       && (wq->thread == NULL)
       && (listcount (wq->items) > 0) )
    {
      wq->thread = thread_add_background (wq->master, work_queue_run, 
                                          wq, delay);
      return 1;
    }
  else
    return 0;
}
  
void
work_queue_add (struct work_queue *wq, void *data)
{
  struct work_queue_item *item;
  
  assert (wq);

  if (!(item = work_queue_item_new (wq)))
    {
      zlog_err ("%s: unable to get new queue item", __func__);
      return;
    }
  
  item->data = data;
  listnode_add (wq->items, item);
  
  work_queue_schedule (wq, wq->spec.hold);
  
  return;
}

static void
work_queue_item_remove (struct work_queue *wq, struct listnode *ln)
{
  struct work_queue_item *item = listgetdata (ln);

  assert (item && item->data);

  /* call private data deletion callback if needed */  
  if (wq->spec.del_item_data)
    wq->spec.del_item_data (wq, item->data);

  list_delete_node (wq->items, ln);
  work_queue_item_free (item);
  
  return;
}

static void
work_queue_item_requeue (struct work_queue *wq, struct listnode *ln)
{
  LISTNODE_DETACH (wq->items, ln);
  LISTNODE_ATTACH (wq->items, ln); /* attach to end of list */
}

DEFUN(show_work_queues,
      show_work_queues_cmd,
      "show work-queues",
      SHOW_STR
      "Work Queue information\n")
{
  struct listnode *node;
  struct work_queue *wq;
  
  vty_out (vty, 
           "%c %8s %5s %8s %21s%s",
           ' ', "List","(ms) ","Q. Runs","Cycle Counts   ",
           VTY_NEWLINE);
  vty_out (vty,
           "%c %8s %5s %8s %7s %6s %6s %s%s",
           'P',
           "Items",
           "Hold",
           "Total",
           "Best","Gran.","Avg.", 
           "Name", 
           VTY_NEWLINE);
 
  for (ALL_LIST_ELEMENTS_RO ((&work_queues), node, wq))
    {
      vty_out (vty,"%c %8d %5d %8ld %7d %6d %6u %s%s",
               (CHECK_FLAG (wq->flags, WQ_UNPLUGGED) ? ' ' : 'P'),
               listcount (wq->items),
               wq->spec.hold,
               wq->runs,
               wq->cycles.best, wq->cycles.granularity,
                 (wq->runs) ? 
                   (unsigned int) (wq->cycles.total / wq->runs) : 0,
               wq->name,
               VTY_NEWLINE);
    }
    
  return CMD_SUCCESS;
}

/* 'plug' a queue: Stop it from being scheduled,
 * ie: prevent the queue from draining.
 */
void
work_queue_plug (struct work_queue *wq)
{
  if (wq->thread)
    thread_cancel (wq->thread);
  
  wq->thread = NULL;
  
  UNSET_FLAG (wq->flags, WQ_UNPLUGGED);
}

/* unplug queue, schedule it again, if appropriate
 * Ie: Allow the queue to be drained again
 */
void
work_queue_unplug (struct work_queue *wq)
{
  SET_FLAG (wq->flags, WQ_UNPLUGGED);

  /* if thread isnt already waiting, add one */
  work_queue_schedule (wq, wq->spec.hold);
}

/* timer thread to process a work queue
 * will reschedule itself if required,
 * otherwise work_queue_item_add 
 */
int
work_queue_run (struct thread *thread)
{
  struct work_queue *wq;
  struct work_queue_item *item;
  wq_item_status ret;
  unsigned int cycles = 0;
  struct listnode *node, *nnode;
  char yielded = 0;

  wq = THREAD_ARG (thread);
  wq->thread = NULL;

  assert (wq && wq->items);

  /* calculate cycle granularity:
   * list iteration == 1 cycle
   * granularity == # cycles between checks whether we should yield.
   *
   * granularity should be > 0, and can increase slowly after each run to
   * provide some hysteris, but not past cycles.best or 2*cycles.
   *
   * Best: starts low, can only increase
   *
   * Granularity: starts at WORK_QUEUE_MIN_GRANULARITY, can be decreased 
   *              if we run to end of time slot, can increase otherwise 
   *              by a small factor.
   *
   * We could use just the average and save some work, however we want to be
   * able to adjust quickly to CPU pressure. Average wont shift much if
   * daemon has been running a long time.
   */
   if (wq->cycles.granularity == 0)
     wq->cycles.granularity = WORK_QUEUE_MIN_GRANULARITY;

  for (ALL_LIST_ELEMENTS (wq->items, node, nnode, item))
  {
    assert (item && item->data);
    
    /* dont run items which are past their allowed retries */
    if (item->ran > wq->spec.max_retries)
      {
        /* run error handler, if any */
	if (wq->spec.errorfunc)
	  wq->spec.errorfunc (wq, item->data);
	work_queue_item_remove (wq, node);
	continue;
      }

    /* run and take care of items that want to be retried immediately */
    do
      {
        ret = wq->spec.workfunc (wq, item->data);
        item->ran++;
      }
    while ((ret == WQ_RETRY_NOW) 
           && (item->ran < wq->spec.max_retries));

    switch (ret)
      {
      case WQ_QUEUE_BLOCKED:
        {
          /* decrement item->ran again, cause this isn't an item
           * specific error, and fall through to WQ_RETRY_LATER
           */
          item->ran--;
        }
      case WQ_RETRY_LATER:
	{
	  goto stats;
	}
      case WQ_REQUEUE:
	{
	  item->ran--;
	  work_queue_item_requeue (wq, node);
	  break;
	}
      case WQ_RETRY_NOW:
        /* a RETRY_NOW that gets here has exceeded max_tries, same as ERROR */
      case WQ_ERROR:
	{
	  if (wq->spec.errorfunc)
	    wq->spec.errorfunc (wq, item);
	}
	/* fall through here is deliberate */
      case WQ_SUCCESS:
      default:
	{
	  work_queue_item_remove (wq, node);
	  break;
	}
      }

    /* completed cycle */
    cycles++;

    /* test if we should yield */
    if ( !(cycles % wq->cycles.granularity) 
        && thread_should_yield (thread))
      {
        yielded = 1;
        goto stats;
      }
  }

stats:

#define WQ_HYSTERESIS_FACTOR 4

  /* we yielded, check whether granularity should be reduced */
  if (yielded && (cycles < wq->cycles.granularity))
    {
      wq->cycles.granularity = ((cycles > 0) ? cycles 
                                             : WORK_QUEUE_MIN_GRANULARITY);
    }
  /* otherwise, should granularity increase? */
  else if (cycles >= (wq->cycles.granularity))
    {
      if (cycles > wq->cycles.best)
        wq->cycles.best = cycles;
      
      /* along with yielded check, provides hysteresis for granularity */
      if (cycles > (wq->cycles.granularity * WQ_HYSTERESIS_FACTOR
                                           * WQ_HYSTERESIS_FACTOR))
        wq->cycles.granularity *= WQ_HYSTERESIS_FACTOR; /* quick ramp-up */
      else if (cycles > (wq->cycles.granularity * WQ_HYSTERESIS_FACTOR))
        wq->cycles.granularity += WQ_HYSTERESIS_FACTOR;
    }
#undef WQ_HYSTERIS_FACTOR
  
  wq->runs++;
  wq->cycles.total += cycles;

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