[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 <zebra.h>
#include "thread.h"
#include "memory.h"
#include "workqueue.h"
#include "linklist.h"
#include "command.h"
#include "log.h"

DEFINE_MTYPE(LIB, WORK_QUEUE,             "Work queue")
DEFINE_MTYPE_STATIC(LIB, WORK_QUEUE_ITEM, "Work queue item")
DEFINE_MTYPE_STATIC(LIB, WORK_QUEUE_NAME, "Work queue name string")

/* master list of work_queues */
static struct list _work_queues;
/* pointer primarily to avoid an otherwise harmless warning on
 * ALL_LIST_ELEMENTS_RO
 */
static struct list *work_queues = &_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;
  new->spec.yield = THREAD_YIELD_TIME_SLOT;
    
  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;
}

bool
work_queue_is_scheduled (struct work_queue *wq)
{
  return (wq->thread != NULL);
}

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, NULL);
      /* set thread yield time, if needed */
      if (wq->thread && wq->spec.yield != THREAD_YIELD_TIME_SLOT)
        thread_set_yield_time (wq->thread, wq->spec.yield);
      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 %8s %21s%s",
           ' ', "List","(ms) ","Q. Runs","Yields","Cycle Counts   ",
           VTY_NEWLINE);
  vty_out (vty,
           "%c %8s %5s %8s %8s %7s %6s %8s %6s %s%s",
           'P',
           "Items",
           "Hold",
           "Total","Total",
           "Best","Gran.","Total","Avg.",
           "Name", 
           VTY_NEWLINE);
 
  for (ALL_LIST_ELEMENTS_RO (work_queues, node, wq))
    {
      vty_out (vty,"%c %8d %5d %8ld %8ld %7d %6d %8ld %6u %s%s",
               (CHECK_FLAG (wq->flags, WQ_UNPLUGGED) ? ' ' : 'P'),
               listcount (wq->items),
               wq->spec.hold,
               wq->runs, wq->yields,
               wq->cycles.best, wq->cycles.granularity, wq->cycles.total,
                 (wq->runs) ? 
                   (unsigned int) (wq->cycles.total / wq->runs) : 0,
               wq->name,
               VTY_NEWLINE);
    }
    
  return CMD_SUCCESS;
}

void
workqueue_cmd_init (void)
{
  install_element (VIEW_NODE, &show_work_queues_cmd);
}

/* '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 run
   * listnode processing == 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);
      /* If a single node is being used with a meta-queue (e.g., zebra),
       * update the next node as we don't want to exit the thread and
       * reschedule it after every node. By definition, WQ_REQUEUE is
       * meant to continue the processing; the yield logic will kick in
       * to terminate the thread when time has exceeded.
       */
      if (nnode == NULL)
        nnode = 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 (yielded)
    wq->yields++;

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