Wired up receiving updates, with update packet sent from BGP to Routing

engine.  Wired up Routing engine side of sending update packets to BGP
engine with flow control.

Eliminated a lot of compilation errors and warnings although a few
remain.

Have skipped out now unused code, particularly in bgp_packet.c that will
be deleted eventually.

11 files changed, 224 insertions, 207 deletions

diff --git a/bgpd/bgp_debug.c b/bgpd/bgp_debug.c
index 9267eebd..d86ddfc7 100644
--- a/bgpd/bgp_debug.c
+++ b/bgpd/bgp_debug.c
@@ -280,14 +280,14 @@ bgp_notify_print(struct peer *peer, struct bgp_notify *bgp_notify,
               peer->host, bgp_notify->code, bgp_notify->subcode,
                LOOKUP (bgp_notify_msg, bgp_notify->code),
               subcode_str, bgp_notify->length,
-              bgp_notify->data ? bgp_notify->data : "");
+              bgp_notify->size ? bgp_notify->data : "");
   else if (BGP_DEBUG (normal, NORMAL))
     plog_debug (peer->log, "%s %s NOTIFICATION %d/%d (%s%s) %d bytes %s",
 	       peer ? peer->host : "",
 	       direct, bgp_notify->code, bgp_notify->subcode,
 	       LOOKUP (bgp_notify_msg, bgp_notify->code),
 	       subcode_str, bgp_notify->length,
-	       bgp_notify->data ? bgp_notify->data : "");
+	       bgp_notify->size ? bgp_notify->data : "");
 }
 
 /* Debug option setting interface. */
diff --git a/bgpd/bgp_fsm.c b/bgpd/bgp_fsm.c
index 8db13c3e..475d2cbf 100644
--- a/bgpd/bgp_fsm.c
+++ b/bgpd/bgp_fsm.c
@@ -2586,6 +2586,8 @@ bgp_keepalive_timer_action(qtimer qtr, void* timer_info, qtime_mono_t when)
 
 /*============================================================================*/
 /* BGP Peer Down Cause */
+/* TODO: this is also defined in bgp_peer.c */
+#if 0
 const char *peer_down_str[] =
 {
   "",
@@ -2612,3 +2614,4 @@ const char *peer_down_str[] =
   "Multihop config change",
   "NSF peer closed the session"
 };
+#endif
diff --git a/bgpd/bgp_msg_read.c b/bgpd/bgp_msg_read.c
index 4c94c999..4e581f4f 100644
--- a/bgpd/bgp_msg_read.c
+++ b/bgpd/bgp_msg_read.c
@@ -1048,7 +1048,7 @@ bgp_msg_capability_orf_not_support (char *host, afi_t afi, safi_t safi,
 static int
 bgp_msg_update_receive (bgp_connection connection, bgp_size_t size)
 {
-  bgp_session_update_recv(connection->session, stream_dup(connection->ibuf));
+  bgp_session_update_recv(connection->session, connection->ibuf, size);
   bgp_fsm_event(connection, bgp_fsm_Receive_UPDATE_message);
   return 0;
 }
diff --git a/bgpd/bgp_packet.c b/bgpd/bgp_packet.c
index 3018daa9..55e7975c 100644
--- a/bgpd/bgp_packet.c
+++ b/bgpd/bgp_packet.c
@@ -102,6 +102,7 @@ bgp_packet_delete (struct peer *peer)
   stream_free (stream_fifo_pop (peer->obuf));
 }
 
+#if 0
 /* Check file descriptor whether connect is established. */
 static void
 bgp_connect_check (struct peer *peer)
@@ -139,6 +140,7 @@ bgp_connect_check (struct peer *peer)
       BGP_EVENT_ADD (peer, TCP_connection_open_failed);
     }
 }
+#endif
 
 /* Make BGP update packet.  */
 static struct stream *
@@ -225,7 +227,7 @@ bgp_update_packet (struct peer *peer, afi_t afi, safi_t safi)
       bgp_packet_set_size (s);
       packet = stream_dup (s);
       bgp_packet_add (peer, packet);
-      BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+      bgp_write(peer);
       stream_reset (s);
       return packet;
     }
@@ -423,8 +425,7 @@ bgp_default_update_send (struct peer *peer, struct attr *attr,
 
   /* Add packet to the peer. */
   bgp_packet_add (peer, packet);
-
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
 
 void
@@ -496,8 +497,7 @@ bgp_default_withdraw_send (struct peer *peer, afi_t afi, safi_t safi)
 
   /* Add packet to the peer. */
   bgp_packet_add (peer, packet);
-
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
 
 /* Get next packet to be written.  */
@@ -565,6 +565,7 @@ bgp_write_packet (struct peer *peer)
   return NULL;
 }
 
+#if 0
 /* Is there partially written packet or updates we can send right
    now.  */
 static int
@@ -590,70 +591,22 @@ bgp_write_proceed (struct peer *peer)
 
   return 0;
 }
+#endif
 
-/* Write packet to the peer. */
+/* Write packets to the peer. */
 int
-bgp_write (struct thread *thread)
+bgp_write (bgp_peer peer)
 {
-  struct peer *peer;
   u_char type;
   struct stream *s;
-  int num;
-  unsigned int count = 0;
-  int write_errno;
 
-  /* Yes first of all get peer pointer. */
-  peer = THREAD_ARG (thread);
-  peer->t_write = NULL;
-
-  /* For non-blocking IO check. */
-  if (peer->status == Connect)
+  while (bgp_session_is_XON(peer))
     {
-      bgp_connect_check (peer);
-      return 0;
-    }
-
-    /* Nonblocking write until TCP output buffer is full.  */
-  while (1)
-    {
-      int writenum;
-      int val;
-
       s = bgp_write_packet (peer);
       if (! s)
-	return 0;
-
-      /* XXX: FIXME, the socket should be NONBLOCK from the start
-       * status shouldnt need to be toggled on each write
-       */
-      val = fcntl (peer->fd, F_GETFL, 0);
-      fcntl (peer->fd, F_SETFL, val|O_NONBLOCK);
-
-      /* Number of bytes to be sent.  */
-      writenum = stream_get_endp (s) - stream_get_getp (s);
-
-      /* Call write() system call.  */
-      num = write (peer->fd, STREAM_PNT (s), writenum);
-      write_errno = errno;
-      fcntl (peer->fd, F_SETFL, val);
-      if (num <= 0)
-	{
-	  /* Partial write. */
-	  if (write_errno == EWOULDBLOCK || write_errno == EAGAIN)
-	      break;
-
-	  BGP_EVENT_ADD (peer, TCP_fatal_error);
-	  return 0;
-	}
-      if (num != writenum)
-	{
-	  stream_forward_getp (s, num);
-
-	  if (write_errno == EAGAIN)
-	    break;
+	break;
 
-	  continue;
-	}
+      bgp_session_update_send(peer->session, s);
 
       /* Retrieve BGP packet type. */
       stream_set_getp (s, BGP_MARKER_SIZE + 2);
@@ -676,8 +629,7 @@ bgp_write (struct thread *thread)
 	  if (peer->v_start >= (60 * 2))
 	    peer->v_start = (60 * 2);
 
-	  /* Flush any existing events */
-	  BGP_EVENT_ADD (peer, BGP_Stop);
+	  assert(0); /* shouldn't get notifies through here */
 	  return 0;
 	case BGP_MSG_KEEPALIVE:
 	  peer->keepalive_out++;
@@ -691,19 +643,14 @@ bgp_write (struct thread *thread)
 	  break;
 	}
 
-      /* OK we send packet so delete it. */
+      /* OK we sent packet so delete it. */
       bgp_packet_delete (peer);
-
-      if (++count >= BGP_WRITE_PACKET_MAX)
-	break;
     }
 
-  if (bgp_write_proceed (peer))
-    BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
-
   return 0;
 }
 
+#if 0
 /* This is only for sending NOTIFICATION message to neighbor. */
 static int
 bgp_write_notify (struct peer *peer)
@@ -746,7 +693,9 @@ bgp_write_notify (struct peer *peer)
 
   return 0;
 }
+#endif
 
+#if 0
 /* Make keepalive packet and send it to the peer. */
 void
 bgp_keepalive_send (struct peer *peer)
@@ -773,10 +722,11 @@ bgp_keepalive_send (struct peer *peer)
 
   /* Add packet to the peer. */
   bgp_packet_add (peer, s);
-
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
+#endif
 
+#if 0
 /* Make open packet and send it to the peer. */
 void
 bgp_open_send (struct peer *peer)
@@ -829,77 +779,25 @@ bgp_open_send (struct peer *peer)
 
   /* Add packet to the peer. */
   bgp_packet_add (peer, s);
-
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
+#endif
 
 /* Send BGP notify packet with data potion. */
 void
 bgp_notify_send_with_data (struct peer *peer, u_char code, u_char sub_code,
 			   u_char *data, size_t datalen)
 {
-  /* TODO: do we still need this? */
-#if 0
-  struct stream *s;
-  int length;
-
-  /* Allocate new stream. */
-  s = stream_new (BGP_MAX_PACKET_SIZE);
-
-  /* Make nitify packet. */
-  bgp_packet_set_marker (s, BGP_MSG_NOTIFY);
-
-  /* Set notify packet values. */
-  stream_putc (s, code);        /* BGP notify code */
-  stream_putc (s, sub_code);	/* BGP notify sub_code */
-
-  /* If notify data is present. */
-  if (data)
-    stream_write (s, data, datalen);
-
-  /* Set BGP packet length. */
-  length = bgp_packet_set_size (s);
-
-  /* Add packet to the peer. */
-  stream_fifo_clean (peer->obuf);
-  bgp_packet_add (peer, s);
+  bgp_notify notification;
+  notification = bgp_notify_new(code, sub_code, datalen);
+  notification = bgp_notify_append_data(notification, data, datalen);
 
   /* For debug */
-  {
-    struct bgp_notify bgp_notify;
-    int first = 0;
-    int i;
-    char c[4];
-
-    bgp_notify.code = code;
-    bgp_notify.subcode = sub_code;
-    bgp_notify.data = NULL;
-    bgp_notify.length = length - BGP_MSG_NOTIFY_MIN_SIZE;
-
-    if (bgp_notify.length)
-      {
-	bgp_notify.data = XMALLOC (MTYPE_TMP, bgp_notify.length * 3);
-	for (i = 0; i < bgp_notify.length; i++)
-	  if (first)
-	    {
-	      sprintf (c, " %02x", data[i]);
-	      strcat (bgp_notify.data, c);
-	    }
-	  else
-	    {
-	      first = 1;
-	      sprintf (c, "%02x", data[i]);
-	      strcpy (bgp_notify.data, c);
-	    }
-      }
-    bgp_notify_print (peer, &bgp_notify, "sending");
-    if (bgp_notify.data)
-      XFREE (MTYPE_TMP, bgp_notify.data);
-  }
+  bgp_notify_print (peer, notification, "sending");
 
   if (BGP_DEBUG (normal, NORMAL))
     zlog_debug ("%s send message type %d, length (incl. header) %d",
-	       peer->host, BGP_MSG_NOTIFY, length);
+	       peer->host, BGP_MSG_NOTIFY, notification->length);
 
   /* peer reset cause */
   if (sub_code != BGP_NOTIFY_CEASE_CONFIG_CHANGE)
@@ -912,11 +810,7 @@ bgp_notify_send_with_data (struct peer *peer, u_char code, u_char sub_code,
       peer->last_reset = PEER_DOWN_NOTIFY_SEND;
     }
 
-  /* Call imidiately. */
-  BGP_WRITE_OFF (peer->t_write);
-
-  bgp_write_notify (peer);
-#endif
+  bgp_peer_reenable(peer, notification);
 }
 
 /* Send BGP notify packet. */
@@ -1019,8 +913,7 @@ bgp_route_refresh_send (struct peer *peer, afi_t afi, safi_t safi,
 
   /* Add packet to the peer. */
   bgp_packet_add (peer, packet);
-
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
 
 /* Send capability message to the peer. */
@@ -1071,9 +964,10 @@ bgp_capability_send (struct peer *peer, afi_t afi, safi_t safi,
     zlog_debug ("%s send message type %d, length (incl. header) %d",
 	       peer->host, BGP_MSG_CAPABILITY, length);
 
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 }
 
+#if 0
 /* RFC1771 6.8 Connection collision detection. */
 static int
 bgp_collision_detect (struct peer *new, struct in_addr remote_id)
@@ -1137,7 +1031,9 @@ bgp_collision_detect (struct peer *new, struct in_addr remote_id)
     }
   return 0;
 }
+#endif
 
+#if 0
 static int
 bgp_open_receive (struct peer *peer, bgp_size_t size)
 {
@@ -1488,9 +1384,10 @@ bgp_open_receive (struct peer *peer, bgp_size_t size)
 
   return 0;
 }
+#endif
 
 /* Parse BGP Update packet and make attribute object. */
-static int
+int
 bgp_update_receive (struct peer *peer, bgp_size_t size)
 {
   int ret;
@@ -1814,17 +1711,19 @@ bgp_update_receive (struct peer *peer, bgp_size_t size)
   peer->update_time = time (NULL);
 
   /* Generate BGP event. */
+  /* TODO: is this needed? */
+#if 0
   BGP_EVENT_ADD (peer, Receive_UPDATE_message);
+#endif
 
   return 0;
 }
 
+#if 0
 /* Notify message treatment function. */
 static void
 bgp_notify_receive (struct peer *peer, bgp_size_t size)
 {
-  /* TODO: do we still need this? */
-#if 0
   struct bgp_notify bgp_notify;
 
   if (peer->notify.data)
@@ -1899,9 +1798,10 @@ bgp_notify_receive (struct peer *peer, bgp_size_t size)
     UNSET_FLAG (peer->sflags, PEER_STATUS_CAPABILITY_OPEN);
 
   BGP_EVENT_ADD (peer, Receive_NOTIFICATION_message);
-#endif
 }
+#endif
 
+#if 0
 /* Keepalive treatment function -- get keepalive send keepalive */
 static void
 bgp_keepalive_receive (struct peer *peer, bgp_size_t size)
@@ -1911,7 +1811,9 @@ bgp_keepalive_receive (struct peer *peer, bgp_size_t size)
 
   BGP_EVENT_ADD (peer, Receive_KEEPALIVE_message);
 }
+#endif
 
+#if 0
 /* Route refresh message is received. */
 static void
 bgp_route_refresh_receive (struct peer *peer, bgp_size_t size)
@@ -2085,6 +1987,7 @@ bgp_route_refresh_receive (struct peer *peer, bgp_size_t size)
   /* Perform route refreshment to the peer */
   bgp_announce_route (peer, afi, safi);
 }
+#endif
 
 static int
 bgp_capability_msg_parse (struct peer *peer, u_char *pnt, bgp_size_t length)
@@ -2180,7 +2083,12 @@ bgp_capability_msg_parse (struct peer *peer, u_char *pnt, bgp_size_t length)
               if (peer_active_nego (peer))
                 bgp_clear_route (peer, afi, safi, BGP_CLEAR_ROUTE_NORMAL);
               else
+                {
+                  /* TODO: only used for unit tests.  Test will need fixing */
+#if 0
                 BGP_EVENT_ADD (peer, BGP_Stop);
+#endif
+                }
             }
         }
       else
@@ -2232,6 +2140,7 @@ bgp_capability_receive (struct peer *peer, bgp_size_t size)
   return bgp_capability_msg_parse (peer, pnt, size);
 }
 
+#if 0
 /* BGP read utility function. */
 static int
 bgp_read_packet (struct peer *peer)
@@ -2300,7 +2209,9 @@ bgp_read_packet (struct peer *peer)
 
   return 0;
 }
+#endif
 
+#if 0
 /* Marker check. */
 static int
 bgp_marker_all_one (struct stream *s, int length)
@@ -2313,7 +2224,9 @@ bgp_marker_all_one (struct stream *s, int length)
 
   return 1;
 }
+#endif
 
+#if 0
 /* Starting point of packet process function. */
 int
 bgp_read (struct thread *thread)
@@ -2477,3 +2390,4 @@ bgp_read (struct thread *thread)
     }
   return 0;
 }
+#endif
diff --git a/bgpd/bgp_packet.h b/bgpd/bgp_packet.h
index 8be95f40..c60bcf04 100644
--- a/bgpd/bgp_packet.h
+++ b/bgpd/bgp_packet.h
@@ -21,6 +21,8 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 #ifndef _QUAGGA_BGP_PACKET_H
 #define _QUAGGA_BGP_PACKET_H
 
+#include "bgpd/bgp_attr.h"
+
 #define BGP_NLRI_LENGTH       1U
 #define BGP_TOTAL_ATTR_LEN    2U
 #define BGP_UNFEASIBLE_LEN    2U
@@ -41,7 +43,7 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 
 /* Packet send and receive function prototypes. */
 extern int bgp_read (struct thread *);
-extern int bgp_write (struct thread *);
+extern int bgp_write (bgp_peer peer);
 
 extern void bgp_keepalive_send (struct peer *);
 extern void bgp_open_send (struct peer *);
@@ -56,4 +58,6 @@ extern void bgp_default_withdraw_send (struct peer *, afi_t, safi_t);
 
 extern int bgp_capability_receive (struct peer *, bgp_size_t);
 
+extern int bgp_update_receive (struct peer *peer, bgp_size_t size);
+
 #endif /* _QUAGGA_BGP_PACKET_H */
diff --git a/bgpd/bgp_peer.c b/bgpd/bgp_peer.c
index f2728438..fa5a0d1c 100644
--- a/bgpd/bgp_peer.c
+++ b/bgpd/bgp_peer.c
@@ -240,9 +240,6 @@ bgp_session_has_established(bgp_peer peer)
                                     REFRESH_IMMEDIATE, 0);
         }
 
-  if (peer->v_keepalive)
-    bgp_keepalive_send (peer);
-
   /* First update is deferred until ORF or ROUTE-REFRESH is received */
   for (afi = AFI_IP ; afi < AFI_MAX ; afi++)
     for (safi = SAFI_UNICAST ; safi < SAFI_MAX ; safi++)
@@ -268,7 +265,7 @@ bgp_session_has_stopped(bgp_peer peer)
   if (session->defer_enable)
     {
       session->defer_enable = 0;
-      bpg_session_enable(peer);
+      bgp_session_enable(peer);
     }
 
   return 0;
@@ -404,6 +401,7 @@ bgp_peer_stop (struct peer *peer)
   return 0;
 }
 
+#if 0
 /* Stop all timers for the given peer
  */
 static void
@@ -416,13 +414,11 @@ bgp_peer_timers_stop(bgp_peer peer)
   BGP_TIMER_OFF (peer->t_gr_stale);
   BGP_TIMER_OFF (peer->t_pmax_restart);
 } ;
+#endif
 
-/* TODO: bgp_timer_set - kill ? */
 static void
 bgp_timer_set (struct peer *peer)
 {
-  int jitter = 0;
-
   switch (peer->status)
     {
     case Idle:
@@ -490,7 +486,7 @@ bgp_routeadv_timer (struct thread *thread)
 
   peer->synctime = time (NULL);
 
-  BGP_WRITE_ON (peer->t_write, bgp_write, peer->fd);
+  bgp_write(peer);
 
   BGP_TIMER_ON (peer->t_routeadv, bgp_routeadv_timer,
 		peer->v_routeadv);
@@ -585,7 +581,7 @@ bgp_graceful_stale_timer_expire (struct thread *thread)
   return 0;
 }
 
-
+#if 0
 /* BGP peer is stopped by the error. */
 static int
 bgp_stop_with_error (struct peer *peer)
@@ -601,6 +597,7 @@ bgp_stop_with_error (struct peer *peer)
 
   return 0;
 }
+#endif
 
 /* Allocate new peer object, implicitly locked.  */
 struct peer *
@@ -759,7 +756,7 @@ peer_delete (struct peer *peer)
    * executed after peer structure is deleted.
    */
   peer->last_reset = PEER_DOWN_NEIGHBOR_DELETE;
-  bgp_stop (peer);
+  bgp_peer_stop (peer);
   bgp_fsm_change_status (peer, Deleted);
 
   /* Password configuration */
@@ -767,9 +764,6 @@ peer_delete (struct peer *peer)
     {
       XFREE (MTYPE_PEER_PASSWORD, peer->password);
       peer->password = NULL;
-
-      if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
-        bgp_md5_set (peer);
     }
 
   bgp_timer_set (peer); /* stops all timers for Deleted */
@@ -903,9 +897,9 @@ peer_free (struct peer *peer)
   XFREE (MTYPE_BGP_PEER, peer);
 }
 
-/* Config change, disable then re-enable the peer */
+/* Disable then enable the peer.  Sends notification. */
 void
-bgp_peer_config_change(bgp_peer peer, bgp_notify notification)
+bgp_peer_reenable(bgp_peer peer, bgp_notify notification)
 {
   bgp_peer_disable(peer, notification);
   bgp_peer_enable(peer); /* may defer if still stopping */
diff --git a/bgpd/bgp_peer.h b/bgpd/bgp_peer.h
index 27cfae68..c3c8b88d 100644
--- a/bgpd/bgp_peer.h
+++ b/bgpd/bgp_peer.h
@@ -329,7 +329,7 @@ struct peer
   u_int32_t established;        /* Established */
   u_int32_t dropped;            /* Dropped */
 
-  /* Syncronization list and time.  */
+  /* Synchronization list and time.  */
   struct bgp_synchronize *sync[AFI_MAX][SAFI_MAX];
   time_t synctime;
 
@@ -472,7 +472,7 @@ extern void
 bgp_session_do_event(mqueue_block mqb, mqb_flag_t flag);
 
 void
-bgp_peer_config_change(bgp_peer peer, bgp_notify notification);
+bgp_peer_reenable(bgp_peer peer, bgp_notify notification);
 
 extern void
 bgp_peer_enable(bgp_peer peer);
diff --git a/bgpd/bgp_session.c b/bgpd/bgp_session.c
index fdba5fd7..c0f1e8a5 100644
--- a/bgpd/bgp_session.c
+++ b/bgpd/bgp_session.c
@@ -25,6 +25,7 @@
 #include "bgpd/bgp_peer_index.h"
 #include "bgpd/bgp_fsm.h"
 #include "bgpd/bgp_open_state.h"
+#include "bgpd/bgp_packet.h"
 
 #include "lib/memory.h"
 #include "lib/sockunion.h"
@@ -32,6 +33,14 @@
 /* prototypes */
 static int
 bgp_session_defer_if_stopping(bgp_session session);
+static void bgp_session_do_enable(mqueue_block mqb, mqb_flag_t flag) ;
+static void bgp_session_do_update_recv(mqueue_block mqb, mqb_flag_t flag);
+static void bgp_session_do_update_send(mqueue_block mqb, mqb_flag_t flag);
+static void bgp_session_do_disable(mqueue_block mqb, mqb_flag_t flag) ;
+static void bgp_session_XON(bgp_session session);
+static void bgp_session_do_XON(mqueue_block mqb, mqb_flag_t flag);
+
+
 
 /*==============================================================================
  * BGP Session.
@@ -181,8 +190,6 @@ bgp_session_free(bgp_session session)
  *
  *
  */
-static void
-bgp_session_do_enable(mqueue_block mqb, mqb_flag_t flag) ;
 
 void
 bgp_session_enable(bgp_peer peer)
@@ -298,8 +305,6 @@ bgp_session_do_enable(mqueue_block mqb, mqb_flag_t flag)
  * When session has been brought to a stop, BGP Engine will respond with an
  * eDisabled event (unless session stopped of its own accord first).
  */
-static void
-bgp_session_do_disable(mqueue_block mqb, mqb_flag_t flag) ;
 
 extern void
 bgp_session_disable(bgp_peer peer, bgp_notify notification)
@@ -356,7 +361,6 @@ bgp_session_do_disable(mqueue_block mqb, mqb_flag_t flag)
       bgp_session session = mqb_get_arg0(mqb) ;
       struct bgp_session_disable_args* args = mqb_get_args(mqb) ;
 
-      /* TODO: disable session */
       bgp_fsm_disable_session(session, args->notification) ;
     } ;
 
@@ -398,10 +402,7 @@ bgp_session_event(bgp_session session, bgp_session_event_t  event,
  * dealt with.
  */
 
-static void
-bgp_session_do_update_send(mqueue_block mqb, mqb_flag_t flag);
-
-extern void
+void
 bgp_session_update_send(bgp_session session, struct stream* upd)
 {
   struct bgp_session_update_args* args ;
@@ -410,12 +411,50 @@ bgp_session_update_send(bgp_session session, struct stream* upd)
   mqb = mqb_init_new(NULL, bgp_session_do_update_send, session) ;
 
   args = mqb_get_args(mqb) ;
+  args->buf   = stream_dup(upd) ;
 
-  args->buf   = upd ;
+  BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
+  session->flow_control++;      /* count them in ... */
+  BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
 
   bgp_to_bgp_engine(mqb) ;
 } ;
 
+static void
+bgp_session_do_update_send(mqueue_block mqb, mqb_flag_t flag)
+{
+  if (flag == mqb_action)
+    {
+      bgp_session session = mqb_get_arg0(mqb) ;
+      struct bgp_session_update_args* args = mqb_get_args(mqb) ;
+      int result;
+
+      /* TODO: process an update packet */
+
+      BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
+      result = session->flow_control--; /* ... count them out */
+      BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
+
+      if (result == 0)
+        bgp_session_XON(session);
+    }
+
+  mqb_free(mqb) ;
+}
+
+/* Are we in XON state ? */
+int
+bgp_session_is_XON(bgp_peer peer)
+{
+  int result = 0;
+  bgp_session session = peer->session;
+
+  BGP_SESSION_LOCK(session) ;   /*<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<*/
+  result = session->flow_control < (int)BGP_WRITE_PACKET_MAX;
+  BGP_SESSION_UNLOCK(session) ; /*>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>*/
+
+  return result;
+}
 /*==============================================================================
  * Forward incoming update -- BGP Engine -> Peering Engine
  *
@@ -425,11 +464,8 @@ bgp_session_update_send(bgp_session session, struct stream* upd)
  * dealt with.
  */
 
-static void
-bgp_session_do_update_recv(mqueue_block mqb, mqb_flag_t flag);
-
-extern void
-bgp_session_update_recv(bgp_session session, struct stream* upd)
+void
+bgp_session_update_recv(bgp_session session, struct stream* buf, bgp_size_t size)
 {
   struct bgp_session_update_args* args ;
   mqueue_block   mqb ;
@@ -437,13 +473,62 @@ bgp_session_update_recv(bgp_session session, struct stream* upd)
   mqb = mqb_init_new(NULL, bgp_session_do_update_recv, session) ;
 
   args = mqb_get_args(mqb) ;
+  args->buf = stream_dup(buf) ;
+  args->size = size;
+
+  bgp_to_peering_engine(mqb) ;
+}
+
+static void
+bgp_session_do_update_recv(mqueue_block mqb, mqb_flag_t flag)
+{
+
+  if (flag == mqb_action)
+    {
+      bgp_session session = mqb_get_arg0(mqb) ;
+      bgp_peer peer = session->peer;
+      struct bgp_session_update_args* args = mqb_get_args(mqb) ;
 
-  args->buf   = upd ;
+      stream_free(peer->ibuf);
+      peer->ibuf = args->buf;
+      bgp_update_receive (peer, args->size);
+    }
+
+  mqb_free(mqb) ;
+}
+
+/*==============================================================================
+ * XON -- BGP Engine -> Peering Engine
+ *
+ * Can be sent more packets now
+ */
+
+static void
+bgp_session_XON(bgp_session session)
+{
+  mqueue_block   mqb ;
+
+  mqb = mqb_init_new(NULL, bgp_session_do_XON, session) ;
+
+  confirm(sizeof(struct bgp_session_XON_args) == 0) ;
 
   bgp_to_peering_engine(mqb) ;
-} ;
+}
+
+static void
+bgp_session_do_XON(mqueue_block mqb, mqb_flag_t flag)
+{
+
+  if (flag == mqb_action)
+    {
+      bgp_session session = mqb_get_arg0(mqb) ;
+      bgp_peer peer = session->peer;
 
+      bgp_write (peer);
+    }
 
+  mqb_free(mqb) ;
+}
 
 /*==============================================================================
  * Session data access functions.
diff --git a/bgpd/bgp_session.h b/bgpd/bgp_session.h
index b54c1be4..1b6f4ad9 100644
--- a/bgpd/bgp_session.h
+++ b/bgpd/bgp_session.h
@@ -106,6 +106,13 @@ struct bgp_session
 
   flag_t                made ;          /* set when -> sEstablished       */
 
+  /* Flow control. Incremented when an update packet is sent
+   * from peering to BGP engine.  Decremented when packet processed
+   * by BGP engine.  On transition to 0 BGP engine should send an XON.
+   */
+
+  int flow_control;
+
   /* These belong to the Peering Engine, and may be set when a session
    * event message is received from the BGP Engine.
    */
@@ -201,6 +208,7 @@ MQB_ARGS_SIZE_OK(bgp_session_enable_args) ;
 struct bgp_session_update_args          /* to and from BGP Engine       */
 {
   struct stream*  buf ;
+  bgp_size_t size;
 } ;
 MQB_ARGS_SIZE_OK(bgp_session_enable_args) ;
 
@@ -214,6 +222,12 @@ struct bgp_session_event_args           /* to Routeing Engine           */
 } ;
 MQB_ARGS_SIZE_OK(bgp_session_enable_args) ;
 
+struct bgp_session_XON_args             /* to Routeing Engine           */
+{
+                                        /* no further arguments         */
+} ;
+MQB_ARGS_SIZE_OK(bgp_session_XON_args) ;
+
 /*==============================================================================
  * Session mutex lock/unlock
  */
@@ -255,7 +269,10 @@ extern void
 bgp_session_update_send(bgp_session session, struct stream* upd) ;
 
 extern void
-bgp_session_update_recv(bgp_session session, struct stream* upd) ;
+bgp_session_update_recv(bgp_session session, struct stream* buf, bgp_size_t size) ;
+
+extern int
+bgp_session_is_XON(bgp_peer peer);
 
 /*==============================================================================
  * Session data access functions.
diff --git a/bgpd/bgp_zebra.c b/bgpd/bgp_zebra.c
index ed4f9e87..c0fc3628 100644
--- a/bgpd/bgp_zebra.c
+++ b/bgpd/bgp_zebra.c
@@ -167,7 +167,7 @@ bgp_interface_down (int command, struct zclient *zclient, zebra_size_t length)
 	      continue;
 
 	    if (ifp == peer_if)
-	      BGP_EVENT_ADD (peer, BGP_Stop);
+	      bgp_peer_disable(peer, NULL);
 	  }
       }
   }
diff --git a/bgpd/bgpd.c b/bgpd/bgpd.c
index 060f850b..442ef607 100644
--- a/bgpd/bgpd.c
+++ b/bgpd/bgpd.c
@@ -293,7 +293,7 @@ bgp_confederation_id_set (struct bgp *bgp, as_t as)
 	    {
 	      peer->local_as = as;
               peer->last_reset = PEER_DOWN_CONFED_ID_CHANGE;
-              bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+              bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                      BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 	    }
 	}
@@ -307,7 +307,7 @@ bgp_confederation_id_set (struct bgp *bgp, as_t as)
 	      if (peer_sort (peer) == BGP_PEER_EBGP)
 		peer->local_as = as;
               peer->last_reset = PEER_DOWN_CONFED_ID_CHANGE;
-              bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+              bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                      BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 	    }
 	}
@@ -331,7 +331,7 @@ bgp_confederation_id_unset (struct bgp *bgp)
 	{
 	  peer->local_as = bgp->as;
           peer->last_reset = PEER_DOWN_CONFED_ID_CHANGE;
-          bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+          bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                     BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 	}
     }
@@ -389,7 +389,7 @@ bgp_confederation_peers_add (struct bgp *bgp, as_t as)
 	    {
 	      peer->local_as = bgp->as;
               peer->last_reset = PEER_DOWN_CONFED_PEER_CHANGE;
-              bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+              bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                      BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 	    }
 	}
@@ -440,7 +440,7 @@ bgp_confederation_peers_remove (struct bgp *bgp, as_t as)
 	    {
 	      peer->local_as = bgp->confed_id;
               peer->last_reset = PEER_DOWN_CONFED_PEER_CHANGE;
-              bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+              bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                      BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 	    }
 	}
@@ -724,7 +724,7 @@ peer_as_change (struct peer *peer, as_t as)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
       peer->last_reset = PEER_DOWN_REMOTE_AS_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
   type = peer_sort (peer);
@@ -883,7 +883,7 @@ peer_activate (struct peer *peer, afi_t afi, safi_t safi)
 #endif
                {
                  peer->last_reset = PEER_DOWN_AF_ACTIVATE;
-                 bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+                 bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
                         BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
                }
 #if 0
@@ -1518,7 +1518,7 @@ peer_group_bind (struct bgp *bgp, union sockunion *su,
   peer_group2peer_config_copy (group, peer, afi, safi);
 
   peer->last_reset = PEER_DOWN_RMAP_BIND;
-  bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+  bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
          BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 
   return 0;
@@ -1556,7 +1556,7 @@ peer_group_unbind (struct bgp *bgp, struct peer *peer,
     }
 
     peer->last_reset = PEER_DOWN_RMAP_UNBIND;
-    bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+    bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
            BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
   return 0;
 }
@@ -2048,7 +2048,7 @@ peer_flag_modify_action (struct peer *peer, u_int32_t flag)
       else if (flag == PEER_FLAG_DISABLE_CONNECTED_CHECK)
 	peer->last_reset = PEER_DOWN_MULTIHOP_CHANGE;
 
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
 }
@@ -2418,7 +2418,7 @@ peer_update_source_if_set (struct peer *peer, const char *ifname)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
        peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-       bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+       bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
               BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return 0;
     }
@@ -2445,7 +2445,7 @@ peer_update_source_if_set (struct peer *peer, const char *ifname)
       peer->update_if = XSTRDUP (MTYPE_PEER_UPDATE_SOURCE, ifname);
 
       peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
   return 0;
@@ -2477,7 +2477,7 @@ peer_update_source_addr_set (struct peer *peer, union sockunion *su)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
       peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return 0;
     }
@@ -2503,7 +2503,7 @@ peer_update_source_addr_set (struct peer *peer, union sockunion *su)
       peer->update_source = sockunion_dup (su);
 
       peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
           BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
   return 0;
@@ -2549,7 +2549,7 @@ peer_update_source_unset (struct peer *peer)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
       peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return 0;
     }
@@ -2574,7 +2574,7 @@ peer_update_source_unset (struct peer *peer)
 	}
 
       peer->last_reset = PEER_DOWN_UPDATE_SOURCE_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
   return 0;
@@ -2998,7 +2998,7 @@ peer_local_as_set (struct peer *peer, as_t as, int no_prepend)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
       peer->last_reset = PEER_DOWN_LOCAL_AS_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return 0;
     }
@@ -3013,7 +3013,7 @@ peer_local_as_set (struct peer *peer, as_t as, int no_prepend)
 	UNSET_FLAG (peer->flags, PEER_FLAG_LOCAL_AS_NO_PREPEND);
 
       peer->last_reset = PEER_DOWN_LOCAL_AS_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
 
@@ -3038,7 +3038,7 @@ peer_local_as_unset (struct peer *peer)
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
       peer->last_reset = PEER_DOWN_LOCAL_AS_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return 0;
     }
@@ -3050,7 +3050,7 @@ peer_local_as_unset (struct peer *peer)
       UNSET_FLAG (peer->flags, PEER_FLAG_LOCAL_AS_NO_PREPEND);
 
       peer->last_reset = PEER_DOWN_LOCAL_AS_CHANGE;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
   return 0;
@@ -3078,7 +3078,7 @@ peer_password_set (struct peer *peer, const char *password)
 
   if (! CHECK_FLAG (peer->sflags, PEER_STATUS_GROUP))
     {
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
           BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
       return BGP_SUCCESS;
     }
@@ -3093,7 +3093,7 @@ peer_password_set (struct peer *peer, const char *password)
 
       peer->password = XSTRDUP(MTYPE_PEER_PASSWORD, password);
 
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
     }
 
@@ -3116,7 +3116,7 @@ peer_password_unset (struct peer *peer)
 	  && strcmp (peer->group->conf->password, peer->password) == 0)
 	return BGP_ERR_PEER_GROUP_HAS_THE_FLAG;
 
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 
       if (peer->password)
@@ -3134,7 +3134,7 @@ peer_password_unset (struct peer *peer)
       if (!peer->password)
 	continue;
 
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
              BGP_NOTIFY_CEASE_CONFIG_CHANGE, 0));
 
       XFREE (MTYPE_PEER_PASSWORD, peer->password);
@@ -3864,12 +3864,12 @@ peer_clear (struct peer *peer)
 		zlog_debug ("%s Maximum-prefix restart timer cancelled",
 			    peer->host);
 	    }
-	  BGP_EVENT_ADD (peer, BGP_Start);
+	  bgp_peer_enable(peer);
 	  return 0;
 	}
 
       peer->v_start = BGP_INIT_START_TIMER;
-      bgp_peer_config_change(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
+      bgp_peer_reenable(peer, bgp_notify_new(BGP_NOTIFY_CEASE,
           BGP_NOTIFY_CEASE_ADMIN_RESET, 0));
     }
   return 0;