lib, bgpd: implement pthread lifecycle management

Removes the WiP shim and implements proper thread lifecycle management.

* Declare necessary pthread_t's in bgp_master
* Define new MTYPE in lib/thread.c for pthreads
* Allocate and free BGP's pthreads appropriately
* Terminate and join threads appropriately

Signed-off-by: Quentin Young <qlyoung@cumulusnetworks.com>

bgpd/bgp_main.c

@@ -194,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);
@@ -395,11 +398,9 @@ int main(int argc, char **argv)
 	snprintf(bgpd_di.startinfo, sizeof(bgpd_di.startinfo), ", bgp@%s:%d",
 		 (bm->address ? bm->address : "<all>"), bm->port);
 
-	pthread_t packet_writes, keepalives;
-	pthread_create(&packet_writes, NULL, &peer_writes_start, NULL);
-	pthread_create(&keepalives, NULL, &keepalives_start, NULL);
-
 	frr_config_fork();
+	/* must be called after fork() */
+	bgp_pthreads_init();
 	frr_run(bm->master);
 
 	/* Not reached. */

bgpd/bgp_packet.c

@@ -2318,6 +2318,9 @@ void *peer_writes_start(void *arg)
 				bgp_write(peer);
 			}
 			pthread_mutex_unlock(&peer->obuf_mtx);
+
+			if (!bgp_packet_writes_thread_run)
+				break;
 		}
 
 		// schedule update packet generation on main thread

bgpd/bgpd.c

@@ -75,6 +75,7 @@
 #include "bgpd/bgp_bfd.h"
 #include "bgpd/bgp_memory.h"
 #include "bgpd/bgp_evpn_vty.h"
+#include "bgpd/bgp_keepalives.h"
 
 
 DEFINE_MTYPE_STATIC(BGPD, PEER_TX_SHUTDOWN_MSG, "Peer shutdown message (TX)");
@@ -7326,6 +7327,8 @@ void bgp_master_init(struct thread_master *master)
 	bm->start_time = bgp_clock();
 	bm->t_rmap_update = NULL;
 	bm->rmap_update_timer = RMAP_DEFAULT_UPDATE_TIMER;
+	bm->t_bgp_keepalives = XCALLOC(MTYPE_PTHREAD, sizeof(pthread_t));
+	bm->t_bgp_packet_writes = XCALLOC(MTYPE_PTHREAD, sizeof(pthread_t));
 
 	bgp_process_queue_init();
 
@@ -7382,6 +7385,32 @@ static const struct cmd_variable_handler bgp_viewvrf_var_handlers[] = {
 	{.completions = NULL},
 };
 
+void bgp_pthreads_init()
+{
+	/* init write & keepalive threads */
+	pthread_create(bm->t_bgp_keepalives, NULL, &peer_keepalives_start,
+		       NULL);
+	pthread_create(bm->t_bgp_packet_writes, NULL, &peer_writes_start, NULL);
+}
+
+void bgp_pthreads_finish()
+{
+	/* set thread cancellation flags */
+	bgp_keepalives_thread_run = false;
+	bgp_packet_writes_thread_run = false;
+
+	/* wake them up */
+	peer_writes_wake();
+	peer_keepalives_wake();
+
+	/* join */
+	pthread_join(*bm->t_bgp_keepalives, NULL);
+	pthread_join(*bm->t_bgp_packet_writes, NULL);
+
+	XFREE(MTYPE_PTHREAD, bm->t_bgp_keepalives);
+	XFREE(MTYPE_PTHREAD, bm->t_bgp_packet_writes);
+}
+
 void bgp_init(void)
 {
 

bgpd/bgpd.h

@@ -100,6 +100,10 @@ struct bgp_master {
 	/* BGP thread master.  */
 	struct thread_master *master;
 
+	/* BGP pthreads. */
+	pthread_t *t_bgp_keepalives;
+	pthread_t *t_bgp_packet_writes;
+
 	/* work queues */
 	struct work_queue *process_main_queue;
 
@@ -1254,6 +1258,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_init(void);
+extern void bgp_pthreads_finish(void);
 extern void bgp_route_map_init(void);
 extern void bgp_session_reset(struct peer *);
 

lib/thread.c

@@ -36,6 +36,7 @@
 DEFINE_MTYPE_STATIC(LIB, THREAD, "Thread")
 DEFINE_MTYPE_STATIC(LIB, THREAD_MASTER, "Thread master")
 DEFINE_MTYPE_STATIC(LIB, THREAD_STATS, "Thread stats")
+DEFINE_MTYPE(LIB, PTHREAD, "POSIX Thread")
 
 #if defined(__APPLE__)
 #include <mach/mach.h>

lib/thread.h

@@ -26,6 +26,9 @@
 #include <poll.h>
 #include "monotime.h"
 
+#include "memory.h"
+DECLARE_MTYPE(PTHREAD)
+
 struct rusage_t {
 	struct rusage cpu;
 	struct timeval real;