Problem description:

  At the end of ng_snd_item(), node queue is processed. In certain
netgraph setups deep recursive calls can occur.
  For example this happens, when two nodes are connected and can send
items to each other in both directions. If, for some reason, both nodes
have a lot of items in their queues, then the processing thread will
recurse between these two nodes, delivering items left and right, going
deeper in the stack. Other setups can suffer from deep recursion, too.
The following factors can influence risk of deep netgraph call:
 - periodical write-access events on node
 - combination of slow link and fast one in one graph
 - net.inet.ip.fastforwarding

Changes made:

 - In ng_acquire_{read,write}() do not dequeue another item. Instead,
   call ng_setisr() for this node.
 - At the end of ng_snd_item(), do not process queue. Call ng_setisr(),
   if there are any dequeueable items on node queue.
 - In ng_setisr() narrow worklist mutex holding.
 - In ng_setisr() assert queue mutex.

Theoretically, the first two changes should negatively affect performance.
To check this, some profiling was made:

1) In general real tasks, no noticable performance difference was found.

2) The following test was made: two multithreaded nodes and one
single-threaded were connected into a ring. A large queues of packets
were sent around this ring. Time to pass the ring N times was measured.
This is a very vacuous test: no items/mbufs are allocated, no upcalls or
downcalls outside of netgraph. It doesn't represent a real load, it is
a stress test for ng_acquire_{read,write}() and item queueing functions.
Surprisingly, the performance impact was positive! New code is 13% faster
on UP and 17% faster on SMP, in this particular test.

The problem was originally found, described, analyzed and original patch
was written by Roselyn Lee from Vernier Networks. Thanks!

Submitted by:		Roselyn Lee <rosel verniernetworks com>

1 files changed, 17 insertions, 67 deletions

diff --git a/sys/netgraph/ng_base.c b/sys/netgraph/ng_base.c
index 5a9f8d9..aa73479 100644
--- a/sys/netgraph/ng_base.c
+++ b/sys/netgraph/ng_base.c
@@ -1973,14 +1973,11 @@ ng_acquire_read(struct ng_queue *ngq, item_p item)
 	 */
 	item->el_flags |= NGQF_READER;
 	ng_queue_rw(ngq, item, NGQRW_R);
-
-	/*
-	 * Ok, so that's the item successfully queued for later. So now we
-	 * see if we can dequeue something to run instead.
-	 */
-	item = ng_dequeue(ngq);
+	if (CAN_GET_WORK(ngq->q_flags))
+		ng_setisr(ngq->q_node);
 	mtx_unlock_spin(&(ngq->q_mtx));
-	return (item);
+
+	return (NULL);
 }
 
 static __inline item_p
@@ -2009,14 +2006,11 @@ restart:
 	 */
 	item->el_flags &= ~NGQF_RW;
 	ng_queue_rw(ngq, item, NGQRW_W);
-
-	/*
-	 * Ok, so that's the item successfully queued for later. So now we
-	 * see if we can dequeue something to run instead.
-	 */
-	item = ng_dequeue(ngq);
+	if (CAN_GET_WORK(ngq->q_flags))
+		ng_setisr(ngq->q_node);
 	mtx_unlock_spin(&(ngq->q_mtx));
-	return (item);
+
+	return (NULL);
 }
 
 static __inline void
@@ -2272,58 +2266,11 @@ ng_snd_item(item_p item, int flags)
 		return (error);
 	}
 
-	/*
-	 * Now we've handled the packet we brought, (or a friend of it) let's
-	 * look for any other packets that may have been queued up. We hold
-	 * no locks, so if someone puts something in the queue after
-	 * we check that it is empty, it is their problem
-	 * to ensure it is processed. If we have the netisr thread cme in here
-	 * while we still say we have stuff to do, we may get a boost
-	 * in SMP systems. :-)
-	 */
-	for (;;) {
-		/*
-		 * dequeue acquires and adjusts the input_queue as it dequeues
-		 * packets. It acquires the rw lock as needed.
-		 */
-		mtx_lock_spin(&ngq->q_mtx);
-		item = ng_dequeue(ngq); /* fixes worklist too*/
-		if (!item) {
-			mtx_unlock_spin(&ngq->q_mtx);
-			return (error);
-		}
-		mtx_unlock_spin(&ngq->q_mtx);
-
-		/*
-		 * Take over the reference frm the item.
-		 * Hold it until the called function returns.
-		 */
-
-		NGI_GET_NODE(item, node); /* zaps stored node */
-
-		/*
-		 * We have the appropriate lock, so run the item.
-		 * When finished it will drop the lock accordingly
-		 */
-		ierror = ng_apply_item(node, item);
-
-		/*
-		 * only return an error if it was our initial
-		 * item.. (compat hack)
-		 */
-		if (oitem == item) {
-			error = ierror;
-		}
+	mtx_lock_spin(&(ngq->q_mtx));
+	if (CAN_GET_WORK(ngq->q_flags))
+		ng_setisr(ngq->q_node);
+	mtx_unlock_spin(&(ngq->q_mtx));
 
-		/*
-		 * If the node goes away when we remove the reference, 
-		 * whatever we just did caused it.. whatever we do, DO NOT
-		 * access the node again!
-		 */
-		if (NG_NODE_UNREF(node) == 0) {
-			break;
-		}
-	}
 	return (error);
 }
 
@@ -3311,17 +3258,20 @@ ng_worklist_remove(node_p node)
 static void
 ng_setisr(node_p node)
 {
-	mtx_lock_spin(&ng_worklist_mtx);
+
+	mtx_assert(&ngq->q_mtx, MA_OWNED);
+
 	if ((node->nd_flags & NGF_WORKQ) == 0) {
 		/*
 		 * If we are not already on the work queue,
 		 * then put us on.
 		 */
 		node->nd_flags |= NGF_WORKQ;
+		mtx_lock_spin(&ng_worklist_mtx);
 		TAILQ_INSERT_TAIL(&ng_worklist, node, nd_work);
+		mtx_unlock_spin(&ng_worklist_mtx);
 		NG_NODE_REF(node); /* XXX fafe in mutex? */
 	}
-	mtx_unlock_spin(&ng_worklist_mtx);
 	schednetisr(NETISR_NETGRAPH);
 }