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_keepalives.c
+ bgp_keepalives.c bgp_io.c
noinst_HEADERS = \
bgp_memory.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_keepalives.h \
+ bgp_io.h
bgpd_SOURCES = bgp_main.c
bgpd_LDADD = libbgp.a $(BGP_VNC_RFP_LIB) ../lib/libfrr.la @LIBCAP@ @LIBM@
#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))
BGP_TIMER_OFF(peer->t_holdtime);
peer_keepalives_off(peer);
BGP_TIMER_OFF(peer->t_routeadv);
+ BGP_TIMER_OFF(peer->t_generate_updgrp_packets);
/* Stream reset. */
peer->packet_size = 0;
--- /dev/null
+/*
+ BGP I/O.
+ Implements a consumer thread to flush packets destined for remote peers.
+
+ Copyright (C) 2017 Cumulus Networks
+
+ 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
+ */
+
+#include <zebra.h>
+#include <sys/time.h>
+#include <pthread.h>
+
+#include "thread.h"
+#include "hash.h"
+#include "stream.h"
+#include "memory.h"
+#include "log.h"
+#include "monotime.h"
+#include "network.h"
+#include "frr_pthread.h"
+
+#include "bgpd/bgpd.h"
+#include "bgpd/bgp_io.h"
+#include "bgpd/bgp_debug.h"
+#include "bgpd/bgp_packet.h"
+#include "bgpd/bgp_fsm.h"
+
+static int bgp_write(struct peer *);
+static void peer_process_writes(struct hash_backet *, void *);
+
+bool bgp_packet_writes_thread_run = false;
+
+/* Hash table of peers to operate on, associated synchronization primitives and
+ * hash table callbacks.
+ * ------------------------------------------------------------------------ */
+static struct hash *peerhash;
+/* Mutex to protect hash table */
+static pthread_mutex_t *peerhash_mtx;
+/* Condition variable used to notify the write thread that there is work to do
+ */
+static pthread_cond_t *write_cond;
+
+static unsigned int peer_hash_key_make(void *p)
+{
+ struct peer *peer = p;
+ return sockunion_hash(&peer->su);
+}
+
+static int peer_hash_cmp(const void *p1, const void *p2)
+{
+ const struct peer *peer1 = p1;
+ const struct peer *peer2 = p2;
+ return (sockunion_same(&peer1->su, &peer2->su)
+ && CHECK_FLAG(peer1->flags, PEER_FLAG_CONFIG_NODE)
+ == CHECK_FLAG(peer2->flags, PEER_FLAG_CONFIG_NODE));
+}
+/* ------------------------------------------------------------------------ */
+
+void peer_writes_init(void)
+{
+ peerhash_mtx = XCALLOC(MTYPE_PTHREAD, sizeof(pthread_mutex_t));
+ write_cond = XCALLOC(MTYPE_PTHREAD, 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(write_cond, &attrs);
+ pthread_condattr_destroy(&attrs);
+
+ // initialize peerhash
+ peerhash = hash_create_size(2048, peer_hash_key_make, peer_hash_cmp);
+}
+
+static void peer_writes_finish(void *arg)
+{
+ bgp_packet_writes_thread_run = false;
+
+ if (peerhash)
+ hash_free(peerhash);
+
+ peerhash = NULL;
+
+ pthread_mutex_unlock(peerhash_mtx);
+ pthread_mutex_destroy(peerhash_mtx);
+ pthread_cond_destroy(write_cond);
+
+ XFREE(MTYPE_PTHREAD, peerhash_mtx);
+ XFREE(MTYPE_PTHREAD, write_cond);
+}
+
+void *peer_writes_start(void *arg)
+{
+ struct timeval currtime = {0, 0};
+ struct timeval sleeptime = {0, 500};
+ struct timespec next_update = {0, 0};
+
+ pthread_mutex_lock(peerhash_mtx);
+
+ // register cleanup handler
+ pthread_cleanup_push(&peer_writes_finish, NULL);
+
+ bgp_packet_writes_thread_run = true;
+
+ while (bgp_packet_writes_thread_run) {
+ // wait around until next update time
+ if (peerhash->count > 0)
+ pthread_cond_timedwait(write_cond, peerhash_mtx,
+ &next_update);
+ else // wait around until we have some peers
+ while (peerhash->count == 0
+ && bgp_packet_writes_thread_run)
+ pthread_cond_wait(write_cond, peerhash_mtx);
+
+ hash_iterate(peerhash, peer_process_writes, NULL);
+
+ monotime(&currtime);
+ timeradd(&currtime, &sleeptime, &currtime);
+ TIMEVAL_TO_TIMESPEC(&currtime, &next_update);
+ }
+
+ // clean up
+ pthread_cleanup_pop(1);
+
+ return NULL;
+}
+
+int peer_writes_stop(void **result)
+{
+ struct frr_pthread *fpt = frr_pthread_get(PTHREAD_WRITE);
+ bgp_packet_writes_thread_run = false;
+ peer_writes_wake();
+ pthread_join(fpt->thread, result);
+ return 0;
+}
+
+void peer_writes_on(struct peer *peer)
+{
+ if (peer->status == Deleted)
+ return;
+
+ pthread_mutex_lock(peerhash_mtx);
+ {
+ if (!hash_lookup(peerhash, peer)) {
+ hash_get(peerhash, peer, hash_alloc_intern);
+ peer_lock(peer);
+ }
+
+ SET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
+ }
+ pthread_mutex_unlock(peerhash_mtx);
+ peer_writes_wake();
+}
+
+void peer_writes_off(struct peer *peer)
+{
+ pthread_mutex_lock(peerhash_mtx);
+ {
+ if (hash_release(peerhash, peer)) {
+ peer_unlock(peer);
+ fprintf(stderr, "Releasing %p\n", peer);
+ }
+
+ UNSET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
+ }
+ pthread_mutex_unlock(peerhash_mtx);
+}
+
+void peer_writes_wake()
+{
+ pthread_cond_signal(write_cond);
+}
+
+/**
+ * Callback for hash_iterate. Takes a hash bucket, unwraps it into a peer and
+ * synchronously calls bgp_write() on the peer.
+ */
+static void peer_process_writes(struct hash_backet *hb, void *arg)
+{
+ static struct peer *peer;
+ peer = hb->data;
+ pthread_mutex_lock(&peer->obuf_mtx);
+ {
+ bgp_write(peer);
+ }
+ pthread_mutex_unlock(&peer->obuf_mtx);
+
+ // dispatch job on main thread
+ BGP_TIMER_ON(peer->t_generate_updgrp_packets,
+ bgp_generate_updgrp_packets, 100);
+}
+
+/**
+ * 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.
+ *
+ * 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 int bgp_write(struct peer *peer)
+{
+ u_char type;
+ struct stream *s;
+ int num;
+ int update_last_write = 0;
+ unsigned int count = 0;
+ unsigned int oc = 0;
+
+ /* Write packets. The number of packets written is the value of
+ * bgp->wpkt_quanta or the size of the output buffer, whichever is
+ * smaller.*/
+ while (count < peer->bgp->wpkt_quanta
+ && (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);
+
+ 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:
+ 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);
+
+ /* Handle Graceful Restart case where the state changes
+ to
+ Connect instead of Idle */
+ /* 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;
+ }
+
+ count++;
+ /* OK we send packet so delete it. */
+ stream_free(stream_fifo_pop(peer->obuf));
+ update_last_write = 1;
+ }
+
+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();
+}
+
+ return count;
+}
--- /dev/null
+/*
+ BGP I/O.
+ Implements a consumer thread to flush packets destined for remote peers.
+
+ Copyright (C) 2017 Cumulus Networks
+
+ 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
+
+#include "bgpd/bgpd.h"
+
+/**
+ * Control variable for write thread.
+ *
+ * Setting this variable to false and calling peer_writes_wake() will
+ * eventually result in thread termination.
+ */
+extern bool bgp_packet_writes_thread_run;
+
+/**
+ * Initializes data structures and flags for the write thread.
+ *
+ * This function should be called from the main thread before
+ * peer_writes_start() is invoked.
+ */
+extern void peer_writes_init(void);
+
+/**
+ * Start function for write thread.
+ *
+ * This function should be passed to pthread_create() during BGP startup.
+ */
+extern void *peer_writes_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 peer_writes_stop(void **result);
+
+/**
+ * Registers a peer with the write thread.
+ *
+ * This function adds the peer to an internal data structure, which must be
+ * locked for write access. This call will block until the structure can be
+ * locked.
+ *
+ * After this function is called, any packets placed on peer->obuf will be
+ * written to peer->fd at regular intervals.
+ *
+ * This function increments the peer reference counter with peer_lock().
+ *
+ * If the peer is already registered, nothing happens.
+ *
+ * @param peer - peer to register
+ */
+extern void peer_writes_on(struct peer *peer);
+
+/**
+ * Deregisters a peer with the write thread.
+ *
+ * This function removes the peer from an internal data structure, which must
+ * be locked for write access. This call will block until the structure can be
+ * locked.
+ *
+ * After this function is called, any packets placed on peer->obuf will not be
+ * written to peer->fd.
+ *
+ * This function decrements the peer reference counter with peer_unlock().
+ *
+ * If the peer is not registered, nothing happens.
+ *
+ * @param peer - peer to deregister
+ */
+extern void peer_writes_off(struct peer *peer);
+
+/**
+ * Notifies the write thread that there is work to be done.
+ *
+ * This function has the effect of waking the write thread if it is sleeping.
+ * If the thread is not sleeping, this signal will be ignored.
+ */
+extern void peer_writes_wake(void);
+
+#endif /* _FRR_BGP_IO_H */
/* 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.
#include "plist.h"
#include "queue.h"
#include "filter.h"
-#include "lib/frr_pthread.h"
#include "bgpd/bgpd.h"
#include "bgpd/bgp_table.h"
#include "bgpd/bgp_vty.h"
#include "bgpd/bgp_updgrp.h"
#include "bgpd/bgp_label.h"
+#include "bgpd/bgp_io.h"
/* Linked list of active peers */
static pthread_mutex_t *plist_mtx;
return cp;
}
-/**
- * Push a packet onto the beginning of the peer's output queue.
- * Must be externally synchronized around 'peer'.
- */
-static void bgp_packet_add_unsafe(struct peer *peer, struct stream *s)
-{
- /* Add packet to the end of list. */
- stream_fifo_push(peer->obuf, s);
- peer_writes_wake();
-}
-
/*
* 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)
{
pthread_mutex_lock(&peer->obuf_mtx);
- bgp_packet_add_unsafe(peer, s);
+ stream_fifo_push(peer->obuf, s);
pthread_mutex_unlock(&peer->obuf_mtx);
-}
-/**
- * Pop a packet off the end of the peer's output queue.
- * Must be externally synchronized around 'peer'.
- */
-static void bgp_packet_delete_unsafe(struct peer *peer)
-{
- stream_free(stream_fifo_pop(peer->obuf));
+ peer_writes_wake();
}
static struct stream *bgp_update_packet_eor(struct peer *peer, afi_t afi,
return s;
}
-/* Get next packet to be written. */
-static struct stream *bgp_write_packet(struct peer *peer)
+/*
+ * Enqueue onto the peer's output buffer any packets which are pending for the
+ * update group it is a member of.
+ *
+ * XXX: Severely needs performance work.
+ */
+int bgp_generate_updgrp_packets(struct thread *thread)
{
- struct stream *s = NULL;
+ struct peer *peer = THREAD_ARG(thread);
+ peer->t_generate_updgrp_packets = NULL;
+
+ struct stream *s;
struct peer_af *paf;
struct bpacket *next_pkt;
afi_t afi;
* update-delay post processing).
*/
if (peer->status != Established)
- return NULL;
+ return 0;
if (peer->bgp && peer->bgp->main_peers_update_hold)
- return 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;
-
- /* 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;
- }
+ return 0;
- /* 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));
+ do {
+ s = NULL;
+ for (afi = AFI_IP; afi < AFI_MAX; afi++)
+ for (safi = SAFI_UNICAST; safi < SAFI_MAX; safi++) {
+ paf = peer_af_find(peer, afi, safi);
+ if (!paf || !PAF_SUBGRP(paf))
+ continue;
next_pkt = paf->next_pkt_to_send;
- }
-
- /* 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);
+ /* 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;
+ }
- return s;
+ /* 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;
}
- }
- continue;
- }
-
- /* 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);
- return s;
- }
-
- return NULL;
-}
-static int bgp_generate_updgrp_packets(struct thread *thread)
-{
- struct listnode *ln;
- struct peer *peer;
- pthread_mutex_lock(plist_mtx);
- {
- for (ALL_LIST_ELEMENTS_RO(plist, ln, peer))
- while (bgp_write_packet(peer))
- ;
+ /* 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);
- t_generate_updgrp_packets = NULL;
- }
- pthread_mutex_unlock(plist_mtx);
return 0;
}
-
/*
* Creates a BGP Keepalive packet and appends it to the peer's output queue.
*/
return 0;
}
-
-/* ------------- write thread ------------------ */
-
-/**
- * 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.
- *
- * 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 int bgp_write(struct peer *peer)
-{
- u_char type;
- struct stream *s;
- int num;
- int update_last_write = 0;
- unsigned int count = 0;
- unsigned int oc = 0;
-
- /* Write packets. The number of packets written is the value of
- * bgp->wpkt_quanta or the size of the output buffer, whichever is
- * smaller.*/
- while (count < peer->bgp->wpkt_quanta
- && (s = stream_fifo_head(peer->obuf))) {
- int writenum;
- do { // write a full packet, or return on error
- 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);
-
- 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:
- 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);
-
- /* Handle Graceful Restart case where the state changes
- to
- Connect instead of Idle */
- /* 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;
- }
-
- count++;
- /* OK we send packet so delete it. */
- bgp_packet_delete_unsafe(peer);
- update_last_write = 1;
- }
-
-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();
-}
-
- return count;
-}
-
-void peer_writes_init(void)
-{
- plist_mtx = XCALLOC(MTYPE_PTHREAD, sizeof(pthread_mutex_t));
- write_cond = XCALLOC(MTYPE_PTHREAD, sizeof(pthread_cond_t));
-
- // initialize mutex
- pthread_mutex_init(plist_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(write_cond, &attrs);
- pthread_condattr_destroy(&attrs);
-
- // initialize peerlist
- plist = list_new();
-}
-
-static void peer_writes_finish(void *arg)
-{
- bgp_packet_writes_thread_run = false;
-
- if (plist)
- list_delete(plist);
-
- plist = NULL;
-
- pthread_mutex_unlock(plist_mtx);
- pthread_mutex_destroy(plist_mtx);
- pthread_cond_destroy(write_cond);
-
- XFREE(MTYPE_PTHREAD, plist_mtx);
- XFREE(MTYPE_PTHREAD, write_cond);
-}
-
-/**
- * Entry function for peer packet flushing pthread.
- *
- * peer_writes_init() must be called prior to this.
- */
-void *peer_writes_start(void *arg)
-{
- struct timeval currtime = {0, 0};
- struct timeval sleeptime = {0, 500};
- struct timespec next_update = {0, 0};
-
- struct listnode *ln;
- struct peer *peer;
-
- pthread_mutex_lock(plist_mtx);
-
- // register cleanup handler
- pthread_cleanup_push(&peer_writes_finish, NULL);
-
- bgp_packet_writes_thread_run = true;
-
- while (bgp_packet_writes_thread_run) { // wait around until next update
- // time
- if (plist->count > 0)
- pthread_cond_timedwait(write_cond, plist_mtx,
- &next_update);
- else // wait around until we have some peers
- while (plist->count == 0
- && bgp_packet_writes_thread_run)
- pthread_cond_wait(write_cond, plist_mtx);
-
- for (ALL_LIST_ELEMENTS_RO(plist, ln, peer)) {
- pthread_mutex_lock(&peer->obuf_mtx);
- {
- bgp_write(peer);
- }
- pthread_mutex_unlock(&peer->obuf_mtx);
-
- if (!bgp_packet_writes_thread_run)
- break;
- }
-
- // schedule update packet generation on main thread
- if (!t_generate_updgrp_packets)
- t_generate_updgrp_packets = thread_add_event(
- bm->master, bgp_generate_updgrp_packets, NULL,
- 0);
-
- monotime(&currtime);
- timeradd(&currtime, &sleeptime, &currtime);
- TIMEVAL_TO_TIMESPEC(&currtime, &next_update);
- }
-
- // clean up
- pthread_cleanup_pop(1);
-
- return NULL;
-}
-
-/**
- * Turns on packet writing for a peer.
- */
-void peer_writes_on(struct peer *peer)
-{
- if (peer->status == Deleted)
- return;
-
- pthread_mutex_lock(plist_mtx);
- {
- struct listnode *ln, *nn;
- struct peer *p;
-
- // make sure this peer isn't already in the list
- for (ALL_LIST_ELEMENTS(plist, ln, nn, p))
- if (p == peer) {
- pthread_mutex_unlock(plist_mtx);
- return;
- }
-
- peer_lock(peer);
- listnode_add(plist, peer);
-
- SET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
- }
- pthread_mutex_unlock(plist_mtx);
- peer_writes_wake();
-}
-
-/**
- * Turns off packet writing for a peer.
- */
-void peer_writes_off(struct peer *peer)
-{
- struct listnode *ln, *nn;
- struct peer *p;
- pthread_mutex_lock(plist_mtx);
- {
- for (ALL_LIST_ELEMENTS(plist, ln, nn, p))
- if (p == peer) {
- list_delete_node(plist, ln);
- peer_unlock(peer);
- break;
- }
-
- UNSET_FLAG(peer->thread_flags, PEER_THREAD_WRITES_ON);
- }
- pthread_mutex_unlock(plist_mtx);
-}
-
-/**
- * Wakes up the write thread to do work.
- */
-void peer_writes_wake()
-{
- pthread_cond_signal(write_cond);
-}
-
-int peer_writes_stop(void **result)
-{
- struct frr_pthread *fpt = frr_pthread_get(PTHREAD_WRITE);
- bgp_packet_writes_thread_run = false;
- peer_writes_wake();
- pthread_join(fpt->thread, result);
- return 0;
-}
/* Control variable for write thread. */
extern bool bgp_packet_writes_thread_run;
-extern void peer_writes_init(void);
-extern void *peer_writes_start(void *arg);
-extern void peer_writes_on(struct peer *peer);
-extern void peer_writes_off(struct peer *peer);
-extern void peer_writes_wake(void);
-extern int peer_writes_stop(void **result);
+extern int bgp_generate_updgrp_packets(struct thread *);
#endif /* _QUAGGA_BGP_PACKET_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)");
struct thread *t_pmax_restart;
struct thread *t_gr_restart;
struct thread *t_gr_stale;
+ struct thread *t_generate_updgrp_packets;
/* Thread flags. */
u_int16_t thread_flags;
projects / mirror_frr.git / commitdiff