Fix retry for connect() connections + debug for mqueue.

  modified:   bgpd/bgp_connection.c
  modified:   bgpd/bgp_connection.h
  modified:   bgpd/bgp_engine.h
  modified:   bgpd/bgp_fsm.c
  modified:   bgpd/bgp_session.c
  modified:   bgpd/bgpd.c

6 files changed, 136 insertions, 57 deletions

diff --git a/bgpd/bgp_connection.c b/bgpd/bgp_connection.c
index d2b6a017..8320ae49 100644
--- a/bgpd/bgp_connection.c
+++ b/bgpd/bgp_connection.c
@@ -536,7 +536,7 @@ bgp_connection_add_pending(bgp_connection connection, mqueue_block mqb,
  *
  *   * if secondary connection, turn off accept()
  *   * sets the qfile and fd ready for use
- *   * clears except, notification and err
+ *   * clears err -- must be OK so far
  *   * discards any open_state and notification
  *   * copies hold_timer_interval and keep_alive_timer_interval from session
  *
@@ -566,9 +566,6 @@ bgp_connection_open(bgp_connection connection, int fd)
   /* Set the file going                                                 */
   qps_add_file(bgp_nexus->selection, &connection->qf, fd, connection) ;
 
-  /* Clear sundry state is clear                                        */
-  connection->except  = bgp_session_null_event ;
-  bgp_notify_unset(&connection->notification) ;
   connection->err     = 0 ;                     /* so far, so good      */
 
   bgp_open_state_unset(&connection->open_recv) ;
@@ -796,7 +793,7 @@ bgp_connection_write(bgp_connection connection, struct stream* s)
     } ;
 
   /* Write nothing if cannot write everything                   */
-  if (!bgp_write_buffer_can(wb, stream_pending(s)))
+  if (bgp_write_buffer_cannot(wb, stream_pending(s)))
     return 0 ;
 
   /* Transfer the obuf contents to the write buffer.            */
@@ -881,7 +878,7 @@ bgp_connection_write_action(qps_file qf, void* file_info)
   int ret ;
 
   /* Try to empty the write buffer.                                     */
-  have = wb->p_out - wb->p_in ;
+  have = bgp_write_buffer_pending(wb) ;
   while (have != 0)
     {
       ret = write(qps_file_fd(qf), wb->p_out, have) ;
diff --git a/bgpd/bgp_connection.h b/bgpd/bgp_connection.h
index 4a0e7f5d..054cd953 100644
--- a/bgpd/bgp_connection.h
+++ b/bgpd/bgp_connection.h
@@ -251,32 +251,41 @@ bgp_connection_add_pending(bgp_connection connection, mqueue_block mqb,
                                                    bgp_connection* is_pending) ;
 
 /*------------------------------------------------------------------------------
- * See if have enough room for what want to write PLUS 1.
+ * See if do NOT have enough room for what want to write PLUS 1.
  *
- * NB: caller must ensure buffer has been allocated, which will be true if
- *     has found that the buffer is not empty !
+ * NB: before using the buffer the caller MUST ensure it has been allocated.
+ *
+ *     Unallocated buffer is made to appear to have room for one maximum
+ *     size BGP message.
  */
 Inline int
-bgp_write_buffer_can(bgp_wbuffer wb, size_t want)
+bgp_write_buffer_cannot(bgp_wbuffer wb, size_t want)
 {
   return ((size_t)(wb->limit - wb->p_in) <= want) ;
 } ;
 
 /*------------------------------------------------------------------------------
- * Full if not enough room for a maximum size BGP message + 1
+ * Full if NOT enough room for a maximum size BGP message + 1
  *
- * NB: this will be true even if the buffer has not been allocated (!).
+ * NB: this will be FALSE if the buffer has not been allocated -- because can
+ *     allocate a buffer and proceed if required.
  */
 enum { bgp_write_buffer_full_threshold = BGP_MSG_MAX_L + 1 } ;
 
 Inline int
 bgp_write_buffer_full(bgp_wbuffer wb)
 {
-  return bgp_write_buffer_can(wb, BGP_MSG_MAX_L) ;
+  return bgp_write_buffer_cannot(wb, BGP_MSG_MAX_L) ;
 } ;
 
 /*------------------------------------------------------------------------------
- * Empty if in and out pointers are equal (but may need to be reset !)
+ * Empty if in and out pointers are equal.
+ *
+ * NB: buffer is empty if it has not yet been allocated.
+ *
+ *     NOT empty => allocated.
+ *
+ * NB: empty does NOT imply that both pointers are at the start of the buffer.
  */
 Inline int
 bgp_write_buffer_empty(bgp_wbuffer wb)
@@ -285,6 +294,22 @@ bgp_write_buffer_empty(bgp_wbuffer wb)
 } ;
 
 /*------------------------------------------------------------------------------
+ * Return how much the write buffer still has to write.
+ *
+ * NB: if returns 0, may not yet have been allocated.
+ *
+ *     > 0 => allocated.
+ *
+ * NB: 0 does NOT imply that both pointers are at the start of the buffer.
+ */
+Inline int
+bgp_write_buffer_pending(bgp_wbuffer wb)
+{
+  dassert(wb->p_out <= wb->p_in) ;
+  return (wb->p_in - wb->p_out) ;
+} ;
+
+/*------------------------------------------------------------------------------
  * As above, for connection
  */
 Inline int
diff --git a/bgpd/bgp_engine.h b/bgpd/bgp_engine.h
index 8d344b22..fdbcef70 100644
--- a/bgpd/bgp_engine.h
+++ b/bgpd/bgp_engine.h
@@ -38,40 +38,65 @@
  *
  */
 
-static unsigned bgp_engine_queue_thresh_up   = 0 ;
-static unsigned bgp_engine_queue_thresh_down = 0 ;
-static unsigned peering_engine_queue_thresh_up   = 0 ;
-static unsigned peering_engine_queue_thresh_down = 0 ;
+struct queue_stats
+{
+  unsigned        count ;
+  unsigned long   total ;
+  unsigned        max ;
+  unsigned        recent ;
+
+  unsigned        xon ;
+  unsigned        event ;
+  unsigned        update ;
+} ;
+
+static struct queue_stats bgp_engine_queue_stats ;
+static struct queue_stats peering_engine_queue_stats ;
 
 Inline void
-bgp_queue_logging(const char* name, unsigned count, unsigned* up,
-                                                    unsigned* down)
+bgp_queue_logging(const char* name, mqueue_queue mq, struct queue_stats* stats)
 {
-  if (count > *up)
+  double average ;
+  unsigned my_count ;
+  mqueue_block mqb ;
+
+  ++stats->count ;
+
+  if (mq->count > stats->max)
+    stats->max    = mq->count ;
+  if (mq->count > stats->recent)
+    stats->recent = mq->count ;
+
+  stats->total += mq->count ;
+
+  if (stats->count < 1000)
+    return ;
+
+  my_count = 0 ;
+
+  mqb = mq->head ;
+  while (mqb != NULL)
     {
-      if (*up != 0)
-        {
-          zlog_debug("%s queue up to %u entries", name, count) ;
-
-          *up  *= 2 ;
-          *down = *up / 4 ;
-        }
-      else
-        *up   = 32 ;
+      ++my_count ;
+      mqb = mqb->next ;
     } ;
 
-  if (count < *down)
-    {
-      zlog_debug("%s queue down to %u entries", name, count) ;
-
-      if (*up == 32)
-        *down = 0 ;
-      else
-        {
-          *up  /= 2 ;
-          *down = *up / 2 ;
-        } ;
-    } ;  ;
+  assert(my_count == mq->count) ;
+
+  average = stats->total ;
+  average /= stats->count ;
+
+  zlog_debug("%s queue: max=%u  recent: max=%u av=%3.1f (%u) [x=%u e=%u u=%u]",
+                    name, stats->max, stats->recent, average, stats->count,
+                                      stats->xon, stats->event, stats->update) ;
+
+  stats->recent = 0 ;
+  stats->count  = 0 ;
+  stats->total  = 0 ;
+
+  stats->event  = 0 ;
+  stats->update = 0 ;
+  stats->xon    = 0 ;
 } ;
 
 /* Send given message to the BGP Engine -- ordinary
@@ -80,8 +105,7 @@ Inline void
 bgp_to_bgp_engine(mqueue_block mqb)
 {
   mqueue_enqueue(bgp_nexus->queue, mqb, 0) ;
-  bgp_queue_logging("BGP Engine", bgp_nexus->queue->count,
-                   &bgp_engine_queue_thresh_up, &bgp_engine_queue_thresh_down) ;
+  bgp_queue_logging("BGP Engine", bgp_nexus->queue, &bgp_engine_queue_stats) ;
 } ;
 
 /* Send given message to the BGP Engine -- priority
@@ -90,8 +114,7 @@ Inline void
 bgp_to_bgp_engine_priority(mqueue_block mqb)
 {
   mqueue_enqueue(bgp_nexus->queue, mqb, 1) ;
-  bgp_queue_logging("BGP Engine", bgp_nexus->queue->count,
-                   &bgp_engine_queue_thresh_up, &bgp_engine_queue_thresh_down) ;
+  bgp_queue_logging("BGP Engine", bgp_nexus->queue, &bgp_engine_queue_stats) ;
 } ;
 
 /*==============================================================================
@@ -104,8 +127,8 @@ Inline void
 bgp_to_peering_engine(mqueue_block mqb)
 {
   mqueue_enqueue(routing_nexus->queue, mqb, 0) ;
-  bgp_queue_logging("Peering Engine", routing_nexus->queue->count,
-           &peering_engine_queue_thresh_up, &peering_engine_queue_thresh_down) ;
+  bgp_queue_logging("Peering Engine", routing_nexus->queue,
+                                                  &peering_engine_queue_stats) ;
 } ;
 
 /* Send given message to the Peering Engine -- priority
@@ -114,8 +137,8 @@ Inline void
 bgp_to_peering_engine_priority(mqueue_block mqb)
 {
   mqueue_enqueue(routing_nexus->queue, mqb, 1) ;
-  bgp_queue_logging("Peering Engine", routing_nexus->queue->count,
-           &peering_engine_queue_thresh_up, &peering_engine_queue_thresh_down) ;
+  bgp_queue_logging("Peering Engine", routing_nexus->queue,
+                                                  &peering_engine_queue_stats) ;
 } ;
 
 #endif /* QUAGGA_BGP_ENGINE_H */
diff --git a/bgpd/bgp_fsm.c b/bgpd/bgp_fsm.c
index 46bd3569..8a8be52d 100644
--- a/bgpd/bgp_fsm.c
+++ b/bgpd/bgp_fsm.c
@@ -336,6 +336,8 @@
  *      write buffer.  For sOpenSent and sOpenConfirm expect it to go directly
  *      to the TCP buffer.
  *
+ *   -- stop the KeepaliveTimer
+ *
  *   -- set HoldTimer to a waiting to clear buffer time -- say 20 secs.
  *
  *      Don't expect to need to wait at all in sOpenSent/sOpenConfirm states.
@@ -594,6 +596,8 @@ bgp_fsm_io_fatal_error(bgp_connection connection, int err)
   plog_err (connection->log, "%s [Error] bgp IO error: %s",
             connection->host, safe_strerror(err)) ;
 
+  assert(err != EFAULT) ;
+
   bgp_fsm_throw(connection, bgp_session_eTCP_error, NULL, err,
                                                      bgp_fsm_eTCP_fatal_error) ;
 } ;
@@ -1971,9 +1975,6 @@ static bgp_fsm_action(bgp_fsm_exit)
  * Uses the posted information and the expected next_state to deal with some
  * exception.  Proceeds:
  *
- *   0) stop any timers -- so if held in sOpenSent/sOpenConfirm by notification
- *      process, won't (eg) have extraneous keepalive going off.
- *
  *  1a) if have notification & not eNOM_recv & is in a suitable state
  *
  *      Suitable states are sOpenSent/sOpenConfirm/sEstablished.
@@ -2030,10 +2031,6 @@ bgp_fsm_catch(bgp_connection connection, bgp_fsm_state_t next_state)
 
   assert(connection->except != bgp_session_null_event) ;
 
-  /* It's bad news, so stop doing whatever was doing.                   */
-  qtimer_unset(&connection->hold_timer) ;
-  qtimer_unset(&connection->keepalive_timer) ;
-
   /* Have a notification to send iff not just received one, and is in a
    * suitable state to send one at all.
    */
@@ -2068,6 +2065,8 @@ bgp_fsm_catch(bgp_connection connection, bgp_fsm_state_t next_state)
       /* Close for reading and flush write buffers.                         */
       bgp_connection_part_close(connection) ;
 
+      qtimer_unset(&connection->keepalive_timer) ;
+
       /* Write the message
        *
        * If the write fails it raises a suitable event, which will now be
diff --git a/bgpd/bgp_session.c b/bgpd/bgp_session.c
index cbbc181e..8836b1b3 100644
--- a/bgpd/bgp_session.c
+++ b/bgpd/bgp_session.c
@@ -320,6 +320,9 @@ bgp_session_enable(bgp_peer peer)
   confirm(sizeof(struct bgp_session_enable_args) == 0) ;
 
   session->state = bgp_session_sEnabled;
+
+  ++bgp_engine_queue_stats.event ;
+
   bgp_to_bgp_engine(mqb) ;
 } ;
 
@@ -418,6 +421,22 @@ bgp_session_disable(bgp_peer peer, bgp_notify notification)
 
   args = mqb_get_args(mqb) ;
   args->notification = notification ;
+  {
+    int c, s ;
+    if (notification != NULL)
+      {
+        c = notification->code ;
+        s = notification->subcode ;
+      }
+    else
+      {
+        c = 0 ;
+        s = 0 ;
+      } ;
+    fprintf(stderr, " session disable %d/%d", c, s) ;
+  } ;
+
+  ++bgp_engine_queue_stats.event ;
 
   bgp_to_bgp_engine_priority(mqb) ;
 } ;
@@ -475,6 +494,8 @@ bgp_session_event(bgp_session session, bgp_session_event_t  event,
   args->ordinal      = ordinal ;
   args->stopped      = stopped,
 
+  ++peering_engine_queue_stats.event ;
+
   bgp_to_peering_engine(mqb) ;
 }
 
@@ -502,6 +523,8 @@ bgp_session_update_send(bgp_session session, struct stream* upd)
   session->flow_control++;      /* count them in ... */
   BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
 
+  ++bgp_engine_queue_stats.update ;
+
   bgp_to_bgp_engine(mqb) ;
 } ;
 
@@ -626,6 +649,8 @@ bgp_session_route_refresh_send(bgp_session session, bgp_route_refresh rr)
   args->rr         = rr ;
   args->is_pending = NULL ;
 
+  ++bgp_engine_queue_stats.event ;
+
   bgp_to_bgp_engine(mqb) ;
 } ;
 
@@ -689,6 +714,8 @@ bgp_session_end_of_rib_send(bgp_session session, qAFI_t afi, qSAFI_t safi)
   args->safi       = get_iSAFI(qafx) ;
   args->is_pending = NULL ;
 
+  ++bgp_engine_queue_stats.xon ;
+
   bgp_to_bgp_engine(mqb) ;
 } ;
 
@@ -753,6 +780,8 @@ bgp_session_update_recv(bgp_session session, struct stream* buf, bgp_size_t size
   args->buf = stream_dup(buf) ;
   args->size = size;
 
+  ++peering_engine_queue_stats.update ;
+
   bgp_to_peering_engine(mqb) ;
 }
 
@@ -831,6 +860,8 @@ bgp_session_XON(bgp_session session)
 
   confirm(sizeof(struct bgp_session_XON_args) == 0) ;
 
+  ++peering_engine_queue_stats.xon ;
+
   bgp_to_peering_engine(mqb) ;
 }
 
@@ -863,6 +894,8 @@ bgp_session_set_ttl(bgp_session session, int ttl)
   args = mqb_get_args(mqb) ;
   args->ttl = ttl ;
 
+  ++bgp_engine_queue_stats.event ;
+
   bgp_to_bgp_engine(mqb) ;
 }
 
diff --git a/bgpd/bgpd.c b/bgpd/bgpd.c
index 238bd01c..154cb28d 100644
--- a/bgpd/bgpd.c
+++ b/bgpd/bgpd.c
@@ -4712,6 +4712,7 @@ bgp_terminate (int terminating, int retain_mode)
   for (ALL_LIST_ELEMENTS (bm->bgp, mnode, mnnode, bgp))
     for (ALL_LIST_ELEMENTS (bgp->peer, node, nnode, peer))
       {
+fprintf(stderr, ">>> %s:", peer->host) ;
         if (retain_mode)
           bgp_peer_disable(peer, NULL);
         else if (terminating)
@@ -4724,6 +4725,7 @@ bgp_terminate (int terminating, int retain_mode)
         else
           bgp_notify_send(peer, BGP_NOTIFY_CEASE,
               BGP_NOTIFY_CEASE_ADMIN_RESET);
+fprintf(stderr, "<<<\n") ;
       }
 
   if (!retain_mode)