Re-implement xon.

4 files changed, 19 insertions, 38 deletions

diff --git a/bgpd/bgp_packet.c b/bgpd/bgp_packet.c
index 9d587428..d565c265 100644
--- a/bgpd/bgp_packet.c
+++ b/bgpd/bgp_packet.c
@@ -629,10 +629,7 @@ bgp_write (bgp_peer peer)
   struct stream *s;
   int free_s ;
 
-  if (bgp_session_is_XOFF(peer))
-    return 0 ;
-
-  do
+  while (bgp_session_is_XON(peer))
     {
       free_s = 0 ;
 
@@ -681,7 +678,7 @@ bgp_write (bgp_peer peer)
       if (free_s)
         bgp_packet_delete (peer);
 
-    } while (bgp_session_is_XON(peer)) ;
+    }
 
   return 0;
 }
diff --git a/bgpd/bgp_peer.c b/bgpd/bgp_peer.c
index 0dac1433..e6be06c7 100644
--- a/bgpd/bgp_peer.c
+++ b/bgpd/bgp_peer.c
@@ -182,6 +182,7 @@ bgp_session_has_established(bgp_peer peer)
   assert(session->state == bgp_session_sEnabled) ;
 
   session->state = bgp_session_sEstablished ;
+  session->flow_control = BGP_XON_REFRESH; /* updates can be sent */
   peer_change_status (peer, bgp_peer_sEstablished);
 
   /* update peer state from received open */
diff --git a/bgpd/bgp_session.c b/bgpd/bgp_session.c
index 8836b1b3..a2b49da5 100644
--- a/bgpd/bgp_session.c
+++ b/bgpd/bgp_session.c
@@ -518,10 +518,8 @@ bgp_session_update_send(bgp_session session, struct stream* upd)
   args = mqb_get_args(mqb) ;
   args->buf        = stream_dup(upd) ;
   args->is_pending = NULL ;
-
-  BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
-  session->flow_control++;      /* count them in ... */
-  BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
+  args->xon_kick = (session->flow_control == BGP_XON_KICK);
+  session->flow_control--;
 
   ++bgp_engine_queue_stats.update ;
 
@@ -583,13 +581,8 @@ bgp_session_do_update_send(mqueue_block mqb, mqb_flag_t flag)
             }
           else if (ret > 0)
             {
-              /* Successfully wrote the message.                        */
-              int xon ;
-              BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
-              xon = --session->flow_control ;   /* ... count them out */
-              BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
-
-              if (xon == 0)
+              /* Successfully wrote the message.  XON if requested      */
+              if (args->xon_kick)
                 bgp_session_XON(session);
             } ;
         } ;
@@ -600,22 +593,6 @@ bgp_session_do_update_send(mqueue_block mqb, mqb_flag_t flag)
 } ;
 
 /*------------------------------------------------------------------------------
- * Peering Engine: are we in XOFF state ?
- */
-extern int
-bgp_session_is_XOFF(bgp_peer peer)
-{
-  int result = 0;
-  bgp_session session = peer->session;
-
-  BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
-  result = session->flow_control > BGP_XOFF_THRESHOLD ;
-  BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
-
-  return result;
-} ;
-
-/*------------------------------------------------------------------------------
  * Peering Engine: are we in XON state ?
  */
 extern int
@@ -624,9 +601,7 @@ bgp_session_is_XON(bgp_peer peer)
   int result = 0;
   bgp_session session = peer->session;
 
-  BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
-  result = session->flow_control < BGP_XON_THRESHOLD ;
-  BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
+  result = (session->flow_control > 0);
 
   return result;
 } ;
@@ -779,6 +754,7 @@ bgp_session_update_recv(bgp_session session, struct stream* buf, bgp_size_t size
   args = mqb_get_args(mqb) ;
   args->buf = stream_dup(buf) ;
   args->size = size;
+  args->xon_kick = 0;
 
   ++peering_engine_queue_stats.update ;
 
@@ -874,7 +850,12 @@ bgp_session_do_XON(mqueue_block mqb, mqb_flag_t flag)
   bgp_session session = mqb_get_arg0(mqb) ;
 
   if ((flag == mqb_action) && (session->state == bgp_session_sEstablished))
-    bgp_write (session->peer) ;
+    {
+      int xoff = (session->flow_control <= 0);
+      session->flow_control = BGP_XON_REFRESH;
+      if (xoff)
+        bgp_write (session->peer) ;
+    }
 
   mqb_free(mqb) ;
 }
diff --git a/bgpd/bgp_session.h b/bgpd/bgp_session.h
index af376f9a..5af81688 100644
--- a/bgpd/bgp_session.h
+++ b/bgpd/bgp_session.h
@@ -241,6 +241,7 @@ struct bgp_session_update_args          /* to and from BGP Engine       */
 {
   struct stream*  buf ;
   bgp_size_t size ;
+  int xon_kick;                         /* send XON when processed this */
 
   bgp_connection  is_pending ;          /* used inside the BGP Engine   */
                                         /* set NULL on message creation */
@@ -281,8 +282,9 @@ struct bgp_session_XON_args             /* to Routeing Engine           */
                                         /* no further arguments         */
 } ;
 MQB_ARGS_SIZE_OK(bgp_session_XON_args) ;
-enum { BGP_XON_THRESHOLD   = 12,
-       BGP_XOFF_THRESHOLD  =  4 } ;
+enum { BGP_XON_REFRESH     = 12,
+       BGP_XON_KICK        =  4,
+} ;
 
 struct bgp_session_ttl_args             /* to bgp Engine                */
 {