SimpleMessenger: Keep a disposable flag for use in reset

pipes marked disposable must not inherit the lossy policy on reconnect.
Also, in Pipe::writer, when sent.empty() && close_on_empty, mark as
lossy to ensure that fault will close the connection.

Signed-off-by: Samuel Just <samuel.just@dreamhost.com>

diff --git a/src/msg/SimpleMessenger.cc b/src/msg/SimpleMessenger.cc
index a96363ee8ab5c7145e136a3377084e8d40677e55..63dbfabdfb9032a19f39b00b5244aa2c49cb8de9 100644 (file)
--- a/src/msg/SimpleMessenger.cc
+++ b/src/msg/SimpleMessenger.cc
@@ -1257,7 +1257,9 @@ int SimpleMessenger::Pipe::connect()
 
       // hooray!
       peer_global_seq = reply.global_seq;
-      policy.lossy = reply.flags & CEPH_MSG_CONNECT_LOSSY;
+      if (!disposable) {
+       policy.lossy = reply.flags & CEPH_MSG_CONNECT_LOSSY;
+      }
       state = STATE_OPEN;
       connect_seq = cseq + 1;
       assert(connect_seq == reply.connect_seq);
@@ -1746,6 +1748,7 @@ void SimpleMessenger::Pipe::writer()
     if (sent.empty() && close_on_empty) {
       // this is slightly hacky
       dout(10) << "writer out and sent queues empty, closing" << dendl;
+      policy.lossy = true;
       fault();
       continue;
     }
@@ -2721,6 +2724,7 @@ void SimpleMessenger::mark_disposable(Connection *con)
     dout(1) << "mark_disposable " << con << " -- " << p << dendl;
     p->pipe_lock.Lock();
     p->policy.lossy = true;
+    p->disposable = true;
     p->pipe_lock.Unlock();
     p->put();
   } else {

diff --git a/src/msg/SimpleMessenger.h b/src/msg/SimpleMessenger.h
index 228ff7321ed629125f92461a9cc7bb631268f3d0..cc9e6efa16d3bf97b0e96ac9356f64152073cb10 100644 (file)
--- a/src/msg/SimpleMessenger.h
+++ b/src/msg/SimpleMessenger.h
@@ -153,6 +153,7 @@ private:
     bool keepalive;
     bool halt_delivery; //if a pipe's queue is destroyed, stop adding to it
     bool close_on_empty;
+    bool disposable;
     
     __u32 connect_seq, peer_global_seq;
     uint64_t out_seq;
@@ -222,7 +223,8 @@ private:
       state(st), 
       connection_state(new Connection),
       reader_running(false), reader_joining(false), writer_running(false),
-      in_qlen(0), keepalive(false), halt_delivery(false), close_on_empty(false),
+      in_qlen(0), keepalive(false), halt_delivery(false), 
+      close_on_empty(false), disposable(false),
       connect_seq(0), peer_global_seq(0),
       out_seq(0), in_seq(0), in_seq_acked(0),
       reader_thread(this), writer_thread(this) {