Merge pull request #1614 from qlyoung/imp-bgpd-pthread-startup-sync

improve bgpd thread startup characteristics

diff --git a/bgpd/bgp_io.c b/bgpd/bgp_io.c
index 5ab14d5cd61db24269c7968455cfb5ab2e4b8ff7..98a8ec6e02cb959031e2502c5a343d57925504b6 100644 (file)
--- a/bgpd/bgp_io.c
+++ b/bgpd/bgp_io.c
@@ -51,19 +51,37 @@ static bool validate_header(struct peer *);
 #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
+/* Plumbing & control variables for thread lifecycle
  * ------------------------------------------------------------------------ */
-_Atomic bool bgp_io_thread_run;
-_Atomic bool bgp_io_thread_started;
+bool bgp_io_thread_run;
+pthread_mutex_t *running_cond_mtx;
+pthread_cond_t *running_cond;
 
-void bgp_io_init()
+/* Unused callback for thread_add_read() */
+static int bgp_io_dummy(struct thread *thread) { return 0; }
+
+/* Poison pill task */
+static int bgp_io_finish(struct thread *thread)
 {
        bgp_io_thread_run = false;
-       bgp_io_thread_started = false;
+       return 0;
 }
 
-/* Unused callback for thread_add_read() */
-static int bgp_io_dummy(struct thread *thread) { return 0; }
+/* Extern lifecycle control functions. init -> start -> stop
+ * ------------------------------------------------------------------------ */
+void bgp_io_init()
+{
+       bgp_io_thread_run = false;
+
+       running_cond_mtx = XCALLOC(MTYPE_PTHREAD_PRIM, sizeof(pthread_mutex_t));
+       running_cond = XCALLOC(MTYPE_PTHREAD_PRIM, sizeof(pthread_cond_t));
+
+       pthread_mutex_init(running_cond_mtx, NULL);
+       pthread_cond_init(running_cond, NULL);
+
+       /* unlocked in bgp_io_wait_running() */
+       pthread_mutex_lock(running_cond_mtx);
+}
 
 void *bgp_io_start(void *arg)
 {
@@ -80,9 +98,12 @@ void *bgp_io_start(void *arg)
 
        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);
+       pthread_mutex_lock(running_cond_mtx);
+       {
+               bgp_io_thread_run = true;
+               pthread_cond_signal(running_cond);
+       }
+       pthread_mutex_unlock(running_cond_mtx);
 
        while (bgp_io_thread_run) {
                if (thread_fetch(fpt->master, &task)) {
@@ -96,29 +117,35 @@ void *bgp_io_start(void *arg)
        return NULL;
 }
 
-static int bgp_io_finish(struct thread *thread)
+void bgp_io_wait_running()
 {
-       atomic_store_explicit(&bgp_io_thread_run, false, memory_order_seq_cst);
-       return 0;
+       while (!bgp_io_thread_run)
+               pthread_cond_wait(running_cond, running_cond_mtx);
+
+       /* locked in bgp_io_init() */
+       pthread_mutex_unlock(running_cond_mtx);
 }
 
 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);
+
+       pthread_mutex_destroy(running_cond_mtx);
+       pthread_cond_destroy(running_cond);
+
+       XFREE(MTYPE_PTHREAD_PRIM, running_cond_mtx);
+       XFREE(MTYPE_PTHREAD_PRIM, running_cond);
+
        return 0;
 }
 
 /* Extern API -------------------------------------------------------------- */
-void bgp_io_running(void)
-{
-       while (!atomic_load_explicit(&bgp_io_thread_started,
-                                    memory_order_seq_cst))
-               frr_pthread_yield();
-}
 
 void bgp_writes_on(struct peer *peer)
 {
+       assert(bgp_io_thread_run);
+
        assert(peer->status != Deleted);
        assert(peer->obuf);
        assert(peer->ibuf);
@@ -136,6 +163,8 @@ void bgp_writes_on(struct peer *peer)
 
 void bgp_writes_off(struct peer *peer)
 {
+       assert(bgp_io_thread_run);
+
        struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO);
 
        thread_cancel_async(fpt->master, &peer->t_write, NULL);
@@ -146,6 +175,8 @@ void bgp_writes_off(struct peer *peer)
 
 void bgp_reads_on(struct peer *peer)
 {
+       assert(bgp_io_thread_run);
+
        assert(peer->status != Deleted);
        assert(peer->ibuf);
        assert(peer->fd);
@@ -165,6 +196,8 @@ void bgp_reads_on(struct peer *peer)
 
 void bgp_reads_off(struct peer *peer)
 {
+       assert(bgp_io_thread_run);
+
        struct frr_pthread *fpt = frr_pthread_get(PTHREAD_IO);
 
        thread_cancel_async(fpt->master, &peer->t_read, NULL);

diff --git a/bgpd/bgp_io.h b/bgpd/bgp_io.h
index 73587366d7871ebb3460f6e4bf62d5628a93fb33..7cfd1db7106982e50bfa6d56daa366bcc25d5f91 100644 (file)
--- a/bgpd/bgp_io.h
+++ b/bgpd/bgp_io.h
@@ -31,27 +31,26 @@
 /**
  * Initializes data structures and flags for the write thread.
  *
- * This function should be called from the main thread before
+ * This function must be called from the main thread before
  * bgp_writes_start() is invoked.
  */
 extern void bgp_io_init(void);
 
 /**
- * Ensure that the BGP IO thread is actually up and running
+ * Start function for write thread.
  *
- * This function must be called immediately after the thread
- * has been created for running.  This is because we want
- * to make sure that the io thread is ready before other
- * threads start attempting to use it.
+ * @param arg - unused
  */
-extern void bgp_io_running(void);
+extern void *bgp_io_start(void *arg);
 
 /**
- * Start function for write thread.
+ * Wait until the IO thread is ready to accept jobs.
  *
- * @param arg - unused
+ * This function must be called immediately after the thread has been created
+ * for running. Use of other functions before calling this one will result in
+ * undefined behavior.
  */
-extern void *bgp_io_start(void *arg);
+extern void bgp_io_wait_running(void);
 
 /**
  * Start function for write thread.

diff --git a/bgpd/bgpd.c b/bgpd/bgpd.c
index 5356cfd2fc9f22fbf99cd9f7520df780c264ce8b..7d54c538af5809bb3d48c3c116800c69655e1f7d 100644 (file)
--- a/bgpd/bgpd.c
+++ b/bgpd/bgpd.c
@@ -7486,13 +7486,11 @@ void bgp_pthreads_run()
        pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
 
        /*
-        * Please ensure that the io thread is running
-        * by calling bgp_io_running.  The BGP threads
-        * depend on it being running when we start
-        * looking for it.
+        * I/O related code assumes the thread is ready for work at all times,
+        * so we wait until it is.
         */
        frr_pthread_run(PTHREAD_IO, &attr, NULL);
-       bgp_io_running();
+       bgp_io_wait_running();
 
        frr_pthread_run(PTHREAD_KEEPALIVES, &attr, NULL);
 }

diff --git a/lib/frr_pthread.c b/lib/frr_pthread.c
index 19dfbaf54bfe46d7011079b28bbb188e6b065edd..de522e5ef9511b8c6e47b88cf624487ecbea711f 100644 (file)
--- a/lib/frr_pthread.c
+++ b/lib/frr_pthread.c
@@ -26,6 +26,7 @@
 #include "hash.h"
 
 DEFINE_MTYPE_STATIC(LIB, FRR_PTHREAD, "FRR POSIX Thread");
+DEFINE_MTYPE(LIB, PTHREAD_PRIM, "POSIX synchronization primitives");
 
 static unsigned int next_id = 0;
 

diff --git a/lib/frr_pthread.h b/lib/frr_pthread.h
index f6000340a70873a2390af5235fef3118862b2db0..7915b43a46d8b59c1d047e1622049d10979baac5 100644 (file)
--- a/lib/frr_pthread.h
+++ b/lib/frr_pthread.h
@@ -21,8 +21,11 @@
 #define _FRR_PTHREAD_H
 
 #include <pthread.h>
+#include "memory.h"
 #include "thread.h"
 
+DECLARE_MTYPE(PTHREAD_PRIM);
+
 struct frr_pthread {
 
        /* pthread id */