bgpd: handle socket read errors in the main pthread

Add a handler for socket errors that runs in the main pthread,
rather than the io pthread. When the io pthread encounters a
read error, capture the error and schedule a task for the main
pthread.

Signed-off-by: Mark Stapp <mjs@voltanet.io>

diff --git a/bgpd/bgp_io.c b/bgpd/bgp_io.c
index 412c8e3e5ec0e13dafef76c3e95cd2ec11d378c7..4190b18518e4f5930b5e2ebf1a82be85069c4ddc 100644 (file)
--- a/bgpd/bgp_io.c
+++ b/bgpd/bgp_io.c
@@ -44,7 +44,7 @@
 
 /* forward declarations */
 static uint16_t bgp_write(struct peer *);
-static uint16_t bgp_read(struct peer *);
+static uint16_t bgp_read(struct peer *peer, int *code_p);
 static int bgp_process_writes(struct thread *);
 static int bgp_process_reads(struct thread *);
 static bool validate_header(struct peer *);
@@ -175,6 +175,7 @@ static int bgp_process_reads(struct thread *thread)
        bool fatal = false;             // whether fatal error occurred
        bool added_pkt = false;         // whether we pushed onto ->ibuf
        /* clang-format on */
+       int code;
 
        peer = THREAD_ARG(thread);
 
@@ -184,7 +185,7 @@ static int bgp_process_reads(struct thread *thread)
        struct frr_pthread *fpt = bgp_pth_io;
 
        frr_with_mutex(&peer->io_mtx) {
-               status = bgp_read(peer);
+               status = bgp_read(peer, &code);
        }
 
        /* error checking phase */
@@ -197,6 +198,12 @@ static int bgp_process_reads(struct thread *thread)
                /* problem; tear down session */
                more = false;
                fatal = true;
+
+               /* Handle the error in the main pthread, include the
+                * specific state change from 'bgp_read'.
+                */
+               thread_add_event(bm->master, bgp_packet_process_error,
+                                peer, code, NULL);
        }
 
        while (more) {
@@ -230,6 +237,7 @@ static int bgp_process_reads(struct thread *thread)
                 */
                if (ringbuf_remain(ibw) >= pktsize) {
                        struct stream *pkt = stream_new(pktsize);
+
                        assert(STREAM_WRITEABLE(pkt) == pktsize);
                        assert(ringbuf_get(ibw, pkt->data, pktsize) == pktsize);
                        stream_set_endp(pkt, pktsize);
@@ -442,7 +450,7 @@ done : {
  *
  * @return status flag (see top-of-file)
  */
-static uint16_t bgp_read(struct peer *peer)
+static uint16_t bgp_read(struct peer *peer, int *code_p)
 {
        size_t readsize; // how many bytes we want to read
        ssize_t nbytes;  // how many bytes we actually read
@@ -455,43 +463,28 @@ static uint16_t bgp_read(struct peer *peer)
        /* EAGAIN or EWOULDBLOCK; come back later */
        if (nbytes < 0 && ERRNO_IO_RETRY(errno)) {
                SET_FLAG(status, BGP_IO_TRANS_ERR);
-               /* Fatal error; tear down session */
        } else if (nbytes < 0) {
+               /* Fatal error; tear down session */
                flog_err(EC_BGP_UPDATE_RCV,
                         "%s [Error] bgp_read_packet error: %s", peer->host,
                         safe_strerror(errno));
 
-               if (peer->status == Established) {
-                       if ((CHECK_FLAG(peer->flags, PEER_FLAG_GRACEFUL_RESTART)
-                            || CHECK_FLAG(peer->flags,
-                                          PEER_FLAG_GRACEFUL_RESTART_HELPER))
-                           && CHECK_FLAG(peer->sflags, PEER_STATUS_NSF_MODE)) {
-                               peer->last_reset = PEER_DOWN_NSF_CLOSE_SESSION;
-                               SET_FLAG(peer->sflags, PEER_STATUS_NSF_WAIT);
-                       } else
-                               peer->last_reset = PEER_DOWN_CLOSE_SESSION;
-               }
+               /* Handle the error in the main pthread. */
+               if (code_p)
+                       *code_p = TCP_fatal_error;
 
-               BGP_EVENT_ADD(peer, TCP_fatal_error);
                SET_FLAG(status, BGP_IO_FATAL_ERR);
-               /* Received EOF / TCP session closed */
+
        } else if (nbytes == 0) {
+               /* Received EOF / TCP session closed */
                if (bgp_debug_neighbor_events(peer))
                        zlog_debug("%s [Event] BGP connection closed fd %d",
                                   peer->host, peer->fd);
 
-               if (peer->status == Established) {
-                       if ((CHECK_FLAG(peer->flags, PEER_FLAG_GRACEFUL_RESTART)
-                            || CHECK_FLAG(peer->flags,
-                                          PEER_FLAG_GRACEFUL_RESTART_HELPER))
-                           && CHECK_FLAG(peer->sflags, PEER_STATUS_NSF_MODE)) {
-                               peer->last_reset = PEER_DOWN_NSF_CLOSE_SESSION;
-                               SET_FLAG(peer->sflags, PEER_STATUS_NSF_WAIT);
-                       } else
-                               peer->last_reset = PEER_DOWN_CLOSE_SESSION;
-               }
+               /* Handle the error in the main pthread. */
+               if (code_p)
+                       *code_p = TCP_connection_closed;
 
-               BGP_EVENT_ADD(peer, TCP_connection_closed);
                SET_FLAG(status, BGP_IO_FATAL_ERR);
        } else {
                assert(ringbuf_put(peer->ibuf_work, ibw, nbytes)

diff --git a/bgpd/bgp_packet.c b/bgpd/bgp_packet.c
index d4715c24337573a8b76cbdc1cf7d80c9e0f4dae1..2c8a375a55e228a01d1dc73d923a259528f06421 100644 (file)
--- a/bgpd/bgp_packet.c
+++ b/bgpd/bgp_packet.c
@@ -2488,3 +2488,37 @@ void bgp_send_delayed_eor(struct bgp *bgp)
        for (ALL_LIST_ELEMENTS(bgp->peer, node, nnode, peer))
                bgp_write_proceed_actions(peer);
 }
+
+/*
+ * Task callback to handle socket error encountered in the io pthread. We avoid
+ * having the io pthread try to enqueue fsm events or mess with the peer
+ * struct.
+ */
+int bgp_packet_process_error(struct thread *thread)
+{
+       struct peer *peer;
+       int code;
+
+       peer = THREAD_ARG(thread);
+       code = THREAD_VAL(thread);
+
+       if (bgp_debug_neighbor_events(peer))
+               zlog_debug("%s [Event] BGP error %d on fd %d",
+                          peer->host, peer->fd, code);
+
+       /* Closed connection or error on the socket */
+       if (peer->status == Established) {
+               if ((CHECK_FLAG(peer->flags, PEER_FLAG_GRACEFUL_RESTART)
+                    || CHECK_FLAG(peer->flags,
+                                  PEER_FLAG_GRACEFUL_RESTART_HELPER))
+                   && CHECK_FLAG(peer->sflags, PEER_STATUS_NSF_MODE)) {
+                       peer->last_reset = PEER_DOWN_NSF_CLOSE_SESSION;
+                       SET_FLAG(peer->sflags, PEER_STATUS_NSF_WAIT);
+               } else
+                       peer->last_reset = PEER_DOWN_CLOSE_SESSION;
+       }
+
+       bgp_event_update(peer, code);
+
+       return 0;
+}

diff --git a/bgpd/bgp_packet.h b/bgpd/bgp_packet.h
index e83f7d950c8f172aa6b38f419bfc1fcc932d290c..48de9d9cbd5026958fa7269ea0f0867a1f99e8aa 100644 (file)
--- a/bgpd/bgp_packet.h
+++ b/bgpd/bgp_packet.h
@@ -82,4 +82,8 @@ extern int bgp_generate_updgrp_packets(struct thread *);
 extern int bgp_process_packet(struct thread *);
 
 extern void bgp_send_delayed_eor(struct bgp *bgp);
+
+/* Task callback to handle socket error encountered in the io pthread */
+int bgp_packet_process_error(struct thread *thread);
+
 #endif /* _QUAGGA_BGP_PACKET_H */