bgpd: add input per-peer priority pkt queue

Under heavy loaded system with many connections, and many BGP updates
received, the system has issues to maintain BGP connectivity with peers.

For instance, the wireshark capture indicates a BGP open packet is received,
and is processed by the device many seconds after. Obviously, the remote
peer had already flushed the connection..

The proposed fix attempts to create a second stream of buffers reserved
for prioritary BGP packets: BGP OPEN, BGP NOTIF, BGP KEEPALIVE.
When running the bgp_packet_receive() job, this prioritary fifo is
dequeued before consuming the standard fifo queue.

Signed-off-by: Philippe Guibert <[email protected]>

bgpd/bgp_fsm.c

@@ -503,7 +503,7 @@ static void bgp_connect_timer(struct event *thread)
 /* BGP holdtime timer. */
 static void bgp_holdtime_timer(struct event *thread)
 {
-	atomic_size_t inq_count;
+	atomic_size_t inq_count, inq_count_priority;
 	struct peer_connection *connection = EVENT_ARG(thread);
 	struct peer *peer = connection->peer;
 
@@ -523,7 +523,11 @@ static void bgp_holdtime_timer(struct event *thread)
 	 */
 	inq_count = atomic_load_explicit(&connection->ibuf->count,
 					 memory_order_relaxed);
-	if (inq_count)
+	inq_count_priority =
+		atomic_load_explicit(&connection->ibuf_priority->count,
+				     memory_order_relaxed);
+
+	if (inq_count + inq_count_priority)
 		BGP_TIMER_ON(connection->t_holdtime, bgp_holdtime_timer,
 			     peer->v_holdtime);
 
@@ -1489,6 +1493,8 @@ enum bgp_fsm_state_progress bgp_stop(struct peer_connection *connection)
 	frr_with_mutex (&connection->io_mtx) {
 		if (connection->ibuf)
 			stream_fifo_clean(connection->ibuf);
+		if (connection->ibuf_priority)
+			stream_fifo_clean(connection->ibuf_priority);
 		if (connection->obuf)
 			stream_fifo_clean(connection->obuf);
 

bgpd/bgp_io.c

@@ -51,6 +51,7 @@ void bgp_writes_on(struct peer_connection *connection)
 	assert(connection->status != Deleted);
 	assert(connection->obuf);
 	assert(connection->ibuf);
+	assert(connection->ibuf_priority);
 	assert(connection->ibuf_work);
 	assert(!connection->t_connect_check_r);
 	assert(!connection->t_connect_check_w);
@@ -80,6 +81,7 @@ void bgp_reads_on(struct peer_connection *connection)
 
 	assert(connection->status != Deleted);
 	assert(connection->ibuf);
+	assert(connection->ibuf_priority);
 	assert(connection->fd);
 	assert(connection->ibuf_work);
 	assert(connection->obuf);
@@ -162,6 +164,7 @@ static int read_ibuf_work(struct peer_connection *connection)
 	/* packet size as given by header */
 	uint16_t pktsize = 0;
 	struct stream *pkt;
+	uint8_t type;
 
 	/* ============================================== */
 	frr_with_mutex (&connection->io_mtx) {
@@ -200,8 +203,18 @@ static int read_ibuf_work(struct peer_connection *connection)
 	stream_set_endp(pkt, pktsize);
 
 	frrtrace(2, frr_bgp, packet_read, connection->peer, pkt);
+
+	/* get the type of message */
+	stream_forward_getp(pkt, BGP_MARKER_SIZE + 2);
+	type = *stream_pnt(pkt);
+	stream_rewind_getp(pkt, BGP_MARKER_SIZE + 2);
+
 	frr_with_mutex (&connection->io_mtx) {
-		stream_fifo_push(connection->ibuf, pkt);
+		if (type == BGP_MSG_OPEN || type == BGP_MSG_KEEPALIVE ||
+		    type == BGP_MSG_NOTIFY)
+			stream_fifo_push(connection->ibuf_priority, pkt);
+		else
+			stream_fifo_push(connection->ibuf, pkt);
 	}
 
 	return pktsize;

bgpd/bgp_packet.c

@@ -4027,7 +4027,9 @@ void bgp_process_packet(struct event *thread)
 		char notify_data_length[2];
 
 		frr_with_mutex (&connection->io_mtx) {
-			peer->curr = stream_fifo_pop(connection->ibuf);
+			peer->curr = stream_fifo_pop(connection->ibuf_priority);
+			if (peer->curr == NULL)
+				peer->curr = stream_fifo_pop(connection->ibuf);
 		}
 
 		if (peer->curr == NULL) // no packets to process, hmm...
@@ -4155,7 +4157,8 @@ void bgp_process_packet(struct event *thread)
 	    && fsm_update_result != FSM_PEER_STOPPED) {
 		frr_with_mutex (&connection->io_mtx) {
 			// more work to do, come back later
-			if (connection->ibuf->count > 0)
+			if (connection->ibuf->count > 0 ||
+			    connection->ibuf_priority->count)
 				event_add_event(bm->master, bgp_process_packet,
 						connection, 0,
 						&connection->t_process_packet);

bgpd/bgpd.c

@@ -1177,6 +1177,11 @@ void bgp_peer_connection_buffers_free(struct peer_connection *connection)
 			connection->ibuf = NULL;
 		}
 
+		if (connection->ibuf_priority) {
+			stream_fifo_free(connection->ibuf_priority);
+			connection->ibuf_priority = NULL;
+		}
+
 		if (connection->obuf) {
 			stream_fifo_free(connection->obuf);
 			connection->obuf = NULL;
@@ -1211,6 +1216,7 @@ struct peer_connection *bgp_peer_connection_new(struct peer *peer)
 	connection->fd = -1;
 
 	connection->ibuf = stream_fifo_new();
+	connection->ibuf_priority = stream_fifo_new();
 	connection->obuf = stream_fifo_new();
 	pthread_mutex_init(&connection->io_mtx, NULL);
 

bgpd/bgpd.h

@@ -1197,6 +1197,8 @@ struct peer_connection {
 	/* Packet receive and send buffer. */
 	pthread_mutex_t io_mtx;	  // guards ibuf, obuf
 	struct stream_fifo *ibuf; // packets waiting to be processed
+	struct stream_fifo *ibuf_priority; // keepalive/open/notify packets
+					   // waiting to be processed
 	struct stream_fifo *obuf; // packets waiting to be written
 
 	struct ringbuf *ibuf_work; // WiP buffer used by bgp_read() only