Merging of euro_ix branch into pipework.

Bring in fixes for bgp dumping with pthreaded BGP Engine.

Bring in new "show nexus xxx" command.

Fix removal of '~' prompt.

17 files changed, 1025 insertions, 188 deletions

diff --git a/lib/memtypes.c b/lib/memtypes.c
index 66d65328..11adeced 100644
--- a/lib/memtypes.c
+++ b/lib/memtypes.c
@@ -140,7 +140,8 @@ struct memory_list memory_list_bgp[] =
   { MTYPE_AS_PATH,		"BGP aspath"			},
   { MTYPE_AS_SEG,		"BGP aspath seg"		},
   { MTYPE_AS_SEG_DATA,		"BGP aspath segment data"	},
-  { MTYPE_AS_STR,		"BGP aspath str"		},
+  { MTYPE_AS_STR,               "BGP aspath str"                },
+  { MTYPE_BGP_DUMP,             "BGP MRT dump item"             },
   { 0, NULL },
   { MTYPE_BGP_TABLE,		"BGP table"			},
   { MTYPE_BGP_NODE,		"BGP node"			},
diff --git a/lib/qfstring.c b/lib/qfstring.c
index eda2aa9a..a4ddb803 100644
--- a/lib/qfstring.c
+++ b/lib/qfstring.c
@@ -222,7 +222,7 @@ qfs_append_n(qf_str qfs, const char* src, uint n)
 } ;
 
 /*------------------------------------------------------------------------------
- * Append upto 'n' copies of the given character to the qf_str.
+ * Append upto 'n' copies of the given character to the qf_str
  *
  * May append nothing at all !
  *
@@ -1374,3 +1374,459 @@ qfs_arg_float(qf_str qfs, va_list* p_va, const char* start, size_t flen,
 
   return (want >= 0) ? pfp_done : pfp_failed ;
 } ;
+
+/*==============================================================================
+ * Construction of scaled numbers.
+ *
+ *
+ *
+ *
+ */
+
+enum { scale_max = 6 } ;
+
+static const char* scale_d_tags [] =
+{
+    [0] = " " ,
+    [1] = "k",
+    [2] = "m",
+    [3] = "g",
+    [4] = "t",          /* Tera 10^12   */
+    [5] = "p",          /* Peta 10^15   */
+    [6] = "e",          /* Exa  10^18   */
+} ;
+CONFIRM((sizeof(scale_d_tags) / sizeof(char*)) == (scale_max + 1)) ;
+
+static const char* scale_b_tags [] =
+{
+    [0] = "   " ,
+    [1] = "KiB",
+    [2] = "MiB",
+    [3] = "GiB",
+    [4] = "TiB",
+    [5] = "PiB",
+    [6] = "EiB",
+} ;
+CONFIRM((sizeof(scale_b_tags) / sizeof(char*)) == (scale_max + 1)) ;
+
+static const ulong p10 [] =
+{
+    [ 0] = 1l,
+    [ 1] = 10l,
+    [ 2] = 100l,
+    [ 3] = 1000l,
+    [ 4] = 10000l,
+    [ 5] = 100000l,
+    [ 6] = 1000000l,
+    [ 7] = 10000000l,
+    [ 8] = 100000000l,
+    [ 9] = 1000000000l,
+    [10] = 10000000000l,
+    [11] = 100000000000l,
+    [12] = 1000000000000l,
+    [13] = 10000000000000l,
+    [14] = 100000000000000l,
+    [15] = 1000000000000000l,
+    [16] = 10000000000000000l,
+    [17] = 100000000000000000l,
+    [18] = 1000000000000000000l,
+    [19] = (ulong)LONG_MAX + 1,         /* all signed values < this     */
+} ;
+CONFIRM((sizeof(p10) / sizeof(ulong)) == ((scale_max * 3) + 2)) ;
+CONFIRM((LONG_MAX / 10) < 1000000000000000000l) ;
+
+static const long q10 [] =
+{
+    [ 0] = 1l / 2,
+    [ 1] = 10l / 2,
+    [ 2] = 100l / 2,
+    [ 3] = 1000l / 2,
+    [ 4] = 10000l / 2,
+    [ 5] = 100000l / 2,
+    [ 6] = 1000000l / 2,
+    [ 7] = 10000000l / 2,
+    [ 8] = 100000000l / 2,
+    [ 9] = 1000000000l / 2,
+    [10] = 10000000000l / 2,
+    [11] = 100000000000l / 2,
+    [12] = 1000000000000l / 2,
+    [13] = 10000000000000l / 2,
+    [14] = 100000000000000l / 2,
+    [15] = 1000000000000000l / 2,
+    [16] = 10000000000000000l / 2,
+    [17] = 100000000000000000l / 2,
+    [18] = 1000000000000000000l / 2,
+} ;
+CONFIRM((sizeof(q10) / sizeof(long)) == ((scale_max * 3) + 1)) ;
+
+static void qfs_form_scaled(qf_str qfs, long v, uint f, uint d, const char* tag,
+                                                          enum pf_flags flags) ;
+
+/*------------------------------------------------------------------------------
+ * Form value scaled to 4 significant digits, or as simple decimal.
+ *
+ * When scaling, scale by powers of 1,000, to produce:
+ *
+ *    0..999                as simple 1, 2 or 3 digits, followed by " "
+ *    1,000..9,999          as 4 digits with comma, followed by " "
+ *
+ *    10,000..99,994        as 99.99k -- rounded
+ *    99,995..999,949       as 999.9k -- rounded
+ *    999,950..9,999,499    as 9,999k -- rounded
+ *
+ *    thereafter, as for 'k', but with 'm', 'g', etc.
+ *
+ * When not scaling, produce simple decimal with no trailing space.
+ *
+ * In any case, produce a leading sign if required.
+ *
+ * Accepts the following pf_xxx flags:
+ *
+ *   pf_scale    -- scale as above (if not, no scaling)
+ *   pf_trailing -- include blank scale for units
+ *   pf_commas   -- format with commas -- implied if pf_scale
+ *   pf_plus     -- add '+' sign if not -ve
+ *   pf_space    -- add ' ' "sign" if not -ve
+ */
+extern qfs_num_str_t
+qfs_dec_value(long val, enum pf_flags flags)
+{
+  qfs_num_str_t num ;
+  qf_str_t qfs ;
+
+  qfs_init(qfs, num.str, sizeof(num.str)) ;
+
+  flags &= (pf_commas | pf_plus | pf_space | pf_scale | pf_trailing) ;
+
+  if ((flags & pf_scale) == 0)
+    {
+      qfs_signed(qfs, val, flags & ~pf_trailing, 0, 0) ;
+    }
+  else
+    {
+      int s ;
+      uint    i, d ;
+      ldiv_t  r ;
+
+      i = 0 ;
+      d = 0 ;
+
+      if (val >= 0)
+        s = +1 ;
+      else
+        {
+          s = -1 ;
+          val = -val ;
+        } ;
+
+      /* Find the power of 1,000 which val is ... */
+
+      while (((ulong)val >= p10[i + 4]) && (i < ((scale_max - 1) * 3)))
+        i += 3 ;
+
+      if (i == 0)
+        {
+          r.quot = val ;
+          r.rem  = 0 ;
+        }
+      else
+        {
+          /* Maximum i == (scale_max - 1) * 3 -- and have p10 upto and
+           * including scale_max * 3.
+           */
+          if      ((ulong)val < p10[i + 1])
+            d = 3 ;
+          else if ((ulong)val < p10[i + 2])
+            d = 2 ;
+          else if ((ulong)val < p10[i + 3])
+            d = 1 ;
+          else
+            d = 0 ;
+
+          /* Scale down to required number of decimals and round.
+           *
+           * If is thousands, then i = 3, if value = 10,000 (smallest possible)
+           * then d == 2.  So divide by 5 (q10[3 - 2]) to make ls bit the
+           * rounding bit, add one and shift off the rounding bit.
+           *
+           * The result should be 1000..9999, unless value is greater than our
+           * ability to scale, or has rounded up one decade.
+           */
+          val = ((val / q10[i - d]) + 1) >> 1 ;
+
+          qassert(val >= 1000) ;
+
+          if (val > 9999)
+            {
+              if (d == 0)
+                {
+                  if (i < (scale_max * 3))
+                    {
+                      qassert(val == 10000) ;
+
+                      val = 1000 ;
+                      d   = 2 ;
+                      i  += 3 ;
+                    } ;
+                }
+              else
+                {
+                  qassert(val == 10000) ;
+
+                  val = 1000 ;
+                  d  -= 1 ;
+                } ;
+            } ;
+
+          r = ldiv(val, p10[d]) ;
+        } ;
+
+      qfs_form_scaled(qfs, r.quot * s, r.rem, d, scale_d_tags[i / 3], flags) ;
+    } ;
+
+  qfs_term(qfs) ;
+
+  return num ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Form value scaled to 4 significant digits, or as simple decimal.
+ *
+ * When scaling, scale by powers of 1,024, to produce:
+ *
+ *    0..999                as simple 1, 2 or 3 digits, followed by "   "
+ *    1,000..9,999          as 4 digits with comma, followed by "   "
+ *
+ *    10,000..99,994        as 99.99KiB -- rounded
+ *    99,995..999,949       as 999.9KiB-- rounded
+ *    999,950..9,999,499    as 9,999KiB-- rounded
+ *
+ *    thereafter, as for 'KiB', but with 'MiB', 'GiB', etc.
+ *
+ * When not scaling, produce simple decimal with no trailing space.
+ *
+ * In any case, produce a leading sign if required.
+ *
+ * Accepts the following pf_xxx flags:
+ *
+ *   pf_scale    -- scale as above (if not, no scaling)
+ *   pf_trailing -- include blank scale for units
+ *   pf_commas   -- format with commas -- implied if pf_scale
+ *   pf_plus     -- add '+' sign if not -ve
+ *   pf_space    -- add ' ' "sign" if not -ve
+ */
+extern qfs_num_str_t
+qfs_bin_value(long val, enum pf_flags flags)
+{
+  qfs_num_str_t num ;
+  qf_str_t qfs ;
+
+  qfs_init(qfs, num.str, sizeof(num.str)) ;
+
+  flags &= (pf_commas | pf_plus | pf_space | pf_scale | pf_trailing) ;
+
+  if ((flags & pf_scale) == 0)
+    {
+      qfs_signed(qfs, val, flags & ~pf_trailing, 0, 0) ;
+    }
+  else
+    {
+      int s ;
+
+      ulong v ;
+      uint i, d, f ;
+
+      i = 0 ;
+      d = 0 ;
+      f = 0 ;
+
+      if (val >= 0)
+        s = +1 ;
+      else
+        {
+          s = -1 ;
+          val = -val ;
+        } ;
+
+      v = val ;
+      while ((v >= 1024) && (i < scale_max))
+        {
+          v >>= 10 ;            /* find power of 1024 scale     */
+          i += 1 ;
+        } ;
+
+      if (i > 0)
+        {
+          ulong e ;
+          int   is ;
+
+          if      (v < 10)
+            d = 3 ;             /* number of decimals expected  */
+          else if (v < 100)
+            d = 2 ;
+          else if (v < 1000)
+            d = 1 ;
+          else
+            d = 0 ;             /* should be already            */
+
+          /* Scale up to the required number of decimals, shift down so that
+           * only ms bit of fraction is left, round and shift off rounding bit.
+           *
+           * If d != 0, then will scale up by 10, 100 or 1000.  If the value is
+           * greater than ULONG_MAX / 1024, then we do the bottom 10 bits
+           * separately, and scale the calculation down by 10 bits.
+           */
+          v = val ;             /* operate on unsigned v        */
+
+          e  = 0 ;              /* assume no extra bits         */
+          is = i * 10 ;         /* the shift down               */
+
+          if ((d != 0) && (v > (ULONG_MAX >> 10)))
+            {
+              e = (v & 0x3FF) * p10[d] ;        /* take bottom 10 bits  */
+              e >>= 10 ;        /* discard 10 bits of extra part        */
+              v >>= 10 ;        /* scale down value                     */
+              is -= 10 ;        /* reduce shift                         */
+            } ;
+
+          v = ((((v * p10[d]) + e) >> (is - 1)) + 1) >> 1 ;
+
+//        qassert(v >= 1000) ;
+
+          if (d == 0)
+            {
+              if ((v == 1024) && (i < scale_max))
+                {
+                  v  = 1000 ;       /* rounded up to next power of 1024     */
+                  d  = 3 ;
+                  i += 1 ;
+                }
+            }
+          else
+            {
+              if (v >= 10000)
+                {
+//                qassert(v == 10000) ;
+
+                  v  = 1000 ;       /* rounded up to one less decimals      */
+                  d -= 1 ;
+                } ;
+            } ;
+
+          val = v / p10[d] ;
+          f   = v % p10[d] ;
+       } ;
+
+      qfs_form_scaled(qfs, val * s, f, d, scale_d_tags[i / 3], flags) ;
+    } ;
+
+  qfs_term(qfs) ;
+
+  return num ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Form a time period value.
+ *
+ *    +/-999d99h99m99h99.999s
+ *
+ * Accepts the following pf_xxx flags:
+ *
+ *   pf_commas   -- format with commas (very unlikely !)
+ *   pf_plus     -- add '+' sign if not -ve
+ *   pf_space    -- add ' ' "sign" if not -ve
+ */
+extern qfs_num_str_t
+qfs_time_period(qtime_t val, enum pf_flags flags)
+{
+  qfs_num_str_t num ;
+  qf_str_t qfs ;
+  int s ;
+  int w ;
+
+  qfs_init(qfs, num.str, sizeof(num.str)) ;
+
+  flags &= (pf_commas | pf_plus | pf_space) ;
+
+  if (val >= 0)
+    s = +1 ;
+  else
+    {
+      s = -1 ;
+      val = -val ;
+    } ;
+
+  /* Round value to milli seconds
+   */
+  val = (val + (QTIME_SECOND / 2000)) / (QTIME_SECOND / 1000) ;
+
+  w = 0 ;
+
+  if (val >= (2 * 24 * 60 * 60 * 1000))
+    {
+      qfs_signed(qfs, (val / (24 * 60 * 60 * 1000)) * s, flags, w, w) ;
+      qfs_append_ch(qfs, 'd') ;
+
+      val %= (24 * 60 * 60 * 1000) ;
+      s = 1 ;
+      flags = pf_zeros ;
+      w = 2 ;
+    } ;
+
+  if ((val >= (2 * 60 * 60 * 1000)) || (w > 0))
+    {
+      qfs_signed(qfs, (val / (60 * 60 * 1000)) * s, flags, w, w) ;
+      qfs_append_ch(qfs, 'h') ;
+
+      val %= (60 * 60 * 1000) ;
+      s = 1 ;
+      flags = pf_zeros ;
+      w = 2 ;
+    } ;
+
+  if ((val >= (2 * 60 * 1000)) || (w > 0))
+    {
+      qfs_signed(qfs, (val / (60 * 1000)) * s, flags, w, w) ;
+      qfs_append_ch(qfs, 'm') ;
+
+      val %= (60 * 1000) ;
+      s = 1 ;
+      flags = pf_zeros ;
+      w = 2 ;
+    } ;
+
+  qfs_signed(qfs, (val / 1000) * s, flags, w, w) ;
+  qfs_append_ch(qfs, '.') ;
+  qfs_unsigned(qfs, val % 1000, pf_zeros, 3, 3) ;
+  qfs_append_ch(qfs, 's') ;
+
+  qfs_term(qfs) ;
+
+  return num ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Form string for number, with commas and "d" decimal digits, followed by
+ * the given tag -- where d = 0..4
+ *
+ * So: val=1234567, d=2, tag="k" -> "12,345.67k".
+ *     val=1234,    d=0, tag=""  -> "1,234"
+ */
+static void
+qfs_form_scaled(qf_str qfs, long v, uint f, uint d, const char* tag,
+                                                            enum pf_flags flags)
+{
+  if (d == 0)
+    qfs_signed(qfs, v, flags | pf_commas, 0, 0) ;
+  else
+    {
+      qfs_signed(qfs, v, flags, 0, 0) ;
+      qfs_append_ch(qfs, '.') ;
+      qfs_unsigned(qfs, f, pf_zeros, d, 0) ;
+    } ;
+
+  if ((*tag != ' ') || ((flags & pf_trailing) != 0))
+    qfs_append(qfs, tag) ;
+} ;
+
+
diff --git a/lib/qfstring.h b/lib/qfstring.h
index be858899..98593685 100644
--- a/lib/qfstring.h
+++ b/lib/qfstring.h
@@ -25,6 +25,11 @@
 #include "misc.h"
 #include "vargs.h"
 
+#include <stddef.h>
+#include <stdint.h>
+
+#include "qtime.h"
+
 /*==============================================================================
  * These "qfstring" address the issues of dealing with *fixed* length
  * strings, particularly where the string handling must be async-signal-safe.
@@ -76,7 +81,11 @@ enum pf_flags
   pf_zeros      = BIT( 3),      /* "0" seen             */
   pf_alt        = BIT( 4),      /* "#" seen             */
 
-  pf_precision  = BIT( 7),      /* '.' seen             */
+  pf_precision  = 1 <<  5,      /* '.' seen             */
+
+  /* For scaled formatting of decimals and byte counts          */
+  pf_scale      = 1 <<  6,
+  pf_trailing   = 1 <<  7,      /* add blank scale if required  */
 
   /* The following signal how to render the value               */
   pf_oct        = BIT( 8),      /* octal                */
@@ -113,7 +122,6 @@ Inline uint  qfs_left(qf_str qfs) ;
 
 extern void qfs_append(qf_str qfs, const char* src) ;
 extern void qfs_append_n(qf_str qfs, const char* src, uint n) ;
-
 extern void qfs_append_ch_x_n(qf_str qfs, char ch, uint n) ;
 extern void qfs_append_justified(qf_str qfs, const char* src, int width) ;
 extern void qfs_append_justified_n(qf_str qfs, const char* src,
@@ -132,6 +140,24 @@ extern uint qfs_vprintf(qf_str qfs, const char *format, va_list args) ;
 
 Inline uint qfs_strlen(const char* str) ;
 
+/* Construction of numbers from long
+ *
+ * Need enough space for groups of 3 decimal digits plus ',' or '\0', and an
+ * extra group for sign and some slack.  For 64 bits comes out at 32 bytes !
+ */
+enum { qfs_number_len = (((64 + 9) / 10) + 1) * (3 + 1) } ;
+
+CONFIRM((sizeof(long) * 8) <= 64) ;
+
+typedef struct qfs_num_str_t
+{
+  char   str[qfs_number_len] ;
+} qfs_num_str_t;
+
+extern qfs_num_str_t qfs_dec_value(long val, enum pf_flags flags) ;
+extern qfs_num_str_t qfs_bin_value(long val, enum pf_flags flags) ;
+extern qfs_num_str_t qfs_time_period(qtime_t val, enum pf_flags flags) ;
+
 /*==============================================================================
  * The Inline functions.
  */
diff --git a/lib/qpnexus.c b/lib/qpnexus.c
index 8a805900..21a4ee8e 100644
--- a/lib/qpnexus.c
+++ b/lib/qpnexus.c
@@ -73,19 +73,23 @@ qpn_self_knowledge(qpn_nexus qpn)
  * Returns the qpn_nexus.
  */
 extern qpn_nexus
-qpn_init_new(qpn_nexus qpn, bool main_thread)
+qpn_init_new(qpn_nexus qpn, bool main_thread, const char* name)
 {
   if (qpn == NULL)
     qpn = XCALLOC(MTYPE_QPN_NEXUS, sizeof(struct qpn_nexus)) ;
   else
     memset(qpn, 0,  sizeof(struct qpn_nexus)) ;
 
+  qpn->name        = name ;
+
   qpn->selection   = qps_selection_init_new(qpn->selection);
   qpn->pile        = qtimer_pile_init_new(qpn->pile);
   qpn->queue       = mqueue_init_new(qpn->queue, mqt_signal_unicast);
   qpn->main_thread = main_thread;
   qpn->start       = qpn_start;
 
+  qpt_spin_init(qpn->stats_slk) ;
+
   if (main_thread)
     {
       qpn->thread_id = qpt_thread_self();
@@ -187,6 +191,7 @@ qpn_exec(qpn_nexus qpn)
  *
  *   6) Timers -- qtimers
  *
+ *   7) Low priority pending work
  */
 static void*
 qpn_start(void* arg)
@@ -198,7 +203,8 @@ qpn_start(void* arg)
   qtime_t      max_wait ;
   unsigned i;
   unsigned done ;
-  unsigned wait ;
+  unsigned prev ;
+  bool     wait ;
 
   /* now in our thread, complete initialisation                         */
   qpn_in_thread_init(qpn);
@@ -208,25 +214,40 @@ qpn_start(void* arg)
     ((qpn_init_function*)(qpn->in_thread_init.hooks[i++]))() ;
 
   /* Until required to terminate, loop                                  */
+  prev = 1 ;
   done = 1 ;
   while (!qpn->terminate)
     {
+      ++qpn->raw.cycles ;
+
       /* Signals are highest priority -- only execute for main thread   */
       if (qpn->main_thread)
-        done |= quagga_sigevent_process() ;
+        {
+          int ret = quagga_sigevent_process() ;
+          if (ret != 0)
+            {
+              ++done ;                  /* count each signal            */
+              ++qpn->raw.signals ;
+            } ;
+        } ;
 
       /* Foreground hooks, if any.                                      */
-      for (i = 0; i < qpn->foreground.count ;)
-        done |= ((qpn_hook_function*)(qpn->foreground.hooks[i++]))() ;
+      for (i = 0; i < qpn->foreground.count ; ++i)
+        {
+          int ret = ((qpn_hook_function*)(qpn->foreground.hooks[i]))() ;
+          if (ret != 0)
+            {
+              ++done ;                  /* count each foreground action */
+              ++qpn->raw.foreg ;
+            } ;
+        } ;
 
       /* take stuff from the message queue
        *
-       * If nothing done the last time around the loop then may wait this
-       * time if the queue is empty first time through.
+       * If nothing done this time and last time around the loop then will
+       * arrange to wait iff the queue is empty first time through.
        */
-      wait = (done == 0) ;      /* may wait this time only if nothing
-                                   found to do on the last pass         */
-      done = 0 ;
+      wait = ((prev == 0) && (done == 0)) ;
       do
         {
           mqb = mqueue_dequeue(qpn->queue, wait, qpn->mts) ;
@@ -235,39 +256,87 @@ qpn_start(void* arg)
 
           mqb_dispatch(mqb, mqb_action);
 
-          ++done ;              /* done another                         */
-          wait = 0 ;            /* done something, so turn off wait     */
+          wait = false ;
+          ++done ;
         } while (done < 200) ;
 
+      qpn->raw.dispatch += done ;
+
       /* block for some input, output, signal or timeout
        *
-       * wait will be true iff did nothing the last time round the loop, and
-       * not found anything to be done up to this point either.
+       * wait will be true iff found nothing to do this and last time round the
+       * loop.
        */
+      now = qt_get_monotonic() ;
+
       if (wait)
-        max_wait = qtimer_pile_top_wait(qpn->pile, QTIME(MAX_PSELECT_WAIT)) ;
+        max_wait = qtimer_pile_top_wait(qpn->pile, QTIME(MAX_PSELECT_WAIT),
+                                                                          now) ;
       else
         max_wait = 0 ;
 
-      actions = qps_pselect(qpn->selection, max_wait) ;
-      done |= actions ;
+      /* We are about to do a pselect, which may wait.  Now is the time to
+       * set the "raw" current time, and publish the stats.
+       */
+      qpn->raw.last_time = now ;
+
+      qpt_spin_lock(qpn->stats_slk) ;
+      qpn->stats = qpn->raw ;
+      qpt_spin_unlock(qpn->stats_slk) ;
+
+      /* Do pselect, which may now wait
+       *
+       * After pselect, if is "wait", then will have set the message queue
+       * waiting, which can now be cleared.  Also, any time since
+       * "raw.last_time" must be counted as idle time.
+       *
+       * Remember current "done" as "prev", and set done 0 or 1, depending on
+       * I/O action count -- processing I/O actions and/or timers is counted as
+       * a single activity.
+       */
+      do
+        actions = qps_pselect(qpn->selection, max_wait) ;
+      while (actions < 0) ;
+
+      now = qt_get_monotonic() ;
 
       if (wait)
-        mqueue_done_waiting(qpn->queue, qpn->mts);
+        {
+          qpn->raw.idle += now - qpn->raw.last_time ;
+
+          mqueue_done_waiting(qpn->queue, qpn->mts) ;
+        } ;
+
+      prev = done ;
+      done = (actions != 0) ? 1 : 0 ;
+      qpn->raw.io_acts += actions ;
 
-      /* process I/O actions                                            */
       while (actions)
         actions = qps_dispatch_next(qpn->selection) ;
 
-      /* process timers                                                 */
-      now = qt_get_monotonic() ;
+      /* process timers -- also counts as one activity
+       */
       while (qtimer_pile_dispatch_next(qpn->pile, now))
-        done = 1 ;
+        {
+          ++qpn->raw.timers ;
+          done = 1 ;                    /* Done something in this pass  */
+        } ;
 
-      /* If nothing done in this pass, see if anything in the background */
-      if (done == 0)
-        for (i = 0; i < qpn->background.count ; ++i)
-          done |= ((qpn_hook_function*)(qpn->background.hooks[i]))() ;
+      /* If nothing done so far in this pass, and nothing in the previous,
+       * see if anything in the background
+       */
+      if ((prev == 0) && (done == 0))
+        {
+          for (i = 0; i < qpn->background.count ; ++i)
+            {
+              int ret = ((qpn_hook_function*)(qpn->background.hooks[i]))() ;
+              if (ret != 0)
+                {
+                  ++done ;
+                  ++qpn->raw.backg ;
+                } ;
+            } ;
+        } ;
     } ;
 
   /* custom in-thread finalization                                      */
@@ -285,6 +354,14 @@ qpn_in_thread_init(qpn_nexus qpn)
 {
   sigset_t sigmask[1];
 
+  memset(&qpn->raw, 0, sizeof(qpn_stats_t)) ;
+  qpn->raw.start_time = qt_get_monotonic() ;
+  qpn->raw.last_time  = qpn->raw.start_time ;
+
+  qpt_spin_lock(qpn->stats_slk) ;
+  qpn->prev_stats = qpn->stats = qpn->raw ;
+  qpt_spin_unlock(qpn->stats_slk) ;
+
   qpn->thread_id = qpt_thread_self();
   qpn_self_knowledge(qpn) ;
 
@@ -332,4 +409,22 @@ qpn_terminate(qpn_nexus qpn)
       if (qpthreads_enabled)
         qpt_thread_signal(qpn->thread_id, SIG_INTERRUPT);
     } ;
-}
+} ;
+
+/*------------------------------------------------------------------------------
+ * Get a copy of the current stats for the given nexus.
+ *
+ * This copies the stats to the "prev_stats" area in the nexus.
+ */
+extern void
+qpn_get_stats(qpn_nexus qpn, qpn_stats curr, qpn_stats prev)
+{
+  qpt_spin_lock(qpn->stats_slk) ;
+
+  *prev = qpn->prev_stats ;
+  *curr = qpn->stats ;
+
+  qpn->prev_stats = qpn->stats ;
+
+  qpt_spin_unlock(qpn->stats_slk) ;
+} ;
diff --git a/lib/qpnexus.h b/lib/qpnexus.h
index 38dcbebc..15125150 100644
--- a/lib/qpnexus.h
+++ b/lib/qpnexus.h
@@ -24,11 +24,9 @@
 
 #include "misc.h"
 #include <time.h>
-#include <unistd.h>
 #include <errno.h>
 #include <signal.h>
 
-#include "zassert.h"
 #include "qpthreads.h"
 #include "qtimers.h"
 #include "mqueue.h"
@@ -67,11 +65,32 @@ struct qpn_hook_list
   unsigned  count ;
 } ;
 
+typedef struct qpn_stats
+{
+  qtime_mono_t  start_time ;
+  qtime_mono_t  last_time ;
+  qtime_mono_t  idle ;
+
+  ulong   cycles ;
+  ulong   signals ;
+  ulong   foreg ;
+  ulong   dispatch ;
+  ulong   io_acts ;
+  ulong   timers ;
+  ulong   backg ;
+
+} qpn_stats_t ;
+
+typedef qpn_stats_t* qpn_stats ;
+
 typedef struct qpn_nexus* qpn_nexus ;
 
 struct qpn_nexus
 {
-  /* set true to terminate the thread (eventually)      */
+  /* name of thread                                             */
+  const char* name ;
+
+  /* set true to terminate the thread (eventually) */
   bool terminate;
 
   /* true if this is the main thread                    */
@@ -141,6 +160,16 @@ struct qpn_nexus
    * and return.  MUST return 0 iff there is no more work to do.
    */
   struct qpn_hook_list background ;
+
+
+  /* statistics gathering
+   */
+  qpt_spin_t    stats_slk[1] ;
+
+  qpn_stats_t   raw ;           /* belongs to thread                    */
+  qpn_stats_t   stats ;         /* set, under spin lock, once per cycle */
+  qpn_stats_t   prev_stats ;    /* set, under spin lock, each time stats
+                                 * are fetched.                         */
 };
 
 /*------------------------------------------------------------------------------
@@ -159,10 +188,13 @@ qpn_find_self(void)
  * Functions
  */
 extern void qpn_init(void) ;
-extern qpn_nexus qpn_init_new(qpn_nexus qpn, bool main_thread);
+extern qpn_nexus qpn_init_new(qpn_nexus qpn, bool main_thread,
+                                                             const char* name) ;
 extern void qpn_add_hook_function(qpn_hook_list list, void* hook) ;
 extern void qpn_exec(qpn_nexus qpn);
 extern void qpn_terminate(qpn_nexus qpn);
 extern qpn_nexus qpn_reset(qpn_nexus qpn, free_keep_b free_structure);
 
+extern void qpn_get_stats(qpn_nexus qpn, qpn_stats curr, qpn_stats prev) ;
+
 #endif /* _ZEBRA_QPNEXUS_H */
diff --git a/lib/qpselect.c b/lib/qpselect.c
index 3332db4f..5ac725b9 100644
--- a/lib/qpselect.c
+++ b/lib/qpselect.c
@@ -136,11 +136,11 @@ static short fd_byte_count[FD_SETSIZE] ;    /* number of bytes for fds 0..fd */
 /* Forward references   */
 static void qps_make_super_set_map(void) ;
 static void qps_selection_re_init(qps_selection qps) ;
-static qps_file qps_file_lookup_fd(qps_selection qps, int fd, qps_file insert) ;
+static qps_file qps_file_lookup_fd(qps_selection qps, fd_t fd, qps_file insert);
 static void qps_file_remove(qps_selection qps, qps_file qf) ;
 static void qps_super_set_zero(fd_super_set* p_set, int n) ;
 static int qps_super_set_cmp(fd_super_set* p_a, fd_super_set* p_b, int n) ;
-static int qps_next_fd_pending(fd_super_set* pending, int fd, int fd_last) ;
+static int qps_next_fd_pending(fd_super_set* pending, fd_t fd, fd_t fd_last) ;
 static void qps_selection_validate(qps_selection qps) ;
 
 /*------------------------------------------------------------------------------
@@ -164,7 +164,7 @@ qps_start_up(void)
  *     is the caller's responsibility to have dealt with its contents before
  *     calling this.
  */
-qps_selection
+extern qps_selection
 qps_selection_init_new(qps_selection qps)
 {
   if (qps == NULL)
@@ -209,7 +209,7 @@ qps_selection_re_init(qps_selection qps)
    *
    * So nothing much else to do:
    */
-  qps->fd_last = -1 ;   /* not an fd in sight.  */
+  qps->fd_last = fd_undef ;     /* not an fd in sight.  */
 } ;
 
 /*------------------------------------------------------------------------------
@@ -222,8 +222,8 @@ qps_selection_re_init(qps_selection qps)
  *
  * Adding a file which is already a member a selection is a FATAL error.
  */
-void
-qps_add_file(qps_selection qps, qps_file qf, int fd, void* file_info)
+extern void
+qps_add_file(qps_selection qps, qps_file qf, fd_t fd, void* file_info)
 {
   passert(qf->selection == NULL) ;
 
@@ -245,7 +245,7 @@ qps_add_file(qps_selection qps, qps_file qf, int fd, void* file_info)
  *
  * When the file is removed it is disabled in all modes.
  */
-void
+extern void
 qps_remove_file(qps_file qf)
 {
   if (qf->selection != NULL)
@@ -274,7 +274,7 @@ qps_remove_file(qps_file qf)
  * NB: once reaming has started, the selection MUST NOT be used for anything,
  *     and the process MUST be run to completion.
  */
-qps_file
+extern qps_file
 qps_selection_ream(qps_selection qps, int free_structure)
 {
   qps_file qf ;
@@ -326,7 +326,7 @@ qps_set_signal(qps_selection qps, const sigset_t* sigmask)
  *
  * The qps_dispatch_next() processes the returns from pselect().
  */
-int
+extern int
 qps_pselect(qps_selection qps, qtime_t max_wait)
 {
   struct timespec ts ;
@@ -423,10 +423,10 @@ qps_pselect(qps_selection qps, qtime_t max_wait)
  *     result bit vectors should be zeroised again.  Also, this allows the
  *     search to proceed from the last known fd -- won't find it again !
  */
-int
+extern int
 qps_dispatch_next(qps_selection qps)
 {
-  int        fd ;
+  fd_t       fd ;
   qps_file   qf ;
   qps_mnum_t mnum ;
 
@@ -489,7 +489,7 @@ qps_dispatch_next(qps_selection qps)
  *
  * Returns the qps_file.
  */
-qps_file
+extern qps_file
 qps_file_init_new(qps_file qf, qps_file template)
 {
   if (qf == NULL)
@@ -534,7 +534,7 @@ qps_file_free(qps_file qf)
       if (qf->selection != NULL)
         qps_remove_file(qf) ;
 
-      if (qf->fd >= 0)
+      if (qf->fd >= fd_first)
         {
           close(qf->fd) ;
           qf->fd = fd_undef ;
@@ -556,7 +556,7 @@ qps_file_free(qps_file qf)
  * NB: It is a FATAL error to enable modes for a file which is not in a
  *     selection.
  */
-void
+extern void
 qps_enable_mode(qps_file qf, qps_mnum_t mnum, qps_action* action)
 {
   qps_mbit_t    mbit = qps_mbit(mnum) ;
@@ -590,7 +590,7 @@ qps_enable_mode(qps_file qf, qps_mnum_t mnum, qps_action* action)
  *
  * NB: it is a fatal error to unset an action for a mode which is enabled.
  */
-void
+extern void
 qps_set_action(qps_file qf, qps_mnum_t mnum, qps_action* action)
 {
   dassert((mnum >= 0) && (mnum <= qps_mnum_count)) ;
@@ -628,7 +628,7 @@ static qps_mnum_t qps_first_mnum[qps_mbit(qps_mnum_count)] =
 
 CONFIRM(qps_mbit(qps_mnum_count) == 8) ;
 
-void
+extern void
 qps_disable_modes(qps_file qf, qps_mbit_t mbits)
 {
   qps_mnum_t  mnum ;
@@ -695,7 +695,7 @@ qps_fd_cmp(const int** pp_fd, const qps_file* p_qf)
  * NB: FATAL error to insert file with same fd as an existing one.
  */
 static qps_file
-qps_file_lookup_fd(qps_selection qps, int fd, qps_file insert)
+qps_file_lookup_fd(qps_selection qps, fd_t fd, qps_file insert)
 {
   qps_file qf ;
   vector_index_t i ;
@@ -773,7 +773,7 @@ static void
 qps_file_remove(qps_selection qps, qps_file qf)
 {
   qps_file qfd ;
-  int   fd_last ;
+  fd_t     fd_last ;
 
   passert((qf->fd >= 0) && (qf->fd <= qps->fd_last) && (qps == qf->selection)) ;
 
@@ -917,13 +917,13 @@ CONFIRM((qps_cc_bit_ord  == 70) || (qps_cc_bit_ord  == 7)) ;
 /* Functions required for the cross check.                              */
 
 static inline int
-qpd_cc_word(int fd)
+qpd_cc_word(fd_t fd)
 {
   return fd / qps_cc_word_bits ;
 } ;
 
 static inline int
-qps_cc_byte(int fd)
+qps_cc_byte(fd_t fd)
 {
   if (qps_cc_byte_ord == 10)
     return (qpd_cc_word(fd) * qps_cc_word_bytes)
@@ -933,7 +933,7 @@ qps_cc_byte(int fd)
 } ;
 
 static inline uint8_t
-qps_cc_bit(int fd)
+qps_cc_bit(fd_t fd)
 {
   if (qps_cc_bit_ord == 70)
     return 0x01 << (fd & 0x7) ;
@@ -942,7 +942,7 @@ qps_cc_bit(int fd)
 } ;
 
 static int
-ccFD_ISSET(int fd, fd_set* set)
+ccFD_ISSET(fd_t fd, fd_set* set)
 {
   return (*((uint8_t*)set + qps_cc_byte(fd)) & qps_cc_bit(fd)) != 0 ;
 } ;
@@ -955,7 +955,7 @@ ccFD_ISSET(int fd, fd_set* set)
  * Returns next fd, or -1 if none.
  */
 static int
-qps_next_fd_pending(fd_super_set* pending, int fd, int fd_last)
+qps_next_fd_pending(fd_super_set* pending, fd_t fd, fd_t fd_last)
 {
   uint8_t b ;
 
@@ -1001,12 +1001,13 @@ static void
 qps_make_super_set_map(void)
 {
   fd_super_set test ;
-  int fd, i, iw, ib ;
+  fd_t fd ;
+  int i, iw, ib ;
 
   /* (1) check that a zeroised fd_super_set is an empty one.    */
   qps_super_set_zero(&test, 1) ;
 
-  for (fd = 0 ; fd < (int)FD_SETSIZE ; ++fd)
+  for (fd = fd_first ; fd < (int)FD_SETSIZE ; ++fd)
     if (FD_ISSET(fd, &test.fdset))
       zabort("Zeroised fd_super_set is not empty") ;
 
@@ -1018,7 +1019,7 @@ qps_make_super_set_map(void)
 
   /* (3) check that setting one fd sets one bit, and construct the      */
   /*     fd_word_map[], fd_byte_map[] and fd_bit_map[].                 */
-  for (fd = 0 ; fd < (int)FD_SETSIZE ; ++fd)
+  for (fd = fd_first ; fd < (int)FD_SETSIZE ; ++fd)
     {
       fd_word_t w ;
 
@@ -1064,7 +1065,7 @@ qps_make_super_set_map(void)
   /*     make sure that have  8 contiguous fd to a byte.                */
   /*     make sure that have 32 contiguous fd to a word.                */
 
-  for (fd = 0 ; fd < (int)FD_SETSIZE ; fd += 8)
+  for (fd = fd_first ; fd < (int)FD_SETSIZE ; fd += 8)
     {
       int fds ;
       ib  = fd_byte_map[fd] ;
@@ -1121,7 +1122,7 @@ qps_make_super_set_map(void)
   /*     include fds 0..fd.                                             */
 
   i = 0 ;
-  for (fd = 0 ; fd < (int)FD_SETSIZE ; ++fd)
+  for (fd = fd_first ; fd < (int)FD_SETSIZE ; ++fd)
     {
       int c = fd_byte_map[fd] + 1 ;
 
@@ -1140,7 +1141,7 @@ qps_make_super_set_map(void)
    * Checking that the maps have been correctly deduced -- where know what
    * the mapping really is !
    */
-  for (fd = 0 ; fd < (int)FD_SETSIZE ; ++fd)
+  for (fd = fd_first ; fd < (int)FD_SETSIZE ; ++fd)
     {
       uint8_t b ;
       short   c ;
@@ -1290,13 +1291,13 @@ qps_super_set_count(fd_super_set* p_set, int n)
 static void
 qps_selection_validate(qps_selection qps)
 {
-  int   fd_last ;
+  fd_t  fd, fd_last ;
   int   enabled_count[qps_mnum_count] ;
   fd_full_set enabled ;
 
   qps_file       qf ;
-  int            fd, n, mnum, p_mnum ;
-  vector_index_t  i ;
+  int            n, mnum, p_mnum ;
+  vector_index_t i ;
 
   /* 1..4)  Run down the selection vector and check.            */
   /*        Collect new enabled_count and enabled bit vectors.  */
diff --git a/lib/qpselect.h b/lib/qpselect.h
index fc9037ce..d56fdf9b 100644
--- a/lib/qpselect.h
+++ b/lib/qpselect.h
@@ -65,7 +65,12 @@ enum qps_mbits                  /* "mode" bits: error/read/write        */
 typedef enum qps_mbits qps_mbit_t ;
 
 /* "fd_undef" -- used when fd is undefined                              */
-enum { fd_undef = -1 } ;
+typedef enum
+{
+  fd_undef = -1,
+  fd_first = 0,
+  fd_huge  = 8192
+} fd_t ;
 
 /* Forward references   */
 typedef struct qps_selection* qps_selection ;
@@ -148,7 +153,7 @@ struct qps_file
   qps_selection selection ;
 
   void*   file_info ;
-  int     fd ;
+  fd_t    fd ;
 
   qps_mbit_t    enabled_bits ;
 
@@ -158,68 +163,71 @@ struct qps_file
 /*==============================================================================
  * qps_selection handling
  */
-
-extern void
-qps_start_up(void) ;
-
-extern qps_selection
-qps_selection_init_new(qps_selection qps) ;
-
-extern void
-qps_add_file(qps_selection qps, qps_file qf, int fd, void* file_info) ;
-
-extern void
-qps_remove_file(qps_file qf) ;
-
-extern qps_file
-qps_selection_ream(qps_selection qps, int free_structure) ;
+extern void qps_start_up(void) ;
+extern qps_selection qps_selection_init_new(qps_selection qps) ;
+extern void qps_add_file(qps_selection qps, qps_file qf, fd_t fd,
+                                                              void* file_info) ;
+extern void qps_remove_file(qps_file qf) ;
+extern qps_file qps_selection_ream(qps_selection qps, int free_structure) ;
 
 /* Ream out selection and free the selection structure.   */
 #define qps_selection_ream_free(qps) qps_selection_ream(qps, 1)
 /* Ream out selection but keep the selection structure.   */
 #define qps_selection_ream_keep(qps) qps_selection_ream(qps, 0)
 
-extern void
-qps_set_signal(qps_selection qps, const sigset_t* sigmask) ;
-
-extern int
-qps_pselect(qps_selection qps, qtime_mono_t timeout) ;
-
-extern int
-qps_dispatch_next(qps_selection qps) ;
+extern void qps_set_signal(qps_selection qps, const sigset_t* sigmask) ;
+extern int qps_pselect(qps_selection qps, qtime_mono_t timeout) ;
+extern int qps_dispatch_next(qps_selection qps) ;
 
 /*==============================================================================
  * qps_file structure handling
  */
+extern qps_file qps_file_init_new(qps_file qf, qps_file template) ;
+extern qps_file qps_file_free(qps_file qf) ;
+extern void qps_enable_mode(qps_file qf, qps_mnum_t mnum, qps_action* action) ;
+extern void qps_set_action(qps_file qf, qps_mnum_t mnum, qps_action* action) ;
+extern void qps_disable_modes(qps_file qf, qps_mbit_t mbits) ;
 
-extern qps_file
-qps_file_init_new(qps_file qf, qps_file template) ;
+Inline void* qps_file_info(qps_file qf) ;
+Inline fd_t qps_file_fd(qps_file qf) ;
+Inline fd_t qps_file_unset_fd(qps_file qf) ;
+Inline void qps_set_file_info(qps_file qf, void* info) ;
 
-extern qps_file
-qps_file_free(qps_file qf) ;
-
-extern void
-qps_enable_mode(qps_file qf, qps_mnum_t mnum, qps_action* action) ;
-
-extern void
-qps_set_action(qps_file qf, qps_mnum_t mnum, qps_action* action) ;
-
-extern void
-qps_disable_modes(qps_file qf, qps_mbit_t mbits) ;
+/*==============================================================================
+ * Inline functions
+ */
 
+/*------------------------------------------------------------------------------
+ * Get the "file_info"
+ */
 Inline void*
 qps_file_info(qps_file qf)
 {
   return qf->file_info ;
 } ;
 
-Inline int
+/*------------------------------------------------------------------------------
+ * Set the "file_info"
+ */
+Inline void
+qps_set_file_info(qps_file qf, void* info)
+{
+  qf->file_info = info ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Get the "fd"
+ */
+Inline fd_t
 qps_file_fd(qps_file qf)
 {
   return qf->fd ;
 } ;
 
-Inline int
+/*------------------------------------------------------------------------------
+ * Unset the "fd" and return previous value
+ */
+Inline fd_t
 qps_file_unset_fd(qps_file qf)
 {
   int fd = qf->fd ;
@@ -228,12 +236,6 @@ qps_file_unset_fd(qps_file qf)
   return fd ;
 } ;
 
-Inline void
-qps_set_file_info(qps_file qf, void* info)
-{
-  qf->file_info = info ;
-} ;
-
 /*==============================================================================
  * Miniature pselect
  *
diff --git a/lib/qpthreads.c b/lib/qpthreads.c
index 8909f64b..9990fc14 100644
--- a/lib/qpthreads.c
+++ b/lib/qpthreads.c
@@ -72,17 +72,21 @@
  *
  *   * qpt_mutex_init_new
  *   * qpt_cond_init_new
+ *   * qpt_splin_init_new
  *
  * This allows the application to decide as late as possible (but no later)
- * whether to enable pthreads.  If a mutex or a condition variable has been
- * initialised before the application gets around to enabling qpthreads, that
- * will be trapped when qpthreads is finally enabled.
+ * whether to enable pthreads.  If a mutex, condition variable or spin lock has
+ * been initialised before the application gets around to enabling qpthreads,
+ * that will be trapped when qpthreads is finally enabled.
  *
  * Pthread Requirements
  * ====================
  *
- * This is assuming support for 1003.1-2004 -- XOPEN Issue 6, with [THR] and
- * [XSI] options.
+ * This is assuming support for 1003.1-2004 -- XOPEN Issue 6, with [THR], [SPI]
+ * and [XSI] options.
+ *
+ * In 1003.1-2008, XOPEN issue 7, [THR], [SPI] and pthread_mutexattr_settype()
+ * have been moved to Base.
  *
  * The [XSI] is required for pthread_mutexattr_settype(), only.
  *
@@ -117,13 +121,8 @@
  *   pthread_rwlockattr_init()/_destroy()                 [THR]      pro tem
  *   pthread_rwlockattr_getpshared()/_setpshared()        [TSH]
  *
- *   pthread_spin_xxx()                                   [SPI]
- *
  * [CS] (Clock Select) is assumed if HAVE_CLOCK_MONOTONIC.
  *
- * In 1003.1-2008, XOPEN issue 7, [THR] and pthread_mutexattr_settype() have
- * been moved to Base.
- *
  * NB: it is essential that pthread_kill() delivers the signal to the target
  *     thread only -- ie, it must be POSIX compliant.  That rules out the old
  *     (2.4) LinuxThreads.  For Linux, 2.6 (or greater) is required, with
@@ -616,9 +615,9 @@ qpt_mutex_destroy(qpt_mutex mx, int free_mutex)
  *   qpt_cond_realtime   -- force CLOCK_REALTIME
  *   qpt_cond_monotonic  -- force CLOCK_MONOTONIC  (if available)
  *
- * NB: FATAL error to attempt this is !qptthreads_enabled.
+ * NB: FATAL error to attempt this if !qptthreads_enabled.
  *
- * Returns the condition variable -- or original cv id !qpthreads_enabled.
+ * Returns the condition variable -- or original cv if !qpthreads_enabled.
  */
 extern qpt_cond
 qpt_cond_init_new(qpt_cond cv, enum qpt_cond_options opts)
@@ -732,6 +731,49 @@ qpt_cond_timedwait(qpt_cond cv, qpt_mutex mx, qtime_mono_t timeout_time)
 } ;
 
 /*==============================================================================
+ * Spinlock initialise and destroy.
+ */
+
+/* Initialise Spinlock -- NB: no allocation option
+ *
+ * Does nothing if !qpthreads_enabled -- but freezes the state.
+ */
+extern void
+qpt_spin_init(qpt_spin slk)
+{
+  int err ;
+
+  if (!qpthreads_enabled_freeze)
+    return ;
+
+  enum {
+#ifndef PTHREAD_PROCESS_PRIVATE
+    pthread_process_private = 0
+#else
+    pthread_process_private = PTHREAD_PROCESS_PRIVATE
+#endif
+  } ;
+
+  err = pthread_spin_init(slk, pthread_process_private) ;
+  if (err != 0)
+    zabort_err("pthread_spin_init failed", err) ;
+} ;
+
+/* Destroy given spin lock -- NB: no free option
+ *                                       -- or do nothing if !qpthreads_enabled.
+ */
+extern void
+qpt_spin_destroy(qpt_spin slk)
+{
+  if (qpthreads_enabled)
+    {
+      int err = pthread_spin_destroy(slk) ;
+      if (err != 0)
+        zabort_err("pthread_spin_destroy failed", err) ;
+    } ;
+} ;
+
+/*==============================================================================
  * Signal Handling.
  */
 
diff --git a/lib/qpthreads.h b/lib/qpthreads.h
index d05f7e34..6b9c5741 100644
--- a/lib/qpthreads.h
+++ b/lib/qpthreads.h
@@ -106,16 +106,18 @@ enum { qpthreads_debug = QPTHREADS_DEBUG } ;
 /*==============================================================================
  * Data types
  */
-typedef pthread_t        qpt_thread_t ;
+typedef pthread_t           qpt_thread_t ;
 
-typedef pthread_mutex_t  qpt_mutex_t[1] ;
-typedef pthread_mutex_t* qpt_mutex ;
+typedef pthread_mutex_t     qpt_mutex_t[1] ;
+typedef pthread_mutex_t*    qpt_mutex ;
 
-typedef pthread_cond_t   qpt_cond_t[1] ;
-typedef pthread_cond_t*  qpt_cond ;
+typedef pthread_cond_t      qpt_cond_t[1] ;
+typedef pthread_cond_t*     qpt_cond ;
 
-typedef pthread_attr_t   qpt_thread_attr_t ;
+typedef pthread_attr_t      qpt_thread_attr_t ;
 
+typedef pthread_spinlock_t  qpt_spin_t ;
+typedef qpt_spin_t*         qpt_spin ;
 
 /*==============================================================================
  * Thread Creation -- see qpthreads.c for further discussion.
@@ -325,6 +327,30 @@ Inline void                       /* do nothing if !qpthreads_enabled   */
 qpt_cond_broadcast(qpt_cond cv) ;
 
 /*==============================================================================
+ * Spinlock handling
+ *
+ * Spinlocks are pretty trivial -- requiring only to be initialised, locked,
+ * unlocked and, finally, destroyed.
+ *
+ * NB: recursive spinlocks are not supported !
+ *
+ * NB: if NOT qpthreads_enabled, locking and unlocking always succeed.  This
+ *     allows code to be made thread-safe for when pthreads is running, but to
+ *     work perfectly well without pthreads.
+ */
+extern void                     /* freezes qpthreads_enabled          */
+qpt_spin_init(qpt_spin slk) ;
+
+extern void                     /* do nothing if !qpthreads_enabled   */
+qpt_spin_destroy(qpt_spin slk) ;
+
+Inline void                     /* do nothing if !qpthreads_enabled   */
+qpt_spin_lock(qpt_spin slk) ;
+
+Inline void                     /* do nothing if !qpthreads_enabled   */
+qpt_spin_unlock(qpt_spin slk) ;
+
+/*==============================================================================
  * Mutex inline functions
  */
 
@@ -443,6 +469,50 @@ qpt_cond_broadcast(qpt_cond cv)
 } ;
 
 /*==============================================================================
+ * Spinlock inline functions
+ */
+
+/* Lock spinlock  -- do nothing if !qpthreads_enabled
+ *
+ * Unless both NCHECK_QPTHREADS and NDEBUG are defined, checks that the
+ * return value is valid -- zabort_errno if it isn't.
+ */
+Inline void
+qpt_spin_lock(qpt_spin slk)
+{
+  if (qpthreads_enabled)
+    {
+#if defined(NDEBUG) && defined(NDEBUG_QPTHREADS)
+      pthread_spin_lock(slk) ;
+#else
+      int err = pthread_spin_lock(slk) ;
+      if (err != 0)
+        zabort_err("pthread_spin_lock failed", err) ;
+#endif
+    } ;
+} ;
+
+/* Unlock spinlock  -- do nothing if !qpthreads_enabled
+ *
+ * Unless both NCHECK_QPTHREADS and NDEBUG are defined, checks that the
+ * return value is valid -- zabort_errno if it isn't.
+ */
+Inline void
+qpt_spin_unlock(qpt_spin slk)
+{
+  if (qpthreads_enabled)
+    {
+#if defined(NDEBUG) && defined(NDEBUG_QPTHREADS)
+      pthread_spin_unlock(slk) ;
+#else
+      int err = pthread_spin_unlock(slk) ;
+      if (err != 0)
+        zabort_err("pthread_spin_unlock failed", err) ;
+#endif
+    } ;
+} ;
+
+/*==============================================================================
  * Signal Handling.
  */
 extern void                     /* sigprocmask() if !qpthreads_enabled  */
diff --git a/lib/qtimers.c b/lib/qtimers.c
index ad8da395..9d23eeda 100644
--- a/lib/qtimers.c
+++ b/lib/qtimers.c
@@ -128,7 +128,7 @@ qtimer_pile_init_new(qtimer_pile qtp)
  * is returned.
  */
 extern qtime_t
-qtimer_pile_top_wait(qtimer_pile qtp, qtime_t max_wait)
+qtimer_pile_top_wait(qtimer_pile qtp, qtime_t max_wait, qtime_t now)
 {
   qtime_t top_wait ;
   qtimer  qtr = heap_top_item(&qtp->timers) ;
@@ -136,7 +136,7 @@ qtimer_pile_top_wait(qtimer_pile qtp, qtime_t max_wait)
   if (qtr == NULL)
     return max_wait ;
 
-  top_wait = qtr->time - qt_get_monotonic() ;
+  top_wait = qtr->time - now ;
 
   return (top_wait < max_wait) ? top_wait : max_wait ;
 } ;
@@ -158,6 +158,7 @@ extern bool
 qtimer_pile_dispatch_next(qtimer_pile qtp, qtime_mono_t upto)
 {
   qtimer   qtr ;
+  qtr_state_t state ;
 
   if (qtimers_debug)
     qtimer_pile_verify(qtp) ;
@@ -167,16 +168,22 @@ qtimer_pile_dispatch_next(qtimer_pile qtp, qtime_mono_t upto)
   if ((qtr == NULL) || (qtr->time > upto))
     return 0 ;
 
-  passert((qtp == qtr->pile) && (qtr->active)) ;
+  passert((qtp == qtr->pile) && (qtr->state == qtrs_active)) ;
 
-  qtp->implicit_unset = qtr ;   /* Timer must be unset if is still here
-                                   when the action function returns       */
+  qtr->state = qtrs_dispatch | qtrs_unset_pending | qtrs_active ;
+                                /* Timer must be unset if is still here
+                                   when the action function returns     */
   qtr->action(qtr, qtr->timer_info, upto) ;
 
-  if (qtp->implicit_unset == qtr)
+  state = qtr->state ;
+  qtr->state &= qtrs_active ;   /* No longer in dispatch                */
+
+  confirm((qtrs_active != 0) && (qtrs_inactive == 0)) ;
+
+  if      ((state & qtrs_free_pending) != 0)
+    qtimer_free(qtr) ;
+  else if ((state & qtrs_unset_pending) != 0)
     qtimer_unset(qtr) ;
-  else
-    assert(qtp->implicit_unset == NULL) ;   /* check for tidy-ness      */
 
   return 1 ;
 } ;
@@ -209,7 +216,7 @@ qtimer_pile_ream(qtimer_pile qtp, free_keep_b free_structure)
 
   qtr = heap_ream(&qtp->timers, keep_it) ; /* ream, keeping the heap    */
   if (qtr != NULL)
-    qtr->active = false ;               /* has been removed from pile         */
+    qtr->state = qtrs_inactive ;        /* has been removed from pile         */
   else
     if (free_structure)                 /* pile is empty, may now free it     */
       XFREE(MTYPE_QTIMER_PILE, qtp) ;
@@ -245,7 +252,7 @@ qtimer_init_new(qtimer qtr, qtimer_pile qtp,
    *   pile        -- NULL -- not in any pile (yet)
    *   backlink    -- unset
    *
-   *   active      -- false
+   *   state       -- qtrs_inactive
    *
    *   time        -- unset
    *   action      -- NULL -- no action set (yet)
@@ -253,6 +260,7 @@ qtimer_init_new(qtimer qtr, qtimer_pile qtp,
    *
    *   interval    -- unset
    */
+  confirm(qtrs_inactive == 0) ;
 
   qtr->pile       = qtp ;
   qtr->action     = action ;
@@ -264,9 +272,13 @@ qtimer_init_new(qtimer qtr, qtimer_pile qtp,
 /*------------------------------------------------------------------------------
  * Free given timer -- if any.
  *
- * Unsets it first if it is active or pending unset.
+ * Unsets it first if it is active.
  *
  * Returns: NULL
+ *
+ * Note: if this is currently a dispatched timer, then does not actually free,
+ *       but leaves that for the dispatch loop to tidy up.  The caller is
+ *       expected to assume that the timer has gone, gone.
  */
 extern qtimer
 qtimer_free(qtimer qtr)
@@ -276,10 +288,13 @@ qtimer_free(qtimer qtr)
    */
   if (qtr != NULL)
     {
-      if (qtr->active)
+      if ((qtr->state & qtrs_active) != 0)
         qtimer_unset(qtr) ;
 
-      XFREE(MTYPE_QTIMER, qtr) ;
+      if ((qtr->state & qtrs_dispatch) == 0)
+        XFREE(MTYPE_QTIMER, qtr) ;
+      else
+        qtr->state = qtrs_dispatch | qtrs_free_pending ;
     } ;
 
   return NULL ;
@@ -301,7 +316,7 @@ qtimer_set_pile(qtimer qtr, qtimer_pile qtp)
   /* Note that if is the current dispatched timer and an unset is still
    * pending, then it must still be active.
    */
-  if (qtr->active)
+  if ((qtr->state & qtrs_active) != 0)
     qtimer_unset(qtr) ;
 
   qtr->pile = qtp ;
@@ -315,7 +330,9 @@ qtimer_set_pile(qtimer qtr, qtimer_pile qtp)
  * Sets any given action -- if the action given is NULL, retains previously set
  * action.
  *
- * If the timer is already active, sets the new time & updates pile.
+ * If the timer is already active, sets the new time & updates pile.  If is the
+ * dispatched timer, and was pending being unset, then no longer needs to be
+ * unset.
  *
  * Otherwise, sets the time and adds to pile -- making timer active.
  *
@@ -337,22 +354,19 @@ qtimer_set(qtimer qtr, qtime_mono_t when, qtimer_action* action)
 
   qtr->time = when ;
 
-  if (qtr->active)
+  if ((qtr->state & qtrs_active) != 0)
     {
       /* Is active, so update the timer in the pile.                    */
       heap_update_item(&qtp->timers, qtr) ;
 
-      if (qtr == qtp->implicit_unset)
-        qtp->implicit_unset = NULL ;        /* no unset required, now   */
+      qtr->state &= ~qtrs_unset_pending ;   /* no unset required, now   */
     }
   else
     {
       /* Is not active, so insert the timer into the pile.              */
       heap_push_item(&qtp->timers, qtr) ;
 
-      assert(qtr != qtp->implicit_unset) ;  /* because it's not active  */
-
-      qtr->active = true ;
+      qtr->state |= qtrs_active ;
     } ;
 
   if (action != NULL)
@@ -362,12 +376,19 @@ qtimer_set(qtimer qtr, qtime_mono_t when, qtimer_action* action)
 
   if (qtimers_debug)
     qtimer_pile_verify(qtp) ;
+
+  if (qdebug)
+    assert( (qtr->state ==  qtrs_active) ||
+            (qtr->state == (qtrs_active | qtrs_dispatch)) ) ;
 } ;
 
 /*------------------------------------------------------------------------------
  * Unset given timer
  *
  * If the timer is active, removes from pile and sets inactive.
+ *
+ * If timer was pending being unset (because is the dispatched timer), then no
+ * longer needs to be unset.
  */
 extern void
 qtimer_unset(qtimer qtr)
@@ -379,20 +400,19 @@ qtimer_unset(qtimer qtr)
   if (qtimers_debug)
     qtimer_pile_verify(qtp) ;
 
-  if (qtr->active)
+  if ((qtr->state & qtrs_active) != 0)
     {
-      if (qtr == qtp->implicit_unset)
-        qtp->implicit_unset = NULL ;        /* no unset required, now       */
-
       heap_delete_item(&qtp->timers, qtr) ;
 
+      qtr->state &= ~(qtrs_unset_pending | qtrs_active);
+                                  /* not active, no unset required, now   */
       if (qtimers_debug)
         qtimer_pile_verify(qtp) ;
+    } ;
 
-      qtr->active = false ;
-    }
-  else
-    assert(qtr != qtp->implicit_unset) ;
+  if (qdebug)
+    assert( (qtr->state ==  qtrs_inactive) ||
+            (qtr->state == (qtrs_inactive | qtrs_dispatch)) ) ;
 } ;
 
 /*==============================================================================
@@ -406,7 +426,7 @@ qtimer_pile_verify(qtimer_pile qtp)
   vector_index_t  i ;
   vector_length_t e ;
   qtimer qtr ;
-  bool seen = false ;
+  bool seen_dispatch ;
 
   assert(qtp != NULL) ;
 
@@ -421,23 +441,23 @@ qtimer_pile_verify(qtimer_pile qtp)
 
   v = th->v ;
   e = vector_end(v) ;
+  seen_dispatch = false ;
   for (i = 0 ; i < e ; ++i)
     {
       qtr = vector_get_item(v, i) ;
       assert(qtr != NULL) ;
 
-      if (qtr == qtp->implicit_unset)
+      if (qtr->state != qtrs_active)
         {
-          assert(!seen) ;
-          seen = true ;
+          assert((qtr->state & qtrs_dispatch) != 0) ;
+          assert((qtr->state & qtrs_free_pending) == 0) ;
+          assert((qtr->state & qtrs_active) != 0) ;
+          assert(!seen_dispatch) ;
+          seen_dispatch = true ;
         } ;
 
-      assert(qtr->active) ;
-
       assert(qtr->pile     == qtp) ;
       assert(qtr->backlink == i) ;
       assert(qtr->action   != NULL) ;
     } ;
-
-  assert(seen || (qtp->implicit_unset == NULL)) ;
 } ;
diff --git a/lib/qtimers.h b/lib/qtimers.h
index 16808f1b..ae0a5c70 100644
--- a/lib/qtimers.h
+++ b/lib/qtimers.h
@@ -77,12 +77,23 @@ typedef struct qtimer_pile*  qtimer_pile ;
 
 typedef void (qtimer_action)(qtimer qtr, void* timer_info, qtime_mono_t when) ;
 
+typedef enum
+{
+  qtrs_inactive      = 0,
+  qtrs_active        = 1,
+
+  qtrs_unset_pending = 2,
+  qtrs_free_pending  = 4,
+
+  qtrs_dispatch      = 8,
+} qtr_state_t ;
+
 struct qtimer
 {
   qtimer_pile     pile ;        /* pile currently allocated to          */
   heap_backlink_t backlink ;
 
-  bool            active ;      /* true => in the pile                  */
+  qtr_state_t     state ;
 
   qtime_mono_t    time ;        /* current time to trigger action       */
   qtimer_action*  action ;
@@ -94,8 +105,6 @@ struct qtimer
 struct qtimer_pile
 {
   struct heap timers ;
-
-  qtimer      implicit_unset ;  /* used during dispatch                 */
 } ;
 
 /*==============================================================================
@@ -104,7 +113,8 @@ struct qtimer_pile
 
 extern qtimer_pile qtimer_pile_init_new(qtimer_pile qtp) ;
 extern bool qtimer_pile_dispatch_next(qtimer_pile qtp, qtime_mono_t upto) ;
-extern qtime_t qtimer_pile_top_wait(qtimer_pile qtp, qtime_t max_wait) ;
+extern qtime_t qtimer_pile_top_wait(qtimer_pile qtp, qtime_t max_wait,
+                                                                  qtime_t now) ;
 extern qtimer qtimer_pile_ream(qtimer_pile qtp, free_keep_b free_structure) ;
 
 /* Ream out qtimer pile and free the qtimer structure.   */
@@ -200,4 +210,13 @@ qtimer_get_interval(qtimer qtr)
   return qtr->interval ;
 } ;
 
+/*------------------------------------------------------------------------------
+ * See if given qtimer (if any) is active
+ */
+Inline bool
+qtimer_is_active(qtimer qtr)
+{
+  return (qtr != NULL) && ((qtr->state & qtrs_active) != 0) ;
+}
+
 #endif /* _ZEBRA_QTIMERS_H */
diff --git a/lib/sockunion.c b/lib/sockunion.c
index 9084c27e..9577b901 100644
--- a/lib/sockunion.c
+++ b/lib/sockunion.c
@@ -256,7 +256,32 @@ sockunion_init_new(sockunion su, sa_family_t family)
 } ;
 
 /*------------------------------------------------------------------------------
- * Get the length of the address in the given sockunion.
+ * Get the AFI for the sockaddr in the given sockunion.
+ *
+ * Returns zero if AF_UNSPEC or not any known address family.
+ */
+extern afi_t
+sockunion_get_afi(sockunion su)
+{
+  switch (su->sa.sa_family)
+  {
+    case AF_INET:
+      return AFI_IP ;
+
+#ifdef HAVE_IPV6
+    case AF_INET6:
+      return AFI_IP6 ;
+#endif
+
+    default:
+      return 0 ;
+  } ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Get the length of the sockaddr in the given sockunion.
+ *
+ * This length includes the family, port number, protocol address, etc.
  *
  * Returns zero if AF_UNSPEC or not any known address family.
  */
@@ -279,6 +304,56 @@ sockunion_get_len(sockunion su)
 } ;
 
 /*------------------------------------------------------------------------------
+ * Get the length of the protocol address in the given sockunion.
+ *
+ * This length is for just the IPv4, IPv6, etc. address.
+ *
+ * Returns zero if AF_UNSPEC or not any known address family.
+ */
+extern int
+sockunion_get_addr_len(sockunion su)
+{
+  switch (su->sa.sa_family)
+  {
+    case AF_INET:
+      return sizeof(su->sin.sin_addr.s_addr) ;
+
+#ifdef HAVE_IPV6
+    case AF_INET6:
+      return sizeof(su->sin6.sin6_addr.s6_addr) ;
+#endif
+
+    default:
+      return 0 ;
+  } ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Get pointer to the protocol address in the given sockunion.
+ *
+ * Note that IP and IPv6 addresses are in Network Order.
+ *
+ * Returns NULL if AF_UNSPEC or not any known address family.
+ */
+extern void*
+sockunion_get_addr(sockunion su)
+{
+  switch (su->sa.sa_family)
+  {
+    case AF_INET:
+      return &su->sin.sin_addr.s_addr ;
+
+#ifdef HAVE_IPV6
+    case AF_INET6:
+      return &su->sin6.sin6_addr.s6_addr ;
+#endif
+
+    default:
+      return 0 ;
+  } ;
+} ;
+
+/*------------------------------------------------------------------------------
  * From the given string, fill in the given sockunion.
  *
  * Returns: 0 => OK -- sockunion filled in
diff --git a/lib/sockunion.h b/lib/sockunion.h
index eeae72d5..54aa6bc3 100644
--- a/lib/sockunion.h
+++ b/lib/sockunion.h
@@ -98,11 +98,18 @@ struct sockunion_string
 #define sock2ip6(X)  (((struct sockaddr_in6 *)(X))->sin6_addr.s6_addr)
 #endif /* HAVE_IPV6 */
 
-#define sockunion_family(X)  (X)->sa.sa_family
+inline static sa_family_t
+sockunion_family(sockunion su)
+{
+  return su->sa.sa_family ;
+} ;
 
 /* Prototypes. */
 extern sockunion sockunion_init_new(sockunion su, sa_family_t family) ;
+extern afi_t sockunion_get_afi(sockunion su) ;
 extern int sockunion_get_len(sockunion su) ;
+extern int sockunion_get_addr_len(sockunion su) ;
+extern void* sockunion_get_addr(sockunion su) ;
 extern int sockunion_set_port(sockunion su, in_port_t port) ;
 extern int str2sockunion (const char * str, sockunion su);
 extern const char *sockunion2str (sockunion su, char* buf, size_t size);
diff --git a/lib/stream.c b/lib/stream.c
index 0183ce78..277ff216 100644
--- a/lib/stream.c
+++ b/lib/stream.c
@@ -928,7 +928,6 @@ stream_recvmsg (struct stream *s, int fd, struct msghdr *msgh, int flags,
 size_t
 stream_write (struct stream *s, const void *ptr, size_t size)
 {
-
   CHECK_SIZE(s, size);
 
   STREAM_VERIFY_SANE(s);
diff --git a/lib/vty_cli.c b/lib/vty_cli.c
index a6a276e8..a4ab8de3 100644
--- a/lib/vty_cli.c
+++ b/lib/vty_cli.c
@@ -261,9 +261,7 @@ uty_cli_close(vty_cli cli, bool final)
   /* Discard any pause_timer, and suppress */
   cli->pause_timer    = qtimer_free(cli->pause_timer) ;
   cli->paused         = false ;
-#if 0
   cli->tilde_enabled  = false ;
-#endif
 
   /* If final, free the CLI object.                                     */
   if (final)
@@ -817,11 +815,7 @@ uty_cli_dispatch(vty_cli cli)
 
       uty_cmd_signal(vio, CMD_SUCCESS) ;
 
-      cli->blocked = (to_do_now != cmd_do_command)
-#if 0
-                                                   || !cli->tilde_enabled
-#endif
-          ;
+      cli->blocked = (to_do_now != cmd_do_command) || !cli->tilde_enabled ;
     }
   else
     {
diff --git a/lib/vty_cli.h b/lib/vty_cli.h
index ec00965a..e366ecdf 100644
--- a/lib/vty_cli.h
+++ b/lib/vty_cli.h
@@ -95,9 +95,7 @@ struct vty_cli
   bool          drawn ;
 
   bool          tilde_prompt ;
-#if 0
   bool          tilde_enabled ;
-#endif
 
   int           prompt_len ;
   int           extra_len ;
diff --git a/lib/zclient.c b/lib/zclient.c
index 6803aa4a..ff95b624 100644
--- a/lib/zclient.c
+++ b/lib/zclient.c
@@ -396,7 +396,7 @@ zclient_start (struct zclient *zclient)
     return 0;
 
   /* Check timer */
-  if (zclient->qtr && zclient->qtr->active)
+  if (qtimer_is_active(zclient->qtr))
     return 0;
 
   /* Make socket. */
@@ -1169,11 +1169,11 @@ zclient_event_r (enum event event, struct zclient *zclient)
   switch (event)
     {
     case ZLOOKUP_SCHEDULE:
-      if (!zclient->qtr->active)
+      if (!qtimer_is_active(zclient->qtr))
         qtimer_set(zclient->qtr, qt_get_monotonic(), zlookup_connect_r) ;
       break;
     case ZCLIENT_SCHEDULE:
-      if (!zclient->qtr->active)
+      if (!qtimer_is_active(zclient->qtr))
         qtimer_set(zclient->qtr, qt_get_monotonic(), zclient_connect_r) ;
       break;
     case ZCLIENT_CONNECT:
@@ -1182,7 +1182,7 @@ zclient_event_r (enum event event, struct zclient *zclient)
       if (zclient_debug)
 	zlog_debug ("zclient connect schedule interval is %d",
 		   zclient->fail < 3 ? 10 : 60);
-      if (!zclient->qtr->active)
+      if (!qtimer_is_active(zclient->qtr))
         qtimer_set(zclient->qtr,
           qt_add_monotonic(QTIME(zclient->fail < 3 ? 10 : 60)), zclient_connect_r) ;
       break;