msgr: preserve incoming message queue when replacing pipes

If we replace an existing pipe with a new one, move the incoming queue
of messages that have not yet been dispatched over to the new Pipe so that
they are not lost.  This prevents messages from being lost.

Alternatively, we could set in_seq = existing->in_seq - existing->in_qlen,
but that would make the other end resend those messages, which is a waste
of bandwidth.

Very easy to reproduce the original bug with 'ms inject socket failures'.

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

diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc
index 6852fc3acfd0b1b5834bee4734eab5a7d983baa0..3514de7fcf28179d72c13263467c7ec6bf8b6d43 100644 (file)
--- a/src/msg/SimpleMessenger.cc
+++ b/src/msg/SimpleMessenger.cc
@@ -853,11 +853,19 @@ int SimpleMessenger::Pipe::accept()
     // do not clear existing->connection_state, since read_message and write_message both
     // dereference it without pipe_lock.
 
-    // steal queue and out_seq
-    existing->requeue_sent();
-    out_seq = existing->out_seq;
+    // steal incoming queue
     in_seq = existing->in_seq;
     in_seq_acked = in_seq;
+    delete in_q;
+    in_q = existing->in_q;
+    in_q->lock.Lock();
+    in_q->pipe = this;
+    in_q->lock.Unlock();
+    existing->in_q = new IncomingQueue(msgr->cct, existing);
+
+    // steal outgoing queue and out_seq
+    existing->requeue_sent();
+    out_seq = existing->out_seq;
     ldout(msgr->cct,10) << "accept re-queuing on out_seq " << out_seq << " in_seq " << in_seq << dendl;
     for (map<int, list<Message*> >::iterator p = existing->out_q.begin();
          p != existing->out_q.end();