Nexus to handle old threads and process the bgp local events. Deals

with having 1 or multiple nexus (and pthreads).

4 files changed, 86 insertions, 16 deletions

diff --git a/lib/qpnexus.c b/lib/qpnexus.c
index 98bd767c..270142be 100644
--- a/lib/qpnexus.c
+++ b/lib/qpnexus.c
@@ -121,7 +121,7 @@ qpn_exec(qpn_nexus qpn)
  *
  *   1) Main thread only -- signals.
  *
- *   2) Pending work -- local queue.
+ *   2) Pending work -- event hooks.
  *
  *   3) messages coming from other pthreads -- mqueue_queue.
  *
@@ -145,7 +145,8 @@ qpn_start(void* arg)
   mqueue_block mqb;
   int actions;
   qtime_mono_t now;
-  struct thread thread;
+  qtime_mono_t max_wait;
+  int i;
 
   /* now in our thread, complete initialisation */
   qpn_in_thread_init(qpn);
@@ -157,6 +158,21 @@ qpn_start(void* arg)
       if (qpn->main_thread)
         quagga_sigevent_process ();
 
+      /* max time to wait in pselect */
+      now = qt_get_monotonic();
+      max_wait = now + QTIME(MAX_PSELECT_TIMOUT);
+
+      /* event hooks, if any */
+      for (i = 0; i < NUM_EVENT_HOOK; ++i)
+        {
+          if (qpn->event_hook[i] != NULL)
+            {
+              qtime_mono_t event_wait = qpn->event_hook[i]();
+              if (event_wait > 0 && event_wait < max_wait)
+                max_wait = event_wait;
+            }
+        }
+
       /* drain the message queue, will be in waiting for signal state
        * when it's empty */
       for (;;)
@@ -169,9 +185,8 @@ qpn_start(void* arg)
         }
 
       /* block for some input, output, signal or timeout */
-      now = qt_get_monotonic();
       actions = qps_pselect(qpn->selection,
-          qtimer_pile_top_time(qpn->pile, now + QTIME(MAX_PSELECT_TIMOUT)) );
+          qtimer_pile_top_time(qpn->pile, max_wait));
 
       /* process I/O actions */
       while (actions)
@@ -184,15 +199,6 @@ qpn_start(void* arg)
       while (qtimer_pile_dispatch_next(qpn->pile, now))
         {
         }
-
-      /* legacy threads */
-      /* TODO: legacy threads must not pselect.  How is the pselect above
-       * to know when to timeout for legacy timers? */
-      if (qpn->master != NULL)
-        {
-          if (thread_fetch (qpn->master, &thread))
-            thread_call (&thread);
-        }
     }
 
   /* last bit of code to run in this thread */
diff --git a/lib/qpnexus.h b/lib/qpnexus.h
index c717ed23..32219df1 100644
--- a/lib/qpnexus.h
+++ b/lib/qpnexus.h
@@ -53,6 +53,9 @@
 /* signal for message queues */
 #define SIGMQUEUE SIGUSR2
 
+/* number of event hooks */
+#define NUM_EVENT_HOOK 2
+
 /*==============================================================================
  * Data Structures.
  */
@@ -80,9 +83,6 @@ struct qpn_nexus
   mqueue_queue queue;
   mqueue_thread_signal mts;
 
-  /* legacy threads */
-  struct thread_master *master;
-
   /* qpthread routine, can override */
   void* (*start)(void*);
 
@@ -95,6 +95,12 @@ struct qpn_nexus
    * thread loop is no longer executed */
   void (*in_thread_final)(void);
 
+  /* thread loop events, can override.  Called before message queue,
+   * I/O and timers.
+   * Returns the time to try again, 0 means default to maximum.
+   */
+  qtime_mono_t (*event_hook[NUM_EVENT_HOOK])(void);
+
 };
 
 /*==============================================================================
diff --git a/lib/thread.c b/lib/thread.c
index 773e9338..62f9669f 100644
--- a/lib/thread.c
+++ b/lib/thread.c
@@ -1028,6 +1028,61 @@ thread_fetch (struct thread_master *m, struct thread *fetch)
     }
 }
 
+
+/* Fetch next ready thread.  Events and timeouts only.  No I/O.
+ * If nothing to do returns NULL and sets event_wait to recommended time
+ * to be called again. */
+struct thread *
+thread_fetch_event (struct thread_master *m, struct thread *fetch,
+    qtime_mono_t *event_wait)
+{
+  struct thread *thread;
+  struct timeval timer_val;
+  struct timeval timer_val_bg;
+  struct timeval *timer_wait;
+  struct timeval *timer_wait_bg;
+
+  /* Normal event are the next highest priority.  */
+  if ((thread = thread_trim_head (&m->event)) != NULL)
+    return thread_run (m, thread, fetch);
+
+  /* If there are any ready threads from previous scheduler runs,
+   * process top of them.
+   */
+  if ((thread = thread_trim_head (&m->ready)) != NULL)
+    return thread_run (m, thread, fetch);
+
+  /* Calculate select wait timer if nothing else to do */
+  quagga_get_relative (NULL);
+  timer_wait = thread_timer_wait (&m->timer, &timer_val);
+  timer_wait_bg = thread_timer_wait (&m->background, &timer_val_bg);
+
+  if (timer_wait_bg &&
+      (!timer_wait || (timeval_cmp (*timer_wait, *timer_wait_bg) > 0)))
+    timer_wait = timer_wait_bg;
+
+  /* When is the next timer due ? */
+  if (timer_wait)
+    {
+      *event_wait = timeval2qtime(timer_wait);
+      return NULL;
+    }
+
+  /* Check foreground timers.  Historically, they have had higher
+     priority than I/O threads, so let's push them onto the ready
+     list in front of the I/O threads. */
+  quagga_get_relative (NULL);
+  thread_timer_process (&m->timer, &relative_time);
+
+  /* Background timer/events, lowest priority */
+  thread_timer_process (&m->background, &relative_time);
+
+  if ((thread = thread_trim_head (&m->ready)) != NULL)
+    return thread_run (m, thread, fetch);
+
+  return NULL;
+}
+
 unsigned long
 thread_consumed_time (RUSAGE_T *now, RUSAGE_T *start, unsigned long *cputime)
 {
diff --git a/lib/thread.h b/lib/thread.h
index 43d4d12f..b0699650 100644
--- a/lib/thread.h
+++ b/lib/thread.h
@@ -23,6 +23,7 @@
 #define _ZEBRA_THREAD_H
 
 #include <sys/resource.h>
+#include "qtime.h"
 
 struct rusage_t
 {
@@ -194,6 +195,8 @@ extern struct thread *funcname_thread_execute (struct thread_master *,
 extern void thread_cancel (struct thread *);
 extern unsigned int thread_cancel_event (struct thread_master *, void *);
 extern struct thread *thread_fetch (struct thread_master *, struct thread *);
+struct thread * thread_fetch_event (struct thread_master *m, struct thread *fetch,
+    qtime_mono_t *event_wait);
 extern void thread_call (struct thread *);
 extern unsigned long thread_timer_remain_second (struct thread *);
 extern int thread_should_yield (struct thread *);