diff options
37 files changed, 2468 insertions, 1072 deletions
diff --git a/bgpd/Makefile.am b/bgpd/Makefile.am index fa1dcbb762..b0d34dc43b 100644 --- a/bgpd/Makefile.am +++ b/bgpd/Makefile.am @@ -85,7 +85,8 @@ libbgp_a_SOURCES = \ bgp_damp.c bgp_table.c bgp_advertise.c bgp_vty.c bgp_mpath.c \ bgp_nht.c bgp_updgrp.c bgp_updgrp_packet.c bgp_updgrp_adv.c bgp_bfd.c \ bgp_encap_tlv.c $(BGP_VNC_RFAPI_SRC) bgp_attr_evpn.c \ - bgp_evpn.c bgp_evpn_vty.c bgp_vpn.c bgp_label.c bgp_rd.c + bgp_evpn.c bgp_evpn_vty.c bgp_vpn.c bgp_label.c bgp_rd.c \ + bgp_keepalives.c bgp_io.c noinst_HEADERS = \ bgp_memory.h \ @@ -97,7 +98,8 @@ noinst_HEADERS = \ bgp_advertise.h bgp_vty.h bgp_mpath.h bgp_nht.h \ bgp_updgrp.h bgp_bfd.h bgp_encap_tlv.h bgp_encap_types.h \ $(BGP_VNC_RFAPI_HD) bgp_attr_evpn.h bgp_evpn.h bgp_evpn_vty.h \ - bgp_vpn.h bgp_label.h bgp_rd.h bgp_evpn_private.h + bgp_vpn.h bgp_label.h bgp_rd.h bgp_evpn_private.h bgp_keepalives.h \ + bgp_io.h bgpd_SOURCES = bgp_main.c bgpd_LDADD = libbgp.a $(BGP_VNC_RFP_LIB) ../lib/libfrr.la @LIBCAP@ @LIBM@ diff --git a/bgpd/bgp_attr.c b/bgpd/bgp_attr.c index 6ddb2ec8a7..1f0662bfb0 100644 --- a/bgpd/bgp_attr.c +++ b/bgpd/bgp_attr.c @@ -1156,7 +1156,7 @@ static int bgp_attr_aspath(struct bgp_attr_parser_args *args) * peer with AS4 => will get 4Byte ASnums * otherwise, will get 16 Bit */ - attr->aspath = aspath_parse(peer->ibuf, length, + attr->aspath = aspath_parse(peer->curr, length, CHECK_FLAG(peer->cap, PEER_CAP_AS4_RCV)); /* In case of IBGP, length will be zero. */ @@ -1230,7 +1230,7 @@ static int bgp_attr_as4_path(struct bgp_attr_parser_args *args, struct attr *const attr = args->attr; const bgp_size_t length = args->length; - *as4_path = aspath_parse(peer->ibuf, length, 1); + *as4_path = aspath_parse(peer->curr, length, 1); /* In case of IBGP, length will be zero. */ if (!*as4_path) { @@ -1271,7 +1271,7 @@ static bgp_attr_parse_ret_t bgp_attr_nexthop(struct bgp_attr_parser_args *args) logged locally (this is implemented somewhere else). The UPDATE message gets ignored in any of these cases. */ - nexthop_n = stream_get_ipv4(peer->ibuf); + nexthop_n = stream_get_ipv4(peer->curr); nexthop_h = ntohl(nexthop_n); if ((IPV4_NET0(nexthop_h) || IPV4_NET127(nexthop_h) || IPV4_CLASS_DE(nexthop_h)) @@ -1307,7 +1307,7 @@ static bgp_attr_parse_ret_t bgp_attr_med(struct bgp_attr_parser_args *args) args->total); } - attr->med = stream_getl(peer->ibuf); + attr->med = stream_getl(peer->curr); attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_MULTI_EXIT_DISC); @@ -1333,11 +1333,11 @@ bgp_attr_local_pref(struct bgp_attr_parser_args *args) external peer, then this attribute MUST be ignored by the receiving speaker. */ if (peer->sort == BGP_PEER_EBGP) { - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); return BGP_ATTR_PARSE_PROCEED; } - attr->local_pref = stream_getl(peer->ibuf); + attr->local_pref = stream_getl(peer->curr); /* Set the local-pref flag. */ attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_LOCAL_PREF); @@ -1386,10 +1386,10 @@ static int bgp_attr_aggregator(struct bgp_attr_parser_args *args) } if (CHECK_FLAG(peer->cap, PEER_CAP_AS4_RCV)) - attr->aggregator_as = stream_getl(peer->ibuf); + attr->aggregator_as = stream_getl(peer->curr); else - attr->aggregator_as = stream_getw(peer->ibuf); - attr->aggregator_addr.s_addr = stream_get_ipv4(peer->ibuf); + attr->aggregator_as = stream_getw(peer->curr); + attr->aggregator_addr.s_addr = stream_get_ipv4(peer->curr); /* Set atomic aggregate flag. */ attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_AGGREGATOR); @@ -1413,8 +1413,8 @@ bgp_attr_as4_aggregator(struct bgp_attr_parser_args *args, 0); } - *as4_aggregator_as = stream_getl(peer->ibuf); - as4_aggregator_addr->s_addr = stream_get_ipv4(peer->ibuf); + *as4_aggregator_as = stream_getl(peer->curr); + as4_aggregator_addr->s_addr = stream_get_ipv4(peer->curr); attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_AS4_AGGREGATOR); @@ -1540,10 +1540,10 @@ bgp_attr_community(struct bgp_attr_parser_args *args) } attr->community = - community_parse((u_int32_t *)stream_pnt(peer->ibuf), length); + community_parse((u_int32_t *)stream_pnt(peer->curr), length); /* XXX: fix community_parse to use stream API and remove this */ - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); if (!attr->community) return bgp_attr_malformed(args, BGP_NOTIFY_UPDATE_OPT_ATTR_ERR, @@ -1570,7 +1570,7 @@ bgp_attr_originator_id(struct bgp_attr_parser_args *args) args->total); } - attr->originator_id.s_addr = stream_get_ipv4(peer->ibuf); + attr->originator_id.s_addr = stream_get_ipv4(peer->curr); attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_ORIGINATOR_ID); @@ -1594,10 +1594,10 @@ bgp_attr_cluster_list(struct bgp_attr_parser_args *args) } attr->cluster = - cluster_parse((struct in_addr *)stream_pnt(peer->ibuf), length); + cluster_parse((struct in_addr *)stream_pnt(peer->curr), length); /* XXX: Fix cluster_parse to use stream API and then remove this */ - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_CLUSTER_LIST); @@ -1778,7 +1778,7 @@ int bgp_mp_unreach_parse(struct bgp_attr_parser_args *args, struct attr *const attr = args->attr; const bgp_size_t length = args->length; - s = peer->ibuf; + s = peer->curr; #define BGP_MP_UNREACH_MIN_SIZE 3 if ((length > STREAM_READABLE(s)) || (length < BGP_MP_UNREACH_MIN_SIZE)) @@ -1832,9 +1832,9 @@ bgp_attr_large_community(struct bgp_attr_parser_args *args) } attr->lcommunity = - lcommunity_parse((u_int8_t *)stream_pnt(peer->ibuf), length); + lcommunity_parse((u_int8_t *)stream_pnt(peer->curr), length); /* XXX: fix ecommunity_parse to use stream API */ - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); if (!attr->lcommunity) return bgp_attr_malformed(args, BGP_NOTIFY_UPDATE_OPT_ATTR_ERR, @@ -1861,9 +1861,9 @@ bgp_attr_ext_communities(struct bgp_attr_parser_args *args) } attr->ecommunity = - ecommunity_parse((u_int8_t *)stream_pnt(peer->ibuf), length); + ecommunity_parse((u_int8_t *)stream_pnt(peer->curr), length); /* XXX: fix ecommunity_parse to use stream API */ - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); if (!attr->ecommunity) return bgp_attr_malformed(args, BGP_NOTIFY_UPDATE_OPT_ATTR_ERR, @@ -1957,7 +1957,7 @@ static int bgp_attr_encap(uint8_t type, struct peer *peer, /* IN */ + sublength); tlv->type = subtype; tlv->length = sublength; - stream_get(tlv->value, peer->ibuf, sublength); + stream_get(tlv->value, peer->curr, sublength); length -= sublength; /* attach tlv to encap chain */ @@ -2025,8 +2025,8 @@ bgp_attr_prefix_sid(struct bgp_attr_parser_args *args, attr->flag |= ATTR_FLAG_BIT(BGP_ATTR_PREFIX_SID); - type = stream_getc(peer->ibuf); - length = stream_getw(peer->ibuf); + type = stream_getc(peer->curr); + length = stream_getw(peer->curr); if (type == BGP_PREFIX_SID_LABEL_INDEX) { if (length != BGP_PREFIX_SID_LABEL_INDEX_LENGTH) { @@ -2039,11 +2039,11 @@ bgp_attr_prefix_sid(struct bgp_attr_parser_args *args, } /* Ignore flags and reserved */ - stream_getc(peer->ibuf); - stream_getw(peer->ibuf); + stream_getc(peer->curr); + stream_getw(peer->curr); /* Fetch the label index and see if it is valid. */ - label_index = stream_getl(peer->ibuf); + label_index = stream_getl(peer->curr); if (label_index == BGP_INVALID_LABEL_INDEX) return bgp_attr_malformed( args, BGP_NOTIFY_UPDATE_OPT_ATTR_ERR, @@ -2074,16 +2074,16 @@ bgp_attr_prefix_sid(struct bgp_attr_parser_args *args, } /* Ignore reserved */ - stream_getc(peer->ibuf); - stream_getw(peer->ibuf); + stream_getc(peer->curr); + stream_getw(peer->curr); - stream_get(&ipv6_sid, peer->ibuf, 16); + stream_get(&ipv6_sid, peer->curr, 16); } /* Placeholder code for the Originator SRGB type */ else if (type == BGP_PREFIX_SID_ORIGINATOR_SRGB) { /* Ignore flags */ - stream_getw(peer->ibuf); + stream_getw(peer->curr); length -= 2; @@ -2099,8 +2099,8 @@ bgp_attr_prefix_sid(struct bgp_attr_parser_args *args, srgb_count = length / BGP_PREFIX_SID_ORIGINATOR_SRGB_LENGTH; for (int i = 0; i < srgb_count; i++) { - stream_get(&srgb_base, peer->ibuf, 3); - stream_get(&srgb_range, peer->ibuf, 3); + stream_get(&srgb_base, peer->curr, 3); + stream_get(&srgb_range, peer->curr, 3); } } @@ -2125,7 +2125,7 @@ static bgp_attr_parse_ret_t bgp_attr_unknown(struct bgp_attr_parser_args *args) peer->host, type, length); /* Forward read pointer of input stream. */ - stream_forward_getp(peer->ibuf, length); + stream_forward_getp(peer->curr, length); /* If any of the mandatory well-known attributes are not recognized, then the Error Subcode is set to Unrecognized Well-known @@ -2247,7 +2247,7 @@ bgp_attr_parse_ret_t bgp_attr_parse(struct peer *peer, struct attr *attr, "%s: error BGP attribute length %lu is smaller than min len", peer->host, (unsigned long)(endp - - STREAM_PNT(BGP_INPUT(peer)))); + - stream_pnt(BGP_INPUT(peer)))); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_ATTR_LENG_ERR); @@ -2269,7 +2269,7 @@ bgp_attr_parse_ret_t bgp_attr_parse(struct peer *peer, struct attr *attr, "%s: Extended length set, but just %lu bytes of attr header", peer->host, (unsigned long)(endp - - STREAM_PNT(BGP_INPUT(peer)))); + - stream_pnt(BGP_INPUT(peer)))); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_ATTR_LENG_ERR); diff --git a/bgpd/bgp_fsm.c b/bgpd/bgp_fsm.c index 8de7e970de..9e58e466e1 100644 --- a/bgpd/bgp_fsm.c +++ b/bgpd/bgp_fsm.c @@ -49,6 +49,8 @@ #include "bgpd/bgp_nht.h" #include "bgpd/bgp_bfd.h" #include "bgpd/bgp_memory.h" +#include "bgpd/bgp_keepalives.h" +#include "bgpd/bgp_io.h" DEFINE_HOOK(peer_backward_transition, (struct peer * peer), (peer)) DEFINE_HOOK(peer_established, (struct peer * peer), (peer)) @@ -86,7 +88,6 @@ int bgp_event(struct thread *); static int bgp_start_timer(struct thread *); static int bgp_connect_timer(struct thread *); static int bgp_holdtime_timer(struct thread *); -static int bgp_keepalive_timer(struct thread *); /* BGP FSM functions. */ static int bgp_start(struct peer *); @@ -125,20 +126,67 @@ static struct peer *peer_xfer_conn(struct peer *from_peer) from_peer->host, from_peer, from_peer->fd, peer, peer->fd); - BGP_WRITE_OFF(peer->t_write); - BGP_READ_OFF(peer->t_read); - BGP_WRITE_OFF(from_peer->t_write); - BGP_READ_OFF(from_peer->t_read); + bgp_writes_off(peer); + bgp_reads_off(peer); + bgp_writes_off(from_peer); + bgp_reads_off(from_peer); BGP_TIMER_OFF(peer->t_routeadv); + BGP_TIMER_OFF(peer->t_connect); + BGP_TIMER_OFF(peer->t_connect_check_r); + BGP_TIMER_OFF(peer->t_connect_check_w); BGP_TIMER_OFF(from_peer->t_routeadv); + BGP_TIMER_OFF(from_peer->t_connect); + BGP_TIMER_OFF(from_peer->t_connect_check_r); + BGP_TIMER_OFF(from_peer->t_connect_check_w); + BGP_TIMER_OFF(from_peer->t_process_packet); + + /* + * At this point in time, it is possible that there are packets pending + * on various buffers. Those need to be transferred or dropped, + * otherwise we'll get spurious failures during session establishment. + */ + pthread_mutex_lock(&peer->io_mtx); + pthread_mutex_lock(&from_peer->io_mtx); + { + fd = peer->fd; + peer->fd = from_peer->fd; + from_peer->fd = fd; + + stream_fifo_clean(peer->ibuf); + stream_fifo_clean(peer->obuf); + stream_reset(peer->ibuf_work); + + /* + * this should never happen, since bgp_process_packet() is the + * only task that sets and unsets the current packet and it + * runs in our pthread. + */ + if (peer->curr) { + zlog_err( + "[%s] Dropping pending packet on connection transfer:", + peer->host); + u_int16_t type = stream_getc_from(peer->curr, + BGP_MARKER_SIZE + 2); + bgp_dump_packet(peer, type, peer->curr); + stream_free(peer->curr); + peer->curr = NULL; + } + + // copy each packet from old peer's output queue to new peer + while (from_peer->obuf->head) + stream_fifo_push(peer->obuf, + stream_fifo_pop(from_peer->obuf)); - fd = peer->fd; - peer->fd = from_peer->fd; - from_peer->fd = fd; - stream_reset(peer->ibuf); - stream_fifo_clean(peer->obuf); - stream_fifo_clean(from_peer->obuf); + // copy each packet from old peer's input queue to new peer + while (from_peer->ibuf->head) + stream_fifo_push(peer->ibuf, + stream_fifo_pop(from_peer->ibuf)); + + stream_copy(peer->ibuf_work, from_peer->ibuf_work); + } + pthread_mutex_unlock(&from_peer->io_mtx); + pthread_mutex_unlock(&peer->io_mtx); peer->as = from_peer->as; peer->v_holdtime = from_peer->v_holdtime; @@ -216,8 +264,10 @@ static struct peer *peer_xfer_conn(struct peer *from_peer) } } - BGP_READ_ON(peer->t_read, bgp_read, peer->fd); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + bgp_reads_on(peer); + bgp_writes_on(peer); + thread_add_timer_msec(bm->master, bgp_process_packet, peer, 0, + &peer->t_process_packet); if (from_peer) peer_xfer_stats(peer, from_peer); @@ -243,7 +293,7 @@ void bgp_timer_set(struct peer *peer) } BGP_TIMER_OFF(peer->t_connect); BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); BGP_TIMER_OFF(peer->t_routeadv); break; @@ -255,7 +305,7 @@ void bgp_timer_set(struct peer *peer) BGP_TIMER_ON(peer->t_connect, bgp_connect_timer, peer->v_connect); BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); BGP_TIMER_OFF(peer->t_routeadv); break; @@ -272,7 +322,7 @@ void bgp_timer_set(struct peer *peer) peer->v_connect); } BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); BGP_TIMER_OFF(peer->t_routeadv); break; @@ -286,7 +336,7 @@ void bgp_timer_set(struct peer *peer) } else { BGP_TIMER_OFF(peer->t_holdtime); } - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); BGP_TIMER_OFF(peer->t_routeadv); break; @@ -299,12 +349,11 @@ void bgp_timer_set(struct peer *peer) timer and KeepAlive timers are not started. */ if (peer->v_holdtime == 0) { BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); } else { BGP_TIMER_ON(peer->t_holdtime, bgp_holdtime_timer, peer->v_holdtime); - BGP_TIMER_ON(peer->t_keepalive, bgp_keepalive_timer, - peer->v_keepalive); + bgp_keepalives_on(peer); } BGP_TIMER_OFF(peer->t_routeadv); break; @@ -319,12 +368,11 @@ void bgp_timer_set(struct peer *peer) and keepalive must be turned off. */ if (peer->v_holdtime == 0) { BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); } else { BGP_TIMER_ON(peer->t_holdtime, bgp_holdtime_timer, peer->v_holdtime); - BGP_TIMER_ON(peer->t_keepalive, bgp_keepalive_timer, - peer->v_keepalive); + bgp_keepalives_on(peer); } break; case Deleted: @@ -336,7 +384,7 @@ void bgp_timer_set(struct peer *peer) BGP_TIMER_OFF(peer->t_start); BGP_TIMER_OFF(peer->t_connect); BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); + bgp_keepalives_off(peer); BGP_TIMER_OFF(peer->t_routeadv); break; } @@ -367,6 +415,10 @@ static int bgp_connect_timer(struct thread *thread) int ret; peer = THREAD_ARG(thread); + + assert(!peer->t_write); + assert(!peer->t_read); + peer->t_connect = NULL; if (bgp_debug_neighbor_events(peer)) @@ -402,24 +454,6 @@ static int bgp_holdtime_timer(struct thread *thread) return 0; } -/* BGP keepalive fire ! */ -static int bgp_keepalive_timer(struct thread *thread) -{ - struct peer *peer; - - peer = THREAD_ARG(thread); - peer->t_keepalive = NULL; - - if (bgp_debug_neighbor_events(peer)) - zlog_debug("%s [FSM] Timer (keepalive timer expire)", - peer->host); - - THREAD_VAL(thread) = KeepAlive_timer_expired; - bgp_event(thread); /* bgp_event unlocks peer */ - - return 0; -} - int bgp_routeadv_timer(struct thread *thread) { struct peer *peer; @@ -433,7 +467,8 @@ int bgp_routeadv_timer(struct thread *thread) peer->synctime = bgp_clock(); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + thread_add_timer_msec(bm->master, bgp_generate_updgrp_packets, peer, 0, + &peer->t_generate_updgrp_packets); /* MRAI timer will be started again when FIFO is built, no need to * do it here. @@ -640,7 +675,9 @@ void bgp_adjust_routeadv(struct peer *peer) BGP_TIMER_OFF(peer->t_routeadv); peer->synctime = bgp_clock(); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + thread_add_timer_msec(bm->master, bgp_generate_updgrp_packets, + peer, 0, + &peer->t_generate_updgrp_packets); return; } @@ -1035,27 +1072,41 @@ int bgp_stop(struct peer *peer) bgp_bfd_deregister_peer(peer); } - /* Stop read and write threads when exists. */ - BGP_READ_OFF(peer->t_read); - BGP_WRITE_OFF(peer->t_write); + /* stop keepalives */ + bgp_keepalives_off(peer); + + /* Stop read and write threads. */ + bgp_writes_off(peer); + bgp_reads_off(peer); + + THREAD_OFF(peer->t_connect_check_r); + THREAD_OFF(peer->t_connect_check_w); /* Stop all timers. */ BGP_TIMER_OFF(peer->t_start); BGP_TIMER_OFF(peer->t_connect); BGP_TIMER_OFF(peer->t_holdtime); - BGP_TIMER_OFF(peer->t_keepalive); BGP_TIMER_OFF(peer->t_routeadv); - /* Stream reset. */ - peer->packet_size = 0; - /* Clear input and output buffer. */ - if (peer->ibuf) - stream_reset(peer->ibuf); - if (peer->work) - stream_reset(peer->work); - if (peer->obuf) - stream_fifo_clean(peer->obuf); + pthread_mutex_lock(&peer->io_mtx); + { + if (peer->ibuf) + stream_fifo_clean(peer->ibuf); + if (peer->obuf) + stream_fifo_clean(peer->obuf); + + if (peer->ibuf_work) + stream_reset(peer->ibuf_work); + if (peer->obuf_work) + stream_reset(peer->obuf_work); + + if (peer->curr) { + stream_free(peer->curr); + peer->curr = NULL; + } + } + pthread_mutex_unlock(&peer->io_mtx); /* Close of file descriptor. */ if (peer->fd >= 0) { @@ -1161,6 +1212,61 @@ static int bgp_stop_with_notify(struct peer *peer, u_char code, u_char sub_code) return (bgp_stop(peer)); } +/** + * Determines whether a TCP session has successfully established for a peer and + * events as appropriate. + * + * This function is called when setting up a new session. After connect() is + * called on the peer's socket (in bgp_start()), the fd is passed to poll() + * to wait for connection success or failure. When poll() returns, this + * function is called to evaluate the result. + * + * Due to differences in behavior of poll() on Linux and BSD - specifically, + * the value of .revents in the case of a closed connection - this function is + * scheduled both for a read and a write event. The write event is triggered + * when the connection is established. A read event is triggered when the + * connection is closed. Thus we need to cancel whichever one did not occur. + */ +static int bgp_connect_check(struct thread *thread) +{ + int status; + socklen_t slen; + int ret; + struct peer *peer; + + peer = THREAD_ARG(thread); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_READS_ON)); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON)); + assert(!peer->t_read); + assert(!peer->t_write); + + THREAD_OFF(peer->t_connect_check_r); + THREAD_OFF(peer->t_connect_check_w); + + /* Check file descriptor. */ + slen = sizeof(status); + ret = getsockopt(peer->fd, SOL_SOCKET, SO_ERROR, (void *)&status, + &slen); + + /* If getsockopt is fail, this is fatal error. */ + if (ret < 0) { + zlog_info("can't get sockopt for nonblocking connect"); + BGP_EVENT_ADD(peer, TCP_fatal_error); + return -1; + } + + /* When status is 0 then TCP connection is established. */ + if (status == 0) { + BGP_EVENT_ADD(peer, TCP_connection_open); + return 1; + } else { + if (bgp_debug_neighbor_events(peer)) + zlog_debug("%s [Event] Connect failed (%s)", peer->host, + safe_strerror(errno)); + BGP_EVENT_ADD(peer, TCP_connection_open_failed); + return 0; + } +} /* TCP connection open. Next we send open message to remote peer. And add read thread for reading open message. */ @@ -1178,10 +1284,11 @@ static int bgp_connect_success(struct peer *peer) __FUNCTION__, peer->host, peer->fd); bgp_notify_send(peer, BGP_NOTIFY_FSM_ERR, 0); /* internal error */ + bgp_writes_on(peer); return -1; } - BGP_READ_ON(peer->t_read, bgp_read, peer->fd); + bgp_reads_on(peer); if (bgp_debug_neighbor_events(peer)) { char buf1[SU_ADDRSTRLEN]; @@ -1285,6 +1392,10 @@ int bgp_start(struct peer *peer) #endif } + assert(!peer->t_write); + assert(!peer->t_read); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON)); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_READS_ON)); status = bgp_connect(peer); switch (status) { @@ -1312,8 +1423,19 @@ int bgp_start(struct peer *peer) peer->fd); return -1; } - BGP_READ_ON(peer->t_read, bgp_read, peer->fd); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + /* + * - when the socket becomes ready, poll() will signify POLLOUT + * - if it fails to connect, poll() will signify POLLHUP + * - POLLHUP is handled as a 'read' event by thread.c + * + * therefore, we schedule both a read and a write event with + * bgp_connect_check() as the handler for each and cancel the + * unused event in that function. + */ + thread_add_read(bm->master, bgp_connect_check, peer, peer->fd, + &peer->t_connect_check_r); + thread_add_write(bm->master, bgp_connect_check, peer, peer->fd, + &peer->t_connect_check_w); break; } return 0; @@ -1340,13 +1462,6 @@ static int bgp_fsm_open(struct peer *peer) return 0; } -/* Keepalive send to peer. */ -static int bgp_fsm_keepalive_expire(struct peer *peer) -{ - bgp_keepalive_send(peer); - return 0; -} - /* FSM error, unexpected event. This is error of BGP connection. So cut the peer and change to Idle status. */ static int bgp_fsm_event_error(struct peer *peer) @@ -1367,8 +1482,12 @@ static int bgp_fsm_holdtime_expire(struct peer *peer) return bgp_stop_with_notify(peer, BGP_NOTIFY_HOLD_ERR, 0); } -/* Status goes to Established. Send keepalive packet then make first - update information. */ +/** + * Transition to Established state. + * + * Convert peer from stub to full fledged peer, set some timers, and generate + * initial updates. + */ static int bgp_establish(struct peer *peer) { afi_t afi; @@ -1458,7 +1577,10 @@ static int bgp_establish(struct peer *peer) hook_call(peer_established, peer); - /* Reset uptime, send keepalive, send current table. */ + /* Reset uptime, turn on keepalives, send current table. */ + if (!peer->v_holdtime) + bgp_keepalives_on(peer); + peer->uptime = bgp_clock(); /* Send route-refresh when ORF is enabled */ @@ -1523,11 +1645,6 @@ static int bgp_establish(struct peer *peer) /* Keepalive packet is received. */ static int bgp_fsm_keepalive(struct peer *peer) { - bgp_update_implicit_eors(peer); - - /* peer count update */ - peer->keepalive_in++; - BGP_TIMER_OFF(peer->t_holdtime); return 0; } @@ -1700,9 +1817,8 @@ static const struct { {bgp_stop, Clearing}, /* TCP_fatal_error */ {bgp_stop, Clearing}, /* ConnectRetry_timer_expired */ {bgp_fsm_holdtime_expire, Clearing}, /* Hold_Timer_expired */ - {bgp_fsm_keepalive_expire, - Established}, /* KeepAlive_timer_expired */ - {bgp_stop, Clearing}, /* Receive_OPEN_message */ + {bgp_ignore, Established}, /* KeepAlive_timer_expired */ + {bgp_stop, Clearing}, /* Receive_OPEN_message */ {bgp_fsm_keepalive, Established}, /* Receive_KEEPALIVE_message */ {bgp_fsm_update, Established}, /* Receive_UPDATE_message */ @@ -1769,6 +1885,9 @@ int bgp_event_update(struct peer *peer, int event) int passive_conn = 0; int dyn_nbr; + /* default return code */ + ret = FSM_PEER_NOOP; + other = peer->doppelganger; passive_conn = (CHECK_FLAG(peer->sflags, PEER_STATUS_ACCEPT_PEER)) ? 1 : 0; @@ -1790,37 +1909,56 @@ int bgp_event_update(struct peer *peer, int event) if (FSM[peer->status - 1][event - 1].func) ret = (*(FSM[peer->status - 1][event - 1].func))(peer); - /* When function do not want proceed next job return -1. */ if (ret >= 0) { if (ret == 1 && next == Established) { /* The case when doppelganger swap accurred in bgp_establish. Update the peer pointer accordingly */ + ret = FSM_PEER_TRANSFERRED; peer = other; } /* If status is changed. */ - if (next != peer->status) + if (next != peer->status) { bgp_fsm_change_status(peer, next); + /* + * If we're going to ESTABLISHED then we executed a + * peer transfer. In this case we can either return + * FSM_PEER_TRANSITIONED or FSM_PEER_TRANSFERRED. + * Opting for TRANSFERRED since transfer implies + * session establishment. + */ + if (ret != FSM_PEER_TRANSFERRED) + ret = FSM_PEER_TRANSITIONED; + } + /* Make sure timer is set. */ bgp_timer_set(peer); - } else if (!dyn_nbr && !passive_conn && peer->bgp) { - /* If we got a return value of -1, that means there was an - * error, restart - * the FSM. If the peer structure was deleted + } else { + /* + * If we got a return value of -1, that means there was an + * error, restart the FSM. Since bgp_stop() was called on the + * peer. only a few fields are safe to access here. In any case + * we need to indicate that the peer was stopped in the return + * code. */ - zlog_err( - "%s [FSM] Failure handling event %s in state %s, " - "prior events %s, %s, fd %d", - peer->host, bgp_event_str[peer->cur_event], - lookup_msg(bgp_status_msg, peer->status, NULL), - bgp_event_str[peer->last_event], - bgp_event_str[peer->last_major_event], peer->fd); - bgp_stop(peer); - bgp_fsm_change_status(peer, Idle); - bgp_timer_set(peer); + if (!dyn_nbr && !passive_conn && peer->bgp) { + zlog_err( + "%s [FSM] Failure handling event %s in state %s, " + "prior events %s, %s, fd %d", + peer->host, bgp_event_str[peer->cur_event], + lookup_msg(bgp_status_msg, peer->status, NULL), + bgp_event_str[peer->last_event], + bgp_event_str[peer->last_major_event], + peer->fd); + bgp_stop(peer); + bgp_fsm_change_status(peer, Idle); + bgp_timer_set(peer); + } + ret = FSM_PEER_STOPPED; } + return ret; } diff --git a/bgpd/bgp_fsm.h b/bgpd/bgp_fsm.h index 51d5d7aaa8..d021c9884a 100644 --- a/bgpd/bgp_fsm.h +++ b/bgpd/bgp_fsm.h @@ -23,36 +23,6 @@ #define _QUAGGA_BGP_FSM_H /* Macro for BGP read, write and timer thread. */ -#define BGP_READ_ON(T, F, V) \ - do { \ - if ((peer->status != Deleted)) \ - thread_add_read(bm->master, (F), peer, (V), &(T)); \ - } while (0) - -#define BGP_READ_OFF(T) \ - do { \ - if (T) \ - THREAD_READ_OFF(T); \ - } while (0) - -#define BGP_WRITE_ON(T, F, V) \ - do { \ - if ((peer)->status != Deleted) \ - thread_add_write(bm->master, (F), (peer), (V), &(T)); \ - } while (0) - -#define BGP_PEER_WRITE_ON(T, F, V, peer) \ - do { \ - if ((peer)->status != Deleted) \ - thread_add_write(bm->master, (F), (peer), (V), &(T)); \ - } while (0) - -#define BGP_WRITE_OFF(T) \ - do { \ - if (T) \ - THREAD_WRITE_OFF(T); \ - } while (0) - #define BGP_TIMER_ON(T, F, V) \ do { \ if ((peer->status != Deleted)) \ @@ -80,6 +50,12 @@ #define BGP_MSEC_JITTER 10 +/* Status codes for bgp_event_update() */ +#define FSM_PEER_NOOP 0 +#define FSM_PEER_STOPPED 1 +#define FSM_PEER_TRANSFERRED 2 +#define FSM_PEER_TRANSITIONED 3 + /* Prototypes. */ extern void bgp_fsm_nht_update(struct peer *, int valid); extern int bgp_event(struct thread *); diff --git a/bgpd/bgp_io.c b/bgpd/bgp_io.c new file mode 100644 index 0000000000..548167b3a3 --- /dev/null +++ b/bgpd/bgp_io.c @@ -0,0 +1,598 @@ +/* BGP I/O. + * Implements packet I/O in a pthread. + * Copyright (C) 2017 Cumulus Networks + * Quentin Young + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; see the file COPYING; if not, write to the + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, + * MA 02110-1301 USA + */ + +/* clang-format off */ +#include <zebra.h> +#include <pthread.h> // for pthread_mutex_unlock, pthread_mutex_lock + +#include "frr_pthread.h" // for frr_pthread_get, frr_pthread +#include "linklist.h" // for list_delete, list_delete_all_node, lis... +#include "log.h" // for zlog_debug, safe_strerror, zlog_err +#include "memory.h" // for MTYPE_TMP, XCALLOC, XFREE +#include "network.h" // for ERRNO_IO_RETRY +#include "stream.h" // for stream_get_endp, stream_getw_from, str... +#include "thread.h" // for THREAD_OFF, THREAD_ARG, thread, thread... +#include "zassert.h" // for assert + +#include "bgpd/bgp_io.h" +#include "bgpd/bgp_debug.h" // for bgp_debug_neighbor_events, bgp_type_str +#include "bgpd/bgp_fsm.h" // for BGP_EVENT_ADD, bgp_event +#include "bgpd/bgp_packet.h" // for bgp_notify_send_with_data, bgp_notify... +#include "bgpd/bgpd.h" // for peer, BGP_MARKER_SIZE, bgp_master, bm +/* clang-format on */ + +/* forward declarations */ +static uint16_t bgp_write(struct peer *); +static uint16_t bgp_read(struct peer *); +static int bgp_process_writes(struct thread *); +static int bgp_process_reads(struct thread *); +static bool validate_header(struct peer *); + +/* generic i/o status codes */ +#define BGP_IO_TRANS_ERR (1 << 0) // EAGAIN or similar occurred +#define BGP_IO_FATAL_ERR (1 << 1) // some kind of fatal TCP error + +/* Start and stop routines for I/O pthread + control variables + * ------------------------------------------------------------------------ */ +_Atomic bool bgp_io_thread_run; +_Atomic bool bgp_io_thread_started; + +void bgp_io_init() +{ + bgp_io_thread_run = false; + bgp_io_thread_started = false; +} + +/* Unused callback for thread_add_read() */ +static int bgp_io_dummy(struct thread *thread) { return 0; } + +void *bgp_io_start(void *arg) +{ + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + fpt->master->owner = pthread_self(); + + // fd so we can sleep in poll() + int sleeper[2]; + pipe(sleeper); + thread_add_read(fpt->master, &bgp_io_dummy, NULL, sleeper[0], NULL); + + // we definitely don't want to handle signals + fpt->master->handle_signals = false; + + struct thread task; + + atomic_store_explicit(&bgp_io_thread_run, true, memory_order_seq_cst); + atomic_store_explicit(&bgp_io_thread_started, true, + memory_order_seq_cst); + + while (bgp_io_thread_run) { + if (thread_fetch(fpt->master, &task)) { + thread_call(&task); + } + } + + close(sleeper[1]); + close(sleeper[0]); + + return NULL; +} + +static int bgp_io_finish(struct thread *thread) +{ + atomic_store_explicit(&bgp_io_thread_run, false, memory_order_seq_cst); + return 0; +} + +int bgp_io_stop(void **result, struct frr_pthread *fpt) +{ + thread_add_event(fpt->master, &bgp_io_finish, NULL, 0, NULL); + pthread_join(fpt->thread, result); + return 0; +} + +/* Extern API -------------------------------------------------------------- */ + +void bgp_writes_on(struct peer *peer) +{ + while ( + !atomic_load_explicit(&bgp_io_thread_started, memory_order_seq_cst)) + ; + + assert(peer->status != Deleted); + assert(peer->obuf); + assert(peer->ibuf); + assert(peer->ibuf_work); + assert(!peer->t_connect_check_r); + assert(!peer->t_connect_check_w); + assert(peer->fd); + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + thread_add_write(fpt->master, bgp_process_writes, peer, peer->fd, + &peer->t_write); + SET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON); +} + +void bgp_writes_off(struct peer *peer) +{ + while ( + !atomic_load_explicit(&bgp_io_thread_started, memory_order_seq_cst)) + ; + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + thread_cancel_async(fpt->master, &peer->t_write, NULL); + THREAD_OFF(peer->t_generate_updgrp_packets); + + UNSET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON); +} + +void bgp_reads_on(struct peer *peer) +{ + while ( + !atomic_load_explicit(&bgp_io_thread_started, memory_order_seq_cst)) + ; + + assert(peer->status != Deleted); + assert(peer->ibuf); + assert(peer->fd); + assert(peer->ibuf_work); + assert(peer->obuf); + assert(!peer->t_connect_check_r); + assert(!peer->t_connect_check_w); + assert(peer->fd); + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + thread_add_read(fpt->master, bgp_process_reads, peer, peer->fd, + &peer->t_read); + + SET_FLAG(peer->thread_flags, PEER_THREAD_READS_ON); +} + +void bgp_reads_off(struct peer *peer) +{ + while ( + !atomic_load_explicit(&bgp_io_thread_started, memory_order_seq_cst)) + ; + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + thread_cancel_async(fpt->master, &peer->t_read, NULL); + THREAD_OFF(peer->t_process_packet); + + UNSET_FLAG(peer->thread_flags, PEER_THREAD_READS_ON); +} + +/* Internal functions ------------------------------------------------------- */ + +/** + * Called from I/O pthread when a file descriptor has become ready for writing. + */ +static int bgp_process_writes(struct thread *thread) +{ + static struct peer *peer; + peer = THREAD_ARG(thread); + uint16_t status; + bool reschedule; + bool fatal = false; + + if (peer->fd < 0) + return -1; + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + pthread_mutex_lock(&peer->io_mtx); + { + status = bgp_write(peer); + reschedule = (stream_fifo_head(peer->obuf) != NULL); + } + pthread_mutex_unlock(&peer->io_mtx); + + if (CHECK_FLAG(status, BGP_IO_TRANS_ERR)) { /* no problem */ + } + + if (CHECK_FLAG(status, BGP_IO_FATAL_ERR)) { + reschedule = false; /* problem */ + fatal = true; + } + + if (reschedule) { + thread_add_write(fpt->master, bgp_process_writes, peer, + peer->fd, &peer->t_write); + } else if (!fatal) { + BGP_TIMER_ON(peer->t_generate_updgrp_packets, + bgp_generate_updgrp_packets, 0); + } + + return 0; +} + +/** + * Called from I/O pthread when a file descriptor has become ready for reading, + * or has hung up. + * + * We read as much data as possible, process as many packets as we can and + * place them on peer->ibuf for secondary processing by the main thread. + */ +static int bgp_process_reads(struct thread *thread) +{ + /* clang-format off */ + static struct peer *peer; // peer to read from + uint16_t status; // bgp_read status code + bool more = true; // whether we got more data + bool fatal = false; // whether fatal error occurred + bool added_pkt = false; // whether we pushed onto ->ibuf + bool header_valid = true; // whether header is valid + /* clang-format on */ + + peer = THREAD_ARG(thread); + + if (peer->fd < 0) + return -1; + + struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO); + + pthread_mutex_lock(&peer->io_mtx); + { + status = bgp_read(peer); + } + pthread_mutex_unlock(&peer->io_mtx); + + /* error checking phase */ + if (CHECK_FLAG(status, BGP_IO_TRANS_ERR)) { + /* no problem; just don't process packets */ + more = false; + } + + if (CHECK_FLAG(status, BGP_IO_FATAL_ERR)) { + /* problem; tear down session */ + more = false; + fatal = true; + } + + while (more) { + /* static buffer for transferring packets */ + static unsigned char pktbuf[BGP_MAX_PACKET_SIZE]; + /* shorter alias to peer's input buffer */ + struct stream *ibw = peer->ibuf_work; + /* offset of start of current packet */ + size_t offset = stream_get_getp(ibw); + /* packet size as given by header */ + u_int16_t pktsize = 0; + + /* check that we have enough data for a header */ + if (STREAM_READABLE(ibw) < BGP_HEADER_SIZE) + break; + + /* validate header */ + header_valid = validate_header(peer); + + if (!header_valid) { + fatal = true; + break; + } + + /* header is valid; retrieve packet size */ + pktsize = stream_getw_from(ibw, offset + BGP_MARKER_SIZE); + + /* if this fails we are seriously screwed */ + assert(pktsize <= BGP_MAX_PACKET_SIZE); + + /* If we have that much data, chuck it into its own + * stream and append to input queue for processing. */ + if (STREAM_READABLE(ibw) >= pktsize) { + struct stream *pkt = stream_new(pktsize); + stream_get(pktbuf, ibw, pktsize); + stream_put(pkt, pktbuf, pktsize); + + pthread_mutex_lock(&peer->io_mtx); + { + stream_fifo_push(peer->ibuf, pkt); + } + pthread_mutex_unlock(&peer->io_mtx); + + added_pkt = true; + } else + break; + } + + /* + * After reading: + * 1. Move unread data to stream start to make room for more. + * 2. Reschedule and return when we have additional data. + * + * XXX: Heavy abuse of stream API. This needs a ring buffer. + */ + if (more && STREAM_WRITEABLE(peer->ibuf_work) < BGP_MAX_PACKET_SIZE) { + void *from = stream_pnt(peer->ibuf_work); + void *to = peer->ibuf_work->data; + size_t siz = STREAM_READABLE(peer->ibuf_work); + memmove(to, from, siz); + stream_set_getp(peer->ibuf_work, 0); + stream_set_endp(peer->ibuf_work, siz); + } + + assert(STREAM_WRITEABLE(peer->ibuf_work) >= BGP_MAX_PACKET_SIZE); + + /* handle invalid header */ + if (fatal) { + /* wipe buffer just in case someone screwed up */ + stream_reset(peer->ibuf_work); + } else { + thread_add_read(fpt->master, bgp_process_reads, peer, peer->fd, + &peer->t_read); + if (added_pkt) + thread_add_timer_msec(bm->master, bgp_process_packet, + peer, 0, &peer->t_process_packet); + } + + return 0; +} + +/** + * Flush peer output buffer. + * + * This function pops packets off of peer->obuf and writes them to peer->fd. + * The amount of packets written is equal to the minimum of peer->wpkt_quanta + * and the number of packets on the output buffer, unless an error occurs. + * + * If write() returns an error, the appropriate FSM event is generated. + * + * The return value is equal to the number of packets written + * (which may be zero). + */ +static uint16_t bgp_write(struct peer *peer) +{ + u_char type; + struct stream *s; + int num; + int update_last_write = 0; + unsigned int count = 0; + uint32_t oc; + uint32_t uo; + uint16_t status = 0; + uint32_t wpkt_quanta_old; + + // save current # updates sent + oc = atomic_load_explicit(&peer->update_out, memory_order_relaxed); + + // cache current write quanta + wpkt_quanta_old = + atomic_load_explicit(&peer->bgp->wpkt_quanta, memory_order_relaxed); + + while (count < wpkt_quanta_old && (s = stream_fifo_head(peer->obuf))) { + int writenum; + do { + writenum = stream_get_endp(s) - stream_get_getp(s); + num = write(peer->fd, STREAM_PNT(s), writenum); + + if (num < 0) { + if (!ERRNO_IO_RETRY(errno)) { + BGP_EVENT_ADD(peer, TCP_fatal_error); + SET_FLAG(status, BGP_IO_FATAL_ERR); + } else { + SET_FLAG(status, BGP_IO_TRANS_ERR); + } + + goto done; + } else if (num != writenum) // incomplete write + stream_forward_getp(s, num); + + } while (num != writenum); + + /* Retrieve BGP packet type. */ + stream_set_getp(s, BGP_MARKER_SIZE + 2); + type = stream_getc(s); + + switch (type) { + case BGP_MSG_OPEN: + atomic_fetch_add_explicit(&peer->open_out, 1, + memory_order_relaxed); + break; + case BGP_MSG_UPDATE: + atomic_fetch_add_explicit(&peer->update_out, 1, + memory_order_relaxed); + break; + case BGP_MSG_NOTIFY: + atomic_fetch_add_explicit(&peer->notify_out, 1, + memory_order_relaxed); + /* Double start timer. */ + peer->v_start *= 2; + + /* Overflow check. */ + if (peer->v_start >= (60 * 2)) + peer->v_start = (60 * 2); + + /* Handle Graceful Restart case where the state changes + * to Connect instead of Idle */ + BGP_EVENT_ADD(peer, BGP_Stop); + goto done; + + case BGP_MSG_KEEPALIVE: + atomic_fetch_add_explicit(&peer->keepalive_out, 1, + memory_order_relaxed); + break; + case BGP_MSG_ROUTE_REFRESH_NEW: + case BGP_MSG_ROUTE_REFRESH_OLD: + atomic_fetch_add_explicit(&peer->refresh_out, 1, + memory_order_relaxed); + break; + case BGP_MSG_CAPABILITY: + atomic_fetch_add_explicit(&peer->dynamic_cap_out, 1, + memory_order_relaxed); + break; + } + + count++; + + stream_free(stream_fifo_pop(peer->obuf)); + update_last_write = 1; + } + +done : { + /* Update last_update if UPDATEs were written. */ + uo = atomic_load_explicit(&peer->update_out, memory_order_relaxed); + if (uo > oc) + atomic_store_explicit(&peer->last_update, bgp_clock(), + memory_order_relaxed); + + /* If we TXed any flavor of packet */ + if (update_last_write) + atomic_store_explicit(&peer->last_write, bgp_clock(), + memory_order_relaxed); +} + + return status; +} + +/** + * Reads a chunk of data from peer->fd into peer->ibuf_work. + * + * @return status flag (see top-of-file) + */ +static uint16_t bgp_read(struct peer *peer) +{ + size_t readsize; // how many bytes we want to read + ssize_t nbytes; // how many bytes we actually read + uint16_t status = 0; + + readsize = STREAM_WRITEABLE(peer->ibuf_work); + + nbytes = stream_read_try(peer->ibuf_work, peer->fd, readsize); + + switch (nbytes) { + /* Fatal error; tear down session */ + case -1: + zlog_err("%s [Error] bgp_read_packet error: %s", peer->host, + safe_strerror(errno)); + + if (peer->status == Established) { + if (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_ADD(peer, TCP_fatal_error); + SET_FLAG(status, BGP_IO_FATAL_ERR); + break; + + /* Received EOF / TCP session closed */ + case 0: + 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->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_ADD(peer, TCP_connection_closed); + SET_FLAG(status, BGP_IO_FATAL_ERR); + break; + + /* EAGAIN or EWOULDBLOCK; come back later */ + case -2: + SET_FLAG(status, BGP_IO_TRANS_ERR); + break; + default: + break; + } + + return status; +} + +/* + * Called after we have read a BGP packet header. Validates marker, message + * type and packet length. If any of these aren't correct, sends a notify. + */ +static bool validate_header(struct peer *peer) +{ + uint16_t size; + uint8_t type; + struct stream *pkt = peer->ibuf_work; + size_t getp = stream_get_getp(pkt); + + static uint8_t marker[BGP_MARKER_SIZE] = { + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}; + + if (memcmp(marker, stream_pnt(pkt), BGP_MARKER_SIZE) != 0) { + bgp_notify_send(peer, BGP_NOTIFY_HEADER_ERR, + BGP_NOTIFY_HEADER_NOT_SYNC); + return false; + } + + /* Get size and type in host byte order. */ + size = stream_getw_from(pkt, getp + BGP_MARKER_SIZE); + type = stream_getc_from(pkt, getp + BGP_MARKER_SIZE + 2); + + /* BGP type check. */ + if (type != BGP_MSG_OPEN && type != BGP_MSG_UPDATE + && type != BGP_MSG_NOTIFY && type != BGP_MSG_KEEPALIVE + && type != BGP_MSG_ROUTE_REFRESH_NEW + && type != BGP_MSG_ROUTE_REFRESH_OLD + && type != BGP_MSG_CAPABILITY) { + if (bgp_debug_neighbor_events(peer)) + zlog_debug("%s unknown message type 0x%02x", peer->host, + type); + + bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR, + BGP_NOTIFY_HEADER_BAD_MESTYPE, + &type, 1); + return false; + } + + /* Minimum packet length check. */ + if ((size < BGP_HEADER_SIZE) || (size > BGP_MAX_PACKET_SIZE) + || (type == BGP_MSG_OPEN && size < BGP_MSG_OPEN_MIN_SIZE) + || (type == BGP_MSG_UPDATE && size < BGP_MSG_UPDATE_MIN_SIZE) + || (type == BGP_MSG_NOTIFY && size < BGP_MSG_NOTIFY_MIN_SIZE) + || (type == BGP_MSG_KEEPALIVE && size != BGP_MSG_KEEPALIVE_MIN_SIZE) + || (type == BGP_MSG_ROUTE_REFRESH_NEW + && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE) + || (type == BGP_MSG_ROUTE_REFRESH_OLD + && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE) + || (type == BGP_MSG_CAPABILITY + && size < BGP_MSG_CAPABILITY_MIN_SIZE)) { + if (bgp_debug_neighbor_events(peer)) { + zlog_debug("%s bad message length - %d for %s", + peer->host, size, + type == 128 ? "ROUTE-REFRESH" + : bgp_type_str[(int) type]); + } + + uint16_t nsize = htons(size); + + bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR, + BGP_NOTIFY_HEADER_BAD_MESLEN, + (unsigned char *) &nsize, 2); + return false; + } + + return true; +} diff --git a/bgpd/bgp_io.h b/bgpd/bgp_io.h new file mode 100644 index 0000000000..c4bd3c2dd9 --- /dev/null +++ b/bgpd/bgp_io.h @@ -0,0 +1,111 @@ +/* BGP I/O. + * Implements packet I/O in a pthread. + * Copyright (C) 2017 Cumulus Networks + * Quentin Young + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; see the file COPYING; if not, write to the + * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, + * MA 02110-1301 USA + */ + +#ifndef _FRR_BGP_IO_H +#define _FRR_BGP_IO_H + +#define BGP_WRITE_PACKET_MAX 10U +#define BGP_READ_PACKET_MAX 10U + +#include "bgpd/bgpd.h" +#include "frr_pthread.h" + +/** + * Initializes data structures and flags for the write thread. + * + * This function should be called from the main thread before + * bgp_writes_start() is invoked. + */ +extern void bgp_io_init(void); + +/** + * Start function for write thread. + * + * @param arg - unused + */ +extern void *bgp_io_start(void *arg); + +/** + * Start function for write thread. + * + * Uninitializes all resources and stops the thread. + * + * @param result - where to store data result, unused + */ +extern int bgp_io_stop(void **result, struct frr_pthread *fpt); + +/** + * Turns on packet writing for a peer. + * + * After this function is called, any packets placed on peer->obuf will be + * written to peer->fd until no more packets remain. + * + * Additionally, it becomes unsafe to perform socket actions on peer->fd. + * + * @param peer - peer to register + */ +extern void bgp_writes_on(struct peer *peer); + +/** + * Turns off packet writing for a peer. + * + * After this function returns, packets placed on peer->obuf will not be + * written to peer->fd by the I/O thread. + * + * After this function returns it becomes safe to perform socket actions on + * peer->fd. + * + * @param peer - peer to deregister + * @param flush - as described + */ +extern void bgp_writes_off(struct peer *peer); + +/** + * Turns on packet reading for a peer. + * + * After this function is called, any packets received on peer->fd will be read + * and copied into the FIFO queue peer->ibuf. + * + * Additionally, it becomes unsafe to perform socket actions on peer->fd. + * + * Whenever one or more packets are placed onto peer->ibuf, a task of type + * THREAD_EVENT will be placed on the main thread whose handler is + * + * bgp_packet.c:bgp_process_packet() + * + * @param peer - peer to register + */ +extern void bgp_reads_on(struct peer *peer); + +/** + * Turns off packet reading for a peer. + * + * After this function is called, any packets received on peer->fd will not be + * read by the I/O thread. + * + * After this function returns it becomes safe to perform socket actions on + * peer->fd. + * + * @param peer - peer to deregister + */ +extern void bgp_reads_off(struct peer *peer); + +#endif /* _FRR_BGP_IO_H */ diff --git a/bgpd/bgp_keepalives.c b/bgpd/bgp_keepalives.c new file mode 100644 index 0000000000..afa280a799 --- /dev/null +++ b/bgpd/bgp_keepalives.c @@ -0,0 +1,292 @@ +/* BGP Keepalives. + * Implements a producer thread to generate BGP keepalives for peers. + * Copyright (C) 2017 Cumulus Networks, Inc. + * Quentin Young + * + * This file is part of FRRouting. + * + * FRRouting is free software; you can redistribute it and/or modify it under + * the terms of the GNU General Public License as published by the Free + * Software Foundation; either version 2, or (at your option) any later + * version. + * + * FRRouting is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more + * details. + * + * You should have received a copy of the GNU General Public License along + * with this program; see the file COPYING; if not, write to the Free Software + * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +/* clang-format off */ +#include <zebra.h> +#include <pthread.h> // for pthread_mutex_lock, pthread_mutex_unlock + +#include "frr_pthread.h" // for frr_pthread +#include "hash.h" // for hash, hash_clean, hash_create_size... +#include "log.h" // for zlog_debug +#include "memory.h" // for MTYPE_TMP, XFREE, XCALLOC, XMALLOC +#include "monotime.h" // for monotime, monotime_since + +#include "bgpd/bgpd.h" // for peer, PEER_THREAD_KEEPALIVES_ON, peer... +#include "bgpd/bgp_debug.h" // for bgp_debug_neighbor_events +#include "bgpd/bgp_packet.h" // for bgp_keepalive_send +#include "bgpd/bgp_keepalives.h" +/* clang-format on */ + +/** + * Peer KeepAlive Timer. + * Associates a peer with the time of its last keepalive. + */ +struct pkat { + // the peer to send keepalives to + struct peer *peer; + // absolute time of last keepalive sent + struct timeval last; +}; + +/* List of peers we are sending keepalives for, and associated mutex. */ +static pthread_mutex_t *peerhash_mtx; +static pthread_cond_t *peerhash_cond; +static struct hash *peerhash; + +/* Thread control flag. */ +bool bgp_keepalives_thread_run = false; + +static struct pkat *pkat_new(struct peer *peer) +{ + struct pkat *pkat = XMALLOC(MTYPE_TMP, sizeof(struct pkat)); + pkat->peer = peer; + monotime(&pkat->last); + return pkat; +} + +static void pkat_del(void *pkat) +{ + XFREE(MTYPE_TMP, pkat); +} + + +/* + * Callback for hash_iterate. Determines if a peer needs a keepalive and if so, + * generates and sends it. + * + * For any given peer, if the elapsed time since its last keepalive exceeds its + * configured keepalive timer, a keepalive is sent to the peer and its + * last-sent time is reset. Additionally, If the elapsed time does not exceed + * the configured keepalive timer, but the time until the next keepalive is due + * is within a hardcoded tolerance, a keepalive is sent as if the configured + * timer was exceeded. Doing this helps alleviate nanosecond sleeps between + * ticks by grouping together peers who are due for keepalives at roughly the + * same time. This tolerance value is arbitrarily chosen to be 100ms. + * + * In addition, this function calculates the maximum amount of time that the + * keepalive thread can sleep before another tick needs to take place. This is + * equivalent to shortest time until a keepalive is due for any one peer. + * + * @return maximum time to wait until next update (0 if infinity) + */ +static void peer_process(struct hash_backet *hb, void *arg) +{ + struct pkat *pkat = hb->data; + + struct timeval *next_update = arg; + + static struct timeval elapsed; // elapsed time since keepalive + static struct timeval ka = {0}; // peer->v_keepalive as a timeval + static struct timeval diff; // ka - elapsed + + static struct timeval tolerance = {0, 100000}; + + // calculate elapsed time since last keepalive + monotime_since(&pkat->last, &elapsed); + + // calculate difference between elapsed time and configured time + ka.tv_sec = pkat->peer->v_keepalive; + timersub(&ka, &elapsed, &diff); + + int send_keepalive = + elapsed.tv_sec >= ka.tv_sec || timercmp(&diff, &tolerance, <); + + if (send_keepalive) { + if (bgp_debug_neighbor_events(pkat->peer)) + zlog_debug("%s [FSM] Timer (keepalive timer expire)", + pkat->peer->host); + + bgp_keepalive_send(pkat->peer); + monotime(&pkat->last); + memset(&elapsed, 0x00, sizeof(struct timeval)); + diff = ka; // time until next keepalive == peer keepalive time + } + + // if calculated next update for this peer < current delay, use it + if (next_update->tv_sec <= 0 || timercmp(&diff, next_update, <)) + *next_update = diff; +} + +static int peer_hash_cmp(const void *f, const void *s) +{ + const struct pkat *p1 = f; + const struct pkat *p2 = s; + return p1->peer == p2->peer; +} + +static unsigned int peer_hash_key(void *arg) +{ + struct pkat *pkat = arg; + return (uintptr_t)pkat->peer; +} + +void bgp_keepalives_init() +{ + peerhash_mtx = XCALLOC(MTYPE_TMP, sizeof(pthread_mutex_t)); + peerhash_cond = XCALLOC(MTYPE_TMP, sizeof(pthread_cond_t)); + + // initialize mutex + pthread_mutex_init(peerhash_mtx, NULL); + + // use monotonic clock with condition variable + pthread_condattr_t attrs; + pthread_condattr_init(&attrs); + pthread_condattr_setclock(&attrs, CLOCK_MONOTONIC); + pthread_cond_init(peerhash_cond, &attrs); + pthread_condattr_destroy(&attrs); + + // initialize peer hashtable + peerhash = hash_create_size(2048, peer_hash_key, peer_hash_cmp, NULL); +} + +static void bgp_keepalives_finish(void *arg) +{ + bgp_keepalives_thread_run = false; + + if (peerhash) { + hash_clean(peerhash, pkat_del); + hash_free(peerhash); + } + + peerhash = NULL; + + pthread_mutex_unlock(peerhash_mtx); + pthread_mutex_destroy(peerhash_mtx); + pthread_cond_destroy(peerhash_cond); + + XFREE(MTYPE_TMP, peerhash_mtx); + XFREE(MTYPE_TMP, peerhash_cond); +} + +/** + * Entry function for peer keepalive generation pthread. + * + * bgp_keepalives_init() must be called prior to this. + */ +void *bgp_keepalives_start(void *arg) +{ + struct timeval currtime = {0, 0}; + struct timeval aftertime = {0, 0}; + struct timeval next_update = {0, 0}; + struct timespec next_update_ts = {0, 0}; + + pthread_mutex_lock(peerhash_mtx); + + // register cleanup handler + pthread_cleanup_push(&bgp_keepalives_finish, NULL); + + bgp_keepalives_thread_run = true; + + while (bgp_keepalives_thread_run) { + if (peerhash->count > 0) + pthread_cond_timedwait(peerhash_cond, peerhash_mtx, + &next_update_ts); + else + while (peerhash->count == 0 + && bgp_keepalives_thread_run) + pthread_cond_wait(peerhash_cond, peerhash_mtx); + + monotime(&currtime); + + next_update.tv_sec = -1; + + hash_iterate(peerhash, peer_process, &next_update); + if (next_update.tv_sec == -1) + memset(&next_update, 0x00, sizeof(next_update)); + + monotime_since(&currtime, &aftertime); + + timeradd(&currtime, &next_update, &next_update); + TIMEVAL_TO_TIMESPEC(&next_update, &next_update_ts); + } + + // clean up + pthread_cleanup_pop(1); + + return NULL; +} + +/* --- thread external functions ------------------------------------------- */ + +void bgp_keepalives_on(struct peer *peer) +{ + /* placeholder bucket data to use for fast key lookups */ + static struct pkat holder = {0}; + + if (!peerhash_mtx) { + zlog_warn("%s: call bgp_keepalives_init() first", __func__); + return; + } + + pthread_mutex_lock(peerhash_mtx); + { + holder.peer = peer; + if (!hash_lookup(peerhash, &holder)) { + struct pkat *pkat = pkat_new(peer); + hash_get(peerhash, pkat, hash_alloc_intern); + peer_lock(peer); + } + SET_FLAG(peer->thread_flags, PEER_THREAD_KEEPALIVES_ON); + } + pthread_mutex_unlock(peerhash_mtx); + bgp_keepalives_wake(); +} + +void bgp_keepalives_off(struct peer *peer) +{ + /* placeholder bucket data to use for fast key lookups */ + static struct pkat holder = {0}; + + if (!peerhash_mtx) { + zlog_warn("%s: call bgp_keepalives_init() first", __func__); + return; + } + + pthread_mutex_lock(peerhash_mtx); + { + holder.peer = peer; + struct pkat *res = hash_release(peerhash, &holder); + if (res) { + pkat_del(res); + peer_unlock(peer); + } + UNSET_FLAG(peer->thread_flags, PEER_THREAD_KEEPALIVES_ON); + } + pthread_mutex_unlock(peerhash_mtx); +} + +void bgp_keepalives_wake() +{ + pthread_mutex_lock(peerhash_mtx); + { + pthread_cond_signal(peerhash_cond); + } + pthread_mutex_unlock(peerhash_mtx); +} + +int bgp_keepalives_stop(void **result, struct frr_pthread *fpt) +{ + bgp_keepalives_thread_run = false; + bgp_keepalives_wake(); + pthread_join(fpt->thread, result); + return 0; +} diff --git a/bgpd/bgp_keepalives.h b/bgpd/bgp_keepalives.h new file mode 100644 index 0000000000..1fbd035b9e --- /dev/null +++ b/bgpd/bgp_keepalives.h @@ -0,0 +1,93 @@ +/* BGP Keepalives. + * Implements a producer thread to generate BGP keepalives for peers. + * Copyright (C) 2017 Cumulus Networks, Inc. + * Quentin Young + * + * This file is part of FRRouting. + * + * FRRouting is free software; you can redistribute it and/or modify it under + * the terms of the GNU General Public License as published by the Free + * Software Foundation; either version 2, or (at your option) any later + * version. + * + * FRRouting is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more + * details. + * + * You should have received a copy of the GNU General Public License along + * with this program; see the file COPYING; if not, write to the Free Software + * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef _FRR_BGP_KEEPALIVES_H +#define _FRR_BGP_KEEPALIVES_H + +#include "frr_pthread.h" +#include "bgpd.h" + +/** + * Turns on keepalives for a peer. + * + * This function adds the peer to an internal list of peers to generate + * keepalives for. + * + * At set intervals, a BGP KEEPALIVE packet is generated and placed on + * peer->obuf. This operation is thread-safe with respect to peer->obuf. + * + * peer->v_keepalive determines the interval. Changing this value before + * unregistering this peer with bgp_keepalives_off() results in undefined + * behavior. + * + * If the peer is already registered for keepalives via this function, nothing + * happens. + */ +extern void bgp_keepalives_on(struct peer *); + +/** + * Turns off keepalives for a peer. + * + * Removes the peer from the internal list of peers to generate keepalives for. + * + * If the peer is already unregistered for keepalives, nothing happens. + */ +extern void bgp_keepalives_off(struct peer *); + +/** + * Pre-run initialization function for keepalives pthread. + * + * Initializes synchronization primitives. This should be called before + * anything else to avoid race conditions. + */ +extern void bgp_keepalives_init(void); + +/** + * Entry function for keepalives pthread. + * + * This function loops over an internal list of peers, generating keepalives at + * regular intervals as determined by each peer's keepalive timer. + * + * See bgp_keepalives_on() for additional details. + * + * @param arg pthread arg, not used + */ +extern void *bgp_keepalives_start(void *arg); + +/** + * Poking function for keepalives pthread. + * + * Under normal circumstances the pthread will automatically wake itself + * whenever it is necessary to do work. This function may be used to force the + * thread to wake up and see if there is any work to do, or if it is time to + * die. + * + * It is not necessary to call this after bgp_keepalives_on(). + */ +extern void bgp_keepalives_wake(void); + +/** + * Stops the thread and blocks until it terminates. + */ +int bgp_keepalives_stop(void **result, struct frr_pthread *fpt); + +#endif /* _FRR_BGP_KEEPALIVES_H */ diff --git a/bgpd/bgp_main.c b/bgpd/bgp_main.c index 1fac2936eb..7dd4253b2e 100644 --- a/bgpd/bgp_main.c +++ b/bgpd/bgp_main.c @@ -20,6 +20,7 @@ #include <zebra.h> +#include <pthread.h> #include "vector.h" #include "command.h" #include "getopt.h" @@ -54,6 +55,8 @@ #include "bgpd/bgp_debug.h" #include "bgpd/bgp_filter.h" #include "bgpd/bgp_zebra.h" +#include "bgpd/bgp_packet.h" +#include "bgpd/bgp_keepalives.h" #ifdef ENABLE_BGP_VNC #include "bgpd/rfapi/rfapi_backend.h" @@ -191,6 +194,9 @@ static __attribute__((__noreturn__)) void bgp_exit(int status) /* reverse bgp_attr_init */ bgp_attr_finish(); + /* stop pthreads */ + bgp_pthreads_finish(); + /* reverse access_list_init */ access_list_add_hook(NULL); access_list_delete_hook(NULL); @@ -393,6 +399,8 @@ int main(int argc, char **argv) (bm->address ? bm->address : "<all>"), bm->port); frr_config_fork(); + /* must be called after fork() */ + bgp_pthreads_run(); frr_run(bm->master); /* Not reached. */ diff --git a/bgpd/bgp_network.c b/bgpd/bgp_network.c index 0d7680ea51..bf39cbe1fc 100644 --- a/bgpd/bgp_network.c +++ b/bgpd/bgp_network.c @@ -550,6 +550,8 @@ static int bgp_update_source(struct peer *peer) /* BGP try to connect to the peer. */ int bgp_connect(struct peer *peer) { + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON)); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_READS_ON)); ifindex_t ifindex = 0; if (peer->conf_if && BGP_PEER_SU_UNSPEC(peer)) { diff --git a/bgpd/bgp_packet.c b/bgpd/bgp_packet.c index a955b3512c..4b018aef4d 100644 --- a/bgpd/bgp_packet.c +++ b/bgpd/bgp_packet.c @@ -1,4 +1,6 @@ /* BGP packet management routine. + * Contains utility functions for constructing and consuming BGP messages. + * Copyright (C) 2017 Cumulus Networks * Copyright (C) 1999 Kunihiro Ishiguro * * This file is part of GNU Zebra. @@ -19,6 +21,7 @@ */ #include <zebra.h> +#include <sys/time.h> #include "thread.h" #include "stream.h" @@ -54,8 +57,16 @@ #include "bgpd/bgp_vty.h" #include "bgpd/bgp_updgrp.h" #include "bgpd/bgp_label.h" +#include "bgpd/bgp_io.h" +#include "bgpd/bgp_keepalives.h" -/* Set up BGP packet marker and packet type. */ +/** + * Sets marker and type fields for a BGP message. + * + * @param s the stream containing the packet + * @param type the packet type + * @return the size of the stream + */ int bgp_packet_set_marker(struct stream *s, u_char type) { int i; @@ -74,8 +85,14 @@ int bgp_packet_set_marker(struct stream *s, u_char type) return stream_get_endp(s); } -/* Set BGP packet header size entry. If size is zero then use current - stream size. */ +/** + * Sets size field for a BGP message. + * + * Size field is set to the size of the stream passed. + * + * @param s the stream containing the packet + * @return the size of the stream + */ int bgp_packet_set_size(struct stream *s) { int cp; @@ -87,54 +104,15 @@ int bgp_packet_set_size(struct stream *s) return cp; } -/* Add new packet to the peer. */ -void bgp_packet_add(struct peer *peer, struct stream *s) +/* + * Push a packet onto the beginning of the peer's output queue. + * This function acquires the peer's write mutex before proceeding. + */ +static void bgp_packet_add(struct peer *peer, struct stream *s) { - /* Add packet to the end of list. */ + pthread_mutex_lock(&peer->io_mtx); stream_fifo_push(peer->obuf, s); -} - -/* Free first packet. */ -static void bgp_packet_delete(struct peer *peer) -{ - stream_free(stream_fifo_pop(peer->obuf)); -} - -/* Check file descriptor whether connect is established. */ -int bgp_connect_check(struct peer *peer, int change_state) -{ - int status; - socklen_t slen; - int ret; - - /* Anyway I have to reset read and write thread. */ - BGP_READ_OFF(peer->t_read); - BGP_WRITE_OFF(peer->t_write); - - /* Check file descriptor. */ - slen = sizeof(status); - ret = getsockopt(peer->fd, SOL_SOCKET, SO_ERROR, (void *)&status, - &slen); - - /* If getsockopt is fail, this is fatal error. */ - if (ret < 0) { - zlog_info("can't get sockopt for nonblocking connect"); - BGP_EVENT_ADD(peer, TCP_fatal_error); - return -1; - } - - /* When status is 0 then TCP connection is established. */ - if (status == 0) { - BGP_EVENT_ADD(peer, TCP_connection_open); - return 1; - } else { - if (bgp_debug_neighbor_events(peer)) - zlog_debug("%s [Event] Connect failed (%s)", peer->host, - safe_strerror(errno)); - if (change_state) - BGP_EVENT_ADD(peer, TCP_connection_open_failed); - return 0; - } + pthread_mutex_unlock(&peer->io_mtx); } static struct stream *bgp_update_packet_eor(struct peer *peer, afi_t afi, @@ -176,106 +154,172 @@ static struct stream *bgp_update_packet_eor(struct peer *peer, afi_t afi, } bgp_packet_set_size(s); - bgp_packet_add(peer, s); return s; } -/* Get next packet to be written. */ -static struct stream *bgp_write_packet(struct peer *peer) +/* Called when there is a change in the EOR(implicit or explicit) status of a + * peer. Ends the update-delay if all expected peers are done with EORs. */ +void bgp_check_update_delay(struct bgp *bgp) { - struct stream *s = NULL; - struct peer_af *paf; - struct bpacket *next_pkt; - afi_t afi; - safi_t safi; + struct listnode *node, *nnode; + struct peer *peer = NULL; - s = stream_fifo_head(peer->obuf); - if (s) - return s; + if (bgp_debug_neighbor_events(peer)) + zlog_debug("Checking update delay, T: %d R: %d I:%d E: %d", + bgp->established, bgp->restarted_peers, + bgp->implicit_eors, bgp->explicit_eors); - /* - * The code beyond this part deals with update packets, proceed only - * if peer is Established and updates are not on hold (as part of - * update-delay post processing). - */ - if (peer->status != Established) - return NULL; + if (bgp->established + <= bgp->restarted_peers + bgp->implicit_eors + bgp->explicit_eors) { + /* + * This is an extra sanity check to make sure we wait for all + * the eligible configured peers. This check is performed if + * establish wait timer is on, or establish wait option is not + * given with the update-delay command + */ + if (bgp->t_establish_wait + || (bgp->v_establish_wait == bgp->v_update_delay)) + for (ALL_LIST_ELEMENTS(bgp->peer, node, nnode, peer)) { + if (CHECK_FLAG(peer->flags, + PEER_FLAG_CONFIG_NODE) + && !CHECK_FLAG(peer->flags, + PEER_FLAG_SHUTDOWN) + && !peer->update_delay_over) { + if (bgp_debug_neighbor_events(peer)) + zlog_debug( + " Peer %s pending, continuing read-only mode", + peer->host); + return; + } + } - if (peer->bgp && peer->bgp->main_peers_update_hold) - return NULL; + zlog_info( + "Update delay ended, restarted: %d, EORs implicit: %d, explicit: %d", + bgp->restarted_peers, bgp->implicit_eors, + bgp->explicit_eors); + bgp_update_delay_end(bgp); + } +} - FOREACH_AFI_SAFI (afi, safi) { - paf = peer_af_find(peer, afi, safi); - if (!paf || !PAF_SUBGRP(paf)) - continue; - next_pkt = paf->next_pkt_to_send; +/* + * Called if peer is known to have restarted. The restart-state bit in + * Graceful-Restart capability is used for that + */ +void bgp_update_restarted_peers(struct peer *peer) +{ + if (!bgp_update_delay_active(peer->bgp)) + return; /* BGP update delay has ended */ + if (peer->update_delay_over) + return; /* This peer has already been considered */ - /* Try to generate a packet for the peer if we are at - * the end of - * the list. Always try to push out WITHDRAWs first. */ - if (!next_pkt || !next_pkt->buffer) { - next_pkt = subgroup_withdraw_packet(PAF_SUBGRP(paf)); - if (!next_pkt || !next_pkt->buffer) - subgroup_update_packet(PAF_SUBGRP(paf)); - next_pkt = paf->next_pkt_to_send; - } + if (bgp_debug_neighbor_events(peer)) + zlog_debug("Peer %s: Checking restarted", peer->host); - /* If we still don't have a packet to send to the peer, - * then - * try to find out out if we have to send eor or if not, - * skip to - * the next AFI, SAFI. - * Don't send the EOR prematurely... if the subgroup's - * coalesce - * timer is running, the adjacency-out structure is not - * created - * yet. - */ - if (!next_pkt || !next_pkt->buffer) { - if (CHECK_FLAG(peer->cap, PEER_CAP_RESTART_RCV)) { - if (!(PAF_SUBGRP(paf))->t_coalesce - && peer->afc_nego[afi][safi] - && peer->synctime - && !CHECK_FLAG(peer->af_sflags[afi][safi], - PEER_STATUS_EOR_SEND)) { - SET_FLAG(peer->af_sflags[afi][safi], - PEER_STATUS_EOR_SEND); - return bgp_update_packet_eor(peer, afi, - safi); - } + if (peer->status == Established) { + peer->update_delay_over = 1; + peer->bgp->restarted_peers++; + bgp_check_update_delay(peer->bgp); + } +} + +/* + * Called as peer receives a keep-alive. Determines if this occurence can be + * taken as an implicit EOR for this peer. + * NOTE: The very first keep-alive after the Established state of a peer is + * considered implicit EOR for the update-delay purposes + */ +void bgp_update_implicit_eors(struct peer *peer) +{ + if (!bgp_update_delay_active(peer->bgp)) + return; /* BGP update delay has ended */ + if (peer->update_delay_over) + return; /* This peer has already been considered */ + + if (bgp_debug_neighbor_events(peer)) + zlog_debug("Peer %s: Checking implicit EORs", peer->host); + + if (peer->status == Established) { + peer->update_delay_over = 1; + peer->bgp->implicit_eors++; + bgp_check_update_delay(peer->bgp); + } +} + +/* + * Should be called only when there is a change in the EOR_RECEIVED status + * for any afi/safi on a peer. + */ +static void bgp_update_explicit_eors(struct peer *peer) +{ + afi_t afi; + safi_t safi; + + if (!bgp_update_delay_active(peer->bgp)) + return; /* BGP update delay has ended */ + if (peer->update_delay_over) + return; /* This peer has already been considered */ + + if (bgp_debug_neighbor_events(peer)) + zlog_debug("Peer %s: Checking explicit EORs", peer->host); + + for (afi = AFI_IP; afi < AFI_MAX; afi++) + for (safi = SAFI_UNICAST; safi < SAFI_MAX; safi++) { + if (peer->afc_nego[afi][safi] + && !CHECK_FLAG(peer->af_sflags[afi][safi], + PEER_STATUS_EOR_RECEIVED)) { + if (bgp_debug_neighbor_events(peer)) + zlog_debug( + " afi %d safi %d didnt receive EOR", + afi, safi); + return; } - continue; } + peer->update_delay_over = 1; + peer->bgp->explicit_eors++; + bgp_check_update_delay(peer->bgp); +} - /* - * Found a packet template to send, overwrite packet - * with appropriate - * attributes from peer and advance peer - */ - s = bpacket_reformat_for_peer(next_pkt, paf); - bpacket_queue_advance_peer(paf); - return s; +/** + * Frontend for NLRI parsing, to fan-out to AFI/SAFI specific parsers. + * + * mp_withdraw, if set, is used to nullify attr structure on most of the + * calling safi function and for evpn, passed as parameter + */ +int bgp_nlri_parse(struct peer *peer, struct attr *attr, + struct bgp_nlri *packet, int mp_withdraw) +{ + switch (packet->safi) { + case SAFI_UNICAST: + case SAFI_MULTICAST: + return bgp_nlri_parse_ip(peer, mp_withdraw ? NULL : attr, + packet); + case SAFI_LABELED_UNICAST: + return bgp_nlri_parse_label(peer, mp_withdraw ? NULL : attr, + packet); + case SAFI_MPLS_VPN: + return bgp_nlri_parse_vpn(peer, mp_withdraw ? NULL : attr, + packet); + case SAFI_EVPN: + return bgp_nlri_parse_evpn(peer, attr, packet, mp_withdraw); } - - return NULL; + return -1; } -/* The next action for the peer from a write perspective */ +/* + * Checks a variety of conditions to determine whether the peer needs to be + * rescheduled for packet generation again, and does so if necessary. + * + * @param peer to check for rescheduling + */ static void bgp_write_proceed_actions(struct peer *peer) { afi_t afi; safi_t safi; struct peer_af *paf; struct bpacket *next_pkt; - int fullq_found = 0; struct update_subgroup *subgrp; - if (stream_fifo_head(peer->obuf)) { - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); - return; - } - FOREACH_AFI_SAFI (afi, safi) { paf = peer_af_find(peer, afi, safi); if (!paf) @@ -286,7 +330,8 @@ static void bgp_write_proceed_actions(struct peer *peer) next_pkt = paf->next_pkt_to_send; if (next_pkt && next_pkt->buffer) { - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + BGP_TIMER_ON(peer->t_generate_updgrp_packets, + bgp_generate_updgrp_packets, 0); return; } @@ -294,10 +339,10 @@ static void bgp_write_proceed_actions(struct peer *peer) * subgroup packets * that need to be generated? */ if (bpacket_queue_is_full(SUBGRP_INST(subgrp), - SUBGRP_PKTQ(subgrp))) - fullq_found = 1; - else if (subgroup_packets_to_build(subgrp)) { - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + SUBGRP_PKTQ(subgrp)) + || subgroup_packets_to_build(subgrp)) { + BGP_TIMER_ON(peer->t_generate_updgrp_packets, + bgp_generate_updgrp_packets, 0); return; } @@ -308,186 +353,119 @@ static void bgp_write_proceed_actions(struct peer *peer) && !CHECK_FLAG(peer->af_sflags[afi][safi], PEER_STATUS_EOR_SEND) && safi != SAFI_MPLS_VPN) { - BGP_WRITE_ON(peer->t_write, bgp_write, - peer->fd); + BGP_TIMER_ON(peer->t_generate_updgrp_packets, + bgp_generate_updgrp_packets, 0); return; } } } - if (fullq_found) { - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); - return; - } } -/* Write packet to the peer. */ -int bgp_write(struct thread *thread) +/* + * Generate advertisement information (withdraws, updates, EOR) from each + * update group a peer belongs to, encode this information into packets, and + * enqueue the packets onto the peer's output buffer. + */ +int bgp_generate_updgrp_packets(struct thread *thread) { - struct peer *peer; - u_char type; + struct peer *peer = THREAD_ARG(thread); + struct stream *s; - int num; - int update_last_write = 0; - unsigned int count = 0; - unsigned int oc = 0; + struct peer_af *paf; + struct bpacket *next_pkt; + uint32_t wpq; + uint32_t generated = 0; + afi_t afi; + safi_t safi; - /* Yes first of all get peer pointer. */ - peer = THREAD_ARG(thread); - peer->t_write = NULL; + wpq = atomic_load_explicit(&peer->bgp->wpkt_quanta, + memory_order_relaxed); - /* For non-blocking IO check. */ - if (peer->status == Connect) { - bgp_connect_check(peer, 1); + /* + * The code beyond this part deals with update packets, proceed only + * if peer is Established and updates are not on hold (as part of + * update-delay post processing). + */ + if (peer->status != Established) return 0; - } - s = bgp_write_packet(peer); - if (!s) { - bgp_write_proceed_actions(peer); + if (peer->bgp && peer->bgp->main_peers_update_hold) return 0; - } - sockopt_cork(peer->fd, 1); - - oc = peer->update_out; - - /* Nonblocking write until TCP output buffer is full. */ do { - int writenum; + s = NULL; + FOREACH_AFI_SAFI (afi, safi) { + paf = peer_af_find(peer, afi, safi); + if (!paf || !PAF_SUBGRP(paf)) + continue; + next_pkt = paf->next_pkt_to_send; - /* Number of bytes to be sent. */ - writenum = stream_get_endp(s) - stream_get_getp(s); + /* + * Try to generate a packet for the peer if we are at + * the end of the list. Always try to push out + * WITHDRAWs first. + */ + if (!next_pkt || !next_pkt->buffer) { + next_pkt = subgroup_withdraw_packet( + PAF_SUBGRP(paf)); + if (!next_pkt || !next_pkt->buffer) + subgroup_update_packet(PAF_SUBGRP(paf)); + next_pkt = paf->next_pkt_to_send; + } - /* Call write() system call. */ - num = write(peer->fd, STREAM_PNT(s), writenum); - if (num < 0) { - /* write failed either retry needed or error */ - if (ERRNO_IO_RETRY(errno)) - break; + /* + * If we still don't have a packet to send to the peer, + * then try to find out out if we have to send eor or + * if not, skip to the next AFI, SAFI. Don't send the + * EOR prematurely; if the subgroup's coalesce timer is + * running, the adjacency-out structure is not created + * yet. + */ + if (!next_pkt || !next_pkt->buffer) { + if (CHECK_FLAG(peer->cap, + PEER_CAP_RESTART_RCV)) { + if (!(PAF_SUBGRP(paf))->t_coalesce + && peer->afc_nego[afi][safi] + && peer->synctime + && !CHECK_FLAG( + peer->af_sflags[afi] + [safi], + PEER_STATUS_EOR_SEND)) { + SET_FLAG(peer->af_sflags[afi] + [safi], + PEER_STATUS_EOR_SEND); + + if ((s = bgp_update_packet_eor( + peer, afi, + safi))) { + bgp_packet_add(peer, s); + } + } + } + continue; + } - BGP_EVENT_ADD(peer, TCP_fatal_error); - return 0; - } - if (num != writenum) { - /* Partial write */ - stream_forward_getp(s, num); - break; + /* Found a packet template to send, overwrite + * packet with appropriate attributes from peer + * and advance peer */ + s = bpacket_reformat_for_peer(next_pkt, paf); + bgp_packet_add(peer, s); + bpacket_queue_advance_peer(paf); } + } while (s && (++generated < wpq)); - /* Retrieve BGP packet type. */ - stream_set_getp(s, BGP_MARKER_SIZE + 2); - type = stream_getc(s); - - switch (type) { - case BGP_MSG_OPEN: - peer->open_out++; - break; - case BGP_MSG_UPDATE: - peer->update_out++; - break; - case BGP_MSG_NOTIFY: - peer->notify_out++; - /* Double start timer. */ - peer->v_start *= 2; - - /* Overflow check. */ - if (peer->v_start >= (60 * 2)) - peer->v_start = (60 * 2); - - /* Flush any existing events */ - BGP_EVENT_ADD(peer, BGP_Stop); - goto done; - - case BGP_MSG_KEEPALIVE: - peer->keepalive_out++; - break; - case BGP_MSG_ROUTE_REFRESH_NEW: - case BGP_MSG_ROUTE_REFRESH_OLD: - peer->refresh_out++; - break; - case BGP_MSG_CAPABILITY: - peer->dynamic_cap_out++; - break; - } - - /* OK we send packet so delete it. */ - bgp_packet_delete(peer); - update_last_write = 1; - } while (++count < peer->bgp->wpkt_quanta - && (s = bgp_write_packet(peer)) != NULL); + if (generated) + bgp_writes_on(peer); bgp_write_proceed_actions(peer); -done: - /* Update last_update if UPDATEs were written. */ - if (peer->update_out > oc) - peer->last_update = bgp_clock(); - - /* If we TXed any flavor of packet update last_write */ - if (update_last_write) - peer->last_write = bgp_clock(); - - sockopt_cork(peer->fd, 0); - return 0; -} - -/* This is only for sending NOTIFICATION message to neighbor. */ -static int bgp_write_notify(struct peer *peer) -{ - int ret, val; - u_char type; - struct stream *s; - - /* There should be at least one packet. */ - s = stream_fifo_head(peer->obuf); - if (!s) - return 0; - assert(stream_get_endp(s) >= BGP_HEADER_SIZE); - - /* Stop collecting data within the socket */ - sockopt_cork(peer->fd, 0); - - /* socket is in nonblocking mode, if we can't deliver the NOTIFY, well, - * we only care about getting a clean shutdown at this point. */ - ret = write(peer->fd, STREAM_DATA(s), stream_get_endp(s)); - - /* only connection reset/close gets counted as TCP_fatal_error, failure - * to write the entire NOTIFY doesn't get different FSM treatment */ - if (ret <= 0) { - BGP_EVENT_ADD(peer, TCP_fatal_error); - return 0; - } - - /* Disable Nagle, make NOTIFY packet go out right away */ - val = 1; - (void)setsockopt(peer->fd, IPPROTO_TCP, TCP_NODELAY, (char *)&val, - sizeof(val)); - - /* Retrieve BGP packet type. */ - stream_set_getp(s, BGP_MARKER_SIZE + 2); - type = stream_getc(s); - - assert(type == BGP_MSG_NOTIFY); - - /* Type should be notify. */ - peer->notify_out++; - - /* Double start timer. */ - peer->v_start *= 2; - - /* Overflow check. */ - if (peer->v_start >= (60 * 2)) - peer->v_start = (60 * 2); - - /* Handle Graceful Restart case where the state changes to - Connect instead of Idle */ - BGP_EVENT_ADD(peer, BGP_Stop); - return 0; } -/* Make keepalive packet and send it to the peer. */ +/* + * Creates a BGP Keepalive packet and appends it to the peer's output queue. + */ void bgp_keepalive_send(struct peer *peer) { struct stream *s; @@ -509,10 +487,13 @@ void 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_writes_on(peer); } -/* Make open packet and send it to the peer. */ +/* + * Creates a BGP Open packet and appends it to the peer's output queue. + * Sets capabilities as necessary. + */ void bgp_open_send(struct peer *peer) { struct stream *s; @@ -561,10 +542,91 @@ void 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_writes_on(peer); } -/* Send BGP notify packet with data potion. */ +/* This is only for sending NOTIFICATION message to neighbor. */ +static int bgp_write_notify(struct peer *peer) +{ + int ret, val; + u_char type; + struct stream *s; + + pthread_mutex_lock(&peer->io_mtx); + { + /* There should be at least one packet. */ + s = stream_fifo_pop(peer->obuf); + } + pthread_mutex_unlock(&peer->io_mtx); + + if (!s) + return 0; + + assert(stream_get_endp(s) >= BGP_HEADER_SIZE); + + /* Stop collecting data within the socket */ + sockopt_cork(peer->fd, 0); + + /* + * socket is in nonblocking mode, if we can't deliver the NOTIFY, well, + * we only care about getting a clean shutdown at this point. + */ + ret = write(peer->fd, STREAM_DATA(s), stream_get_endp(s)); + + /* + * only connection reset/close gets counted as TCP_fatal_error, failure + * to write the entire NOTIFY doesn't get different FSM treatment + */ + if (ret <= 0) { + stream_free(s); + BGP_EVENT_ADD(peer, TCP_fatal_error); + return 0; + } + + /* Disable Nagle, make NOTIFY packet go out right away */ + val = 1; + (void)setsockopt(peer->fd, IPPROTO_TCP, TCP_NODELAY, (char *)&val, + sizeof(val)); + + /* Retrieve BGP packet type. */ + stream_set_getp(s, BGP_MARKER_SIZE + 2); + type = stream_getc(s); + + assert(type == BGP_MSG_NOTIFY); + + /* Type should be notify. */ + peer->notify_out++; + + /* Double start timer. */ + peer->v_start *= 2; + + /* Overflow check. */ + if (peer->v_start >= (60 * 2)) + peer->v_start = (60 * 2); + + /* + * Handle Graceful Restart case where the state changes to + * Connect instead of Idle + */ + BGP_EVENT_ADD(peer, BGP_Stop); + + stream_free(s); + + return 0; +} + +/* + * Creates a BGP Notify and appends it to the peer's output queue. + * + * This function attempts to write the packet from the thread it is called + * from, to ensure the packet gets out ASAP. + * + * @param peer + * @param code BGP error code + * @param sub_code BGP error subcode + * @param data Data portion + * @param datalen length of data portion + */ void bgp_notify_send_with_data(struct peer *peer, u_char code, u_char sub_code, u_char *data, size_t datalen) { @@ -574,7 +636,7 @@ void bgp_notify_send_with_data(struct peer *peer, u_char code, u_char sub_code, /* Allocate new stream. */ s = stream_new(BGP_MAX_PACKET_SIZE); - /* Make nitify packet. */ + /* Make notify packet. */ bgp_packet_set_marker(s, BGP_MSG_NOTIFY); /* Set notify packet values. */ @@ -588,9 +650,32 @@ void bgp_notify_send_with_data(struct peer *peer, u_char code, u_char sub_code, /* 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); + /* + * Turn off keepalive generation for peer. This is necessary because + * otherwise between the time we wipe the output buffer and the time we + * push the NOTIFY onto it, the KA generation thread could have pushed + * a KEEPALIVE in the middle. + */ + bgp_keepalives_off(peer); + + /* wipe output buffer */ + pthread_mutex_lock(&peer->io_mtx); + { + stream_fifo_clean(peer->obuf); + } + pthread_mutex_unlock(&peer->io_mtx); + + /* + * If possible, store last packet for debugging purposes. This check is + * in place because we are sometimes called with a doppelganger peer, + * who tends to have a plethora of fields nulled out. + */ + if (peer->curr && peer->last_reset_cause_size) { + size_t packetsize = stream_get_endp(peer->curr); + assert(packetsize <= peer->last_reset_cause_size); + memcpy(peer->last_reset_cause, peer->curr->data, packetsize); + peer->last_reset_cause_size = packetsize; + } /* For debug */ { @@ -641,19 +726,37 @@ void bgp_notify_send_with_data(struct peer *peer, u_char code, u_char sub_code, } else peer->last_reset = PEER_DOWN_NOTIFY_SEND; - /* Call immediately. */ - BGP_WRITE_OFF(peer->t_write); + /* Add packet to peer's output queue */ + bgp_packet_add(peer, s); bgp_write_notify(peer); } -/* Send BGP notify packet. */ +/* + * Creates a BGP Notify and appends it to the peer's output queue. + * + * This function attempts to write the packet from the thread it is called + * from, to ensure the packet gets out ASAP. + * + * @param peer + * @param code BGP error code + * @param sub_code BGP error subcode + */ void bgp_notify_send(struct peer *peer, u_char code, u_char sub_code) { bgp_notify_send_with_data(peer, code, sub_code, NULL, 0); } -/* Send route refresh message to the peer. */ +/* + * Creates BGP Route Refresh packet and appends it to the peer's output queue. + * + * @param peer + * @param afi Address Family Identifier + * @param safi Subsequent Address Family Identifier + * @param orf_type Outbound Route Filtering type + * @param when_to_refresh Whether to refresh immediately or defer + * @param remove Whether to remove ORF for specified AFI/SAFI + */ void bgp_route_refresh_send(struct peer *peer, afi_t afi, safi_t safi, u_char orf_type, u_char when_to_refresh, int remove) { @@ -742,10 +845,18 @@ void bgp_route_refresh_send(struct peer *peer, afi_t afi, safi_t safi, /* Add packet to the peer. */ bgp_packet_add(peer, s); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + bgp_writes_on(peer); } -/* Send capability message to the peer. */ +/* + * Create a BGP Capability packet and append it to the peer's output queue. + * + * @param peer + * @param afi Address Family Identifier + * @param safi Subsequent Address Family Identifier + * @param capability_code BGP Capability Code + * @param action Set or Remove capability + */ void bgp_capability_send(struct peer *peer, afi_t afi, safi_t safi, int capability_code, int action) { @@ -785,7 +896,7 @@ void bgp_capability_send(struct peer *peer, afi_t afi, safi_t safi, /* Add packet to the peer. */ bgp_packet_add(peer, s); - BGP_WRITE_ON(peer->t_write, bgp_write, peer->fd); + bgp_writes_on(peer); } /* RFC1771 6.8 Connection collision detection. */ @@ -872,6 +983,42 @@ static int bgp_collision_detect(struct peer *new, struct in_addr remote_id) return 0; } +/* Packet processing routines ---------------------------------------------- */ +/* + * This is a family of functions designed to be called from + * bgp_process_packet(). These functions all share similar behavior and should + * adhere to the following invariants and restrictions: + * + * Return codes + * ------------ + * The return code of any one of those functions should be one of the FSM event + * codes specified in bgpd.h. If a NOTIFY was sent, this event code MUST be + * BGP_Stop. Otherwise, the code SHOULD correspond to the function's expected + * packet type. For example, bgp_open_receive() should return BGP_Stop upon + * error and Receive_OPEN_message otherwise. + * + * If no action is necessary, the correct return code is BGP_PACKET_NOOP as + * defined below. + * + * Side effects + * ------------ + * - May send NOTIFY messages + * - May not modify peer->status + * - May not call bgp_event_update() + */ + +#define BGP_PACKET_NOOP 0 + +/** + * Process BGP OPEN message for peer. + * + * If any errors are encountered in the OPEN message, immediately sends NOTIFY + * and returns BGP_Stop. + * + * @param peer + * @param size size of the packet + * @return as in summary + */ static int bgp_open_receive(struct peer *peer, bgp_size_t size) { int ret; @@ -889,13 +1036,13 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) u_int16_t *holdtime_ptr; /* Parse open packet. */ - version = stream_getc(peer->ibuf); - memcpy(notify_data_remote_as, stream_pnt(peer->ibuf), 2); - remote_as = stream_getw(peer->ibuf); - holdtime_ptr = (u_int16_t *)stream_pnt(peer->ibuf); - holdtime = stream_getw(peer->ibuf); - memcpy(notify_data_remote_id, stream_pnt(peer->ibuf), 4); - remote_id.s_addr = stream_get_ipv4(peer->ibuf); + version = stream_getc(peer->curr); + memcpy(notify_data_remote_as, stream_pnt(peer->curr), 2); + remote_as = stream_getw(peer->curr); + holdtime_ptr = (u_int16_t *)stream_pnt(peer->curr); + holdtime = stream_getw(peer->curr); + memcpy(notify_data_remote_id, stream_pnt(peer->curr), 4); + remote_id.s_addr = stream_get_ipv4(peer->curr); /* Receive OPEN message log */ if (bgp_debug_neighbor_events(peer)) @@ -907,14 +1054,14 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) /* BEGIN to read the capability here, but dont do it yet */ mp_capability = 0; - optlen = stream_getc(peer->ibuf); + optlen = stream_getc(peer->curr); if (optlen != 0) { /* If not enough bytes, it is an error. */ - if (STREAM_READABLE(peer->ibuf) < optlen) { + if (STREAM_READABLE(peer->curr) < optlen) { bgp_notify_send(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_MALFORMED_ATTR); - return -1; + return BGP_Stop; } /* We need the as4 capability value *right now* because @@ -934,7 +1081,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as4, 4); - return -1; + return BGP_Stop; } if (remote_as == BGP_AS_TRANS) { @@ -949,7 +1096,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as4, 4); - return -1; + return BGP_Stop; } if (!as4 && BGP_DEBUG(as4, AS4)) @@ -979,7 +1126,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as4, 4); - return -1; + return BGP_Stop; } } @@ -992,7 +1139,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_BGP_IDENT, notify_data_remote_id, 4); - return -1; + return BGP_Stop; } /* Set remote router-id */ @@ -1010,7 +1157,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_UNSUP_VERSION, (u_int8_t *)&maxver, 2); - return -1; + return BGP_Stop; } /* Check neighbor as number. */ @@ -1022,7 +1169,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as, 2); - return -1; + return BGP_Stop; } else if (peer->as_type == AS_INTERNAL) { if (remote_as != peer->bgp->as) { if (bgp_debug_neighbor_events(peer)) @@ -1032,7 +1179,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as, 2); - return -1; + return BGP_Stop; } peer->as = peer->local_as; } else if (peer->as_type == AS_EXTERNAL) { @@ -1044,7 +1191,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as, 2); - return -1; + return BGP_Stop; } peer->as = remote_as; } else if ((peer->as_type == AS_SPECIFIED) && (remote_as != peer->as)) { @@ -1054,7 +1201,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_BAD_PEER_AS, notify_data_remote_as, 2); - return -1; + return BGP_Stop; } /* From the rfc: Upon receipt of an OPEN message, a BGP speaker MUST @@ -1068,7 +1215,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) bgp_notify_send_with_data(peer, BGP_NOTIFY_OPEN_ERR, BGP_NOTIFY_OPEN_UNACEP_HOLDTIME, (u_char *)holdtime_ptr, 2); - return -1; + return BGP_Stop; } /* From the rfc: A reasonable maximum time between KEEPALIVE messages @@ -1097,7 +1244,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) if (optlen != 0) { if ((ret = bgp_open_option_parse(peer, optlen, &mp_capability)) < 0) - return ret; + return BGP_Stop; } else { if (bgp_debug_neighbor_events(peer)) zlog_debug("%s rcvd OPEN w/ OPTION parameter len: 0", @@ -1131,13 +1278,13 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) immidiately. */ ret = bgp_collision_detect(peer, remote_id); if (ret < 0) - return ret; + return BGP_Stop; /* Get sockname. */ if ((ret = bgp_getsockname(peer)) < 0) { zlog_err("%s: bgp_getsockname() failed for peer: %s", __FUNCTION__, peer->host); - return (ret); + return BGP_Stop; } /* Verify valid local address present based on negotiated @@ -1154,7 +1301,7 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) peer->host, peer->fd); bgp_notify_send(peer, BGP_NOTIFY_CEASE, BGP_NOTIFY_SUBCODE_UNSPECIFIC); - return -1; + return BGP_Stop; #endif } } @@ -1170,170 +1317,42 @@ static int bgp_open_receive(struct peer *peer, bgp_size_t size) peer->host, peer->fd); bgp_notify_send(peer, BGP_NOTIFY_CEASE, BGP_NOTIFY_SUBCODE_UNSPECIFIC); - return -1; + return BGP_Stop; #endif } } peer->rtt = sockopt_tcp_rtt(peer->fd); - if ((ret = bgp_event_update(peer, Receive_OPEN_message)) < 0) { - zlog_err("%s: BGP event update failed for peer: %s", - __FUNCTION__, peer->host); - /* DD: bgp send notify and reset state */ - return (ret); - } - - peer->packet_size = 0; - if (peer->ibuf) - stream_reset(peer->ibuf); - - return 0; -} - -/* Called when there is a change in the EOR(implicit or explicit) status of a - peer. - Ends the update-delay if all expected peers are done with EORs. */ -void bgp_check_update_delay(struct bgp *bgp) -{ - struct listnode *node, *nnode; - struct peer *peer = NULL; - - if (bgp_debug_neighbor_events(peer)) - zlog_debug("Checking update delay, T: %d R: %d I:%d E: %d", - bgp->established, bgp->restarted_peers, - bgp->implicit_eors, bgp->explicit_eors); - - if (bgp->established - <= bgp->restarted_peers + bgp->implicit_eors + bgp->explicit_eors) { - /* This is an extra sanity check to make sure we wait for all - the - eligible configured peers. This check is performed if - establish wait - timer is on, or establish wait option is not given with the - update-delay command */ - if (bgp->t_establish_wait - || (bgp->v_establish_wait == bgp->v_update_delay)) - for (ALL_LIST_ELEMENTS(bgp->peer, node, nnode, peer)) { - if (CHECK_FLAG(peer->flags, - PEER_FLAG_CONFIG_NODE) - && !CHECK_FLAG(peer->flags, - PEER_FLAG_SHUTDOWN) - && !peer->update_delay_over) { - if (bgp_debug_neighbor_events(peer)) - zlog_debug( - " Peer %s pending, continuing read-only mode", - peer->host); - return; - } - } - - zlog_info( - "Update delay ended, restarted: %d, EORs implicit: %d, explicit: %d", - bgp->restarted_peers, bgp->implicit_eors, - bgp->explicit_eors); - bgp_update_delay_end(bgp); - } + return Receive_OPEN_message; } -/* Called if peer is known to have restarted. The restart-state bit in - Graceful-Restart capability is used for that */ -void bgp_update_restarted_peers(struct peer *peer) -{ - if (!bgp_update_delay_active(peer->bgp)) - return; /* BGP update delay has ended */ - if (peer->update_delay_over) - return; /* This peer has already been considered */ - - if (bgp_debug_neighbor_events(peer)) - zlog_debug("Peer %s: Checking restarted", peer->host); - - if (peer->status == Established) { - peer->update_delay_over = 1; - peer->bgp->restarted_peers++; - bgp_check_update_delay(peer->bgp); - } -} - -/* Called as peer receives a keep-alive. Determines if this occurence can be - taken as an implicit EOR for this peer. - NOTE: The very first keep-alive after the Established state of a peer is - considered implicit EOR for the update-delay purposes */ -void bgp_update_implicit_eors(struct peer *peer) +/** + * Process BGP KEEPALIVE message for peer. + * + * @param peer + * @param size size of the packet + * @return as in summary + */ +static int bgp_keepalive_receive(struct peer *peer, bgp_size_t size) { - if (!bgp_update_delay_active(peer->bgp)) - return; /* BGP update delay has ended */ - if (peer->update_delay_over) - return; /* This peer has already been considered */ + if (bgp_debug_keepalive(peer)) + zlog_debug("%s KEEPALIVE rcvd", peer->host); - if (bgp_debug_neighbor_events(peer)) - zlog_debug("Peer %s: Checking implicit EORs", peer->host); + bgp_update_implicit_eors(peer); - if (peer->status == Established) { - peer->update_delay_over = 1; - peer->bgp->implicit_eors++; - bgp_check_update_delay(peer->bgp); - } + return Receive_KEEPALIVE_message; } -/* Should be called only when there is a change in the EOR_RECEIVED status - for any afi/safi on a peer */ -static void bgp_update_explicit_eors(struct peer *peer) -{ - afi_t afi; - safi_t safi; - - if (!bgp_update_delay_active(peer->bgp)) - return; /* BGP update delay has ended */ - if (peer->update_delay_over) - return; /* This peer has already been considered */ - - if (bgp_debug_neighbor_events(peer)) - zlog_debug("Peer %s: Checking explicit EORs", peer->host); - - FOREACH_AFI_SAFI (afi, safi) { - if (peer->afc_nego[afi][safi] - && !CHECK_FLAG(peer->af_sflags[afi][safi], - PEER_STATUS_EOR_RECEIVED)) { - if (bgp_debug_neighbor_events(peer)) - zlog_debug( - " afi %d safi %d didnt receive EOR", - afi, safi); - return; - } - } - - peer->update_delay_over = 1; - peer->bgp->explicit_eors++; - bgp_check_update_delay(peer->bgp); -} -/* Frontend for NLRI parsing, to fan-out to AFI/SAFI specific parsers - * mp_withdraw, if set, is used to nullify attr structure on most of the calling - * safi function - * and for evpn, passed as parameter +/** + * Process BGP UPDATE message for peer. + * + * Parses UPDATE and creates attribute object. + * + * @param peer + * @param size size of the packet + * @return as in summary */ -int bgp_nlri_parse(struct peer *peer, struct attr *attr, - struct bgp_nlri *packet, int mp_withdraw) -{ - switch (packet->safi) { - case SAFI_UNICAST: - case SAFI_MULTICAST: - return bgp_nlri_parse_ip(peer, mp_withdraw ? NULL : attr, - packet); - case SAFI_LABELED_UNICAST: - return bgp_nlri_parse_label(peer, mp_withdraw ? NULL : attr, - packet); - case SAFI_MPLS_VPN: - return bgp_nlri_parse_vpn(peer, mp_withdraw ? NULL : attr, - packet); - case SAFI_EVPN: - return bgp_nlri_parse_evpn(peer, attr, packet, mp_withdraw); - default: - return -1; - } -} - -/* Parse BGP Update packet and make attribute object. */ static int bgp_update_receive(struct peer *peer, bgp_size_t size) { int ret, nlri_ret; @@ -1359,7 +1378,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) peer->host, lookup_msg(bgp_status_msg, peer->status, NULL)); bgp_notify_send(peer, BGP_NOTIFY_FSM_ERR, 0); - return -1; + return BGP_Stop; } /* Set initial values. */ @@ -1370,7 +1389,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) memset(peer->rcvd_attr_str, 0, BUFSIZ); peer->rcvd_attr_printed = 0; - s = peer->ibuf; + s = peer->curr; end = stream_pnt(s) + size; /* RFC1771 6.3 If the Unfeasible Routes Length or Total Attribute @@ -1384,7 +1403,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) peer->host); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_MAL_ATTR); - return -1; + return BGP_Stop; } /* Unfeasible Route Length. */ @@ -1398,7 +1417,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) peer->host, withdraw_len); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_MAL_ATTR); - return -1; + return BGP_Stop; } /* Unfeasible Route packet format check. */ @@ -1418,7 +1437,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) peer->host); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_MAL_ATTR); - return -1; + return BGP_Stop; } /* Fetch attribute total length. */ @@ -1432,7 +1451,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) peer->host, attribute_len); bgp_notify_send(peer, BGP_NOTIFY_UPDATE_ERR, BGP_NOTIFY_UPDATE_MAL_ATTR); - return -1; + return BGP_Stop; } /* Certain attribute parsing errors should not be considered bad enough @@ -1455,7 +1474,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) &nlris[NLRI_MP_WITHDRAW]); if (attr_parse_ret == BGP_ATTR_PARSE_ERROR) { bgp_attr_unintern_sub(&attr); - return -1; + return BGP_Stop; } } @@ -1534,7 +1553,7 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) ? BGP_NOTIFY_UPDATE_INVAL_NETWORK : BGP_NOTIFY_UPDATE_OPT_ATTR_ERR); bgp_attr_unintern_sub(&attr); - return -1; + return BGP_Stop; } } @@ -1590,24 +1609,23 @@ static int bgp_update_receive(struct peer *peer, bgp_size_t size) interned in bgp_attr_parse(). */ bgp_attr_unintern_sub(&attr); - /* If peering is stopped due to some reason, do not generate BGP - event. */ - if (peer->status != Established) - return 0; - - /* Increment packet counter. */ - peer->update_in++; peer->update_time = bgp_clock(); /* Rearm holdtime timer */ BGP_TIMER_OFF(peer->t_holdtime); bgp_timer_set(peer); - return 0; + return Receive_UPDATE_message; } -/* Notify message treatment function. */ -static void bgp_notify_receive(struct peer *peer, bgp_size_t size) +/** + * Process BGP NOTIFY message for peer. + * + * @param peer + * @param size size of the packet + * @return as in summary + */ +static int bgp_notify_receive(struct peer *peer, bgp_size_t size) { struct bgp_notify bgp_notify; @@ -1617,8 +1635,8 @@ static void bgp_notify_receive(struct peer *peer, bgp_size_t size) peer->notify.length = 0; } - bgp_notify.code = stream_getc(peer->ibuf); - bgp_notify.subcode = stream_getc(peer->ibuf); + bgp_notify.code = stream_getc(peer->curr); + bgp_notify.subcode = stream_getc(peer->curr); bgp_notify.length = size - 2; bgp_notify.data = NULL; @@ -1629,7 +1647,7 @@ static void bgp_notify_receive(struct peer *peer, bgp_size_t size) if (bgp_notify.length) { peer->notify.length = size - 2; peer->notify.data = XMALLOC(MTYPE_TMP, size - 2); - memcpy(peer->notify.data, stream_pnt(peer->ibuf), size - 2); + memcpy(peer->notify.data, stream_pnt(peer->curr), size - 2); } /* For debug */ @@ -1644,12 +1662,12 @@ static void bgp_notify_receive(struct peer *peer, bgp_size_t size) for (i = 0; i < bgp_notify.length; i++) if (first) { sprintf(c, " %02x", - stream_getc(peer->ibuf)); + stream_getc(peer->curr)); strcat(bgp_notify.data, c); } else { first = 1; sprintf(c, "%02x", - stream_getc(peer->ibuf)); + stream_getc(peer->curr)); strcpy(bgp_notify.data, c); } bgp_notify.raw_data = (u_char *)peer->notify.data; @@ -1676,20 +1694,17 @@ static void bgp_notify_receive(struct peer *peer, bgp_size_t size) && bgp_notify.subcode == BGP_NOTIFY_OPEN_UNSUP_PARAM) UNSET_FLAG(peer->sflags, PEER_STATUS_CAPABILITY_OPEN); - BGP_EVENT_ADD(peer, Receive_NOTIFICATION_message); -} - -/* Keepalive treatment function -- get keepalive send keepalive */ -static void bgp_keepalive_receive(struct peer *peer, bgp_size_t size) -{ - if (bgp_debug_keepalive(peer)) - zlog_debug("%s KEEPALIVE rcvd", peer->host); - - BGP_EVENT_ADD(peer, Receive_KEEPALIVE_message); + return Receive_NOTIFICATION_message; } -/* Route refresh message is received. */ -static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) +/** + * Process BGP ROUTEREFRESH message for peer. + * + * @param peer + * @param size size of the packet + * @return as in summary + */ +static int bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) { iana_afi_t pkt_afi; afi_t afi; @@ -1706,7 +1721,7 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) peer->host); bgp_notify_send(peer, BGP_NOTIFY_HEADER_ERR, BGP_NOTIFY_HEADER_BAD_MESTYPE); - return; + return BGP_Stop; } /* Status must be Established. */ @@ -1716,10 +1731,10 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) peer->host, lookup_msg(bgp_status_msg, peer->status, NULL)); bgp_notify_send(peer, BGP_NOTIFY_FSM_ERR, 0); - return; + return BGP_Stop; } - s = peer->ibuf; + s = peer->curr; /* Parse packet. */ pkt_afi = stream_getw(s); @@ -1735,7 +1750,7 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) zlog_info( "%s REFRESH_REQ for unrecognized afi/safi: %d/%d - ignored", peer->host, pkt_afi, pkt_safi); - return; + return BGP_PACKET_NOOP; } if (size != BGP_MSG_ROUTE_REFRESH_MIN_SIZE - BGP_HEADER_SIZE) { @@ -1749,7 +1764,7 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) zlog_info("%s ORF route refresh length error", peer->host); bgp_notify_send(peer, BGP_NOTIFY_CEASE, 0); - return; + return BGP_Stop; } when_to_refresh = stream_getc(s); @@ -1920,7 +1935,7 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) ? "Defer" : "Immediate"); if (when_to_refresh == REFRESH_DEFER) - return; + return BGP_PACKET_NOOP; } /* First update is deferred until ORF or ROUTE-REFRESH is received */ @@ -1951,8 +1966,18 @@ static void bgp_route_refresh_receive(struct peer *peer, bgp_size_t size) /* Perform route refreshment to the peer */ bgp_announce_route(peer, afi, safi); + + /* No FSM action necessary */ + return BGP_PACKET_NOOP; } +/** + * Parse BGP CAPABILITY message for peer. + * + * @param peer + * @param size size of the packet + * @return as in summary + */ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, bgp_size_t length) { @@ -1973,7 +1998,7 @@ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, if (pnt + 3 > end) { zlog_info("%s Capability length error", peer->host); bgp_notify_send(peer, BGP_NOTIFY_CEASE, 0); - return -1; + return BGP_Stop; } action = *pnt; hdr = (struct capability_header *)(pnt + 1); @@ -1984,7 +2009,7 @@ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, zlog_info("%s Capability Action Value error %d", peer->host, action); bgp_notify_send(peer, BGP_NOTIFY_CEASE, 0); - return -1; + return BGP_Stop; } if (bgp_debug_neighbor_events(peer)) @@ -1996,7 +2021,7 @@ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, if ((pnt + hdr->length + 3) > end) { zlog_info("%s Capability length error", peer->host); bgp_notify_send(peer, BGP_NOTIFY_CEASE, 0); - return -1; + return BGP_Stop; } /* Fetch structure to the byte stream. */ @@ -2047,7 +2072,7 @@ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, if (peer_active_nego(peer)) bgp_clear_route(peer, afi, safi); else - BGP_EVENT_ADD(peer, BGP_Stop); + return BGP_Stop; } } else { zlog_warn( @@ -2055,19 +2080,26 @@ static int bgp_capability_msg_parse(struct peer *peer, u_char *pnt, peer->host, hdr->code); } } - return 0; + + /* No FSM action necessary */ + return BGP_PACKET_NOOP; } -/* Dynamic Capability is received. +/** + * Parse BGP CAPABILITY message for peer. + * + * Exported for unit testing. * - * This is exported for unit-test purposes + * @param peer + * @param size size of the packet + * @return as in summary */ int bgp_capability_receive(struct peer *peer, bgp_size_t size) { u_char *pnt; /* Fetch pointer. */ - pnt = stream_pnt(peer->ibuf); + pnt = stream_pnt(peer->curr); if (bgp_debug_neighbor_events(peer)) zlog_debug("%s rcv CAPABILITY", peer->host); @@ -2078,7 +2110,7 @@ int bgp_capability_receive(struct peer *peer, bgp_size_t size) peer->host); bgp_notify_send(peer, BGP_NOTIFY_HEADER_ERR, BGP_NOTIFY_HEADER_BAD_MESTYPE); - return -1; + return BGP_Stop; } /* Status must be Established. */ @@ -2088,254 +2120,169 @@ int bgp_capability_receive(struct peer *peer, bgp_size_t size) peer->host, lookup_msg(bgp_status_msg, peer->status, NULL)); bgp_notify_send(peer, BGP_NOTIFY_FSM_ERR, 0); - return -1; + return BGP_Stop; } /* Parse packet. */ return bgp_capability_msg_parse(peer, pnt, size); } -/* BGP read utility function. */ -static int bgp_read_packet(struct peer *peer) +/** + * Processes a peer's input buffer. + * + * This function sidesteps the event loop and directly calls bgp_event_update() + * after processing each BGP message. This is necessary to ensure proper + * ordering of FSM events and unifies the behavior that was present previously, + * whereby some of the packet handling functions would update the FSM and some + * would not, making event flow difficult to understand. Please think twice + * before hacking this. + * + * Thread type: THREAD_EVENT + * @param thread + * @return 0 + */ +int bgp_process_packet(struct thread *thread) { - int nbytes; - int readsize; + /* Yes first of all get peer pointer. */ + struct peer *peer; // peer + uint32_t rpkt_quanta_old; // how many packets to read + int fsm_update_result; // return code of bgp_event_update() + int mprc; // message processing return code - readsize = peer->packet_size - stream_get_endp(peer->ibuf); + peer = THREAD_ARG(thread); + rpkt_quanta_old = atomic_load_explicit(&peer->bgp->rpkt_quanta, + memory_order_relaxed); + fsm_update_result = 0; - /* If size is zero then return. */ - if (!readsize) + /* Guard against scheduled events that occur after peer deletion. */ + if (peer->status == Deleted || peer->status == Clearing) return 0; - /* Read packet from fd. */ - nbytes = stream_read_try(peer->ibuf, peer->fd, readsize); - - /* If read byte is smaller than zero then error occured. */ - if (nbytes < 0) { - /* Transient error should retry */ - if (nbytes == -2) - return -1; + unsigned int processed = 0; - zlog_err("%s [Error] bgp_read_packet error: %s", peer->host, - safe_strerror(errno)); + while (processed < rpkt_quanta_old) { + u_char type = 0; + bgp_size_t size; + char notify_data_length[2]; - if (peer->status == Established) { - if (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; + pthread_mutex_lock(&peer->io_mtx); + { + peer->curr = stream_fifo_pop(peer->ibuf); } + pthread_mutex_unlock(&peer->io_mtx); - BGP_EVENT_ADD(peer, TCP_fatal_error); - return -1; - } - - /* When read byte is zero : clear bgp peer and return */ - if (nbytes == 0) { - 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->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_ADD(peer, TCP_connection_closed); - return -1; - } - - /* We read partial packet. */ - if (stream_get_endp(peer->ibuf) != peer->packet_size) - return -1; - - return 0; -} - -/* Marker check. */ -static int bgp_marker_all_one(struct stream *s, int length) -{ - int i; - - for (i = 0; i < length; i++) - if (s->data[i] != 0xff) + if (peer->curr == NULL) // no packets to process, hmm... return 0; - return 1; -} + /* skip the marker and copy the packet length */ + stream_forward_getp(peer->curr, BGP_MARKER_SIZE); + memcpy(notify_data_length, stream_pnt(peer->curr), 2); -/* Starting point of packet process function. */ -int bgp_read(struct thread *thread) -{ - int ret; - u_char type = 0; - struct peer *peer; - bgp_size_t size; - char notify_data_length[2]; - u_int32_t notify_out; - - /* Yes first of all get peer pointer. */ - peer = THREAD_ARG(thread); - peer->t_read = NULL; - - /* Note notify_out so we can check later to see if we sent another one - */ - notify_out = peer->notify_out; + /* read in the packet length and type */ + size = stream_getw(peer->curr); + type = stream_getc(peer->curr); - /* For non-blocking IO check. */ - if (peer->status == Connect) { - bgp_connect_check(peer, 1); - goto done; - } else { - if (peer->fd < 0) { - zlog_err("bgp_read peer's fd is negative value %d", - peer->fd); - return -1; - } - BGP_READ_ON(peer->t_read, bgp_read, peer->fd); - } + /* BGP packet dump function. */ + bgp_dump_packet(peer, type, peer->curr); - /* Read packet header to determine type of the packet */ - if (peer->packet_size == 0) - peer->packet_size = BGP_HEADER_SIZE; - - if (stream_get_endp(peer->ibuf) < BGP_HEADER_SIZE) { - ret = bgp_read_packet(peer); - - /* Header read error or partial read packet. */ - if (ret < 0) - goto done; - - /* Get size and type. */ - stream_forward_getp(peer->ibuf, BGP_MARKER_SIZE); - memcpy(notify_data_length, stream_pnt(peer->ibuf), 2); - size = stream_getw(peer->ibuf); - type = stream_getc(peer->ibuf); - - /* Marker check */ - if (((type == BGP_MSG_OPEN) || (type == BGP_MSG_KEEPALIVE)) - && !bgp_marker_all_one(peer->ibuf, BGP_MARKER_SIZE)) { - bgp_notify_send(peer, BGP_NOTIFY_HEADER_ERR, - BGP_NOTIFY_HEADER_NOT_SYNC); - goto done; - } + /* adjust size to exclude the marker + length + type */ + size -= BGP_HEADER_SIZE; - /* BGP type check. */ - if (type != BGP_MSG_OPEN && type != BGP_MSG_UPDATE - && type != BGP_MSG_NOTIFY && type != BGP_MSG_KEEPALIVE - && type != BGP_MSG_ROUTE_REFRESH_NEW - && type != BGP_MSG_ROUTE_REFRESH_OLD - && type != BGP_MSG_CAPABILITY) { - if (bgp_debug_neighbor_events(peer)) - zlog_debug("%s unknown message type 0x%02x", - peer->host, type); - bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR, - BGP_NOTIFY_HEADER_BAD_MESTYPE, - &type, 1); - goto done; - } - /* Mimimum packet length check. */ - if ((size < BGP_HEADER_SIZE) || (size > BGP_MAX_PACKET_SIZE) - || (type == BGP_MSG_OPEN && size < BGP_MSG_OPEN_MIN_SIZE) - || (type == BGP_MSG_UPDATE - && size < BGP_MSG_UPDATE_MIN_SIZE) - || (type == BGP_MSG_NOTIFY - && size < BGP_MSG_NOTIFY_MIN_SIZE) - || (type == BGP_MSG_KEEPALIVE - && size != BGP_MSG_KEEPALIVE_MIN_SIZE) - || (type == BGP_MSG_ROUTE_REFRESH_NEW - && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE) - || (type == BGP_MSG_ROUTE_REFRESH_OLD - && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE) - || (type == BGP_MSG_CAPABILITY - && size < BGP_MSG_CAPABILITY_MIN_SIZE)) { - if (bgp_debug_neighbor_events(peer)) - zlog_debug("%s bad message length - %d for %s", - peer->host, size, - type == 128 - ? "ROUTE-REFRESH" - : bgp_type_str[(int)type]); - bgp_notify_send_with_data(peer, BGP_NOTIFY_HEADER_ERR, - BGP_NOTIFY_HEADER_BAD_MESLEN, - (u_char *)notify_data_length, - 2); - goto done; + /* Read rest of the packet and call each sort of packet routine + */ + switch (type) { + case BGP_MSG_OPEN: + peer->open_in++; + mprc = bgp_open_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP OPEN receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + case BGP_MSG_UPDATE: + peer->update_in++; + peer->readtime = monotime(NULL); + mprc = bgp_update_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP UPDATE receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + case BGP_MSG_NOTIFY: + peer->notify_in++; + mprc = bgp_notify_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP NOTIFY receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + case BGP_MSG_KEEPALIVE: + peer->readtime = monotime(NULL); + peer->keepalive_in++; + mprc = bgp_keepalive_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP KEEPALIVE receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + case BGP_MSG_ROUTE_REFRESH_NEW: + case BGP_MSG_ROUTE_REFRESH_OLD: + peer->refresh_in++; + mprc = bgp_route_refresh_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP ROUTEREFRESH receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + case BGP_MSG_CAPABILITY: + peer->dynamic_cap_in++; + mprc = bgp_capability_receive(peer, size); + if (mprc == BGP_Stop) + zlog_err( + "%s: BGP CAPABILITY receipt failed for peer: %s", + __FUNCTION__, peer->host); + break; + default: + /* + * The message type should have been sanitized before + * we ever got here. Receipt of a message with an + * invalid header at this point is indicative of a + * security issue. + */ + assert (!"Message of invalid type received during input processing"); } - /* Adjust size to message length. */ - peer->packet_size = size; - } + /* delete processed packet */ + stream_free(peer->curr); + peer->curr = NULL; + processed++; - ret = bgp_read_packet(peer); - if (ret < 0) - goto done; - - /* Get size and type again. */ - (void)stream_getw_from(peer->ibuf, BGP_MARKER_SIZE); - type = stream_getc_from(peer->ibuf, BGP_MARKER_SIZE + 2); - - /* BGP packet dump function. */ - bgp_dump_packet(peer, type, peer->ibuf); - - size = (peer->packet_size - BGP_HEADER_SIZE); - - /* Read rest of the packet and call each sort of packet routine */ - switch (type) { - case BGP_MSG_OPEN: - peer->open_in++; - bgp_open_receive(peer, size); /* XXX return value ignored! */ - break; - case BGP_MSG_UPDATE: - peer->readtime = monotime(NULL); - bgp_update_receive(peer, size); - break; - case BGP_MSG_NOTIFY: - bgp_notify_receive(peer, size); - break; - case BGP_MSG_KEEPALIVE: - peer->readtime = monotime(NULL); - bgp_keepalive_receive(peer, size); - break; - case BGP_MSG_ROUTE_REFRESH_NEW: - case BGP_MSG_ROUTE_REFRESH_OLD: - peer->refresh_in++; - bgp_route_refresh_receive(peer, size); - break; - case BGP_MSG_CAPABILITY: - peer->dynamic_cap_in++; - bgp_capability_receive(peer, size); - break; - } + /* Update FSM */ + if (mprc != BGP_PACKET_NOOP) + fsm_update_result = bgp_event_update(peer, mprc); + else + continue; - /* If reading this packet caused us to send a NOTIFICATION then store a - * copy - * of the packet for troubleshooting purposes - */ - if (notify_out < peer->notify_out) { - memcpy(peer->last_reset_cause, peer->ibuf->data, - peer->packet_size); - peer->last_reset_cause_size = peer->packet_size; - notify_out = peer->notify_out; + /* + * If peer was deleted, do not process any more packets. This + * is usually due to executing BGP_Stop or a stub deletion. + */ + if (fsm_update_result == FSM_PEER_TRANSFERRED + || fsm_update_result == FSM_PEER_STOPPED) + break; } - /* Clear input buffer. */ - peer->packet_size = 0; - if (peer->ibuf) - stream_reset(peer->ibuf); - -done: - /* If reading this packet caused us to send a NOTIFICATION then store a - * copy - * of the packet for troubleshooting purposes - */ - if (notify_out < peer->notify_out) { - memcpy(peer->last_reset_cause, peer->ibuf->data, - peer->packet_size); - peer->last_reset_cause_size = peer->packet_size; + if (fsm_update_result != FSM_PEER_TRANSFERRED + && fsm_update_result != FSM_PEER_STOPPED) { + pthread_mutex_lock(&peer->io_mtx); + { + // more work to do, come back later + if (peer->ibuf->count > 0) + thread_add_timer_msec( + bm->master, bgp_process_packet, peer, 0, + &peer->t_process_packet); + } + pthread_mutex_unlock(&peer->io_mtx); } return 0; diff --git a/bgpd/bgp_packet.h b/bgpd/bgp_packet.h index 7bf498c37c..008f2b814b 100644 --- a/bgpd/bgp_packet.h +++ b/bgpd/bgp_packet.h @@ -24,7 +24,6 @@ #define BGP_NLRI_LENGTH 1U #define BGP_TOTAL_ATTR_LEN 2U #define BGP_UNFEASIBLE_LEN 2U -#define BGP_WRITE_PACKET_MAX 10U /* When to refresh */ #define REFRESH_IMMEDIATE 1 @@ -38,10 +37,6 @@ #define ORF_COMMON_PART_DENY 0x20 /* Packet send and receive function prototypes. */ -extern int bgp_read(struct thread *); -extern int bgp_write(struct thread *); -extern int bgp_connect_check(struct peer *, int change_state); - extern void bgp_keepalive_send(struct peer *); extern void bgp_open_send(struct peer *); extern void bgp_notify_send(struct peer *, u_int8_t, u_int8_t); @@ -65,6 +60,8 @@ extern void bgp_check_update_delay(struct bgp *); extern int bgp_packet_set_marker(struct stream *s, u_char type); extern int bgp_packet_set_size(struct stream *s); -extern void bgp_packet_add(struct peer *peer, struct stream *s); + +extern int bgp_generate_updgrp_packets(struct thread *); +extern int bgp_process_packet(struct thread *); #endif /* _QUAGGA_BGP_PACKET_H */ diff --git a/bgpd/bgp_updgrp.c b/bgpd/bgp_updgrp.c index 8f67290600..1c589f7960 100644 --- a/bgpd/bgp_updgrp.c +++ b/bgpd/bgp_updgrp.c @@ -53,6 +53,7 @@ #include "bgpd/bgp_updgrp.h" #include "bgpd/bgp_route.h" #include "bgpd/bgp_filter.h" +#include "bgpd/bgp_io.h" /******************** * PRIVATE FUNCTIONS @@ -1871,17 +1872,16 @@ void subgroup_trigger_write(struct update_subgroup *subgrp) { struct peer_af *paf; -#if 0 - if (bgp_debug_update(NULL, NULL, subgrp->update_group, 0)) - zlog_debug("u%llu:s%llu scheduling write thread for peers", - subgrp->update_group->id, subgrp->id); -#endif - SUBGRP_FOREACH_PEER (subgrp, paf) { - if (paf->peer->status == Established) { - BGP_PEER_WRITE_ON(paf->peer->t_write, bgp_write, - paf->peer->fd, paf->peer); - } - } + /* + * For each peer in the subgroup, schedule a job to pull packets from + * the subgroup output queue into their own output queue. This action + * will trigger a write job on the I/O thread. + */ + SUBGRP_FOREACH_PEER(subgrp, paf) + if (paf->peer->status == Established) + thread_add_timer_msec(bm->master, bgp_generate_updgrp_packets, + paf->peer, 0, + &paf->peer->t_generate_updgrp_packets); } int update_group_clear_update_dbg(struct update_group *updgrp, void *arg) diff --git a/bgpd/bgp_updgrp.h b/bgpd/bgp_updgrp.h index 52a21679b8..e941fecb61 100644 --- a/bgpd/bgp_updgrp.h +++ b/bgpd/bgp_updgrp.h @@ -29,7 +29,27 @@ #include "bgp_advertise.h" -#define BGP_DEFAULT_SUBGROUP_COALESCE_TIME 200 +/* + * The following three heuristic constants determine how long advertisement to + * a subgroup will be delayed after it is created. The intent is to allow + * transient changes in peer state (primarily session establishment) to settle, + * so that more peers can be grouped together and benefit from sharing + * advertisement computations with the subgroup. + * + * These values have a very large impact on initial convergence time; any + * changes should be accompanied by careful performance testing at all scales. + * + * The coalesce time 'C' for a new subgroup within a particular BGP instance + * 'B' with total number of known peers 'P', established or not, is computed as + * follows: + * + * C = MIN(BGP_MAX_SUBGROUP_COALESCE_TIME, + * BGP_DEFAULT_SUBGROUP_COALESCE_TIME + + * (P*BGP_PEER_ADJUST_SUBGROUP_COALESCE_TIME)) + */ +#define BGP_DEFAULT_SUBGROUP_COALESCE_TIME 1000 +#define BGP_MAX_SUBGROUP_COALESCE_TIME 10000 +#define BGP_PEER_ADJUST_SUBGROUP_COALESCE_TIME 50 #define PEER_UPDGRP_FLAGS \ (PEER_FLAG_LOCAL_AS_NO_PREPEND | PEER_FLAG_LOCAL_AS_REPLACE_AS) @@ -179,7 +199,7 @@ struct update_subgroup { struct stream *work; /* We use a separate stream to encode MP_REACH_NLRI for efficient - * NLRI packing. peer->work stores all the other attributes. The + * NLRI packing. peer->obuf_work stores all the other attributes. The * actual packet is then constructed by concatenating the two. */ struct stream *scratch; diff --git a/bgpd/bgp_updgrp_adv.c b/bgpd/bgp_updgrp_adv.c index b4f18c9f5e..705cb152f0 100644 --- a/bgpd/bgp_updgrp_adv.c +++ b/bgpd/bgp_updgrp_adv.c @@ -483,7 +483,7 @@ void bgp_adj_out_unset_subgroup(struct bgp_node *rn, { struct bgp_adj_out *adj; struct bgp_advertise *adv; - char trigger_write; + bool trigger_write; if (DISABLE_BGP_ANNOUNCE) return; @@ -502,17 +502,13 @@ void bgp_adj_out_unset_subgroup(struct bgp_node *rn, adv->adj = adj; /* Note if we need to trigger a packet write */ - if (BGP_ADV_FIFO_EMPTY(&subgrp->sync->withdraw)) - trigger_write = 1; - else - trigger_write = 0; + trigger_write = + BGP_ADV_FIFO_EMPTY(&subgrp->sync->withdraw); /* Add to synchronization entry for withdraw * announcement. */ BGP_ADV_FIFO_ADD(&subgrp->sync->withdraw, &adv->fifo); - /* Schedule packet write, if FIFO is getting its first - * entry. */ if (trigger_write) subgroup_trigger_write(subgrp); } else { diff --git a/bgpd/bgp_updgrp_packet.c b/bgpd/bgp_updgrp_packet.c index a35d814e47..b63dfbed0a 100644 --- a/bgpd/bgp_updgrp_packet.c +++ b/bgpd/bgp_updgrp_packet.c @@ -633,7 +633,6 @@ struct stream *bpacket_reformat_for_peer(struct bpacket *pkt, } } - bgp_packet_add(peer, s); return s; } @@ -963,7 +962,7 @@ struct bpacket *subgroup_withdraw_packet(struct update_subgroup *subgrp) addpath_tx_id = adj->addpath_tx_id; space_remaining = - STREAM_REMAIN(s) - BGP_MAX_PACKET_SIZE_OVERFLOW; + STREAM_WRITEABLE(s) - BGP_MAX_PACKET_SIZE_OVERFLOW; space_needed = BGP_NLRI_LENGTH + addpath_overhead + BGP_TOTAL_ATTR_LEN + bgp_packet_mpattr_prefix_size(afi, safi, &rn->p); diff --git a/bgpd/bgp_vty.c b/bgpd/bgp_vty.c index 9159bc683d..a673810738 100644 --- a/bgpd/bgp_vty.c +++ b/bgpd/bgp_vty.c @@ -57,6 +57,7 @@ #include "bgpd/bgp_packet.h" #include "bgpd/bgp_updgrp.h" #include "bgpd/bgp_bfd.h" +#include "bgpd/bgp_io.h" static struct peer_group *listen_range_exists(struct bgp *bgp, struct prefix *range, int exact); @@ -1332,25 +1333,55 @@ static int bgp_wpkt_quanta_config_vty(struct vty *vty, const char *num, { VTY_DECLVAR_CONTEXT(bgp, bgp); - if (set) - bgp->wpkt_quanta = strtoul(num, NULL, 10); - else - bgp->wpkt_quanta = BGP_WRITE_PACKET_MAX; + if (set) { + uint32_t quanta = strtoul(num, NULL, 10); + atomic_store_explicit(&bgp->wpkt_quanta, quanta, + memory_order_relaxed); + } else { + atomic_store_explicit(&bgp->wpkt_quanta, BGP_WRITE_PACKET_MAX, + memory_order_relaxed); + } + + return CMD_SUCCESS; +} + +static int bgp_rpkt_quanta_config_vty(struct vty *vty, const char *num, + char set) +{ + VTY_DECLVAR_CONTEXT(bgp, bgp); + + if (set) { + uint32_t quanta = strtoul(num, NULL, 10); + atomic_store_explicit(&bgp->rpkt_quanta, quanta, + memory_order_relaxed); + } else { + atomic_store_explicit(&bgp->rpkt_quanta, BGP_READ_PACKET_MAX, + memory_order_relaxed); + } return CMD_SUCCESS; } void bgp_config_write_wpkt_quanta(struct vty *vty, struct bgp *bgp) { - if (bgp->wpkt_quanta != BGP_WRITE_PACKET_MAX) - vty_out(vty, " write-quanta %d\n", bgp->wpkt_quanta); + uint32_t quanta = + atomic_load_explicit(&bgp->wpkt_quanta, memory_order_relaxed); + if (quanta != BGP_WRITE_PACKET_MAX) + vty_out(vty, " write-quanta %d\n", quanta); } +void bgp_config_write_rpkt_quanta(struct vty *vty, struct bgp *bgp) +{ + uint32_t quanta = + atomic_load_explicit(&bgp->rpkt_quanta, memory_order_relaxed); + if (quanta != BGP_READ_PACKET_MAX) + vty_out(vty, " read-quanta %d\n", quanta); +} -/* Update-delay configuration */ +/* Packet quanta configuration */ DEFUN (bgp_wpkt_quanta, bgp_wpkt_quanta_cmd, - "write-quanta (1-10000)", + "write-quanta (1-10)", "How many packets to write to peer socket per run\n" "Number of packets\n") { @@ -1358,18 +1389,38 @@ DEFUN (bgp_wpkt_quanta, return bgp_wpkt_quanta_config_vty(vty, argv[idx_number]->arg, 1); } -/* Update-delay deconfiguration */ DEFUN (no_bgp_wpkt_quanta, no_bgp_wpkt_quanta_cmd, - "no write-quanta (1-10000)", + "no write-quanta (1-10)", NO_STR - "How many packets to write to peer socket per run\n" + "How many packets to write to peer socket per I/O cycle\n" "Number of packets\n") { int idx_number = 2; return bgp_wpkt_quanta_config_vty(vty, argv[idx_number]->arg, 0); } +DEFUN (bgp_rpkt_quanta, + bgp_rpkt_quanta_cmd, + "read-quanta (1-10)", + "How many packets to read from peer socket per I/O cycle\n" + "Number of packets\n") +{ + int idx_number = 1; + return bgp_rpkt_quanta_config_vty(vty, argv[idx_number]->arg, 1); +} + +DEFUN (no_bgp_rpkt_quanta, + no_bgp_rpkt_quanta_cmd, + "no read-quanta (1-10)", + NO_STR + "How many packets to read from peer socket per I/O cycle\n" + "Number of packets\n") +{ + int idx_number = 2; + return bgp_rpkt_quanta_config_vty(vty, argv[idx_number]->arg, 0); +} + void bgp_config_write_coalesce_time(struct vty *vty, struct bgp *bgp) { if (bgp->coalesce_time != BGP_DEFAULT_SUBGROUP_COALESCE_TIME) @@ -7068,14 +7119,40 @@ static int bgp_show_summary(struct vty *vty, struct bgp *bgp, int afi, int safi, vty_out(vty, "4 %10u %7u %7u %8" PRIu64 " %4d %4zd %8s", peer->as, - peer->open_in + peer->update_in - + peer->keepalive_in + peer->notify_in - + peer->refresh_in - + peer->dynamic_cap_in, - peer->open_out + peer->update_out - + peer->keepalive_out + peer->notify_out - + peer->refresh_out - + peer->dynamic_cap_out, + atomic_load_explicit(&peer->open_in, + memory_order_relaxed) + + atomic_load_explicit( + &peer->update_in, + memory_order_relaxed) + + atomic_load_explicit( + &peer->keepalive_in, + memory_order_relaxed) + + atomic_load_explicit( + &peer->notify_in, + memory_order_relaxed) + + atomic_load_explicit( + &peer->refresh_in, + memory_order_relaxed) + + atomic_load_explicit( + &peer->dynamic_cap_in, + memory_order_relaxed), + atomic_load_explicit(&peer->open_out, + memory_order_relaxed) + + atomic_load_explicit( + &peer->update_out, + memory_order_relaxed) + + atomic_load_explicit( + &peer->keepalive_out, + memory_order_relaxed) + + atomic_load_explicit( + &peer->notify_out, + memory_order_relaxed) + + atomic_load_explicit( + &peer->refresh_out, + memory_order_relaxed) + + atomic_load_explicit( + &peer->dynamic_cap_out, + memory_order_relaxed), peer->version[afi][safi], 0, peer->obuf->count, peer_uptime(peer->uptime, timebuf, BGP_UPTIME_LEN, 0, NULL)); @@ -9657,7 +9734,8 @@ static void bgp_show_peer(struct vty *vty, struct peer *p, u_char use_json, json_object_string_add(json_neigh, "readThread", "on"); else json_object_string_add(json_neigh, "readThread", "off"); - if (p->t_write) + + if (CHECK_FLAG(p->thread_flags, PEER_THREAD_WRITES_ON)) json_object_string_add(json_neigh, "writeThread", "on"); else json_object_string_add(json_neigh, "writeThread", @@ -9683,7 +9761,10 @@ static void bgp_show_peer(struct vty *vty, struct peer *p, u_char use_json, vty_out(vty, "Peer Authentication Enabled\n"); vty_out(vty, "Read thread: %s Write thread: %s\n", - p->t_read ? "on" : "off", p->t_write ? "on" : "off"); + p->t_read ? "on" : "off", + CHECK_FLAG(p->thread_flags, PEER_THREAD_WRITES_ON) + ? "on" + : "off"); } if (p->notify.code == BGP_NOTIFY_OPEN_ERR @@ -11345,6 +11426,8 @@ void bgp_vty_init(void) install_element(BGP_NODE, &bgp_wpkt_quanta_cmd); install_element(BGP_NODE, &no_bgp_wpkt_quanta_cmd); + install_element(BGP_NODE, &bgp_rpkt_quanta_cmd); + install_element(BGP_NODE, &no_bgp_rpkt_quanta_cmd); install_element(BGP_NODE, &bgp_coalesce_time_cmd); install_element(BGP_NODE, &no_bgp_coalesce_time_cmd); diff --git a/bgpd/bgp_vty.h b/bgpd/bgp_vty.h index 59bc012661..e456f7caed 100644 --- a/bgpd/bgp_vty.h +++ b/bgpd/bgp_vty.h @@ -48,6 +48,7 @@ extern const char *afi_safi_print(afi_t, safi_t); extern const char *afi_safi_json(afi_t, safi_t); extern void bgp_config_write_update_delay(struct vty *, struct bgp *); extern void bgp_config_write_wpkt_quanta(struct vty *vty, struct bgp *bgp); +extern void bgp_config_write_rpkt_quanta(struct vty *vty, struct bgp *bgp); extern void bgp_config_write_listen(struct vty *vty, struct bgp *bgp); extern void bgp_config_write_coalesce_time(struct vty *vty, struct bgp *bgp); extern int bgp_vty_return(struct vty *vty, int ret); diff --git a/bgpd/bgpd.c b/bgpd/bgpd.c index a4952be8a6..8aaf19c7e1 100644 --- a/bgpd/bgpd.c +++ b/bgpd/bgpd.c @@ -42,6 +42,7 @@ #include "jhash.h" #include "table.h" #include "lib/json.h" +#include "frr_pthread.h" #include "bgpd/bgpd.h" #include "bgpd/bgp_table.h" @@ -75,6 +76,8 @@ #include "bgpd/bgp_bfd.h" #include "bgpd/bgp_memory.h" #include "bgpd/bgp_evpn_vty.h" +#include "bgpd/bgp_keepalives.h" +#include "bgpd/bgp_io.h" DEFINE_MTYPE_STATIC(BGPD, PEER_TX_SHUTDOWN_MSG, "Peer shutdown message (TX)"); @@ -989,10 +992,14 @@ static void peer_free(struct peer *peer) * but just to be sure.. */ bgp_timer_set(peer); - BGP_READ_OFF(peer->t_read); - BGP_WRITE_OFF(peer->t_write); + bgp_reads_off(peer); + bgp_writes_off(peer); + assert(!peer->t_write); + assert(!peer->t_read); BGP_EVENT_FLUSH(peer); + pthread_mutex_destroy(&peer->io_mtx); + /* Free connected nexthop, if present */ if (CHECK_FLAG(peer->flags, PEER_FLAG_CONFIG_NODE) && !peer_dynamic_neighbor(peer)) @@ -1135,10 +1142,11 @@ struct peer *peer_new(struct bgp *bgp) SET_FLAG(peer->sflags, PEER_STATUS_CAPABILITY_OPEN); /* Create buffers. */ - peer->ibuf = stream_new(BGP_MAX_PACKET_SIZE); + peer->ibuf = stream_fifo_new(); peer->obuf = stream_fifo_new(); + pthread_mutex_init(&peer->io_mtx, NULL); - /* We use a larger buffer for peer->work in the event that: + /* We use a larger buffer for peer->obuf_work in the event that: * - We RX a BGP_UPDATE where the attributes alone are just * under BGP_MAX_PACKET_SIZE * - The user configures an outbound route-map that does many as-path @@ -1152,11 +1160,11 @@ struct peer *peer_new(struct bgp *bgp) * bounds * checking for every single attribute as we construct an UPDATE. */ - peer->work = + peer->obuf_work = stream_new(BGP_MAX_PACKET_SIZE + BGP_MAX_PACKET_SIZE_OVERFLOW); + peer->ibuf_work = stream_new(BGP_MAX_PACKET_SIZE * BGP_READ_PACKET_MAX); peer->scratch = stream_new(BGP_MAX_PACKET_SIZE); - bgp_sync_init(peer); /* Get service port number. */ @@ -1466,6 +1474,11 @@ struct peer *peer_create(union sockunion *su, const char *conf_if, listnode_add_sort(bgp->peer, peer); hash_get(bgp->peerhash, peer, hash_alloc_intern); + /* Adjust update-group coalesce timer heuristics for # peers. */ + long ct = BGP_DEFAULT_SUBGROUP_COALESCE_TIME + + (bgp->peer->count * BGP_PEER_ADJUST_SUBGROUP_COALESCE_TIME); + bgp->coalesce_time = MIN(BGP_MAX_SUBGROUP_COALESCE_TIME, ct); + active = peer_active(peer); /* Last read and reset time set */ @@ -2082,6 +2095,11 @@ int peer_delete(struct peer *peer) bgp = peer->bgp; accept_peer = CHECK_FLAG(peer->sflags, PEER_STATUS_ACCEPT_PEER); + bgp_reads_off(peer); + bgp_writes_off(peer); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON)); + assert(!CHECK_FLAG(peer->thread_flags, PEER_THREAD_READS_ON)); + if (CHECK_FLAG(peer->sflags, PEER_STATUS_NSF_WAIT)) peer_nsf_stop(peer); @@ -2143,7 +2161,7 @@ int peer_delete(struct peer *peer) /* Buffers. */ if (peer->ibuf) { - stream_free(peer->ibuf); + stream_fifo_free(peer->ibuf); peer->ibuf = NULL; } @@ -2152,9 +2170,14 @@ int peer_delete(struct peer *peer) peer->obuf = NULL; } - if (peer->work) { - stream_free(peer->work); - peer->work = NULL; + if (peer->ibuf_work) { + stream_free(peer->ibuf_work); + peer->ibuf_work = NULL; + } + + if (peer->obuf_work) { + stream_free(peer->obuf_work); + peer->obuf_work = NULL; } if (peer->scratch) { @@ -2890,7 +2913,10 @@ static struct bgp *bgp_create(as_t *as, const char *name, bgp->restart_time, &bgp->t_startup); } - bgp->wpkt_quanta = BGP_WRITE_PACKET_MAX; + atomic_store_explicit(&bgp->wpkt_quanta, BGP_WRITE_PACKET_MAX, + memory_order_relaxed); + atomic_store_explicit(&bgp->rpkt_quanta, BGP_READ_PACKET_MAX, + memory_order_relaxed); bgp->coalesce_time = BGP_DEFAULT_SUBGROUP_COALESCE_TIME; QOBJ_REG(bgp, bgp); @@ -7174,6 +7200,8 @@ int bgp_config_write(struct vty *vty) /* write quanta */ bgp_config_write_wpkt_quanta(vty, bgp); + /* read quanta */ + bgp_config_write_rpkt_quanta(vty, bgp); /* coalesce time */ bgp_config_write_coalesce_time(vty, bgp); @@ -7381,12 +7409,45 @@ static const struct cmd_variable_handler bgp_viewvrf_var_handlers[] = { {.completions = NULL}, }; +static void bgp_pthreads_init() +{ + frr_pthread_init(); + + frr_pthread_new("BGP i/o thread", PTHREAD_IO, bgp_io_start, + bgp_io_stop); + frr_pthread_new("BGP keepalives thread", PTHREAD_KEEPALIVES, + bgp_keepalives_start, bgp_keepalives_stop); + + /* pre-run initialization */ + bgp_keepalives_init(); + bgp_io_init(); +} + +void bgp_pthreads_run() +{ + pthread_attr_t attr; + pthread_attr_init(&attr); + pthread_attr_setschedpolicy(&attr, SCHED_FIFO); + + frr_pthread_run(PTHREAD_IO, &attr, NULL); + frr_pthread_run(PTHREAD_KEEPALIVES, &attr, NULL); +} + +void bgp_pthreads_finish() +{ + frr_pthread_stop_all(); + frr_pthread_finish(); +} + void bgp_init(void) { /* allocates some vital data structures used by peer commands in * vty_init */ + /* pre-init pthreads */ + bgp_pthreads_init(); + /* Init zebra. */ bgp_zebra_init(bm->master); diff --git a/bgpd/bgpd.h b/bgpd/bgpd.h index 36bdaf0125..e5e363ef52 100644 --- a/bgpd/bgpd.h +++ b/bgpd/bgpd.h @@ -22,6 +22,8 @@ #define _QUAGGA_BGPD_H #include "qobj.h" +#include <pthread.h> + #include "lib/json.h" #include "vrf.h" #include "vty.h" @@ -98,6 +100,10 @@ struct bgp_master { /* BGP thread master. */ struct thread_master *master; +/* BGP pthreads. */ +#define PTHREAD_IO (1 << 1) +#define PTHREAD_KEEPALIVES (1 << 2) + /* work queues */ struct work_queue *process_main_queue; @@ -372,7 +378,9 @@ struct bgp { #define BGP_FLAG_IBGP_MULTIPATH_SAME_CLUSTERLEN (1 << 0) } maxpaths[AFI_MAX][SAFI_MAX]; - u_int32_t wpkt_quanta; /* per peer packet quanta to write */ + _Atomic uint32_t wpkt_quanta; // max # packets to write per i/o cycle + _Atomic uint32_t rpkt_quanta; // max # packets to read per i/o cycle + u_int32_t coalesce_time; u_int32_t addpath_tx_id; @@ -583,12 +591,17 @@ struct peer { struct in_addr local_id; /* Packet receive and send buffer. */ - struct stream *ibuf; - struct stream_fifo *obuf; - struct stream *work; + pthread_mutex_t io_mtx; // guards ibuf, obuf + struct stream_fifo *ibuf; // packets waiting to be processed + struct stream_fifo *obuf; // packets waiting to be written + + struct stream *ibuf_work; // WiP buffer used by bgp_read() only + struct stream *obuf_work; // WiP buffer used to construct packets + + struct stream *curr; // the current packet being parsed /* We use a separate stream to encode MP_REACH_NLRI for efficient - * NLRI packing. peer->work stores all the other attributes. The + * NLRI packing. peer->obuf_work stores all the other attributes. The * actual packet is then constructed by concatenating the two. */ struct stream *scratch; @@ -776,49 +789,57 @@ struct peer { (CHECK_FLAG(peer->config, PEER_CONFIG_TIMER) \ || CHECK_FLAG(peer->config, PEER_GROUP_CONFIG_TIMER)) - u_int32_t holdtime; - u_int32_t keepalive; - u_int32_t connect; - u_int32_t routeadv; + _Atomic uint32_t holdtime; + _Atomic uint32_t keepalive; + _Atomic uint32_t connect; + _Atomic uint32_t routeadv; /* Timer values. */ - u_int32_t v_start; - u_int32_t v_connect; - u_int32_t v_holdtime; - u_int32_t v_keepalive; - u_int32_t v_routeadv; - u_int32_t v_pmax_restart; - u_int32_t v_gr_restart; + _Atomic uint32_t v_start; + _Atomic uint32_t v_connect; + _Atomic uint32_t v_holdtime; + _Atomic uint32_t v_keepalive; + _Atomic uint32_t v_routeadv; + _Atomic uint32_t v_pmax_restart; + _Atomic uint32_t v_gr_restart; /* Threads. */ struct thread *t_read; struct thread *t_write; struct thread *t_start; + struct thread *t_connect_check_r; + struct thread *t_connect_check_w; struct thread *t_connect; struct thread *t_holdtime; - struct thread *t_keepalive; struct thread *t_routeadv; struct thread *t_pmax_restart; struct thread *t_gr_restart; struct thread *t_gr_stale; - + struct thread *t_generate_updgrp_packets; + struct thread *t_process_packet; + + /* Thread flags. */ + _Atomic uint16_t thread_flags; +#define PEER_THREAD_WRITES_ON (1 << 0) +#define PEER_THREAD_READS_ON (1 << 1) +#define PEER_THREAD_KEEPALIVES_ON (1 << 2) /* workqueues */ struct work_queue *clear_node_queue; /* Statistics field */ - u_int32_t open_in; /* Open message input count */ - u_int32_t open_out; /* Open message output count */ - u_int32_t update_in; /* Update message input count */ - u_int32_t update_out; /* Update message ouput count */ - time_t update_time; /* Update message received time. */ - u_int32_t keepalive_in; /* Keepalive input count */ - u_int32_t keepalive_out; /* Keepalive output count */ - u_int32_t notify_in; /* Notify input count */ - u_int32_t notify_out; /* Notify output count */ - u_int32_t refresh_in; /* Route Refresh input count */ - u_int32_t refresh_out; /* Route Refresh output count */ - u_int32_t dynamic_cap_in; /* Dynamic Capability input count. */ - u_int32_t dynamic_cap_out; /* Dynamic Capability output count. */ + _Atomic uint32_t open_in; /* Open message input count */ + _Atomic uint32_t open_out; /* Open message output count */ + _Atomic uint32_t update_in; /* Update message input count */ + _Atomic uint32_t update_out; /* Update message ouput count */ + _Atomic time_t update_time; /* Update message received time. */ + _Atomic uint32_t keepalive_in; /* Keepalive input count */ + _Atomic uint32_t keepalive_out; /* Keepalive output count */ + _Atomic uint32_t notify_in; /* Notify input count */ + _Atomic uint32_t notify_out; /* Notify output count */ + _Atomic uint32_t refresh_in; /* Route Refresh input count */ + _Atomic uint32_t refresh_out; /* Route Refresh output count */ + _Atomic uint32_t dynamic_cap_in; /* Dynamic Capability input count. */ + _Atomic uint32_t dynamic_cap_out; /* Dynamic Capability output count. */ /* BGP state count */ u_int32_t established; /* Established */ @@ -831,8 +852,10 @@ struct peer { /* Syncronization list and time. */ struct bgp_synchronize *sync[AFI_MAX][SAFI_MAX]; time_t synctime; - time_t last_write; /* timestamp when the last msg was written */ - time_t last_update; /* timestamp when the last UPDATE msg was written */ + /* timestamp when the last UPDATE msg was written */ + _Atomic time_t last_write; + /* timestamp when the last msg was written */ + _Atomic time_t last_update; /* Send prefix count. */ unsigned long scount[AFI_MAX][SAFI_MAX]; @@ -843,9 +866,6 @@ struct peer { /* Notify data. */ struct bgp_notify notify; - /* Whole packet size to be read. */ - unsigned long packet_size; - /* Filter structure. */ struct bgp_filter filter[AFI_MAX][SAFI_MAX]; @@ -1139,8 +1159,8 @@ enum bgp_clear_type { }; /* Macros. */ -#define BGP_INPUT(P) ((P)->ibuf) -#define BGP_INPUT_PNT(P) (STREAM_PNT(BGP_INPUT(P))) +#define BGP_INPUT(P) ((P)->curr) +#define BGP_INPUT_PNT(P) (stream_pnt(BGP_INPUT(P))) #define BGP_IS_VALID_STATE_FOR_NOTIF(S) \ (((S) == OpenSent) || ((S) == OpenConfirm) || ((S) == Established)) @@ -1251,6 +1271,8 @@ extern int bgp_config_write(struct vty *); extern void bgp_master_init(struct thread_master *master); extern void bgp_init(void); +extern void bgp_pthreads_run(void); +extern void bgp_pthreads_finish(void); extern void bgp_route_map_init(void); extern void bgp_session_reset(struct peer *); diff --git a/bgpd/rfapi/rfapi.c b/bgpd/rfapi/rfapi.c index 15a29442f4..fa3da9c283 100644 --- a/bgpd/rfapi/rfapi.c +++ b/bgpd/rfapi/rfapi.c @@ -1304,18 +1304,31 @@ static int rfapi_open_inner(struct rfapi_descriptor *rfd, struct bgp *bgp, rfd->peer = peer_new(bgp); rfd->peer->status = Established; /* keep bgp core happy */ bgp_sync_delete(rfd->peer); /* don't need these */ - if (rfd->peer->ibuf) { - stream_free(rfd->peer->ibuf); /* don't need it */ + + /* + * since this peer is not on the I/O thread, this lock is not strictly + * necessary, but serves as a reminder to those who may meddle... + */ + pthread_mutex_lock(&rfd->peer->io_mtx); + { + // we don't need any I/O related facilities + if (rfd->peer->ibuf) + stream_fifo_free(rfd->peer->ibuf); + if (rfd->peer->obuf) + stream_fifo_free(rfd->peer->obuf); + + if (rfd->peer->ibuf_work) + stream_free(rfd->peer->ibuf_work); + if (rfd->peer->obuf_work) + stream_free(rfd->peer->obuf_work); + rfd->peer->ibuf = NULL; - } - if (rfd->peer->obuf) { - stream_fifo_free(rfd->peer->obuf); /* don't need it */ rfd->peer->obuf = NULL; + rfd->peer->obuf_work = NULL; + rfd->peer->ibuf_work = NULL; } - if (rfd->peer->work) { - stream_free(rfd->peer->work); /* don't need it */ - rfd->peer->work = NULL; - } + pthread_mutex_unlock(&rfd->peer->io_mtx); + { /* base code assumes have valid host pointer */ char buf[BUFSIZ]; buf[0] = 0; diff --git a/bgpd/rfapi/vnc_zebra.c b/bgpd/rfapi/vnc_zebra.c index 5c71df238f..07be7833b6 100644 --- a/bgpd/rfapi/vnc_zebra.c +++ b/bgpd/rfapi/vnc_zebra.c @@ -183,22 +183,32 @@ static void vnc_redistribute_add(struct prefix *p, u_int32_t metric, vncHD1VR.peer->status = Established; /* keep bgp core happy */ bgp_sync_delete(vncHD1VR.peer); /* don't need these */ - if (vncHD1VR.peer->ibuf) { - stream_free(vncHD1VR.peer - ->ibuf); /* don't need it */ + + /* + * since this peer is not on the I/O thread, this lock + * is not strictly necessary, but serves as a reminder + * to those who may meddle... + */ + pthread_mutex_lock(&vncHD1VR.peer->io_mtx); + { + // we don't need any I/O related facilities + if (vncHD1VR.peer->ibuf) + stream_fifo_free(vncHD1VR.peer->ibuf); + if (vncHD1VR.peer->obuf) + stream_fifo_free(vncHD1VR.peer->obuf); + + if (vncHD1VR.peer->ibuf_work) + stream_free(vncHD1VR.peer->ibuf_work); + if (vncHD1VR.peer->obuf_work) + stream_free(vncHD1VR.peer->obuf_work); + vncHD1VR.peer->ibuf = NULL; - } - if (vncHD1VR.peer->obuf) { - stream_fifo_free( - vncHD1VR.peer - ->obuf); /* don't need it */ vncHD1VR.peer->obuf = NULL; + vncHD1VR.peer->obuf_work = NULL; + vncHD1VR.peer->ibuf_work = NULL; } - if (vncHD1VR.peer->work) { - stream_free(vncHD1VR.peer - ->work); /* don't need it */ - vncHD1VR.peer->work = NULL; - } + pthread_mutex_unlock(&vncHD1VR.peer->io_mtx); + /* base code assumes have valid host pointer */ vncHD1VR.peer->host = XSTRDUP(MTYPE_BGP_PEER_HOST, ".zebra."); diff --git a/debianpkg/frr-dbg.lintian-overrides b/debianpkg/frr-dbg.lintian-overrides new file mode 100644 index 0000000000..7880bba29a --- /dev/null +++ b/debianpkg/frr-dbg.lintian-overrides @@ -0,0 +1 @@ +frr-dbg: debug-file-with-no-debug-symbols usr/lib/debug/usr/lib/libfrrfpm_pb.so.0.0.0 diff --git a/eigrpd/eigrp_packet.c b/eigrpd/eigrp_packet.c index ecabee4aa7..ea6f1f3f62 100644 --- a/eigrpd/eigrp_packet.c +++ b/eigrpd/eigrp_packet.c @@ -425,7 +425,7 @@ int eigrp_write(struct thread *thread) iov[0].iov_base = (char *)&iph; iov[0].iov_len = iph.ip_hl << EIGRP_WRITE_IPHL_SHIFT; - iov[1].iov_base = STREAM_PNT(ep->s); + iov[1].iov_base = stream_pnt(ep->s); iov[1].iov_len = ep->length; /* send final fragment (could be first) */ @@ -555,7 +555,7 @@ int eigrp_read(struct thread *thread) by eigrp_recv_packet() to be correct). */ stream_forward_getp(ibuf, (iph->ip_hl * 4)); - eigrph = (struct eigrp_header *)STREAM_PNT(ibuf); + eigrph = (struct eigrp_header *)stream_pnt(ibuf); if (IS_DEBUG_EIGRP_TRANSMIT(0, RECV) && IS_DEBUG_EIGRP_TRANSMIT(0, PACKET_DETAIL)) diff --git a/lib/stream.h b/lib/stream.h index 1048180fac..4d387f9564 100644 --- a/lib/stream.h +++ b/lib/stream.h @@ -123,10 +123,15 @@ struct stream_fifo { #define STREAM_CONCAT_REMAIN(S1, S2, size) ((size) - (S1)->endp - (S2)->endp) /* deprecated macros - do not use in new code */ +#if CONFDATE > 20181128 +CPP_NOTICE("lib: time to remove deprecated stream.h macros") +#endif #define STREAM_PNT(S) stream_pnt((S)) -#define STREAM_DATA(S) ((S)->data) #define STREAM_REMAIN(S) STREAM_WRITEABLE((S)) +/* this macro is deprecated, but not slated for removal anytime soon */ +#define STREAM_DATA(S) ((S)->data) + /* Stream prototypes. * For stream_{put,get}S, the S suffix mean: * diff --git a/lib/thread.c b/lib/thread.c index cb5d1d47ae..d26db88550 100644 --- a/lib/thread.c +++ b/lib/thread.c @@ -1045,7 +1045,8 @@ static void do_thread_cancel(struct thread_master *master) if (queue) { assert(thread->index >= 0); - pqueue_remove(thread, queue); + assert(thread == queue->array[thread->index]); + pqueue_remove_at(thread->index, queue); } else if (list) { thread_list_delete(list, thread); } else if (thread_array) { diff --git a/ospfd/.gitignore b/ospfd/.gitignore index 018a363a93..f0d800efb4 100644 --- a/ospfd/.gitignore +++ b/ospfd/.gitignore @@ -15,4 +15,4 @@ TAGS *~ *.loT *.a -*.clippy.c +*clippy.c diff --git a/ospfd/ospf_dump.c b/ospfd/ospf_dump.c index 6876054a63..6a410f4ed3 100644 --- a/ospfd/ospf_dump.c +++ b/ospfd/ospf_dump.c @@ -230,7 +230,7 @@ static void ospf_packet_hello_dump(struct stream *s, u_int16_t length) struct ospf_hello *hello; int i; - hello = (struct ospf_hello *)STREAM_PNT(s); + hello = (struct ospf_hello *)stream_pnt(s); zlog_debug("Hello"); zlog_debug(" NetworkMask %s", inet_ntoa(hello->network_mask)); @@ -278,7 +278,7 @@ static void ospf_router_lsa_dump(struct stream *s, u_int16_t length) struct router_lsa *rl; int i, len; - rl = (struct router_lsa *)STREAM_PNT(s); + rl = (struct router_lsa *)stream_pnt(s); zlog_debug(" Router-LSA"); zlog_debug(" flags %s", @@ -303,7 +303,7 @@ static void ospf_network_lsa_dump(struct stream *s, u_int16_t length) struct network_lsa *nl; int i, cnt; - nl = (struct network_lsa *)STREAM_PNT(s); + nl = (struct network_lsa *)stream_pnt(s); cnt = (ntohs(nl->header.length) - (OSPF_LSA_HEADER_SIZE + 4)) / 4; zlog_debug(" Network-LSA"); @@ -325,7 +325,7 @@ static void ospf_summary_lsa_dump(struct stream *s, u_int16_t length) int size; int i; - sl = (struct summary_lsa *)STREAM_PNT(s); + sl = (struct summary_lsa *)stream_pnt(s); zlog_debug(" Summary-LSA"); zlog_debug(" Network Mask %s", inet_ntoa(sl->mask)); @@ -342,7 +342,7 @@ static void ospf_as_external_lsa_dump(struct stream *s, u_int16_t length) int size; int i; - al = (struct as_external_lsa *)STREAM_PNT(s); + al = (struct as_external_lsa *)stream_pnt(s); zlog_debug(" %s", ospf_lsa_type_msg[al->header.type].str); zlog_debug(" Network Mask %s", inet_ntoa(al->mask)); @@ -366,7 +366,7 @@ static void ospf_lsa_header_list_dump(struct stream *s, u_int16_t length) /* LSA Headers. */ while (length > 0) { - lsa = (struct lsa_header *)STREAM_PNT(s); + lsa = (struct lsa_header *)stream_pnt(s); ospf_lsa_header_dump(lsa); stream_forward_getp(s, OSPF_LSA_HEADER_SIZE); @@ -382,7 +382,7 @@ static void ospf_packet_db_desc_dump(struct stream *s, u_int16_t length) u_int32_t gp; gp = stream_get_getp(s); - dd = (struct ospf_db_desc *)STREAM_PNT(s); + dd = (struct ospf_db_desc *)stream_pnt(s); zlog_debug("Database Description"); zlog_debug(" Interface MTU %d", ntohs(dd->mtu)); @@ -452,7 +452,7 @@ static void ospf_packet_ls_upd_dump(struct stream *s, u_int16_t length) break; } - lsa = (struct lsa_header *)STREAM_PNT(s); + lsa = (struct lsa_header *)stream_pnt(s); lsa_len = ntohs(lsa->length); ospf_lsa_header_dump(lsa); @@ -566,7 +566,7 @@ void ospf_packet_dump(struct stream *s) gp = stream_get_getp(s); /* OSPF Header dump. */ - ospfh = (struct ospf_header *)STREAM_PNT(s); + ospfh = (struct ospf_header *)stream_pnt(s); /* Until detail flag is set, return. */ if (!(term_debug_ospf_packet[ospfh->type - 1] & OSPF_DEBUG_DETAIL)) diff --git a/ospfd/ospf_lsa.c b/ospfd/ospf_lsa.c index c28e500d5b..a2961992de 100644 --- a/ospfd/ospf_lsa.c +++ b/ospfd/ospf_lsa.c @@ -49,6 +49,7 @@ #include "ospfd/ospf_route.h" #include "ospfd/ospf_ase.h" #include "ospfd/ospf_zebra.h" +#include "ospfd/ospf_abr.h" u_int32_t get_metric(u_char *metric) @@ -437,7 +438,7 @@ static char link_info_set(struct stream *s, struct in_addr id, if (ret == OSPF_MAX_LSA_SIZE) { zlog_warn( "%s: Out of space in LSA stream, left %zd, size %zd", - __func__, STREAM_REMAIN(s), STREAM_SIZE(s)); + __func__, STREAM_WRITEABLE(s), STREAM_SIZE(s)); return 0; } } @@ -2503,6 +2504,7 @@ static struct ospf_lsa *ospf_external_lsa_install(struct ospf *ospf, * abr_task. */ ospf_translated_nssa_refresh(ospf, new, NULL); + ospf_schedule_abr_task(ospf); } } diff --git a/ospfd/ospf_opaque.c b/ospfd/ospf_opaque.c index 5a1f28b036..6f9da92542 100644 --- a/ospfd/ospf_opaque.c +++ b/ospfd/ospf_opaque.c @@ -1181,7 +1181,7 @@ void ospf_opaque_lsa_dump(struct stream *s, u_int16_t length) { struct ospf_lsa lsa; - lsa.data = (struct lsa_header *)STREAM_PNT(s); + lsa.data = (struct lsa_header *)stream_pnt(s); show_opaque_info_detail(NULL, &lsa); return; } diff --git a/ospfd/ospf_packet.c b/ospfd/ospf_packet.c index 33792bbff3..65e9cac837 100644 --- a/ospfd/ospf_packet.c +++ b/ospfd/ospf_packet.c @@ -617,7 +617,7 @@ static void ospf_write_frags(int fd, struct ospf_packet *op, struct ip *iph, iph->ip_off += offset; stream_forward_getp(op->s, iovp->iov_len); - iovp->iov_base = STREAM_PNT(op->s); + iovp->iov_base = stream_pnt(op->s); } /* setup for final fragment */ @@ -763,7 +763,7 @@ static int ospf_write(struct thread *thread) iov[0].iov_base = (char *)&iph; iov[0].iov_len = iph.ip_hl << OSPF_WRITE_IPHL_SHIFT; - iov[1].iov_base = STREAM_PNT(op->s); + iov[1].iov_base = stream_pnt(op->s); iov[1].iov_len = op->length; #ifdef GNU_LINUX @@ -891,7 +891,7 @@ static void ospf_hello(struct ip *iph, struct ospf_header *ospfh, /* increment statistics. */ oi->hello_in++; - hello = (struct ospf_hello *)STREAM_PNT(s); + hello = (struct ospf_hello *)stream_pnt(s); /* If Hello is myself, silently discard. */ if (IPV4_ADDR_SAME(&ospfh->router_id, &oi->ospf->router_id)) { @@ -1119,7 +1119,7 @@ static void ospf_db_desc_proc(struct stream *s, struct ospf_interface *oi, stream_forward_getp(s, OSPF_DB_DESC_MIN_SIZE); for (size -= OSPF_DB_DESC_MIN_SIZE; size >= OSPF_LSA_HEADER_SIZE; size -= OSPF_LSA_HEADER_SIZE) { - lsah = (struct lsa_header *)STREAM_PNT(s); + lsah = (struct lsa_header *)stream_pnt(s); stream_forward_getp(s, OSPF_LSA_HEADER_SIZE); /* Unknown LS type. */ @@ -1268,7 +1268,7 @@ static void ospf_db_desc(struct ip *iph, struct ospf_header *ospfh, /* Increment statistics. */ oi->db_desc_in++; - dd = (struct ospf_db_desc *)STREAM_PNT(s); + dd = (struct ospf_db_desc *)stream_pnt(s); nbr = ospf_nbr_lookup(oi, iph, ospfh); if (nbr == NULL) { @@ -1661,7 +1661,7 @@ static struct list *ospf_ls_upd_list_lsa(struct ospf_neighbor *nbr, for (; size >= OSPF_LSA_HEADER_SIZE && count > 0; size -= length, stream_forward_getp(s, length), count--) { - lsah = (struct lsa_header *)STREAM_PNT(s); + lsah = (struct lsa_header *)stream_pnt(s); length = ntohs(lsah->length); if (length > size) { @@ -2219,10 +2219,10 @@ static void ospf_ls_ack(struct ip *iph, struct ospf_header *ospfh, struct ospf_lsa *lsa, *lsr; lsa = ospf_lsa_new(); - lsa->data = (struct lsa_header *)STREAM_PNT(s); + lsa->data = (struct lsa_header *)stream_pnt(s); lsa->vrf_id = oi->ospf->vrf_id; - /* lsah = (struct lsa_header *) STREAM_PNT (s); */ + /* lsah = (struct lsa_header *) stream_pnt (s); */ size -= OSPF_LSA_HEADER_SIZE; stream_forward_getp(s, OSPF_LSA_HEADER_SIZE); @@ -2936,7 +2936,7 @@ int ospf_read(struct thread *thread) by ospf_recv_packet() to be correct). */ stream_forward_getp(ibuf, iph->ip_hl * 4); - ospfh = (struct ospf_header *)STREAM_PNT(ibuf); + ospfh = (struct ospf_header *)stream_pnt(ibuf); if (MSG_OK != ospf_packet_examin( ospfh, stream_get_endp(ibuf) - stream_get_getp(ibuf))) diff --git a/pimd/pim_msdp_packet.c b/pimd/pim_msdp_packet.c index 11efc158e9..978d979245 100644 --- a/pimd/pim_msdp_packet.c +++ b/pimd/pim_msdp_packet.c @@ -221,7 +221,7 @@ int pim_msdp_write(struct thread *thread) writenum = stream_get_endp(s) - stream_get_getp(s); /* Call write() system call */ - num = write(mp->fd, STREAM_PNT(s), writenum); + num = write(mp->fd, stream_pnt(s), writenum); if (num < 0) { /* write failed either retry needed or error */ if (ERRNO_IO_RETRY(errno)) { diff --git a/tests/bgpd/test_aspath.c b/tests/bgpd/test_aspath.c index 46462d79c4..56808bc8ad 100644 --- a/tests/bgpd/test_aspath.c +++ b/tests/bgpd/test_aspath.c @@ -29,6 +29,7 @@ #include "bgpd/bgpd.h" #include "bgpd/bgp_aspath.h" #include "bgpd/bgp_attr.h" +#include "bgpd/bgp_packet.h" #define VT100_RESET "\x1b[0m" #define VT100_RED "\x1b[31m" @@ -1273,20 +1274,20 @@ static int handle_attr_test(struct aspath_tests *t) asp = make_aspath(t->segment->asdata, t->segment->len, 0); - peer.ibuf = stream_new(BGP_MAX_PACKET_SIZE); + peer.curr = stream_new(BGP_MAX_PACKET_SIZE); peer.obuf = stream_fifo_new(); peer.bgp = &bgp; peer.host = (char *)"none"; peer.fd = -1; peer.cap = t->cap; - stream_write(peer.ibuf, t->attrheader, t->len); - datalen = aspath_put(peer.ibuf, asp, t->as4 == AS4_DATA); + stream_write(peer.curr, t->attrheader, t->len); + datalen = aspath_put(peer.curr, asp, t->as4 == AS4_DATA); if (t->old_segment) { char dummyaspath[] = {BGP_ATTR_FLAG_TRANS, BGP_ATTR_AS_PATH, t->old_segment->len}; - stream_write(peer.ibuf, dummyaspath, sizeof(dummyaspath)); - stream_write(peer.ibuf, t->old_segment->asdata, + stream_write(peer.curr, dummyaspath, sizeof(dummyaspath)); + stream_write(peer.curr, t->old_segment->asdata, t->old_segment->len); datalen += sizeof(dummyaspath) + t->old_segment->len; } diff --git a/tests/bgpd/test_capability.c b/tests/bgpd/test_capability.c index e8700a8b4a..a5092708e2 100644 --- a/tests/bgpd/test_capability.c +++ b/tests/bgpd/test_capability.c @@ -796,14 +796,15 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) int oldfailed = failed; int len = t->len; #define RANDOM_FUZZ 35 - stream_reset(peer->ibuf); - stream_put(peer->ibuf, NULL, RANDOM_FUZZ); - stream_set_getp(peer->ibuf, RANDOM_FUZZ); + + stream_reset(peer->curr); + stream_put(peer->curr, NULL, RANDOM_FUZZ); + stream_set_getp(peer->curr, RANDOM_FUZZ); switch (type) { case CAPABILITY: - stream_putc(peer->ibuf, BGP_OPEN_OPT_CAP); - stream_putc(peer->ibuf, t->len); + stream_putc(peer->curr, BGP_OPEN_OPT_CAP); + stream_putc(peer->curr, t->len); break; case DYNCAP: /* for (i = 0; i < BGP_MARKER_SIZE; i++) @@ -812,7 +813,7 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) stream_putc (s, BGP_MSG_CAPABILITY);*/ break; } - stream_write(peer->ibuf, t->data, t->len); + stream_write(peer->curr, t->data, t->len); printf("%s: %s\n", t->name, t->desc); @@ -825,7 +826,7 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) as4 = peek_for_as4_capability(peer, len); printf("peek_for_as4: as4 is %u\n", as4); /* and it should leave getp as it found it */ - assert(stream_get_getp(peer->ibuf) == RANDOM_FUZZ); + assert(stream_get_getp(peer->curr) == RANDOM_FUZZ); ret = bgp_open_option_parse(peer, len, &capability); break; @@ -837,7 +838,7 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) exit(1); } - if (!ret && t->validate_afi) { + if (ret != BGP_Stop && t->validate_afi) { afi_t afi; safi_t safi; @@ -865,10 +866,20 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) failed++; } + /* + * Some of the functions used return BGP_Stop on error and some return + * -1. If we have -1, keep it; if we have BGP_Stop, transform it to the + * correct pass/fail code + */ + if (ret != -1) + ret = (ret == BGP_Stop) ? -1 : 0; + printf("parsed?: %s\n", ret ? "no" : "yes"); - if (ret != t->parses) + if (ret != t->parses) { + printf("t->parses: %d\nret: %d\n", t->parses, ret); failed++; + } if (tty) printf("%s", @@ -919,6 +930,8 @@ int main(void) peer->afc_adv[i][j] = 1; } + peer->curr = stream_new(BGP_MAX_PACKET_SIZE); + i = 0; while (mp_segments[i].name) parse_test(peer, &mp_segments[i++], CAPABILITY); diff --git a/tests/bgpd/test_mp_attr.c b/tests/bgpd/test_mp_attr.c index 30d5fdd6cd..6df784b984 100644 --- a/tests/bgpd/test_mp_attr.c +++ b/tests/bgpd/test_mp_attr.c @@ -36,6 +36,7 @@ #include "bgpd/bgp_packet.h" #include "bgpd/bgp_mplsvpn.h" #include "bgpd/bgp_nexthop.h" +#include "bgpd/bgp_vty.h" #define VT100_RESET "\x1b[0m" #define VT100_RED "\x1b[31m" @@ -1045,11 +1046,11 @@ static void parse_test(struct peer *peer, struct test_segment *t, int type) .startp = BGP_INPUT_PNT(peer), }; #define RANDOM_FUZZ 35 - stream_reset(peer->ibuf); - stream_put(peer->ibuf, NULL, RANDOM_FUZZ); - stream_set_getp(peer->ibuf, RANDOM_FUZZ); + stream_reset(peer->curr); + stream_put(peer->curr, NULL, RANDOM_FUZZ); + stream_set_getp(peer->curr, RANDOM_FUZZ); - stream_write(peer->ibuf, t->data, t->len); + stream_write(peer->curr, t->data, t->len); printf("%s: %s\n", t->name, t->desc); @@ -1097,7 +1098,9 @@ int main(void) term_bgp_debug_as4 = -1UL; qobj_init(); - master = thread_master_create(NULL); + cmd_init(0); + bgp_vty_init(); + master = thread_master_create("test mp attr"); bgp_master_init(master); vrf_init(NULL, NULL, NULL, NULL); bgp_option_set(BGP_OPT_NO_LISTEN); @@ -1112,6 +1115,7 @@ int main(void) peer = peer_create_accept(bgp); peer->host = (char *)"foo"; peer->status = Established; + peer->curr = stream_new(BGP_MAX_PACKET_SIZE); for (i = AFI_IP; i < AFI_MAX; i++) for (j = SAFI_UNICAST; j < SAFI_MAX; j++) { diff --git a/tests/bgpd/test_packet.c b/tests/bgpd/test_packet.c index 298dd1e185..c58a85eed3 100644 --- a/tests/bgpd/test_packet.c +++ b/tests/bgpd/test_packet.c @@ -80,6 +80,6 @@ int main(int argc, char *argv[]) peer->fd = open(argv[1], O_RDONLY|O_NONBLOCK); t.arg = peer; peer->t_read = &t; - - printf("bgp_read_packet returns: %d\n", bgp_read(&t)); + + // printf("bgp_read_packet returns: %d\n", bgp_read(&t)); } diff --git a/zebra/zebra_fpm.c b/zebra/zebra_fpm.c index 0ffa55f1e4..7448292d9f 100644 --- a/zebra/zebra_fpm.c +++ b/zebra/zebra_fpm.c @@ -996,7 +996,7 @@ static int zfpm_write_cb(struct thread *thread) break; bytes_written = - write(zfpm_g->sock, STREAM_PNT(s), bytes_to_write); + write(zfpm_g->sock, stream_pnt(s), bytes_to_write); zfpm_g->stats.write_calls++; num_writes++; |