Revised thread/timer handling, work queue and scheduling.

Updated quagga thread handling to use qtimers when using the new
qpnexus -- so all timers are qtimers in the new scheme.

Updated work queue handling so that each work queue item is a single
malloced structure, not three.  (Only bgpd and zebra use the work
queue system.)

When using qpnexus the background thread queue is no longer a timer
queue, but simply a list of pending background threads.  When a
background thread is waiting on a timer, it is in the qtimer pile,
same like any other thread.

When using qpnexus, the only remaining quagga thread queues are the
event and ready queues.

Revised the qpnexus loop so that only when there is nothing else to
do will it consider the background threads.

Revised write I/O in the BGP Engine so that all writing is via the
connection's write buffer.  Revised the write I/O in the Routeing
Engine, so that it passes groups of updates in a single mqueue
message.  This all reduces the number of TCP packets sent (because
BGP messages are collected together in the connection's write buffer)
and reduces the number of mqueue messages involved.
(No need for TCP_CORK.)

Code and comments review for the new code.

  modified:   bgpd/bgp_advertise.c
  modified:   bgpd/bgp_common.h
  modified:   bgpd/bgp_connection.c
  modified:   bgpd/bgp_connection.h
  modified:   bgpd/bgp_engine.h
  modified:   bgpd/bgp_fsm.c
  modified:   bgpd/bgp_main.c
  modified:   bgpd/bgp_msg_read.c
  modified:   bgpd/bgp_msg_write.c
  modified:   bgpd/bgp_network.c
  modified:   bgpd/bgp_packet.c
  modified:   bgpd/bgp_packet.h
  modified:   bgpd/bgp_peer.c
  modified:   bgpd/bgp_peer_index.h
  modified:   bgpd/bgp_route.c
  modified:   bgpd/bgp_route_refresh.h
  modified:   bgpd/bgp_session.c
  modified:   bgpd/bgp_session.h
  modified:   bgpd/bgpd.c
  new file:   bgpd/bgpd.cx
  modified:   lib/mqueue.h
  modified:   lib/qpnexus.c
  modified:   lib/qpnexus.h
  modified:   lib/qpselect.c
  modified:   lib/qtimers.c
  modified:   lib/qtimers.h
  modified:   lib/sigevent.c
  modified:   lib/stream.c
  modified:   lib/stream.h
  modified:   lib/thread.c
  modified:   lib/thread.h
  modified:   lib/workqueue.c
  modified:   lib/workqueue.h
  modified:   tests/heavy-wq.c
  modified:   zebra/zebra_rib.c

1 files changed, 44 insertions, 41 deletions

diff --git a/bgpd/bgp_route.c b/bgpd/bgp_route.c
index 7d3ad901..c5191b18 100644
--- a/bgpd/bgp_route.c
+++ b/bgpd/bgp_route.c
@@ -1457,10 +1457,12 @@ struct bgp_process_queue
   safi_t safi;
 };
 
+WQ_ARGS_SIZE_OK(bgp_process_queue) ;
+
 static wq_item_status
-bgp_process_rsclient (struct work_queue *wq, void *data)
+bgp_process_rsclient (struct work_queue *wq, work_queue_item item)
 {
-  struct bgp_process_queue *pq = data;
+  struct bgp_process_queue *pq = work_queue_item_args(item) ;
   struct bgp *bgp = pq->bgp;
   struct bgp_node *rn = pq->rn;
   afi_t afi = pq->afi;
@@ -1518,9 +1520,9 @@ bgp_process_rsclient (struct work_queue *wq, void *data)
 }
 
 static wq_item_status
-bgp_process_main (struct work_queue *wq, void *data)
+bgp_process_main (struct work_queue *wq, work_queue_item item)
 {
-  struct bgp_process_queue *pq = data;
+  struct bgp_process_queue *pq = work_queue_item_args(item) ;
   struct bgp *bgp = pq->bgp;
   struct bgp_node *rn = pq->rn;
   afi_t afi = pq->afi;
@@ -1592,15 +1594,14 @@ bgp_process_main (struct work_queue *wq, void *data)
 }
 
 static void
-bgp_processq_del (struct work_queue *wq, void *data)
+bgp_processq_del (struct work_queue *wq, work_queue_item item)
 {
-  struct bgp_process_queue *pq = data;
+  struct bgp_process_queue *pq = work_queue_item_args(item);
   struct bgp_table *table = pq->rn->table;
 
   bgp_unlock (pq->bgp);
   bgp_unlock_node (pq->rn);
   bgp_table_unlock (table);
-  XFREE (MTYPE_BGP_PROCESS_QUEUE, pq);
 }
 
 static void
@@ -1617,21 +1618,23 @@ bgp_process_queue_init (void)
       exit (1);
     }
 
-  bm->process_main_queue->spec.workfunc = &bgp_process_main;
-  bm->process_rsclient_queue->spec.workfunc = &bgp_process_rsclient;
-  bm->process_main_queue->spec.del_item_data = &bgp_processq_del;
-  bm->process_rsclient_queue->spec.del_item_data
-    =  bm->process_main_queue->spec.del_item_data;
-  bm->process_main_queue->spec.max_retries
-    = bm->process_main_queue->spec.max_retries = 0;
-  bm->process_rsclient_queue->spec.hold
-    = bm->process_main_queue->spec.hold = 50;
+  bm->process_main_queue->spec.data            = bm->master ;
+  bm->process_main_queue->spec.errorfunc       = NULL ;
+  bm->process_main_queue->spec.workfunc        = &bgp_process_main;
+  bm->process_main_queue->spec.del_item_data   = &bgp_processq_del;
+  bm->process_main_queue->spec.completion_func = NULL ;
+  bm->process_main_queue->spec.max_retries     = 0;
+  bm->process_main_queue->spec.hold            = 50;
+
+  bm->process_rsclient_queue->spec = bm->process_main_queue->spec ;
+  bm->process_rsclient_queue->spec.workfunc    = &bgp_process_rsclient;
 }
 
 void
 bgp_process (struct bgp *bgp, struct bgp_node *rn, afi_t afi, safi_t safi)
 {
   struct bgp_process_queue *pqnode;
+  struct work_queue* wq ;
 
   /* already scheduled for processing? */
   if (CHECK_FLAG (rn->flags, BGP_NODE_PROCESS_SCHEDULED))
@@ -1641,29 +1644,31 @@ bgp_process (struct bgp *bgp, struct bgp_node *rn, afi_t afi, safi_t safi)
        (bm->process_rsclient_queue == NULL) )
     bgp_process_queue_init ();
 
-  pqnode = XCALLOC (MTYPE_BGP_PROCESS_QUEUE,
-                    sizeof (struct bgp_process_queue));
-  if (!pqnode)
-    return;
-
-  /* all unlocked in bgp_processq_del */
-  bgp_table_lock (rn->table);
-  pqnode->rn = bgp_lock_node (rn);
-  pqnode->bgp = bgp;
-  bgp_lock (bgp);
-  pqnode->afi = afi;
-  pqnode->safi = safi;
-
   switch (rn->table->type)
     {
       case BGP_TABLE_MAIN:
-        work_queue_add (bm->process_main_queue, pqnode);
+        wq = bm->process_main_queue ;
         break;
       case BGP_TABLE_RSCLIENT:
-        work_queue_add (bm->process_rsclient_queue, pqnode);
+        wq = bm->process_rsclient_queue ;
         break;
+      default:
+        zabort("invalid rn->table->type") ;
     }
 
+  pqnode = work_queue_item_add(wq);
+
+  if (!pqnode)
+    return;
+
+  /* all unlocked in bgp_processq_del */
+  bgp_table_lock (rn->table);
+  pqnode->rn   = bgp_lock_node (rn);
+  pqnode->bgp  = bgp;
+  bgp_lock (bgp);
+  pqnode->afi  = afi;
+  pqnode->safi = safi;
+
   return;
 }
 
@@ -2672,17 +2677,18 @@ bgp_soft_reconfig_in (struct peer *peer, afi_t afi, safi_t safi)
 	bgp_soft_reconfig_table (peer, afi, safi, table);
 }
 
-
 struct bgp_clear_node_queue
 {
   struct bgp_node *rn;
   enum bgp_clear_route_type purpose;
 };
 
+WQ_ARGS_SIZE_OK(bgp_clear_node_queue) ;
+
 static wq_item_status
-bgp_clear_route_node (struct work_queue *wq, void *data)
+bgp_clear_route_node (struct work_queue *wq, work_queue_item item)
 {
-  struct bgp_clear_node_queue *cnq = data;
+  struct bgp_clear_node_queue *cnq = work_queue_item_args(item) ;
   struct bgp_node *rn = cnq->rn;
   struct peer *peer = wq->spec.data;
   struct bgp_info *ri;
@@ -2708,15 +2714,14 @@ bgp_clear_route_node (struct work_queue *wq, void *data)
 }
 
 static void
-bgp_clear_node_queue_del (struct work_queue *wq, void *data)
+bgp_clear_node_queue_del (struct work_queue *wq, work_queue_item item)
 {
-  struct bgp_clear_node_queue *cnq = data;
+  struct bgp_clear_node_queue *cnq = work_queue_item_args(item) ;
   struct bgp_node *rn = cnq->rn;
   struct bgp_table *table = rn->table;
 
   bgp_unlock_node (rn);
   bgp_table_unlock (table);
-  XFREE (MTYPE_BGP_CLEAR_NODE_QUEUE, cnq);
 }
 
 static void
@@ -2823,11 +2828,9 @@ bgp_clear_route_table (struct peer *peer, afi_t afi, safi_t safi,
             /* both unlocked in bgp_clear_node_queue_del */
             bgp_table_lock (rn->table);
             bgp_lock_node (rn);
-            cnq = XCALLOC (MTYPE_BGP_CLEAR_NODE_QUEUE,
-                           sizeof (struct bgp_clear_node_queue));
-            cnq->rn = rn;
+            cnq = work_queue_item_add(peer->clear_node_queue) ;
+            cnq->rn      = rn;
             cnq->purpose = purpose;
-            work_queue_add (peer->clear_node_queue, cnq);
             break;
           }