msg/Pipe: join previous reader threads

We may stop and then restart the reader thread.  Join previous threads
before we create new ones.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/msg/Pipe.cc b/src/msg/Pipe.cc
index 2846937ee40f664af8480c11930d0c9368bb865d..243cf7a82fe907347883e76a4058dd9cee6eaabf 100644 (file)
--- a/src/msg/Pipe.cc
+++ b/src/msg/Pipe.cc
@@ -62,7 +62,7 @@ Pipe::Pipe(SimpleMessenger *r, int st, Connection *con)
     state(st),
     session_security(NULL),
     connection_state(NULL),
-    reader_running(false), reader_joining(false), writer_running(false),
+    reader_running(false), reader_needs_join(false), reader_joining(false), writer_running(false),
     in_q(&(r->dispatch_queue)),
     keepalive(false),
     close_on_empty(false),
@@ -118,6 +118,10 @@ void Pipe::start_reader()
 {
   assert(pipe_lock.is_locked());
   assert(!reader_running);
+  if (reader_needs_join) {
+    reader_thread.join();
+    reader_needs_join = false;
+  }
   reader_running = true;
   reader_thread.create(msgr->cct->_conf->ms_rwthread_stack_bytes);
 }
@@ -142,6 +146,7 @@ void Pipe::join_reader()
   pipe_lock.Lock();
   assert(reader_joining);
   reader_joining = false;
+  reader_needs_join = false;
 }
 
 
@@ -1264,6 +1269,7 @@ void Pipe::reader()
  
   // reap?
   reader_running = false;
+  reader_needs_join = true;
   unlock_maybe_reap();
   ldout(msgr->cct,10) << "reader done" << dendl;
 }

diff --git a/src/msg/Pipe.h b/src/msg/Pipe.h
index 2689d4813bc2c670d1a19d5d41c68a4c8ff7154c..aa2dc1b42e257e1a5483e99e7184b12f0fe18a41 100644 (file)
--- a/src/msg/Pipe.h
+++ b/src/msg/Pipe.h
@@ -113,7 +113,7 @@ class DispatchQueue;
 
     utime_t backoff;         // backoff time
 
-    bool reader_running, reader_joining;
+    bool reader_running, reader_needs_join, reader_joining;
     bool writer_running;
 
     map<int, list<Message*> > out_q;  // priority queue for outbound msgs