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.

23 files changed, 2522 insertions, 737 deletions

diff --git a/bgpd/bgp_connection.c b/bgpd/bgp_connection.c
index 24c86230..f735080a 100644
--- a/bgpd/bgp_connection.c
+++ b/bgpd/bgp_connection.c
@@ -29,6 +29,7 @@
 #include "bgpd/bgp_session.h"
 #include "bgpd/bgp_notification.h"
 #include "bgpd/bgp_msg_read.h"
+#include "bgpd/bgp_dump.h"
 
 #include "lib/memory.h"
 #include "lib/mqueue.h"
@@ -1099,6 +1100,9 @@ bgp_connection_read_action(qps_file qf, void* file_info)
      {
        BGP_CONNECTION_SESSION_LOCK(connection) ;    /*<<<<<<<<<<<<<<<<<<<<<<<<*/
 
+       if (bgp_dump_packet_flag)
+         bgp_dump_packet(connection) ;
+
        connection->msg_func(connection, connection->msg_body_size) ;
 
        BGP_CONNECTION_SESSION_UNLOCK(connection) ;  /*>>>>>>>>>>>>>>>>>>>>>>>>*/
diff --git a/bgpd/bgp_dump.c b/bgpd/bgp_dump.c
index 7c36825c..860ed0cb 100644
--- a/bgpd/bgp_dump.c
+++ b/bgpd/bgp_dump.c
@@ -1,4 +1,4 @@
-/* BGP-4 dump routine
+/* BGP-4 bd routine
    Copyright (C) 1999 Kunihiro Ishiguro
 
 This file is part of GNU Zebra.
@@ -26,660 +26,1212 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 #include "sockunion.h"
 #include "command.h"
 #include "prefix.h"
-#include "thread.h"
 #include "linklist.h"
-#include "bgpd/bgp_table.h"
 #include "qpath.h"
 #include "qstring.h"
+#include "memory.h"
+#include "qtimers.h"
+#include "mqueue.h"
+#include "qiovec.h"
 
+#include "bgpd/bgp_dump.h"
 #include "bgpd/bgpd.h"
+#include "bgpd/bgp_table.h"
 #include "bgpd/bgp_route.h"
 #include "bgpd/bgp_attr.h"
-#include "bgpd/bgp_dump.h"
+#include "bgpd/bgp_engine.h"
+
+/*==============================================================================
+ * This will bd BGP state in MRT ("Multi-threaded Routing Toolkit") form.
+ * See draft-ietf-grow-mrt.
+ *
+ * Three, independent, dumps are supported (simultaneously):
+ *
+ *   ALL:     dumps all BGP messages received, plus FSM state transitions.
+ *
+ *   UPDATES: dumps all BGP UPDATE messages received
+ *
+ *   TABLE:   periodically dumps entire BGP RIB
+ *
+ * For ALL and UPDATES, a new file is opened periodically.  For TABLE, a new
+ * file is opened for each bd.
+ *
+ * The ALL and UPDATES dumps are done in the BGP Engine, though the periodic
+ * opening of new files is done in the Routing Engine and the result passed to
+ * the BGP Engine.
+ */
+
+/* MRT message types
+ */
+enum MRT_MT_TYPES {
+   MRT_MT_NULL              =  0,       /* deprecated   */
+   MRT_MT_START             =  1,       /* deprecated   */
+   MRT_MT_DIE               =  2,       /* deprecated   */
+   MRT_MT_I_AM_DEAD         =  3,       /* deprecated   */
+   MRT_MT_PEER_DOWN         =  4,       /* deprecated   */
+   MRT_MT_BGP               =  5,       /* deprecated   */
+   MRT_MT_RIP               =  6,       /* deprecated   */
+   MRT_MT_IDRP              =  7,       /* deprecated   */
+   MRT_MT_RIPNG             =  8,       /* deprecated   */
+   MRT_MT_BGP4PLUS          =  9,       /* deprecated   */
+   MRT_MT_BGP4PLUS_01       = 10,       /* deprecated   */
+
+   MRT_MT_OSPFv2            = 11,
+   MRT_MT_TABLE_DUMP        = 12,       /* BGP routing table dump       */
+   MRT_MT_TABLE_DUMP_V2     = 13,       /* BGP routing table dump, v2   */
+   MRT_MT_BGP4MP            = 16,       /* BGP4 with MP extensions      */
+   MRT_MT_BGP4MP_ET         = 17,       /* as above with Extended Times */
+   MRT_MT_ISIS              = 32,
+   MRT_MT_ISIS_ET           = 33,       /* as above with Extended Times */
+   MRT_MT_OSPFv3            = 48,
+   MRT_MT_OSPFv3_ET         = 49,       /* as above with Extended Times */
+
+} ;
+
+/* MRT Common Header and other sizes
+ */
+enum
+{
+  MRT_COMMON_HEADER_SIZE  = 12,         /* Timestamp(4), Type(2), Subtype(2)
+                                         * Length(4)
+                                         * NB: length excludes header   */
+  MRT_BGP4MP_HEADER_SIZE  = 44          /* Peer AS(4), Local AS(4),
+                                         * Interface Index(2),
+                                         * Address Family(2),
+                                         * Peer IP(16), Local IP(16)    */
+} ;
+
+/* MRT subtypes of MRT_MT_BGP4MP
+ */
+enum MRT_MT_BGP4MP_SUBTYPES
+{
+  MRT_MST_BGP4MP_STATE_CHANGE      = 0,
+  MRT_MST_BGP4MP_MESSAGE           = 1,
+
+  MRT_MST_BGP4MP_ENTRY             = 2, /* deprecated   */
+  MRT_MST_BGP4MP_SNAPSHOT          = 3, /* deprecated   */
+
+  MRT_MST_BGP4MP_MESSAGE_AS4       = 4,
+  MRT_MST_BGP4MP_STATE_CHANGE_AS4  = 5,
+
+  MRT_MST_BGP4MP_MESSAGE_LOCAL     = 6,
+  MRT_MST_BGP4MP_MESSAGE_AS4_LOCAL = 7,
+} ;
 
-enum bgp_dump_type
+/* MRT subtypes of MRT_MT_TABLE_DUMP_V2
+ */
+enum MRT_MT_TABLE_DUMP_V2_SUBTYPES
+{
+  MRT_MST_TDV2_PEER_INDEX_TABLE    = 1,
+  MRT_MST_TDV2_RIB_IPV4_UNICAST    = 2,
+  MRT_MST_TDV2_RIB_IPV4_MULTICAST  = 3,
+  MRT_MST_TDV2_RIB_IPV6_UNICAST    = 4,
+  MRT_MST_TDV2_RIB_IPV6_MULTICAST  = 5,
+  MRT_MST_TDV2_RIB_GENERIC         = 6,
+} ;
+
+/* Values for MRT_MST_TDV2_PEER_INDEX_TABLE message
+ */
+enum
+{
+  MRT_TDV2_PEER_INDEX_TABLE_IPV4   = 0,
+  MRT_TDV2_PEER_INDEX_TABLE_IPV6   = 1,
+  MRT_TDV2_PEER_INDEX_TABLE_AS2    = 0,
+  MRT_TDV2_PEER_INDEX_TABLE_AS4    = 2,
+} ;
+
+/* Values for FSM states
+ */
+enum
+{
+  MRT_FSM_UNDEF        = 0,     /* Not defined in the standard  */
+
+  MRT_FSM_Idle         = 1,
+  MRT_FSM_Connect      = 2,
+  MRT_FSM_Active       = 3,
+  MRT_FSM_OpenSent     = 4,
+  MRT_FSM_OpenConfirm  = 5,
+  MRT_FSM_Established  = 6,
+} ;
+
+/* Values for AFI in BGP4MP messages
+ */
+enum
+{
+  MRT_AFI_IPv4         = 1,
+  MRT_AFI_IPv6         = 2,
+} ;
+
+/*------------------------------------------------------------------------------
+ * Dump control definitions, structures etc.
+ */
+enum
+{
+  /* This is the size allocated for MRT messages.
+   *
+   * For dumping packets in MRT_MST_BGP4MP_MESSAGE_AS4 form, the maximum
+   * required is as per the CONFIRM below.
+   *
+   * The previous code used a buffer of that size.  UNFORTUNATELY, the TABLE
+   * dumps actually require indefinite size buffering -- so we here allocate
+   * a sizeable lump.
+   *
+   * TODO: fix TABLE dump and indefinite size MRT messages.
+   */
+  BGP_DUMP_BUFFER_SIZE  = 16 * 1024
+} ;
+
+CONFIRM(BGP_DUMP_BUFFER_SIZE >= (MRT_COMMON_HEADER_SIZE +
+                                 MRT_BGP4MP_HEADER_SIZE +
+                                 (BGP_MAX_PACKET_SIZE * 2))) ;
+
+typedef enum bgp_dump_type
 {
   BGP_DUMP_ALL,
   BGP_DUMP_UPDATES,
-  BGP_DUMP_ROUTES
-};
+  BGP_DUMP_TABLE,
 
-enum MRT_MSG_TYPES {
-   MSG_NULL,
-   MSG_START,                   /* sender is starting up */
-   MSG_DIE,                     /* receiver should shut down */
-   MSG_I_AM_DEAD,               /* sender is shutting down */
-   MSG_PEER_DOWN,               /* sender's peer is down */
-   MSG_PROTOCOL_BGP,            /* msg is a BGP packet */
-   MSG_PROTOCOL_RIP,            /* msg is a RIP packet */
-   MSG_PROTOCOL_IDRP,           /* msg is an IDRP packet */
-   MSG_PROTOCOL_RIPNG,          /* msg is a RIPNG packet */
-   MSG_PROTOCOL_BGP4PLUS,       /* msg is a BGP4+ packet */
-   MSG_PROTOCOL_BGP4PLUS_01,    /* msg is a BGP4+ (draft 01) packet */
-   MSG_PROTOCOL_OSPF,           /* msg is an OSPF packet */
-   MSG_TABLE_DUMP,              /* routing table dump */
-   MSG_TABLE_DUMP_V2            /* routing table dump, version 2 */
-};
+  BGP_DUMP_TABLE_NOW,           /* One shot                     */
 
-static int bgp_dump_interval_func (struct thread *);
+  BGP_DUMP_TYPE_COUNT,          /* for arrays of dumps          */
+} bgp_dump_type_t ;
 
 struct bgp_dump
 {
-  enum bgp_dump_type type;
+  const char* typename ;        /* for logging                  */
 
-  char *filename;
+  char* filename ;      /* complete path, as opened             */
+  int   fd ;            /* < 0 if not open                      */
 
-  FILE *fp;
+  struct stream* obuf ; /* set up when file is opened           */
 
-  unsigned int interval;
+  uint  seq ;           /* for TABLE (and TABLE_NOW) dumps      */
 
-  char *interval_str;
+  char* buf ;           /* for TABLE (and TABLE_NOW) dumps...   */
+  char* p ;             /* ...put pointer                       */
+  uint  s ;             /* ...buffer size                       */
+  uint  h ;             /* ...amount of space have left         */
+} ;
 
-  struct thread *t_interval;
-};
+typedef struct bgp_dump  bgp_dump_t ;
+typedef struct bgp_dump* bgp_dump ;
 
-/* BGP packet dump output buffer. */
-struct stream *bgp_dump_obuf;
+struct bgp_dump_control
+{
+  bgp_dump_type_t type ;
+
+  uint      interval;
+
+  char*     template ;
+  char*     interval_str;
+
+  qtimer    qtr ;
+  bgp_dump  bd ;
+} ;
 
-/* BGP dump strucuture for 'dump bgp all' */
-struct bgp_dump bgp_dump_all;
+typedef struct bgp_dump_control  bgp_dump_control_t ;
+typedef struct bgp_dump_control* bgp_dump_control ;
 
-/* BGP dump structure for 'dump bgp updates' */
-struct bgp_dump bgp_dump_updates;
+/* The control structures for the various bd types -- these belong to the
+ * Routing Engine.
+ *
+ * Start with none at all.
+ */
+static bgp_dump_control bgp_dumps[BGP_DUMP_TYPE_COUNT] = { NULL } ;
 
-/* BGP dump structure for 'dump bgp routes' */
-struct bgp_dump bgp_dump_routes;
+/* These belong to the BGP Engine
+ */
+static bgp_dump bd_all     = NULL ;
+static bgp_dump bd_updates = NULL ;
 
-/* Dump whole BGP table is very heavy process.  */
-struct thread *t_bgp_dump_routes;
+bool bgp_dump_state_flag   = false ;
+bool bgp_dump_packet_flag  = false ;
 
-/* Some define for BGP packet dump. */
-static FILE *
-bgp_dump_open_file (struct bgp_dump *bgp_dump)
+/* Types of dump -- in CLI form for ALL, UPDATES and TABLE
+ */
+static const char* bgp_dump_name[] =
+{
+    [BGP_DUMP_ALL]        = "all",
+    [BGP_DUMP_UPDATES]    = "updates",
+    [BGP_DUMP_TABLE]      = "routes-mrt",
+
+    [BGP_DUMP_TABLE_NOW]  = "one-shot routes-mrt",
+} ;
+
+/*==============================================================================
+ * Dump control
+ */
+static void bgp_dump_engine_set(bgp_dump bd, bgp_dump_type_t type) ;
+static void bgp_dump_timer_expired(qtimer qtr, void* timer_info,
+                                                            qtime_mono_t when) ;
+static void bgp_dump_table(bgp_dump_control bdc) ;
+static void bgp_dump_control_free(bgp_dump_control bdc) ;
+static bgp_dump bgp_dump_free(bgp_dump bd) ;
+
+static bgp_dump bgp_dump_bytes(bgp_dump bd, void* bytes, uint n) ;
+static bgp_dump bgp_dump_flush(bgp_dump bd) ;
+static bgp_dump bgp_dump_put(bgp_dump bd, void* p0, uint n0, void* p1, uint n1);
+static bgp_dump bgp_dump_truncate(bgp_dump bd) ;
+
+/*------------------------------------------------------------------------------
+ * Open new file for the given dump
+ *
+ * This is called when an interval timer goes off, or a new ALL or UPDATES
+ * dump is set, or a TABLES_NOW dump is set.
+ *
+ * If this is ALL or UPDATES, then after successful file open, pass the bd
+ * to the BGP Engine !  Note that does nothing if file open fails, so any
+ * existing dump will continue.
+ *
+ * NB: no file can currently be open on the Routing Engine side.
+ *
+ * NB: creates file if required.  Does not, however, truncate -- that is done
+ *     just before the file is to be used.  (This means that if the same file
+ *     is opened for ALL or UPDATES, then the file is truncated on the
+ *     BGP Engine side, so between writes to the file !)
+ */
+static bool
+bgp_dump_open_file (bgp_dump_control bdc, struct vty* vty)
 {
   int ret;
   time_t clock;
   struct tm tm;
   qpath   path ;
   qstring name ;
-  mode_t oldumask;
+  const char* typename ;
+  int fd ;
+  bgp_dump bd ;
+
+  assert(bdc->bd == NULL) ;             /* Nothing should be open       */
 
+  typename = bgp_dump_name[bdc->type] ;
+
+  /* Construct filename from template and attempt to open the file.
+   */
   time (&clock);
   localtime_r(&clock, &tm);
 
-  if (bgp_dump->filename[0] != DIRECTORY_SEP)
-    {
-      path = vty_getcwd(NULL) ;
-      qpath_append_str(path, bgp_dump->filename) ;
-    }
-  else
-    path = qpath_set(NULL, bgp_dump->filename) ;
+  name = qs_new_size(NULL, PATH_MAX) ;
 
-  name = qs_new_size(NULL, qpath_len(path) + 60) ;
+  ret = strftime (qs_char_nn(name), qs_size_nn(name), bdc->template, &tm);
 
-  ret = strftime (qs_char_nn(name), qs_len_nn(name), qpath_string(path), &tm);
+  if (ret != 0)
+    {
+      mode_t oldumask ;
 
-  qpath_free(path) ;
+      qs_set_len_nn(name, ret) ;
+      path = qpath_complete(qpath_set_qs(NULL, name), vty_getcwd(NULL)) ;
 
-  if (ret == 0)
-    {
-      zlog_warn ("bgp_dump_open_file: strftime error");
-      return NULL;
-    }
+      oldumask = umask(0777 & ~LOGFILE_MASK);
+      fd = open(qpath_string(path), O_WRONLY | O_CREAT, LOGFILE_MASK) ;
 
-  if (bgp_dump->fp)
-    fclose (bgp_dump->fp);
+      if (fd < 0)
+        {
+          int err ;
 
+          err = errno ;
 
-  oldumask = umask(0777 & ~LOGFILE_MASK);
-  bgp_dump->fp = fopen (qs_char_nn(name), "w");
+          zlog_warn("Failed to open %s dump file %s: %s", typename,
+                                       qpath_string(path), errtoa(err, 0).str) ;
+          if (vty != NULL)
+            vty_out(vty, "Failed to open file %s: %s\n",
+                                       qpath_string(path), errtoa(err, 0).str) ;
+        } ;
 
-  if (bgp_dump->fp == NULL)
-    {
-      zlog_warn("bgp_dump_open_file: %s: %s", qs_char_nn(name),
-                                                          errtoa(errno, 0).str);
-      umask(oldumask);
-      return NULL;
+      umask(oldumask) ;
     }
-  umask(oldumask);
+  else
+    {
+      fd = -1 ;
+      path = NULL ;
+
+      zlog_warn ("Failed to open %s dump file: strftime failed on '%s'",
+                                                      typename, bdc->template) ;
+      if (vty != NULL)
+        vty_out(vty, "Failed in strftime on '%s'\n", bdc->template) ;
+    } ;
+
+  /* If opened OK, construct the bgp_dump object, log success and then either
+   * set the bdc->bd or pass the bd to the BGP Engine.
+   */
+  if (fd >= 0)
+    {
+      bd = XCALLOC(MTYPE_BGP_DUMP, sizeof(bgp_dump_t)) ;
+
+      /* Zeroising sets bd->buf etc NULL
+       */
+      bd->typename = typename ;
+      bd->filename = XSTRDUP(MTYPE_BGP_DUMP, qpath_string(path)) ;
+      bd->fd       = fd ;
+      bd->obuf     = stream_new(BGP_DUMP_BUFFER_SIZE) ;
+
+      zlog_info("Opened %s dump file: %s", bd->typename, bd->filename) ;
+
+      /* If this is ALL or UPDATE, then we now pass the open file to the
+       * BGP Engine, and forget all about it.
+       *
+       * Otherwise, set the bgp_dump_control's bgp_dump
+       */
+      if ( (bdc->type == BGP_DUMP_ALL) || (bdc->type == BGP_DUMP_UPDATES) )
+        bgp_dump_engine_set(bd, bdc->type) ;
+      else
+        bdc->bd = bd ;
+    } ;
 
-  return bgp_dump->fp;
+  /* Release buffer and return success/failure.
+   */
+  qs_free(name) ;
+  qpath_free(path) ;
+
+  return (fd >= 0) ;
 }
 
-static int
-bgp_dump_interval_add (struct bgp_dump *bgp_dump, int interval)
+/*------------------------------------------------------------------------------
+ * Set a new interval -- if zero, turn off timer.
+ */
+static void
+bgp_dump_set_timer(bgp_dump_control bdc)
 {
-  int secs_into_day;
-  time_t t;
-  struct tm tm;
+  uint interval = bdc->interval ;
 
   if (interval > 0)
     {
-      /* Periodic dump every interval seconds */
-      if ((interval < 86400) && ((86400 % interval) == 0))
+      /* Periodic dump every interval seconds
+       */
+      if ((interval < (24 * 60 * 60)) && (((24 * 60 * 60) % interval) == 0))
 	{
-	  /* Dump at predictable times: if a day has a whole number of
-	   * intervals, dump every interval seconds starting from midnight
-	   */
+          /* Dump at predictable times: if a day has a whole number of
+           * intervals, dump every interval seconds starting from midnight
+           */
+          int secs_into_day;
+          time_t t;
+          struct tm tm;
+
 	  (void) time(&t);
 	  localtime_r(&t, &tm);
-	  secs_into_day = tm.tm_sec + 60*tm.tm_min + 60*60*tm.tm_hour;
+	  secs_into_day = tm.tm_sec + 60 * (tm.tm_min + 60 * tm.tm_hour) ;
 	  interval = interval - secs_into_day % interval; /* always > 0 */
 	}
-      bgp_dump->t_interval = thread_add_timer (master, bgp_dump_interval_func,
-					       bgp_dump, interval);
+
+      if (bdc->qtr == NULL)
+        bdc->qtr = qtimer_init_new(NULL, routing_nexus->pile,
+                                                 bgp_dump_timer_expired, bdc) ;
+      qtimer_set(bdc->qtr, qt_add_monotonic(QTIME(interval)), NULL) ;
     }
   else
     {
-      /* One-off dump: execute immediately, don't affect any scheduled dumps */
-      bgp_dump->t_interval = thread_add_event (master, bgp_dump_interval_func,
-					       bgp_dump, 0);
-    }
+      if (bdc->qtr != NULL)
+        qtimer_unset(bdc->qtr) ;
+    } ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Timer expired function -- called by qtimer.
+ *
+ * For all except BGP_DUMP_TABLE_NOW, open a new dump file.
+ *
+ * For BGP_DUMP_TABLE and BGP_DUMP_TABLE_NOW dump the table and close the
+ * dump file.
+ *
+ * For BGP_DUMP_ALL and BGP_DUMP_UPDATES, pass the new file to the BGP Engine,
+ * which will close the current and adopt the new.
+ *
+ * For all but BGP_DUMP_TABLE_NOW, set a new interval time.
+ */
+static void
+bgp_dump_timer_expired(qtimer qtr, void* timer_info, qtime_mono_t when)
+{
+  bgp_dump_control bdc = timer_info ;
 
-  return 0;
-}
+  assert(bdc->qtr == qtr) ;
 
-/* Dump common header. */
+  switch(bdc->type)
+    {
+      case BGP_DUMP_ALL:
+      case BGP_DUMP_UPDATES:
+        bgp_dump_open_file(bdc, NULL) ; /* rotate -- posts new file to
+                                         * the BGP Engine               */
+        break ;
+
+      case BGP_DUMP_TABLE:
+        if (bgp_dump_open_file(bdc, NULL))
+          bgp_dump_table(bdc) ;         /* if open succeeds             */
+        break ;
+
+      case BGP_DUMP_TABLE_NOW:
+        bgp_dump_table(bdc) ;           /* file already open            */
+        bdc->interval = 0 ;             /* do not repeat                */
+        break ;
+
+      default:                          /* should not happen !          */
+        bdc->interval = 0 ;
+        break ;
+    } ;
+
+  /* Unless is BGP_DUMP_TABLE_NOW, set a new interval time.  Note that for a
+   * TABLE dump, if the current dump took longer than the interval, then will
+   * skip one or more dumps.
+   *
+   * Otherwise, this is a BGP_DUMP_TABLE_NOW, which has completed, and can
+   * now be forgotten.
+   */
+  if (bdc->type != BGP_DUMP_TABLE_NOW)
+    bgp_dump_set_timer(bdc) ;
+  else
+    bgp_dump_control_free(bdc) ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Free the given bgp_dump control, completely.
+ *
+ * Closes and frees any bgp_dump.
+ */
 static void
-bgp_dump_header (struct stream *obuf, int type, int subtype)
+bgp_dump_control_free(bgp_dump_control bdc)
 {
-  time_t now;
+  if (bdc != NULL)
+    {
+      bgp_dump_free(bdc->bd) ;
 
-  /* Set header. */
-  time (&now);
+      XFREE(MTYPE_BGP_DUMP, bdc->template) ;
+      XFREE(MTYPE_BGP_DUMP, bdc->interval_str) ;
 
-  /* Put dump packet header. */
-  stream_putl (obuf, now);
-  stream_putw (obuf, type);
-  stream_putw (obuf, subtype);
+      qtimer_free(bdc->qtr) ;   /* unsets any running timer     */
 
-  stream_putl (obuf, 0);	/* len */
-}
+      assert(bgp_dumps[bdc->type] == bdc) ;
+      bgp_dumps[bdc->type] = NULL ;
 
-static void
-bgp_dump_set_size (struct stream *s, int type)
+      XFREE(MTYPE_BGP_DUMP, bdc) ;
+    } ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Free the given bgp_dump, closing any file.
+ *
+ * Touches nothing beyond the bgp_dump object, so can be used in both the
+ * Routing Engine and the BGP Engine.
+ */
+static bgp_dump
+bgp_dump_free(bgp_dump bd)
 {
-  stream_putl_at (s, 8, stream_get_endp (s) - BGP_DUMP_HEADER_SIZE);
-}
+  if (bd != NULL)
+    {
+      if (bd->fd >= 0)
+        {
+          close(bd->fd) ;
+          zlog_info("Closed %s dump file: %s", bd->typename, bd->filename) ;
+        } ;
+
+      XFREE(MTYPE_BGP_DUMP, bd->filename) ;
+      stream_free(bd->obuf) ;
+      XFREE(MTYPE_TMP, bd->buf) ;
+
+      XFREE(MTYPE_BGP_DUMP, bd) ;
+    } ;
+
+  return NULL ;
+} ;
 
+/*==============================================================================
+ * TABLE (and TABLE_NOW) dumps
+ */
+static bgp_dump bgp_dump_routes_index_table(bgp_dump bd, struct bgp *bgp) ;
+static bgp_dump bgp_dump_routes_family(bgp_dump bd, struct bgp *bgp, afi_t afi);
+
+static struct stream* bgp_dump_header (bgp_dump bd, int type, int subtype) ;
+static void bgp_dump_set_size (struct stream *s, int plus) ;
+
+/*------------------------------------------------------------------------------
+ * Perform a TABLE or TABLE_NOW dump, and close the dump file when finished.
+ */
 static void
-bgp_dump_routes_index_table(struct bgp *bgp)
+bgp_dump_table(bgp_dump_control bdc)
 {
-  struct peer *peer;
-  struct listnode *node;
-  uint16_t peerno = 0;
-  struct stream *obuf;
+  struct bgp *bgp ;
+  bgp_dump bd ;
 
-  obuf = bgp_dump_obuf;
-  stream_reset (obuf);
+  if (qdebug)
+    assert((bdc->bd != NULL) && (bdc->bd->fd >= 0)) ;   /* will be open */
 
-  /* MRT header */
-  bgp_dump_header (obuf, MSG_TABLE_DUMP_V2, TABLE_DUMP_V2_PEER_INDEX_TABLE);
+  bd = bgp_dump_truncate(bdc->bd) ;     /* could (conceivably) fail     */
 
-  /* Collector BGP ID */
-  stream_put_in_addr (obuf, &bgp->router_id);
+  bgp = bgp_get_default();
 
-  /* View name */
-  if(bgp->name)
+  if ((bgp != NULL) && (bd != NULL))
     {
-      stream_putw (obuf, strlen(bgp->name));
-      stream_put(obuf, bgp->name, strlen(bgp->name));
-    }
-  else
-    {
-      stream_putw(obuf, 0);
-    }
+      /* Set up buffer for dumping
+       */
+      bd->s   = 128 * 1024 ;    /* a chunky buffer      */
 
-  /* Peer count */
-  stream_putw (obuf, listcount(bgp->peer));
-
-  /* Walk down all peers */
-  for(ALL_LIST_ELEMENTS_RO (bgp->peer, node, peer))
-    {
+      bd->buf = XMALLOC(MTYPE_TMP, bd->s) ;
+      bd->p   = bd->buf ;
+      bd->h   = bd->s ;
 
-      /* Peer's type */
-      if (sockunion_family(&peer->su) == AF_INET)
-        {
-          stream_putc (obuf, TABLE_DUMP_V2_PEER_INDEX_TABLE_AS4+TABLE_DUMP_V2_PEER_INDEX_TABLE_IP);
-        }
-#ifdef HAVE_IPV6
-      else if (sockunion_family(&peer->su) == AF_INET6)
-        {
-          stream_putc (obuf, TABLE_DUMP_V2_PEER_INDEX_TABLE_AS4+TABLE_DUMP_V2_PEER_INDEX_TABLE_IP6);
-        }
-#endif /* HAVE_IPV6 */
+      bd->seq = 0 ;
 
-      /* Peer's BGP ID */
-      stream_put_in_addr (obuf, &peer->remote_id);
+      /* Construct the index table for all peers and output it.
+       */
+      bd = bgp_dump_routes_index_table(bd, bgp) ;
 
-      /* Peer's IP address */
-      if (sockunion_family(&peer->su) == AF_INET)
-        {
-          stream_put_in_addr (obuf, &peer->su.sin.sin_addr);
-        }
+      /* Now dump all the routes
+       */
+      bd = bgp_dump_routes_family(bd, bgp, AFI_IP) ;
 #ifdef HAVE_IPV6
-      else if (sockunion_family(&peer->su) == AF_INET6)
-        {
-          stream_write (obuf, (u_char *)&peer->su.sin6.sin6_addr,
-                        IPV6_MAX_BYTELEN);
-        }
+      bd = bgp_dump_routes_family(bd, bgp, AFI_IP6) ;
 #endif /* HAVE_IPV6 */
 
-      /* Peer's AS number. */
-      /* Note that, as this is an AS4 compliant quagga, the RIB is always AS4 */
-      stream_putl (obuf, peer->as);
+      /* Flush anything left to go
+       */
+      bgp_dump_flush(bd) ;
+    } ;
+
+  bdc->bd = bgp_dump_free(bd) ;         /* Frees the buf, if any        */
+} ;
+
+/*------------------------------------------------------------------------------
+ * Put the Peer Index Table for the coming route dump
+ *
+ * Returns:  bd if OK, or NULL if failed (or was NULL already).
+ */
+static bgp_dump
+bgp_dump_routes_index_table(bgp_dump bd, struct bgp *bgp)
+{
+  struct stream* s ;
+  struct peer *peer ;
+  struct listnode *node ;
+  uint16_t peerno, len ;
+
+  if (bd == NULL)
+    return NULL ;               /* get out, quick, if no file   */
+
+  /* MRT header
+   */
+  s = bgp_dump_header(bd, MRT_MT_TABLE_DUMP_V2, MRT_MST_TDV2_PEER_INDEX_TABLE) ;
+
+  stream_put_in_addr(s, &bgp->router_id) ;      /* Collector BGP ID     */
+
+  len = (bgp->name != NULL) ? strlen(bgp->name) : 0 ;
+  stream_putw(s, len) ;                         /* View name            */
+  if (len != 0)
+    stream_put(s, bgp->name, len) ;
+
+  stream_putw (s, listcount(bgp->peer));        /* Peer count           */
+
+  /* Walk down all peers and construct peer index entries for the known
+   * address families.
+   *
+   * Note that the type field gives only the type of the address of the peer
+   * and the form of the AS number -- it has nothing to do with the
+   * capabilities of the peer; in particular nothing to do with whether the
+   * peer is an AS4 speaker.
+   */
+  peerno = 0 ;                  /* index for first entry        */
+  for(ALL_LIST_ELEMENTS_RO (bgp->peer, node, peer))
+    {
+      sockunion su ;
+      uint  type ;
 
-      /* Store the peer number for this peer */
-      peer->table_dump_index = peerno;
-      peerno++;
-    }
+      su = &peer->su ;
 
-  bgp_dump_set_size(obuf, MSG_TABLE_DUMP_V2);
+      type = MRT_TDV2_PEER_INDEX_TABLE_AS4 ;    /* always, for simplicity */
 
-  fwrite (STREAM_DATA (obuf), stream_get_endp (obuf), 1, bgp_dump_routes.fp);
-  fflush (bgp_dump_routes.fp);
-}
+      switch (sockunion_family(su))
+        {
+          case AF_INET:
+            type |= MRT_TDV2_PEER_INDEX_TABLE_IPV4 ;
+            break ;
 
+#ifdef HAVE_IPV6
+          case AF_INET6:
+            type |= MRT_TDV2_PEER_INDEX_TABLE_IPV6 ;
+            break ;
+#endif /* HAVE_IPV6 */
 
-/* Runs under child process. */
-static unsigned int
-bgp_dump_routes_func (int afi, int first_run, unsigned int seq)
+          default:
+            continue ;          /* Ignore if not known family ! */
+        } ;
+
+      stream_putc (s, type) ;                   /* Peer's type          */
+      stream_put_in_addr (s, &peer->remote_id); /* Peer's BGP ID        */
+      stream_write (s, sockunion_get_addr(su),  /* Peer's IP address    */
+                       sockunion_get_addr_len(su)) ;
+      stream_putl (s, peer->as);                /* Peer's AS (AS4-wise) */
+
+      peer->table_dump_index = peerno ;         /* set peer number      */
+
+      ++peerno ;
+    } ;
+
+  bgp_dump_set_size(s, 0);
+
+  return bgp_dump_bytes(bd, STREAM_DATA(s), stream_get_endp (s)) ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Dump all routes for the given address family.
+ *
+ * Note: previous comments suggest that this might run under a child
+ *       process -- which would be a good idea -- but does not appear to
+ *       ever have been the case ?
+ *
+ * Returns:  bd if OK, or NULL if failed (or was NULL already).
+ *
+ * Updates the sequence number.
+ *
+ * NB: assumes that th afi is known !
+ */
+static bgp_dump
+bgp_dump_routes_family(bgp_dump bd, struct bgp *bgp, afi_t afi)
 {
-  struct stream *obuf;
-  struct bgp_info *info;
   struct bgp_node *rn;
-  struct bgp *bgp;
   struct bgp_table *table;
+  int subtype ;
 
-  bgp = bgp_get_default ();
-  if (!bgp)
-    return seq;
-
-  if (bgp_dump_routes.fp == NULL)
-    return seq;
+  if (bd == NULL)
+    return NULL ;               /* get out, quick, if no file   */
 
-  /* Note that bgp_dump_routes_index_table will do ipv4 and ipv6 peers,
-     so this should only be done on the first call to bgp_dump_routes_func.
-     ( this function will be called once for ipv4 and once for ipv6 ) */
-  if(first_run)
-    bgp_dump_routes_index_table(bgp);
-
-  obuf = bgp_dump_obuf;
-  stream_reset(obuf);
-
-  /* Walk down each BGP route. */
-  table = bgp->rib[afi][SAFI_UNICAST];
-
-  for (rn = bgp_table_top (table); rn; rn = bgp_route_next (rn))
+  /* Establish subtype of dump entries.
+   */
+  switch (afi)
     {
-      if(!rn->info)
-        continue;
+      case AFI_IP:
+        subtype = MRT_MST_TDV2_RIB_IPV4_UNICAST ;
+        break ;
 
-      stream_reset(obuf);
-
-      /* MRT header */
-      if (afi == AFI_IP)
-        {
-          bgp_dump_header (obuf, MSG_TABLE_DUMP_V2, TABLE_DUMP_V2_RIB_IPV4_UNICAST);
-        }
 #ifdef HAVE_IPV6
-      else if (afi == AFI_IP6)
-        {
-          bgp_dump_header (obuf, MSG_TABLE_DUMP_V2, TABLE_DUMP_V2_RIB_IPV6_UNICAST);
-        }
+      case AFI_IP6:
+        subtype = MRT_MST_TDV2_RIB_IPV6_UNICAST ;
+        break ;
 #endif /* HAVE_IPV6 */
 
-      /* Sequence number */
-      stream_putl(obuf, seq);
+      default:
+        return bd ;             /* do nothing if family unknown */
+    } ;
 
-      /* Prefix length */
-      stream_putc (obuf, rn->p.prefixlen);
+  /* Get the required table -- exit, quick, if none
+   */
+  table = bgp->rib[afi][SAFI_UNICAST] ;
 
-      /* Prefix */
-      if (afi == AFI_IP)
-        {
-          /* We'll dump only the useful bits (those not 0), but have to align on 8 bits */
-          stream_write(obuf, (u_char *)&rn->p.u.prefix4, (rn->p.prefixlen+7)/8);
-        }
-#ifdef HAVE_IPV6
-      else if (afi == AFI_IP6)
-        {
-          /* We'll dump only the useful bits (those not 0), but have to align on 8 bits */
-          stream_write (obuf, (u_char *)&rn->p.u.prefix6, (rn->p.prefixlen+7)/8);
-        }
-#endif /* HAVE_IPV6 */
+  if (table == NULL)
+    return bd ;
+
+  /* Walk down each BGP route
+   */
+  for (rn = bgp_table_top (table) ; rn != NULL ; rn = bgp_route_next (rn))
+    {
+      struct stream* s;
+      struct bgp_info *info;
+      int sizep ;
+      uint16_t entry_count ;
+
+      if (rn->info == NULL)
+        continue;
 
-      /* Save where we are now, so we can overwride the entry count later */
-      int sizep = stream_get_endp(obuf);
+      /* MRT header for MRT_MT_TABLE_DUMP_V2 type message
+       */
+      s = bgp_dump_header (bd, MRT_MT_TABLE_DUMP_V2, subtype) ;
 
-      /* Entry count */
-      uint16_t entry_count = 0;
+      stream_putl(s, bd->seq) ;                 /* Sequence number      */
+      stream_putc(s, rn->p.prefixlen) ;         /* Prefix length        */
+      stream_write(s, &rn->p.u.prefix,          /* Prefix               */
+                      (rn->p.prefixlen+7)/8) ;  /* (zero is OK)         */
 
-      /* Entry count, note that this is overwritten later */
-      stream_putw(obuf, 0);
+      sizep = stream_get_endp(s);               /* will set count later */
+      entry_count = 0;
+      stream_putw(s, entry_count);              /* entry count, so far  */
 
-      for (info = rn->info; info; info = info->info_next)
+      /* Cycle through the known attributes for this prefix             */
+      for (info = rn->info ; info != NULL ; info = info->info_next)
         {
           entry_count++;
 
           /* Peer index */
-          stream_putw(obuf, info->peer->table_dump_index);
+          stream_putw(s, info->peer->table_dump_index);
 
           /* Originated */
-          stream_putl (obuf, bgp_wall_clock(info->uptime));
+          stream_putl (s, bgp_wall_clock(info->uptime));
 
-          /* Dump attribute. */
-          /* Skip prefix & AFI/SAFI for MP_NLRI */
-          bgp_dump_routes_attr (obuf, info->attr, &rn->p);
+          /* Dump attribute.                                            */
+          /* Skip prefix & AFI/SAFI for MP_NLRI                         */
+          bgp_dump_routes_attr (s, info->attr, &rn->p);
         }
 
       /* Overwrite the entry count, now that we know the right number */
-      stream_putw_at (obuf, sizep, entry_count);
+      stream_putw_at (s, sizep, entry_count);
 
-      seq++;
+      bgp_dump_set_size(s, 0) ;
 
-      bgp_dump_set_size(obuf, MSG_TABLE_DUMP_V2);
-      fwrite (STREAM_DATA (obuf), stream_get_endp (obuf), 1, bgp_dump_routes.fp);
+      bd = bgp_dump_bytes(bd, STREAM_DATA(s), stream_get_endp (s)) ;
+      if (bd == NULL)
+        break ;
 
+      ++bd->seq ;
     }
 
-  fflush (bgp_dump_routes.fp);
+  return bd ;
+} ;
 
-  return seq;
-}
+/*==============================================================================
+ * TABLE (and TABLE_NOW) dumps
+ */
 
-static int
-bgp_dump_interval_func (struct thread *t)
+/*------------------------------------------------------------------------------
+ * Reset given stream and construct common header for all MRT packets
+ */
+static struct stream*
+bgp_dump_header (bgp_dump bd, int type, int subtype)
 {
-  struct bgp_dump *bgp_dump;
-  bgp_dump = THREAD_ARG (t);
-  bgp_dump->t_interval = NULL;
+  static struct stream* s ;
 
-  /* Reschedule dump even if file couldn't be opened this time... */
-  if (bgp_dump_open_file (bgp_dump) != NULL)
-    {
-      /* In case of bgp_dump_routes, we need special route dump function. */
-      if (bgp_dump->type == BGP_DUMP_ROUTES)
-	{
-	  unsigned int seq = bgp_dump_routes_func (AFI_IP, 1, 0);
-#ifdef HAVE_IPV6
-	  bgp_dump_routes_func (AFI_IP6, 0, seq);
-#endif /* HAVE_IPV6 */
-	  /* Close the file now. For a RIB dump there's no point in leaving
-	   * it open until the next scheduled dump starts. */
-	  fclose(bgp_dump->fp); bgp_dump->fp = NULL;
-	}
-    }
+  s = bd->obuf ;
+  stream_reset (s);
 
-  /* if interval is set reschedule */
-  if (bgp_dump->interval > 0)
-    bgp_dump_interval_add (bgp_dump, bgp_dump->interval);
+  stream_putl (s, time(NULL));
+  stream_putw (s, type);
+  stream_putw (s, subtype);
+  stream_putl (s, 0);            /* len */
 
-  return 0;
+  return s ;
 }
 
-/* Dump common information. */
+/*------------------------------------------------------------------------------
+ * Set size of MRT packet to be size of stuff in the obuf, less the header,
+ * plus the given size.
+ *
+ * NB: depends on the header being at the start of the given stream.
+ */
 static void
-bgp_dump_common (struct stream *obuf, struct peer *peer, int forceas4)
+bgp_dump_set_size (struct stream *s, int plus)
 {
-  char empty[16] = {0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0};
-
-  /* Source AS number and Destination AS number. */
-  if (forceas4 || CHECK_FLAG (peer->cap, PEER_CAP_AS4_RCV) )
-    {
-      stream_putl (obuf, peer->as);
-      stream_putl (obuf, peer->local_as);
-    }
-  else
-    {
-      stream_putw (obuf, peer->as);
-      stream_putw (obuf, peer->local_as);
-    }
-
-  if (peer->su.sa.sa_family == AF_INET)
-    {
-      stream_putw (obuf, peer->ifindex);
-      stream_putw (obuf, AFI_IP);
+  stream_putl_at (s, 8, stream_get_endp (s) - MRT_COMMON_HEADER_SIZE + plus);
+}
 
-      stream_put (obuf, &peer->su.sin.sin_addr, IPV4_MAX_BYTELEN);
+/*------------------------------------------------------------------------------
+ * Put bytes to dump buffer
+ *
+ * Returns:  bd if OK, or NULL if failed (or was NULL already).
+ *
+ * If fails, closes the file and frees the bd.
+ */
+static bgp_dump
+bgp_dump_bytes(bgp_dump bd, void* bytes, uint n)
+{
+  if (bd == NULL)
+    return NULL ;               /* get out, quick, if no file   */
 
-      if (peer->su_local)
-	stream_put (obuf, &peer->su_local->sin.sin_addr, IPV4_MAX_BYTELEN);
-      else
-	stream_put (obuf, empty, IPV4_MAX_BYTELEN);
-    }
-#ifdef HAVE_IPV6
-  else if (peer->su.sa.sa_family == AF_INET6)
+  while (n > 0)
     {
-      /* Interface Index and Address family. */
-      stream_putw (obuf, peer->ifindex);
-      stream_putw (obuf, AFI_IP6);
+      uint t ;
 
-      /* Source IP Address and Destination IP Address. */
-      stream_put (obuf, &peer->su.sin6.sin6_addr, IPV6_MAX_BYTELEN);
+      t = bd->h ;
 
-      if (peer->su_local)
-	stream_put (obuf, &peer->su_local->sin6.sin6_addr, IPV6_MAX_BYTELEN);
+      if (t >= n)
+        t = n ;
       else
-	stream_put (obuf, empty, IPV6_MAX_BYTELEN);
-    }
-#endif /* HAVE_IPV6 */
-}
-
-/* Dump BGP status change. */
-void
-bgp_dump_state (struct peer *peer, int status_old, int status_new)
+        {
+          if (t == 0)
+            {
+              if ((bd = bgp_dump_flush(bd)) == NULL)
+                break ;
+
+              t = bd->h ;
+              if (t >= n)
+                t = n ;
+            } ;
+        } ;
+
+      memcpy(bd->p, bytes, t) ;
+
+      bd->p += t ;
+      bd->h -= t ;
+
+      bytes = (char*)bytes + t ;
+      n -= t ;
+    } ;
+
+  return bd ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Flush dump buffer to dump
+ *
+ * Returns:  bd if OK, or NULL if failed (or was NULL already).
+ *
+ * If fails, closes the file and frees the bd.
+ */
+static bgp_dump
+bgp_dump_flush(bgp_dump bd)
 {
-  struct stream *obuf;
-
-  /* If dump file pointer is disabled return immediately. */
-  if (bgp_dump_all.fp == NULL)
-    return;
-
-  /* Make dump stream. */
-  obuf = bgp_dump_obuf;
-  stream_reset (obuf);
-
-  bgp_dump_header (obuf, MSG_PROTOCOL_BGP4MP, BGP4MP_STATE_CHANGE_AS4);
-  bgp_dump_common (obuf, peer, 1);/* force this in as4speak*/
+  uint n ;
 
-  stream_putw (obuf, status_old);
-  stream_putw (obuf, status_new);
-
-  /* Set length. */
-  bgp_dump_set_size (obuf, MSG_PROTOCOL_BGP4MP);
-
-  /* Write to the stream. */
-  fwrite (STREAM_DATA (obuf), stream_get_endp (obuf), 1, bgp_dump_all.fp);
-  fflush (bgp_dump_all.fp);
-}
+  if (bd != NULL)
+    {
+      n = bd->p - bd->buf ;
+      if (n > 0)
+        bd = bgp_dump_put(bd, bd->buf, n, NULL, 0) ;
 
-static void
-bgp_dump_packet_func (struct bgp_dump *bgp_dump, struct peer *peer,
-		      struct stream *packet)
+      if (bd != NULL)
+        {
+          bd->p = bd->buf ;
+          bd->h = bd->s ;
+        } ;
+    } ;
+
+  return bd ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Put one or two lumps of data to dump -- first lump may NOT be empty.
+ *
+ * Returns:  bd if OK, or NULL if failed (or was NULL already).
+ *
+ * If fails, closes the file and frees the bd.
+ */
+static bgp_dump
+bgp_dump_put(bgp_dump bd, void* p0, uint n0, void* p1, uint n1)
 {
-  struct stream *obuf;
+  struct iovec iov[2] ;
+  int n ;
+  int w ;
 
-  /* If dump file pointer is disabled return immediately. */
-  if (bgp_dump->fp == NULL)
-    return;
+  if (bd == NULL)
+    return NULL ;               /* get out, quick, if no file   */
 
-  /* Make dump stream. */
-  obuf = bgp_dump_obuf;
-  stream_reset (obuf);
+  iov[0].iov_base = p0 ;
+  iov[0].iov_len  = n0 ;
 
-  /* Dump header and common part. */
-  if (CHECK_FLAG (peer->cap, PEER_CAP_AS4_RCV) )
+  if (n1 > 0)
     {
-      bgp_dump_header (obuf, MSG_PROTOCOL_BGP4MP, BGP4MP_MESSAGE_AS4);
+      iov[1].iov_base = p1 ;
+      iov[1].iov_len  = n1 ;
+
+      n = 2 ;
     }
   else
+    n = 1 ;
+
+  do
+    w = iovec_write_nb(bd->fd, iov, n) ;
+  while (w > 0) ;
+
+  if (w == 0)
+    return bd ;
+
+  /* Error writing to the dump file.
+   */
+  zlog_warn("failed writing %s dump file %s: %s", bd->typename, bd->filename,
+                                                        errtoa(errno, 0).str) ;
+  return bgp_dump_free(bd) ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Truncate the dump file (if any) and seek to start of (now) empty file
+ *
+ * Returns:  bd if OK, or NULL if failed (or was already NULL).
+ *
+ * If fails, closes the file and frees the bd.
+ */
+static bgp_dump
+bgp_dump_truncate(bgp_dump bd)
+{
+  if (bd != NULL)
     {
-      bgp_dump_header (obuf, MSG_PROTOCOL_BGP4MP, BGP4MP_MESSAGE);
-    }
-  bgp_dump_common (obuf, peer, 0);
+      /* Truncate: extremely unlikely to fail, but let's be careful out there.
+       *
+       * Something very odd is going on if does fail, so log as "error".
+       */
+      lseek(bd->fd, 0, SEEK_SET) ;      /* start from the beginning     */
 
-  /* Packet contents. */
-  stream_put (obuf, STREAM_DATA (packet), stream_get_endp (packet));
+      while (1)
+        {
+          int r ;
 
-  /* Set length. */
-  bgp_dump_set_size (obuf, MSG_PROTOCOL_BGP4MP);
+          r = ftruncate(bd->fd, 0) ;    /* redundant in most cases      */
 
-  /* Write to the stream. */
-  fwrite (STREAM_DATA (obuf), stream_get_endp (obuf), 1, bgp_dump->fp);
-  fflush (bgp_dump->fp);
-}
+          if (r >= 0)
+            break ;                     /* OK                   */
 
-/* Called from bgp_packet.c when BGP packet is received. */
-void
-bgp_dump_packet (struct peer *peer, int type, struct stream *packet)
-{
-  /* bgp_dump_all. */
-  bgp_dump_packet_func (&bgp_dump_all, peer, packet);
+          if (errno == EINTR)
+            continue ;                  /* Try again            */
 
-  /* bgp_dump_updates. */
-  if (type == BGP_MSG_UPDATE)
-    bgp_dump_packet_func (&bgp_dump_updates, peer, packet);
-}
+          zlog_err("failed truncating %s dump file %s: %s", bd->typename,
+                                           bd->filename, errtoa(errno, 0).str) ;
+          bd = bgp_dump_free(bd) ;
+
+          break ;                       /* Failed immediately ! */
+        } ;
+    } ;
 
-static unsigned int
+  return bd ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Parse interval string:
+ *
+ *   (\d+[hH])?(\d+[mM])\d+[sS]?  returns -1 if invalid
+ */
+static int
 bgp_dump_parse_time (const char *str)
 {
   int i;
   int len;
-  int seen_h;
-  int seen_m;
+  bool seen_h, seen_m, seen_d ;
   int time;
-  unsigned int total;
+  int total;
 
   time = 0;
   total = 0;
-  seen_h = 0;
-  seen_m = 0;
+  seen_h = false ;
+  seen_m = false ;
+  seen_d = false ;
   len = strlen (str);
 
   for (i = 0; i < len; i++)
     {
       if (isdigit ((int) str[i]))
 	{
+          seen_d = true ;
 	  time *= 10;
 	  time += str[i] - '0';
 	}
-      else if (str[i] == 'H' || str[i] == 'h')
-	{
-	  if (seen_h)
-	    return 0;
-	  if (seen_m)
-	    return 0;
-	  total += time * 60 *60;
-	  time = 0;
-	  seen_h = 1;
-	}
-      else if (str[i] == 'M' || str[i] == 'm')
-	{
-	  if (seen_m)
-	    return 0;
-	  total += time * 60;
-	  time = 0;
-	  seen_h = 1;
-	}
       else
-	return 0;
-    }
-  return total + time;
-}
-
+	{
+          if (!seen_d)
+            return -1 ;         /* must have digit before non-digit     */
+
+          switch (str[i])
+            {
+              case 'H':
+              case 'h':
+                if (seen_h || seen_m)
+                  return -1 ;   /* must not have seen 'h' or 'm'        */
+
+                seen_h = true ;
+                total += time * 60 * 60;
+                break ;
+
+              case 'M':
+              case 'm':
+                if (seen_m)
+                  return -1 ;   /* must not have seen 'm'               */
+
+                seen_m = true ;
+                total += time * 60;
+                break ;
+
+              case 'S':
+              case 's':
+                if (i != (len - 1))
+                  return -1 ;   /* must be at end                       */
+
+              default:
+                return -1 ;     /* unknown character                    */
+            } ;
+
+          seen_d = false ;      /* accept only digit or end             */
+          time = 0 ;            /* no further value, yet                */
+	} ;
+    } ;
+
+  return total + time ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Command function to set (or change) state of a dump.
+ *
+ * If an interval is given, then:
+ *
+ *   If a dump is set, and both the interval and the path are unchanged, then
+ *   do nothing -- debounce.
+ *
+ *   Otherwise, close any existing dump, and set the given one running.
+ *
+ * If no interval is given, then:
+ *
+ *   TABLE:    schedule an immediate, one off TABLE_NOW dump to the given file.
+ *             Leaves any existing periodic dump alone.
+ *
+ *   ALL:
+ *   UPDATES:  set a new dump going.
+ *
+ *             If a dump is already exists with the same file name, then the
+ *             existing file will be truncated.
+ *
+ *             In any case, leave with no rotation interval set.
+ */
 static int
-bgp_dump_set (struct vty *vty, struct bgp_dump *bgp_dump,
-              enum bgp_dump_type type, const char *path,
-              const char *interval_str)
+bgp_dump_set (struct vty *vty, bgp_dump_type_t type, const char *template,
+                                                     const char *interval_str)
 {
-  unsigned int interval;
-
-  if (interval_str)
+  enum cmd_return_code ret ;
+  uint interval;
+  bgp_dump_control bdc ;
+  bool unchanged ;
+
+  assert((type >= 0) && (type < BGP_DUMP_TYPE_COUNT)) ;
+
+  bdc = bgp_dumps[type] ;
+  unchanged = false ;
+
+  /* If we have an interval string then extract interval, and if no actual
+   * change in the bd specification, do nothing.
+   *
+   * If no interval string, then look out for special TABLE_NOW
+   */
+  if (interval_str != NULL)
     {
-
       /* Check interval string. */
-      interval = bgp_dump_parse_time (interval_str);
-      if (interval == 0)
+      int get = bgp_dump_parse_time (interval_str);
+      if (get < 60)
 	{
-	  vty_out (vty, "Malformed interval string%s", VTY_NEWLINE);
-	  return CMD_WARNING;
-	}
+          if (get < 0)
+            vty_out (vty, "Malformed interval string%s", VTY_NEWLINE) ;
+          else
+            vty_out (vty, "Interval < 60 seconds%s", VTY_NEWLINE) ;
 
-      /* Don't schedule duplicate dumps if the dump command is given twice */
-      if (interval == bgp_dump->interval &&
-	  type == bgp_dump->type &&
-          path && bgp_dump->filename && !strcmp (path, bgp_dump->filename))
-	{
-          return CMD_SUCCESS;
-	}
+          return CMD_WARNING;
+	} ;
 
-      /* Set interval. */
-      bgp_dump->interval = interval;
-      if (bgp_dump->interval_str)
-	free (bgp_dump->interval_str);
-      bgp_dump->interval_str = strdup (interval_str);
+      interval = get ;
 
+      /* Don't schedule duplicate dumps if the dump command is given twice  */
+      if (bdc != NULL)
+        unchanged = (interval == bdc->interval) &&
+                                        (strcmp(template, bdc->template) == 0) ;
     }
   else
     {
-      interval = 0;
-    }
-
-  /* Create interval thread. */
-  bgp_dump_interval_add (bgp_dump, interval);
-
-  /* Set type. */
-  bgp_dump->type = type;
+      interval = 0 ;
 
-  /* Set file name. */
-  if (bgp_dump->filename)
-    free (bgp_dump->filename);
-  bgp_dump->filename = strdup (path);
-
-  /* This should be called when interval is expired. */
-  bgp_dump_open_file (bgp_dump);
-
-  return CMD_SUCCESS;
+      if (type == BGP_DUMP_TABLE)
+        {
+          /* A "routes-mrt" type dump with no interval is a one-shot dump, so
+           * set actual type.
+           *
+           * There can be at most one one-shot dump active at any time !
+           *
+           * NB: it's a bit klunky, but we arrange for BGP_DUMP_TABLE_NOW to
+           *     run in 1 second's time -- so that command completes before the
+           *     big work starts.
+           */
+          type = BGP_DUMP_TABLE_NOW ;
+          interval = 1 ;
+          bdc  = NULL ;
+
+          if (bgp_dumps[BGP_DUMP_TABLE_NOW] != NULL)
+            {
+              vty_out (vty, "one-shot routes-mrt dump already pending%s",
+                                                                  VTY_NEWLINE) ;
+              return CMD_WARNING;
+            } ;
+        } ;
+    } ;
+
+  /* If required, create the bgp bd control.
+   *
+   * Zeroising sets:
+   *
+   *    type          -- 0     -- set below
+   *
+   *    interval      -- 0,    none, yet
+   *    template      -- NULL, none, yet
+   *    interval_str  -- NULL  none, yet
+   *    qtr           -- NULL  none, yet
+   *
+   *    bd            -- NULL, no file, yet
+   */
+  if (bdc == NULL)
+    {
+      bdc = (bgp_dump_control)XCALLOC(MTYPE_BGP_DUMP,
+                                                  sizeof(bgp_dump_control_t)) ;
+      bdc->type = type ;
+
+      bgp_dumps[type] = bdc ;
+    } ;
+
+  /* Set interval, interval string and template.
+   *
+   * Even if the interval has not changed, we keep the latest interval string,
+   * just in case that has changed.
+   */
+  bdc->interval = interval ;
+
+  XFREE(MTYPE_BGP_DUMP, bdc->interval_str) ;    /* sets NULL    */
+  if (interval_str != NULL)
+    bdc->interval_str = XSTRDUP(MTYPE_BGP_DUMP, interval_str) ;
+
+  XFREE(MTYPE_BGP_DUMP, bdc->template) ;
+  bdc->template = XSTRDUP(MTYPE_BGP_DUMP, template) ;
+
+  /* If is, in fact, unchanged, get out now.
+   */
+  if (unchanged)
+    return CMD_SUCCESS ;
+
+  /* Unless this is TABLE dump, open a file now if either the template or
+   * the interval have changed.
+   *
+   * Note that for BGP_DUMP_TABLE_NOW changed is always true.
+   *
+   * For ALL and UPDATES dumps, the new file is passed to the BGP Engine.  If
+   * the file open fails, any existing file will continue.
+   *
+   * If fails to open, and no interval was set (so won't be able to try again),
+   * then will return CMD_WARNING.  For TABLE_NOW, will discard the abortive
+   * bdc -- inter alia, it will not then appear in the configuration output.
+   */
+  ret = CMD_SUCCESS ;
+
+  if (type != BGP_DUMP_TABLE)
+    {
+      if (!bgp_dump_open_file(bdc, vty))
+        {
+          if (interval_str == NULL)
+            {
+              ret = CMD_WARNING ;
+
+              if (type == BGP_DUMP_TABLE_NOW)
+                {
+                  bgp_dump_control_free(bdc) ;  /* Discard abortive bdc */
+                  return ret ;
+                } ;
+            } ;
+        } ;
+    } ;
+
+  /* Set interval timer if required, or stop the current timer
+   */
+  bgp_dump_set_timer(bdc) ;
+
+  return ret ;
 }
 
+/*------------------------------------------------------------------------------
+ * Command function to unset a dump.
+ */
 static int
-bgp_dump_unset (struct vty *vty, struct bgp_dump *bgp_dump)
+bgp_dump_unset (struct vty *vty, bgp_dump_type_t type)
 {
-  /* Set file name. */
-  if (bgp_dump->filename)
-    {
-      free (bgp_dump->filename);
-      bgp_dump->filename = NULL;
-    }
-
-  /* This should be called when interval is expired. */
-  if (bgp_dump->fp)
-    {
-      fclose (bgp_dump->fp);
-      bgp_dump->fp = NULL;
-    }
+  bgp_dump_control bdc ;
 
-  /* Create interval thread. */
-  if (bgp_dump->t_interval)
-    {
-      thread_cancel (bgp_dump->t_interval);
-      bgp_dump->t_interval = NULL;
-    }
+  assert((type >= 0) && (type < BGP_DUMP_TYPE_COUNT)) ;
 
-  bgp_dump->interval = 0;
-
-  if (bgp_dump->interval_str)
+  bdc = bgp_dumps[type] ;
+  if (bdc != NULL)
     {
-      free (bgp_dump->interval_str);
-      bgp_dump->interval_str = NULL;
-    }
+      bgp_dump_control_free(bdc) ;
 
+      if ( (type == BGP_DUMP_ALL) || (type == BGP_DUMP_UPDATES) )
+        bgp_dump_engine_set(NULL, type) ;
+    } ;
 
   return CMD_SUCCESS;
 }
 
+/*==============================================================================
+ * The CLI Commands
+ */
+
 DEFUN (dump_bgp_all,
        dump_bgp_all_cmd,
        "dump bgp all PATH",
@@ -688,7 +1240,7 @@ DEFUN (dump_bgp_all,
        "Dump all BGP packets\n"
        "Output filename\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_all, BGP_DUMP_ALL, argv[0], NULL);
+  return bgp_dump_set (vty, BGP_DUMP_ALL, argv[0], NULL);
 }
 
 DEFUN (dump_bgp_all_interval,
@@ -700,7 +1252,7 @@ DEFUN (dump_bgp_all_interval,
        "Output filename\n"
        "Interval of output\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_all, BGP_DUMP_ALL, argv[0], argv[1]);
+  return bgp_dump_set (vty, BGP_DUMP_ALL, argv[0], argv[1]);
 }
 
 DEFUN (no_dump_bgp_all,
@@ -711,7 +1263,7 @@ DEFUN (no_dump_bgp_all,
        "BGP packet dump\n"
        "Dump all BGP packets\n")
 {
-  return bgp_dump_unset (vty, &bgp_dump_all);
+  return bgp_dump_unset (vty, BGP_DUMP_ALL);
 }
 
 DEFUN (dump_bgp_updates,
@@ -722,7 +1274,7 @@ DEFUN (dump_bgp_updates,
        "Dump BGP updates only\n"
        "Output filename\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_updates, BGP_DUMP_UPDATES, argv[0], NULL);
+  return bgp_dump_set (vty, BGP_DUMP_UPDATES, argv[0], NULL);
 }
 
 DEFUN (dump_bgp_updates_interval,
@@ -734,7 +1286,7 @@ DEFUN (dump_bgp_updates_interval,
        "Output filename\n"
        "Interval of output\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_updates, BGP_DUMP_UPDATES, argv[0], argv[1]);
+  return bgp_dump_set (vty, BGP_DUMP_UPDATES, argv[0], argv[1]);
 }
 
 DEFUN (no_dump_bgp_updates,
@@ -745,7 +1297,7 @@ DEFUN (no_dump_bgp_updates,
        "BGP packet dump\n"
        "Dump BGP updates only\n")
 {
-  return bgp_dump_unset (vty, &bgp_dump_updates);
+  return bgp_dump_unset (vty, BGP_DUMP_UPDATES);
 }
 
 DEFUN (dump_bgp_routes,
@@ -756,7 +1308,7 @@ DEFUN (dump_bgp_routes,
        "Dump whole BGP routing table\n"
        "Output filename\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_routes, BGP_DUMP_ROUTES, argv[0], NULL);
+  return bgp_dump_set (vty, BGP_DUMP_TABLE, argv[0], NULL);
 }
 
 DEFUN (dump_bgp_routes_interval,
@@ -768,7 +1320,7 @@ DEFUN (dump_bgp_routes_interval,
        "Output filename\n"
        "Interval of output\n")
 {
-  return bgp_dump_set (vty, &bgp_dump_routes, BGP_DUMP_ROUTES, argv[0], argv[1]);
+  return bgp_dump_set (vty, BGP_DUMP_TABLE, argv[0], argv[1]);
 }
 
 DEFUN (no_dump_bgp_routes,
@@ -779,7 +1331,7 @@ DEFUN (no_dump_bgp_routes,
        "BGP packet dump\n"
        "Dump whole BGP routing table\n")
 {
-  return bgp_dump_unset (vty, &bgp_dump_routes);
+  return bgp_dump_unset (vty, BGP_DUMP_TABLE);
 }
 
 /* BGP node structure. */
@@ -816,53 +1368,41 @@ config_time2str (unsigned int interval)
 }
 #endif
 
+/*------------------------------------------------------------------------------
+ * Output the configuration for bgp dumping
+ */
 static int
 config_write_bgp_dump (struct vty *vty)
 {
-  if (bgp_dump_all.filename)
-    {
-      if (bgp_dump_all.interval_str)
-	vty_out (vty, "dump bgp all %s %s%s",
-		 bgp_dump_all.filename, bgp_dump_all.interval_str,
-		 VTY_NEWLINE);
-      else
-	vty_out (vty, "dump bgp all %s%s",
-		 bgp_dump_all.filename, VTY_NEWLINE);
-    }
-  if (bgp_dump_updates.filename)
-    {
-      if (bgp_dump_updates.interval_str)
-	vty_out (vty, "dump bgp updates %s %s%s",
-		 bgp_dump_updates.filename, bgp_dump_updates.interval_str,
-		 VTY_NEWLINE);
-      else
-	vty_out (vty, "dump bgp updates %s%s",
-		 bgp_dump_updates.filename, VTY_NEWLINE);
-    }
-  if (bgp_dump_routes.filename)
+  bgp_dump_type_t   type ;
+
+  for (type = 0 ; type < BGP_DUMP_TYPE_COUNT ; ++type)
     {
-      if (bgp_dump_routes.interval_str)
-	vty_out (vty, "dump bgp routes-mrt %s %s%s",
-		 bgp_dump_routes.filename, bgp_dump_routes.interval_str,
-		 VTY_NEWLINE);
-      else
-	vty_out (vty, "dump bgp routes-mrt %s%s",
-		 bgp_dump_routes.filename, VTY_NEWLINE);
-    }
+      bgp_dump_control  bdc ;
+
+      bdc = bgp_dumps[type] ;
+
+      if ((bdc == NULL) || (type == BGP_DUMP_TABLE_NOW))
+        continue ;
+
+      vty_out (vty, "dump bgp %s %s", bgp_dump_name[type], bdc->template) ;
+
+      if (bdc->interval_str != NULL)
+        vty_out (vty, " %s", bdc->interval_str) ;
+
+      vty_out (vty, VTY_NEWLINE) ;
+    } ;
   return 0;
 }
 
-/* Initialize BGP packet dump functionality. */
-void
+/*------------------------------------------------------------------------------
+ * Initialize BGP MRT dumping.
+ *
+ * NB: second stage initialisation -- after pthreads start.
+ */
+extern void
 bgp_dump_init (void)
 {
-  memset (&bgp_dump_all, 0, sizeof (struct bgp_dump));
-  memset (&bgp_dump_updates, 0, sizeof (struct bgp_dump));
-  memset (&bgp_dump_routes, 0, sizeof (struct bgp_dump));
-
-  bgp_dump_obuf = stream_new (BGP_MAX_PACKET_SIZE + BGP_DUMP_MSG_HEADER
-                              + BGP_DUMP_HEADER_SIZE);
-
   install_node (&bgp_dump_node, config_write_bgp_dump);
 
   install_element (CONFIG_NODE, &dump_bgp_all_cmd);
@@ -874,11 +1414,318 @@ bgp_dump_init (void)
   install_element (CONFIG_NODE, &dump_bgp_routes_cmd);
   install_element (CONFIG_NODE, &dump_bgp_routes_interval_cmd);
   install_element (CONFIG_NODE, &no_dump_bgp_routes_cmd);
-}
-
-void
+} ;
+
+/*------------------------------------------------------------------------------
+ * Close down all dumping.
+ *
+ * This is called after the BGP Engine has stopped -- so can here free the
+ * BGP Engine stuff, too.
+ */
+extern void
 bgp_dump_finish (void)
 {
-  stream_free (bgp_dump_obuf);
-  bgp_dump_obuf = NULL;
-}
+  uint d ;
+
+  for (d = 0 ; d < BGP_DUMP_TYPE_COUNT ; ++d)
+    bgp_dump_control_free(bgp_dumps[d]) ;
+
+  bd_all     = bgp_dump_free(bd_all) ;
+  bd_updates = bgp_dump_free(bd_updates) ;
+} ;
+
+/*==============================================================================
+ * The BGP_Engine side of bgp_dump.
+ *
+ * For ALL and UPDATES dumps the BGP_Engine is responsible for all actual
+ * I/O.  The Routing Engine will open/rotate files, but pass those to the
+ * BGP_Engine for action.  The BGP_Engine looks after the bgp_dump structures,
+ * and is responsible for closing files.
+ */
+struct bgp_dump_engine_set_args         /* to BGP Engine                */
+{
+  bgp_dump_type_t type ;
+  bgp_dump        bd ;
+} ;
+MQB_ARGS_SIZE_OK(bgp_dump_engine_set_args) ;
+
+static void bgp_dump_engine_do_set(mqueue_block mqb, mqb_flag_t flag) ;
+static struct stream* bgp_dump_common (bgp_dump bd, bgp_connection connection,
+                                                        int subtype, bool as4) ;
+static void bgp_dump_set_flags(void) ;
+static int bgp_dump_fsm_state(bgp_fsm_state_t state) ;
+
+/*------------------------------------------------------------------------------
+ * Set given dump in BGP Engine.
+ *
+ * Passes the given bgp_dump to the BGP Engine.  Responsibility for the fd,
+ * the filename and the stream buffer pass to the BGP Engine.
+ *
+ * Pass NULL to stop the relevant dump.
+ */
+static void
+bgp_dump_engine_set(bgp_dump bd, bgp_dump_type_t type)
+{
+  struct bgp_dump_engine_set_args* args ;
+  mqueue_block mqb ;
+
+  mqb = mqb_init_new(NULL, bgp_dump_engine_do_set, bgp_dumps) ;
+
+  args = mqb_get_args(mqb) ;
+
+  args->type = type ;
+  args->bd   = bd ;
+
+  bgp_to_bgp_engine(mqb, mqb_priority) ;        /* change file ASAP     */
+} ;
+
+/*------------------------------------------------------------------------------
+* BGP Engine: set the given dump.
+*
+* Note that this explicitly truncates the dump file.  This deals with the
+* fringe case of a dump being started using the same name as an existing,
+* active dump !!  (If the old dump wrote something to the file after the
+* new dump had opened the file, then we want to discard the old stuff, now.)
+*/
+static void
+bgp_dump_engine_do_set(mqueue_block mqb, mqb_flag_t flag)
+{
+  bgp_dump* pbd ;
+  bgp_dump  bd ;
+  struct bgp_dump_engine_set_args* args = mqb_get_args(mqb) ;
+
+  pbd = (args->type == BGP_DUMP_UPDATES) ? &bd_updates : &bd_all ;
+
+  *pbd = bgp_dump_free(*pbd) ;          /* close down any existing dump */
+
+  bd = bgp_dump_truncate(args->bd) ;    /* truncate before use          */
+
+  if (flag == mqb_action)
+    *pbd = bd ;                         /* set new dump                 */
+  else
+    *pbd = bgp_dump_free(bd) ;          /* close down new dump          */
+
+  bgp_dump_set_flags() ;                /* reflect current state        */
+
+  mqb_free(mqb) ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Dump BGP status change, if required -- BGP Engine
+ *
+ * Does nothing if no bd_all.
+ *
+ * Frees the dump in the event of any I/O error.
+ */
+extern void
+bgp_dump_state (bgp_connection connection, bgp_fsm_state_t new_state)
+{
+  struct stream *s;
+
+  if (bd_all != NULL)
+    {
+      s = bgp_dump_common(bd_all, connection, MRT_MST_BGP4MP_STATE_CHANGE_AS4,
+                                                                         true) ;
+      stream_putw(s, bgp_dump_fsm_state(connection->state));
+      stream_putw(s, bgp_dump_fsm_state(new_state)) ;
+
+      bgp_dump_set_size (s, 0);
+
+      bd_all = bgp_dump_put(bd_all, STREAM_DATA (s),
+                                                 stream_get_endp (s), NULL, 0) ;
+      if (bd_all != NULL)
+        return ;                /* OK, so no flag change        */
+    } ;
+
+  bgp_dump_set_flags() ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Dump BGP packet received, if required -- BGP Engine
+ *
+ * Does nothing if no bd_all and no bd_updates.
+ *
+ * Frees a dump in the event of any I/O error on it.
+ */
+extern void
+bgp_dump_packet (bgp_connection connection)
+{
+  bgp_dump  bd ;
+  struct stream* s ;
+  uint plen ;
+  bool du ;
+
+  /* If we have nothing to do, get out, quick.
+   *
+   * If we have something to do, select one of the stream buffers for the
+   * MRT header.  Note that in the (unlikely) event of having both ALL and
+   * UPDATES dumps, selects the UPDATES obuf.
+   */
+  du = (bd_updates != NULL) && (connection->msg_type == BGP_MSG_UPDATE) ;
+
+  if (du)
+    bd = bd_updates ;
+  else if (bd_all != NULL)
+    bd = bd_all ;
+  else
+    return bgp_dump_set_flags() ;       /* nothing to do        */
+
+  /* Construct message header for packet dump
+   *
+   * Note that in the (unlikely) event of having both ALL and UPDATES dumps,
+   * we construct just the one header.
+   */
+  s = bgp_dump_common(bd, connection,
+                                connection->as4 ? MRT_MST_BGP4MP_MESSAGE_AS4
+                                                : MRT_MST_BGP4MP_MESSAGE,
+                                                              connection->as4) ;
+  plen = stream_get_endp(connection->ibuf) ;
+  bgp_dump_set_size (s, plen) ;
+
+  /* Output the MRT header and the packet
+   *
+   * In the event of an I/O failure, free the dump.
+   *
+   * Note that in the (unlikely) event of having both ALL and UPDATES dumps,
+   * we output the UPDATES last -- so that if the ALL fails the message
+   * header for the UPDATES dump is still there !
+   */
+  if (bd_all != NULL)
+    {
+      bd_all = bgp_dump_put(bd_all, STREAM_DATA(s),
+                                         stream_get_endp(s),
+                                          STREAM_DATA(connection->ibuf), plen) ;
+      if (bd_all == NULL)
+        bgp_dump_set_flags() ;
+    } ;
+
+  if (du)
+    {
+      bd_updates = bgp_dump_put(bd_updates, STREAM_DATA(s),
+                                         stream_get_endp(s),
+                                          STREAM_DATA(connection->ibuf), plen) ;
+      if (bd_updates == NULL)
+        bgp_dump_set_flags() ;
+    } ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Construct header and common parts of a MSG_PROTOCOL_BGP4MP MRT message
+ */
+static struct stream*
+bgp_dump_common (bgp_dump bd, bgp_connection connection, int subtype, bool as4)
+{
+  static const char empty[16] = {0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0};
+
+  struct stream *s ;
+  asn_t remote_as ;
+  asn_t local_as ;
+  sockunion su_remote ;
+  sockunion su_local ;
+  int sal ;
+
+  s = bgp_dump_header(bd, MRT_MT_BGP4MP, subtype) ;
+
+  /* Source AS number and Destination AS number
+   */
+  remote_as = connection->session->as_peer ;
+  local_as  = connection->session->open_send->my_as ;
+
+  if (as4)
+    {
+      stream_putl (s, remote_as);
+      stream_putl (s, local_as);
+    }
+  else
+    {
+      stream_putw (s, remote_as);
+      stream_putw (s, local_as);
+    }
+
+  /* Interface index
+   */
+  stream_putw (s, connection->session->ifindex);
+
+  /* Remote and local IP addresses
+   *
+   * If no connection has been made, there will not be a local IP address.
+   *
+   * Note that we expect the local and remote addresses to be of the same
+   * family.  We use the length from the remote address in both cases -- so if
+   * something "impossible" has happened, then the message will still be
+   * "syntactically" well formed.
+   */
+  su_remote = connection->session->su_peer ;
+  su_local  = connection->su_local ;
+
+  stream_putw (s, sockunion_get_afi(su_remote)) ;
+
+  confirm(MRT_AFI_IPv4 == AFI_IP) ;
+#ifdef HAVE_IPV6
+  confirm(MRT_AFI_IPv6 == AFI_IP6) ;
+#endif
+
+  sal = sockunion_get_addr_len(su_remote) ;
+  stream_put(s, sockunion_get_addr(su_remote), sal) ;
+
+  if (su_local != NULL)
+    stream_put(s, sockunion_get_addr(su_local), sal) ;
+  else
+    stream_put(s, empty, sal) ;
+
+  return s ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Set the bgp_dump_state_flag and the bgp_dump_packet_flag as required.
+ *
+ * These flags simply reflect the state of the ALL and UPDATES dumps,
+ * but are used to avoid calling the dump functions when not required (which
+ * is most of the time !).
+ *
+ * Note that it doesn't matter if these flags were to not properly reflect the
+ * dump state !
+ */
+static void
+bgp_dump_set_flags(void)
+{
+  bgp_dump_state_flag   = (bd_all != NULL) ;
+  bgp_dump_packet_flag  = (bd_all != NULL) || (bd_updates != NULL) ;
+} ;
+
+/*------------------------------------------------------------------------------
+ * Map an internal bgp_fsm_state_t value to the MRT Values
+ */
+static int
+bgp_dump_fsm_state(bgp_fsm_state_t state)
+{
+  switch (state)
+    {
+      case bgp_fsm_sInitial:
+        return MRT_FSM_UNDEF ;
+
+      case bgp_fsm_sIdle:
+        return MRT_FSM_Idle ;
+
+      case bgp_fsm_sConnect:
+        return MRT_FSM_Connect ;
+
+      case bgp_fsm_sActive:
+        return MRT_FSM_Active ;
+
+      case bgp_fsm_sOpenSent:
+        return MRT_FSM_OpenSent ;
+
+      case bgp_fsm_sOpenConfirm:
+        return MRT_FSM_OpenConfirm ;
+
+      case bgp_fsm_sEstablished:
+        return MRT_FSM_Established ;
+
+      case bgp_fsm_sStopping:
+        return MRT_FSM_UNDEF ;
+
+      default:
+        return MRT_FSM_UNDEF ;
+    } ;
+} ;
diff --git a/bgpd/bgp_dump.h b/bgpd/bgp_dump.h
index e097c784..46eb9e59 100644
--- a/bgpd/bgp_dump.h
+++ b/bgpd/bgp_dump.h
@@ -1,4 +1,4 @@
-/* BGP dump routine.
+/* BGP bd routine.
    Copyright (C) 1999 Kunihiro Ishiguro
 
 This file is part of GNU Zebra.
@@ -21,35 +21,22 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 #ifndef _QUAGGA_BGP_DUMP_H
 #define _QUAGGA_BGP_DUMP_H
 
-/* MRT compatible packet dump values.  */
-/* type value */
-#define MSG_PROTOCOL_BGP4MP  16
-/* subtype value */
-#define BGP4MP_STATE_CHANGE          0
-#define BGP4MP_MESSAGE               1
-#define BGP4MP_ENTRY                 2
-#define BGP4MP_SNAPSHOT              3
-#define BGP4MP_MESSAGE_AS4           4
-#define BGP4MP_STATE_CHANGE_AS4      5
-
-#define BGP_DUMP_HEADER_SIZE 12
-#define BGP_DUMP_MSG_HEADER  40
-
-#define TABLE_DUMP_V2_PEER_INDEX_TABLE   1
-#define TABLE_DUMP_V2_RIB_IPV4_UNICAST   2
-#define TABLE_DUMP_V2_RIB_IPV4_MULTICAST 3
-#define TABLE_DUMP_V2_RIB_IPV6_UNICAST   4
-#define TABLE_DUMP_V2_RIB_IPV6_MULTICAST 5
-#define TABLE_DUMP_V2_RIB_GENERIC        6
-
-#define TABLE_DUMP_V2_PEER_INDEX_TABLE_IP  0
-#define TABLE_DUMP_V2_PEER_INDEX_TABLE_IP6 1
-#define TABLE_DUMP_V2_PEER_INDEX_TABLE_AS2 0
-#define TABLE_DUMP_V2_PEER_INDEX_TABLE_AS4 2
+#include "bgp_connection.h"
+#include <stdbool.h>
 
+/*------------------------------------------------------------------------------
+ */
 extern void bgp_dump_init (void);
 extern void bgp_dump_finish (void);
-extern void bgp_dump_state (struct peer *, int, int);
-extern void bgp_dump_packet (struct peer *, int, struct stream *);
+
+extern void bgp_dump_state (bgp_connection connection,
+                                                    bgp_fsm_state_t new_state) ;
+extern void bgp_dump_packet (bgp_connection connection) ;
+
+/*------------------------------------------------------------------------------
+ * These flags are set iff the respective dump function should be called.
+ */
+extern bool bgp_dump_state_flag ;
+extern bool bgp_dump_packet_flag ;
 
 #endif /* _QUAGGA_BGP_DUMP_H */
diff --git a/bgpd/bgp_fsm.c b/bgpd/bgp_fsm.c
index a03f1b2c..701396b7 100644
--- a/bgpd/bgp_fsm.c
+++ b/bgpd/bgp_fsm.c
@@ -32,6 +32,7 @@
 #include "bgpd/bgp_fsm.h"
 #include "bgpd/bgp_msg_write.h"
 #include "bgpd/bgp_msg_read.h"
+#include "bgpd/bgp_dump.h"
 
 #include "lib/qtimers.h"
 #include "lib/sockunion.h"
@@ -2326,9 +2327,13 @@ static void
 bgp_fsm_state_change(bgp_connection connection, bgp_fsm_state_t new_state)
 {
   bgp_connection sibling ;
-  unsigned  interval ;
+  uint interval ;
+
   bgp_session    session = connection->session ;
 
+  if (bgp_dump_state_flag)
+    bgp_dump_state(connection, new_state) ;
+
   switch (new_state)
     {
     /* Base state of connection's finite state machine -- when a session has
diff --git a/bgpd/bgp_main.c b/bgpd/bgp_main.c
index 38fbb7e4..40e6c158 100644
--- a/bgpd/bgp_main.c
+++ b/bgpd/bgp_main.c
@@ -38,6 +38,7 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 #include "plist.h"
 #include "qpnexus.h"
 #include "qlib_init.h"
+#include "qfstring.h"
 
 #include "bgpd/bgpd.h"
 #include "bgpd/bgp_attr.h"
@@ -194,6 +195,7 @@ static void sighup_action(mqueue_block mqb, mqb_flag_t flag);
 static void sighup_enqueue(void);
 static void sigterm_action(mqueue_block mqb, mqb_flag_t flag);
 static void sigterm_enqueue(void);
+static void bgp_show_nexus_init(void) ;
 
 static struct quagga_signal_t bgp_signals[] =
 {
@@ -267,7 +269,7 @@ sigusr1 (void)
  * For example, "retain_mode" may be set.
  *
  * Note that by the time reach here, only the main (CLI) thread is running,
- *
+ * so may release things that belong to any thread !
  */
 static void
 bgp_exit (int status)
@@ -421,13 +423,15 @@ init_second_stage(bool pthreads)
   bgp_peer_index_mutex_init();
 
   /* Make nexus for main thread, always needed */
-  cli_nexus = qpn_init_new(cli_nexus, true);    /* main thread          */
+  cli_nexus = qpn_init_new(cli_nexus, 1,                /* main thread  */
+                         qpthreads_enabled ? "CLI thread"
+                                           : "bgpd (no threads)");
 
   /* if using pthreads create additional nexus */
   if (qpthreads_enabled)
     {
-      bgp_nexus     = qpn_init_new(bgp_nexus,     false);
-      routing_nexus = qpn_init_new(routing_nexus, false);
+      bgp_nexus     = qpn_init_new(bgp_nexus, 0, "BGP Engine thread");
+      routing_nexus = qpn_init_new(routing_nexus, 0, "Routing Engine thread");
     }
   else
     {
@@ -466,6 +470,8 @@ init_second_stage(bool pthreads)
   /* BGP related initialization.  */
   bgp_init ();
 
+  bgp_show_nexus_init() ;
+
   /* Sort CLI commands. */
   sort_node ();
 }
@@ -808,3 +814,104 @@ sigterm_action(mqueue_block mqb, mqb_flag_t flag)
   mqb_free(mqb);
 } ;
 
+/*==============================================================================
+ * CLI for showing what nexuses are up to
+ */
+static void bgp_do_show_nexus(struct vty* vty, qpn_nexus qpn) ;
+
+DEFUN_CALL(bgp_show_nexus,
+      bgp_show_nexus_cmd,
+      "show nexus (all|cli|routing|bgp)",
+      SHOW_STR
+      "For nexus activity\n"
+      "all nexuses\n"
+      "CLI nexus\n"
+      "Routing Engine\n"
+      "BGP Engine\n")
+{
+  if ((*(argv[0]) == 'a') && qpthreads_enabled)
+    {
+      bgp_do_show_nexus(vty, cli_nexus) ;
+      bgp_do_show_nexus(vty, bgp_nexus) ;
+      bgp_do_show_nexus(vty, routing_nexus) ;
+    }
+  else
+    {
+      qpn_nexus qpn = cli_nexus ;
+
+      if (qpthreads_enabled)
+        {
+          if (*(argv[0]) == 'r')
+            qpn = routing_nexus ;
+          else if (*(argv[0]) == 'b')
+            qpn = bgp_nexus ;
+        } ;
+
+      bgp_do_show_nexus(vty, qpn) ;
+    } ;
+
+  return CMD_SUCCESS ;
+} ;
+
+static void
+bgp_show_nexus_init(void)
+{
+  install_element (VIEW_NODE, &bgp_show_nexus_cmd);
+  install_element (ENABLE_NODE, &bgp_show_nexus_cmd);
+} ;
+
+static void
+bgp_do_show_nexus(struct vty* vty, qpn_nexus qpn)
+{
+  qpn_stats_t prev ;
+  qpn_stats_t curr ;
+
+  qtime_t  now ;
+  qtime_t  delta ;
+  qtime_t  idle_delta ;
+
+  qpn_get_stats(qpn, &curr, &prev) ;
+  now = qt_get_monotonic() ;
+
+  delta = curr.last_time - prev.last_time ;
+  idle_delta = curr.idle - prev.idle ;
+
+  vty_out(vty, "%s updated %s ago: %s(%s/%s) cycles\n",
+          qpn->name,
+          qfs_time_period(curr.last_time - now, 0).str,
+          qfs_dec_value(curr.cycles, pf_scale).str,
+          qfs_dec_value(curr.cycles - prev.cycles, pf_scale | pf_plus).str,
+          qfs_time_period(delta, pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) active  %s(%s) idle\n",
+      qfs_time_period((curr.last_time - curr.start_time) - curr.idle, 0).str,
+      qfs_time_period(delta - idle_delta, pf_plus).str,
+      qfs_time_period(curr.idle, 0).str,
+      qfs_time_period(idle_delta, pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) signals",
+      qfs_dec_value(curr.signals, pf_scale).str,
+      qfs_dec_value(curr.signals - prev.signals, pf_scale | pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) foreg.",
+      qfs_dec_value(curr.foreg, pf_scale).str,
+      qfs_dec_value(curr.foreg - prev.foreg, pf_scale | pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) dispatch\n",
+      qfs_dec_value(curr.dispatch, pf_scale).str,
+      qfs_dec_value(curr.dispatch - prev.dispatch, pf_scale | pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) i/o act.",
+      qfs_dec_value(curr.io_acts, pf_scale).str,
+      qfs_dec_value(curr.io_acts - prev.io_acts, pf_scale | pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) timers",
+      qfs_dec_value(curr.timers, pf_scale).str,
+      qfs_dec_value(curr.timers - prev.timers, pf_scale | pf_plus).str) ;
+
+  vty_out(vty, "  %s(%s) backg.\n",
+      qfs_dec_value(curr.backg, pf_scale).str,
+      qfs_dec_value(curr.backg - prev.backg, pf_scale | pf_plus).str) ;
+} ;
+
+
diff --git a/bgpd/bgp_peer.c b/bgpd/bgp_peer.c
index 196f0d54..f07a30f5 100644
--- a/bgpd/bgp_peer.c
+++ b/bgpd/bgp_peer.c
@@ -35,7 +35,6 @@
 #include "bgpd/bgp_packet.h"
 #include "bgpd/bgp_network.h"
 #include "bgpd/bgp_route.h"
-#include "bgpd/bgp_dump.h"
 #include "bgpd/bgp_open.h"
 #include "bgpd/bgp_advertise.h"
 
@@ -1595,7 +1594,8 @@ bgp_peer_clearing_completed(struct peer *peer)
 /*------------------------------------------------------------------------------
  * Set new peer state.
  *
- * If state changes, do dump new state and log state change if required.
+ * If state changes log state change if required and deal with dropping back to
+ * pIdle.
  *
  * In any case, set timers for the new state -- so if state hasn't changed,
  * will restart those timers.
@@ -1605,8 +1605,6 @@ bgp_peer_change_status (bgp_peer peer, bgp_peer_state_t new_state)
 {
   if (peer->state != new_state)
     {
-      bgp_dump_state (peer, peer->state, new_state);
-
       /* Preserve old status and change into new status. */
       peer->ostate = peer->state ;
       peer->state  = new_state ;
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;