msgr: maintain list of accepting pipes

New pipes exist in a sort of limbo before we know who the peer is and
add them to rank_pipe.  Keep a list of them in accepting_pipes for that
period.

Signed-off-by: Sage Weil <sage@inktank.com>
(cherry picked from commit dd4addef2d5b457cc9a58782fe42af6b13c68b81)

3 files changed, 11 insertions, 0 deletions

diff --git a/src/msg/Pipe.cc b/src/msg/Pipe.cc
index d2bb5141294..1d6bc157ec2 100644
--- a/src/msg/Pipe.cc
+++ b/src/msg/Pipe.cc
@@ -589,6 +589,7 @@ int Pipe::accept()
   // open
   connect_seq = connect.connect_seq + 1;
   peer_global_seq = connect.global_seq;
+  assert(state == STATE_ACCEPTING);
   state = STATE_OPEN;
   ldout(msgr->cct,10) << "accept success, connect_seq = " << connect_seq << ", sending READY" << dendl;
 
@@ -615,6 +616,8 @@ int Pipe::accept()
   // ok!
   if (msgr->dispatch_queue.stop)
     goto shutting_down;
+  inr removed = msgr->accepting_pipes.erase(this);
+  assert(removed == 1);
   register_pipe();
   msgr->lock.Unlock();
 
@@ -1092,6 +1095,7 @@ void Pipe::unregister_pipe()
     msgr->rank_pipe.erase(p);
   } else {
     ldout(msgr->cct,10) << "unregister_pipe - not registered" << dendl;
+    msgr->accepting_pipes.erase(this);  // somewhat overkill, but safe.
   }
 }
 
diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc
index 71bdb8b341d..09bbb79ec92 100644
--- a/src/msg/SimpleMessenger.cc
+++ b/src/msg/SimpleMessenger.cc
@@ -307,6 +307,7 @@ Pipe *SimpleMessenger::add_accept_pipe(int sd)
   p->start_reader();
   p->pipe_lock.Unlock();
   pipes.insert(p);
+  accepting_pipes.insert(p);
   lock.Unlock();
   return p;
 }
diff --git a/src/msg/SimpleMessenger.h b/src/msg/SimpleMessenger.h
index 6be1a0a9539..049c7edda65 100644
--- a/src/msg/SimpleMessenger.h
+++ b/src/msg/SimpleMessenger.h
@@ -514,6 +514,12 @@ private:
    * invalid and can be replaced by anyone holding the msgr lock
    */
   hash_map<entity_addr_t, Pipe*> rank_pipe;
+  /**
+   * list of pipes are in teh process of accepting
+   *
+   * These are not yet in the rank_pipe map.
+   */
+  set<Pipe*> accepting_pipes;
   /// a set of all the Pipes we have which are somehow active
   set<Pipe*>      pipes;
   /// a list of Pipes we want to tear down