ceph: preserve seq # on requeued messages after transient transport errors

If the tcp connection drops and we reconnect to reestablish a stateful
session (with the mds), we need to resend previously sent (and possibly
received) messages with the _same_ seq # so that they can be dropped on
the other end if needed.  Only assign a new seq once after the message is
queued.

Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/fs/ceph/messenger.c b/fs/ceph/messenger.c
index a3a8f368845b4acbeafed2fa3841184e4294567d..cd4fadb6491afe3a48ad5081b5f4ab2a41e4b0bf 100644 (file)
--- a/fs/ceph/messenger.c
+++ b/fs/ceph/messenger.c
@@ -492,7 +492,14 @@ static void prepare_write_message(struct ceph_connection *con)
                list_move_tail(&m->list_head, &con->out_sent);
        }
 
-       m->hdr.seq = cpu_to_le64(++con->out_seq);
+       /*
+        * only assign outgoing seq # if we haven't sent this message
+        * yet.  if it is requeued, resend with it's original seq.
+        */
+       if (m->needs_out_seq) {
+               m->hdr.seq = cpu_to_le64(++con->out_seq);
+               m->needs_out_seq = false;
+       }
 
        dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
             m, con->out_seq, le16_to_cpu(m->hdr.type),
@@ -1986,6 +1993,8 @@ void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
 
        BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
 
+       msg->needs_out_seq = true;
+
        /* queue */
        mutex_lock(&con->mutex);
        BUG_ON(!list_empty(&msg->list_head));

diff --git a/fs/ceph/messenger.h b/fs/ceph/messenger.h
index a343dae73cdcf4f6f458480e4b910e5b9889093e..a5caf91cc97178039689b1f0efc49ef42e56d981 100644 (file)
--- a/fs/ceph/messenger.h
+++ b/fs/ceph/messenger.h
@@ -86,6 +86,7 @@ struct ceph_msg {
        struct kref kref;
        bool front_is_vmalloc;
        bool more_to_follow;
+       bool needs_out_seq;
        int front_max;
 
        struct ceph_msgpool *pool;