libceph: pass the message pointer instead of loading con->out_msg

This pointer is in a register anyway, so let's use that instead of
reloading from memory everywhere.

[ idryomov: formatting ]

Signed-off-by: Max Kellermann <max.kellermann@ionos.com>
Reviewed-by: Viacheslav Dubeyko <Slava.Dubeyko@ibm.com>
Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 9ebcac2981fd4d35a654caf86fb2a1d2650c1dde..6aa4c6478c9f61853cdba58266ce2f44e7d54fbb 100644 (file)
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -555,7 +555,7 @@ struct ceph_msg *ceph_con_get_out_msg(struct ceph_connection *con);
 /* messenger_v1.c */
 int ceph_con_v1_try_read(struct ceph_connection *con);
 int ceph_con_v1_try_write(struct ceph_connection *con);
-void ceph_con_v1_revoke(struct ceph_connection *con);
+void ceph_con_v1_revoke(struct ceph_connection *con, struct ceph_msg *msg);
 void ceph_con_v1_revoke_incoming(struct ceph_connection *con);
 bool ceph_con_v1_opened(struct ceph_connection *con);
 void ceph_con_v1_reset_session(struct ceph_connection *con);
@@ -564,7 +564,7 @@ void ceph_con_v1_reset_protocol(struct ceph_connection *con);
 /* messenger_v2.c */
 int ceph_con_v2_try_read(struct ceph_connection *con);
 int ceph_con_v2_try_write(struct ceph_connection *con);
-void ceph_con_v2_revoke(struct ceph_connection *con);
+void ceph_con_v2_revoke(struct ceph_connection *con, struct ceph_msg *msg);
 void ceph_con_v2_revoke_incoming(struct ceph_connection *con);
 bool ceph_con_v2_opened(struct ceph_connection *con);
 void ceph_con_v2_reset_session(struct ceph_connection *con);

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index b6c7bfc0350361a799cbfd1d894a8eb54ab50719..08a6a083609f967b6013d68da08bbc5505e6b0d1 100644 (file)
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -1793,9 +1793,9 @@ void ceph_msg_revoke(struct ceph_msg *msg)
                WARN_ON(con->state != CEPH_CON_S_OPEN);
                dout("%s con %p msg %p was sending\n", __func__, con, msg);
                if (ceph_msgr2(from_msgr(con->msgr)))
-                       ceph_con_v2_revoke(con);
+                       ceph_con_v2_revoke(con, msg);
                else
-                       ceph_con_v1_revoke(con);
+                       ceph_con_v1_revoke(con, msg);
                ceph_msg_put(con->out_msg);
                con->out_msg = NULL;
        } else {

diff --git a/net/ceph/messenger_v1.c b/net/ceph/messenger_v1.c
index eebe4e19d75afc46a143d577d4bd580b1aa51164..cc4a36ef8462bad42c1f8a90d1aabff57eff97c0 100644 (file)
--- a/net/ceph/messenger_v1.c
+++ b/net/ceph/messenger_v1.c
@@ -169,10 +169,9 @@ static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
  * Prepare footer for currently outgoing message, and finish things
  * off.  Assumes out_kvec* are already valid.. we just add on to the end.
  */
-static void prepare_write_message_footer(struct ceph_connection *con)
+static void prepare_write_message_footer(struct ceph_connection *con,
+                                        struct ceph_msg *m)
 {
-       struct ceph_msg *m = con->out_msg;
-
        m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;
 
        dout("prepare_write_message_footer %p\n", con);
@@ -230,31 +229,31 @@ static void prepare_write_message(struct ceph_connection *con)
 
        /* fill in hdr crc and finalize hdr */
        crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
-       con->out_msg->hdr.crc = cpu_to_le32(crc);
-       memcpy(&con->v1.out_hdr, &con->out_msg->hdr, sizeof(con->v1.out_hdr));
+       m->hdr.crc = cpu_to_le32(crc);
+       memcpy(&con->v1.out_hdr, &m->hdr, sizeof(con->v1.out_hdr));
 
        /* fill in front and middle crc, footer */
        crc = crc32c(0, m->front.iov_base, m->front.iov_len);
-       con->out_msg->footer.front_crc = cpu_to_le32(crc);
+       m->footer.front_crc = cpu_to_le32(crc);
        if (m->middle) {
                crc = crc32c(0, m->middle->vec.iov_base,
                                m->middle->vec.iov_len);
-               con->out_msg->footer.middle_crc = cpu_to_le32(crc);
+               m->footer.middle_crc = cpu_to_le32(crc);
        } else
-               con->out_msg->footer.middle_crc = 0;
+               m->footer.middle_crc = 0;
        dout("%s front_crc %u middle_crc %u\n", __func__,
-            le32_to_cpu(con->out_msg->footer.front_crc),
-            le32_to_cpu(con->out_msg->footer.middle_crc));
-       con->out_msg->footer.flags = 0;
+            le32_to_cpu(m->footer.front_crc),
+            le32_to_cpu(m->footer.middle_crc));
+       m->footer.flags = 0;
 
        /* is there a data payload? */
-       con->out_msg->footer.data_crc = 0;
+       m->footer.data_crc = 0;
        if (m->data_length) {
-               prepare_message_data(con->out_msg, m->data_length);
+               prepare_message_data(m, m->data_length);
                con->v1.out_more = 1;  /* data + footer will follow */
        } else {
                /* no, queue up footer too and be done */
-               prepare_write_message_footer(con);
+               prepare_write_message_footer(con, m);
        }
 
        ceph_con_flag_set(con, CEPH_CON_F_WRITE_PENDING);
@@ -461,9 +460,9 @@ out:
  *  0 -> socket full, but more to do
  * <0 -> error
  */
-static int write_partial_message_data(struct ceph_connection *con)
+static int write_partial_message_data(struct ceph_connection *con,
+                                     struct ceph_msg *msg)
 {
-       struct ceph_msg *msg = con->out_msg;
        struct ceph_msg_data_cursor *cursor = &msg->cursor;
        bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
        u32 crc;
@@ -515,7 +514,7 @@ static int write_partial_message_data(struct ceph_connection *con)
        else
                msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
        con_out_kvec_reset(con);
-       prepare_write_message_footer(con);
+       prepare_write_message_footer(con, msg);
 
        return 1;       /* must return > 0 to indicate success */
 }
@@ -1471,6 +1470,7 @@ bad_tag:
  */
 int ceph_con_v1_try_write(struct ceph_connection *con)
 {
+       struct ceph_msg *msg;
        int ret = 1;
 
        dout("try_write start %p state %d\n", con, con->state);
@@ -1517,14 +1517,15 @@ more:
        }
 
        /* msg pages? */
-       if (con->out_msg) {
+       msg = con->out_msg;
+       if (msg) {
                if (con->v1.out_msg_done) {
-                       ceph_msg_put(con->out_msg);
+                       ceph_msg_put(msg);
                        con->out_msg = NULL;   /* we're done with this one */
                        goto do_next;
                }
 
-               ret = write_partial_message_data(con);
+               ret = write_partial_message_data(con, msg);
                if (ret == 1)
                        goto more;  /* we need to send the footer, too! */
                if (ret == 0)
@@ -1563,10 +1564,8 @@ out:
        return ret;
 }
 
-void ceph_con_v1_revoke(struct ceph_connection *con)
+void ceph_con_v1_revoke(struct ceph_connection *con, struct ceph_msg *msg)
 {
-       struct ceph_msg *msg = con->out_msg;
-
        WARN_ON(con->v1.out_skip);
        /* footer */
        if (con->v1.out_msg_done) {

diff --git a/net/ceph/messenger_v2.c b/net/ceph/messenger_v2.c
index c54c8b5a652612546ec3e1491e569358d5b6b69f..b44e936f3865e89558093a420006c24a0cbdcce6 100644 (file)
--- a/net/ceph/messenger_v2.c
+++ b/net/ceph/messenger_v2.c
@@ -1560,10 +1560,11 @@ static int prepare_ack(struct ceph_connection *con)
        return prepare_control(con, FRAME_TAG_ACK, con->v2.out_buf, 8);
 }
 
-static void prepare_epilogue_plain(struct ceph_connection *con, bool aborted)
+static void prepare_epilogue_plain(struct ceph_connection *con,
+                                  struct ceph_msg *msg, bool aborted)
 {
        dout("%s con %p msg %p aborted %d crcs %u %u %u\n", __func__, con,
-            con->out_msg, aborted, con->v2.out_epil.front_crc,
+            msg, aborted, con->v2.out_epil.front_crc,
             con->v2.out_epil.middle_crc, con->v2.out_epil.data_crc);
 
        encode_epilogue_plain(con, aborted);
@@ -1574,10 +1575,9 @@ static void prepare_epilogue_plain(struct ceph_connection *con, bool aborted)
  * For "used" empty segments, crc is -1.  For unused (trailing)
  * segments, crc is 0.
  */
-static void prepare_message_plain(struct ceph_connection *con)
+static void prepare_message_plain(struct ceph_connection *con,
+                                 struct ceph_msg *msg)
 {
-       struct ceph_msg *msg = con->out_msg;
-
        prepare_head_plain(con, con->v2.out_buf,
                           sizeof(struct ceph_msg_header2), NULL, 0, false);
 
@@ -1618,7 +1618,7 @@ static void prepare_message_plain(struct ceph_connection *con)
                con->v2.out_state = OUT_S_QUEUE_DATA;
        } else {
                con->v2.out_epil.data_crc = 0;
-               prepare_epilogue_plain(con, false);
+               prepare_epilogue_plain(con, msg, false);
                con->v2.out_state = OUT_S_FINISH_MESSAGE;
        }
 }
@@ -1630,7 +1630,8 @@ static void prepare_message_plain(struct ceph_connection *con)
  * allocate pages for the entire tail of the message (currently up
  * to ~32M) and two sgs arrays (up to ~256K each)...
  */
-static int prepare_message_secure(struct ceph_connection *con)
+static int prepare_message_secure(struct ceph_connection *con,
+                                 struct ceph_msg *msg)
 {
        void *zerop = page_address(ceph_zero_page);
        struct sg_table enc_sgt = {};
@@ -1645,7 +1646,7 @@ static int prepare_message_secure(struct ceph_connection *con)
        if (ret)
                return ret;
 
-       tail_len = tail_onwire_len(con->out_msg, true);
+       tail_len = tail_onwire_len(msg, true);
        if (!tail_len) {
                /*
                 * Empty message: once the head is written,
@@ -1656,7 +1657,7 @@ static int prepare_message_secure(struct ceph_connection *con)
        }
 
        encode_epilogue_secure(con, false);
-       ret = setup_message_sgs(&sgt, con->out_msg, zerop, zerop, zerop,
+       ret = setup_message_sgs(&sgt, msg, zerop, zerop, zerop,
                                &con->v2.out_epil, NULL, 0, false);
        if (ret)
                goto out;
@@ -1685,7 +1686,7 @@ static int prepare_message_secure(struct ceph_connection *con)
                goto out;
 
        dout("%s con %p msg %p sg_cnt %d enc_page_cnt %d\n", __func__, con,
-            con->out_msg, sgt.orig_nents, enc_page_cnt);
+            msg, sgt.orig_nents, enc_page_cnt);
        con->v2.out_state = OUT_S_QUEUE_ENC_PAGE;
 
 out:
@@ -1694,19 +1695,19 @@ out:
        return ret;
 }
 
-static int prepare_message(struct ceph_connection *con)
+static int prepare_message(struct ceph_connection *con, struct ceph_msg *msg)
 {
        int lens[] = {
                sizeof(struct ceph_msg_header2),
-               front_len(con->out_msg),
-               middle_len(con->out_msg),
-               data_len(con->out_msg)
+               front_len(msg),
+               middle_len(msg),
+               data_len(msg)
        };
        struct ceph_frame_desc desc;
        int ret;
 
        dout("%s con %p msg %p logical %d+%d+%d+%d\n", __func__, con,
-            con->out_msg, lens[0], lens[1], lens[2], lens[3]);
+            msg, lens[0], lens[1], lens[2], lens[3]);
 
        if (con->in_seq > con->in_seq_acked) {
                dout("%s con %p in_seq_acked %llu -> %llu\n", __func__, con,
@@ -1717,15 +1718,15 @@ static int prepare_message(struct ceph_connection *con)
        reset_out_kvecs(con);
        init_frame_desc(&desc, FRAME_TAG_MESSAGE, lens, 4);
        encode_preamble(&desc, con->v2.out_buf);
-       fill_header2(CTRL_BODY(con->v2.out_buf), &con->out_msg->hdr,
+       fill_header2(CTRL_BODY(con->v2.out_buf), &msg->hdr,
                     con->in_seq_acked);
 
        if (con_secure(con)) {
-               ret = prepare_message_secure(con);
+               ret = prepare_message_secure(con, msg);
                if (ret)
                        return ret;
        } else {
-               prepare_message_plain(con);
+               prepare_message_plain(con, msg);
        }
 
        ceph_con_flag_set(con, CEPH_CON_F_WRITE_PENDING);
@@ -3153,20 +3154,20 @@ int ceph_con_v2_try_read(struct ceph_connection *con)
        }
 }
 
-static void queue_data(struct ceph_connection *con)
+static void queue_data(struct ceph_connection *con, struct ceph_msg *msg)
 {
        struct bio_vec bv;
 
        con->v2.out_epil.data_crc = -1;
-       ceph_msg_data_cursor_init(&con->v2.out_cursor, con->out_msg,
-                                 data_len(con->out_msg));
+       ceph_msg_data_cursor_init(&con->v2.out_cursor, msg,
+                                 data_len(msg));
 
        get_bvec_at(&con->v2.out_cursor, &bv);
        set_out_bvec(con, &bv, true);
        con->v2.out_state = OUT_S_QUEUE_DATA_CONT;
 }
 
-static void queue_data_cont(struct ceph_connection *con)
+static void queue_data_cont(struct ceph_connection *con, struct ceph_msg *msg)
 {
        struct bio_vec bv;
 
@@ -3187,7 +3188,7 @@ static void queue_data_cont(struct ceph_connection *con)
         * we are done.
         */
        reset_out_kvecs(con);
-       prepare_epilogue_plain(con, false);
+       prepare_epilogue_plain(con, msg, false);
        con->v2.out_state = OUT_S_FINISH_MESSAGE;
 }
 
@@ -3219,7 +3220,7 @@ static void queue_enc_page(struct ceph_connection *con)
        con->v2.out_state = OUT_S_FINISH_MESSAGE;
 }
 
-static void queue_zeros(struct ceph_connection *con)
+static void queue_zeros(struct ceph_connection *con, struct ceph_msg *msg)
 {
        dout("%s con %p out_zero %d\n", __func__, con, con->v2.out_zero);
 
@@ -3236,7 +3237,7 @@ static void queue_zeros(struct ceph_connection *con)
         * Once it's written, we are done patching up for the revoke.
         */
        reset_out_kvecs(con);
-       prepare_epilogue_plain(con, true);
+       prepare_epilogue_plain(con, msg, true);
        con->v2.out_state = OUT_S_FINISH_MESSAGE;
 }
 
@@ -3263,6 +3264,7 @@ static void finish_message(struct ceph_connection *con)
 
 static int populate_out_iter(struct ceph_connection *con)
 {
+       struct ceph_msg *msg;
        int ret;
 
        dout("%s con %p state %d out_state %d\n", __func__, con, con->state,
@@ -3278,18 +3280,18 @@ static int populate_out_iter(struct ceph_connection *con)
        switch (con->v2.out_state) {
        case OUT_S_QUEUE_DATA:
                WARN_ON(!con->out_msg);
-               queue_data(con);
+               queue_data(con, con->out_msg);
                goto populated;
        case OUT_S_QUEUE_DATA_CONT:
                WARN_ON(!con->out_msg);
-               queue_data_cont(con);
+               queue_data_cont(con, con->out_msg);
                goto populated;
        case OUT_S_QUEUE_ENC_PAGE:
                queue_enc_page(con);
                goto populated;
        case OUT_S_QUEUE_ZEROS:
                WARN_ON(con->out_msg);  /* revoked */
-               queue_zeros(con);
+               queue_zeros(con, con->out_msg);
                goto populated;
        case OUT_S_FINISH_MESSAGE:
                finish_message(con);
@@ -3309,8 +3311,8 @@ static int populate_out_iter(struct ceph_connection *con)
                        return ret;
                }
        } else if (!list_empty(&con->out_queue)) {
-               ceph_con_get_out_msg(con);
-               ret = prepare_message(con);
+               msg = ceph_con_get_out_msg(con);
+               ret = prepare_message(con, msg);
                if (ret) {
                        pr_err("prepare_message failed: %d\n", ret);
                        return ret;
@@ -3422,17 +3424,18 @@ static u32 crc32c_zeros(u32 crc, int zero_len)
        return crc;
 }
 
-static void prepare_zero_front(struct ceph_connection *con, int resid)
+static void prepare_zero_front(struct ceph_connection *con,
+                              struct ceph_msg *msg, int resid)
 {
        int sent;
 
-       WARN_ON(!resid || resid > front_len(con->out_msg));
-       sent = front_len(con->out_msg) - resid;
+       WARN_ON(!resid || resid > front_len(msg));
+       sent = front_len(msg) - resid;
        dout("%s con %p sent %d resid %d\n", __func__, con, sent, resid);
 
        if (sent) {
                con->v2.out_epil.front_crc =
-                       crc32c(-1, con->out_msg->front.iov_base, sent);
+                       crc32c(-1, msg->front.iov_base, sent);
                con->v2.out_epil.front_crc =
                        crc32c_zeros(con->v2.out_epil.front_crc, resid);
        } else {
@@ -3443,17 +3446,18 @@ static void prepare_zero_front(struct ceph_connection *con, int resid)
        out_zero_add(con, resid);
 }
 
-static void prepare_zero_middle(struct ceph_connection *con, int resid)
+static void prepare_zero_middle(struct ceph_connection *con,
+                               struct ceph_msg *msg, int resid)
 {
        int sent;
 
-       WARN_ON(!resid || resid > middle_len(con->out_msg));
-       sent = middle_len(con->out_msg) - resid;
+       WARN_ON(!resid || resid > middle_len(msg));
+       sent = middle_len(msg) - resid;
        dout("%s con %p sent %d resid %d\n", __func__, con, sent, resid);
 
        if (sent) {
                con->v2.out_epil.middle_crc =
-                       crc32c(-1, con->out_msg->middle->vec.iov_base, sent);
+                       crc32c(-1, msg->middle->vec.iov_base, sent);
                con->v2.out_epil.middle_crc =
                        crc32c_zeros(con->v2.out_epil.middle_crc, resid);
        } else {
@@ -3464,61 +3468,64 @@ static void prepare_zero_middle(struct ceph_connection *con, int resid)
        out_zero_add(con, resid);
 }
 
-static void prepare_zero_data(struct ceph_connection *con)
+static void prepare_zero_data(struct ceph_connection *con,
+                             struct ceph_msg *msg)
 {
        dout("%s con %p\n", __func__, con);
-       con->v2.out_epil.data_crc = crc32c_zeros(-1, data_len(con->out_msg));
-       out_zero_add(con, data_len(con->out_msg));
+       con->v2.out_epil.data_crc = crc32c_zeros(-1, data_len(msg));
+       out_zero_add(con, data_len(msg));
 }
 
-static void revoke_at_queue_data(struct ceph_connection *con)
+static void revoke_at_queue_data(struct ceph_connection *con,
+                                struct ceph_msg *msg)
 {
        int boundary;
        int resid;
 
-       WARN_ON(!data_len(con->out_msg));
+       WARN_ON(!data_len(msg));
        WARN_ON(!iov_iter_is_kvec(&con->v2.out_iter));
        resid = iov_iter_count(&con->v2.out_iter);
 
-       boundary = front_len(con->out_msg) + middle_len(con->out_msg);
+       boundary = front_len(msg) + middle_len(msg);
        if (resid > boundary) {
                resid -= boundary;
                WARN_ON(resid > MESSAGE_HEAD_PLAIN_LEN);
                dout("%s con %p was sending head\n", __func__, con);
-               if (front_len(con->out_msg))
-                       prepare_zero_front(con, front_len(con->out_msg));
-               if (middle_len(con->out_msg))
-                       prepare_zero_middle(con, middle_len(con->out_msg));
-               prepare_zero_data(con);
+               if (front_len(msg))
+                       prepare_zero_front(con, msg, front_len(msg));
+               if (middle_len(msg))
+                       prepare_zero_middle(con, msg, middle_len(msg));
+               prepare_zero_data(con, msg);
                WARN_ON(iov_iter_count(&con->v2.out_iter) != resid);
                con->v2.out_state = OUT_S_QUEUE_ZEROS;
                return;
        }
 
-       boundary = middle_len(con->out_msg);
+       boundary = middle_len(msg);
        if (resid > boundary) {
                resid -= boundary;
                dout("%s con %p was sending front\n", __func__, con);
-               prepare_zero_front(con, resid);
-               if (middle_len(con->out_msg))
-                       prepare_zero_middle(con, middle_len(con->out_msg));
-               prepare_zero_data(con);
-               queue_zeros(con);
+               prepare_zero_front(con, msg, resid);
+               if (middle_len(msg))
+                       prepare_zero_middle(con, msg, middle_len(msg));
+               prepare_zero_data(con, msg);
+               queue_zeros(con, msg);
                return;
        }
 
        WARN_ON(!resid);
        dout("%s con %p was sending middle\n", __func__, con);
-       prepare_zero_middle(con, resid);
-       prepare_zero_data(con);
-       queue_zeros(con);
+       prepare_zero_middle(con, msg, resid);
+       prepare_zero_data(con, msg);
+       queue_zeros(con, msg);
 }
 
-static void revoke_at_queue_data_cont(struct ceph_connection *con)
+static void revoke_at_queue_data_cont(struct ceph_connection *con,
+                                     struct ceph_msg *msg)
 {
        int sent, resid;  /* current piece of data */
 
-       WARN_ON(!data_len(con->out_msg));
+       WARN_ON(!data_len(msg));
        WARN_ON(!iov_iter_is_bvec(&con->v2.out_iter));
        resid = iov_iter_count(&con->v2.out_iter);
        WARN_ON(!resid || resid > con->v2.out_bvec.bv_len);
@@ -3537,10 +3544,11 @@ static void revoke_at_queue_data_cont(struct ceph_connection *con)
 
        con->v2.out_iter.count -= resid;
        out_zero_add(con, con->v2.out_cursor.total_resid);
-       queue_zeros(con);
+       queue_zeros(con, msg);
 }
 
-static void revoke_at_finish_message(struct ceph_connection *con)
+static void revoke_at_finish_message(struct ceph_connection *con,
+                                    struct ceph_msg *msg)
 {
        int boundary;
        int resid;
@@ -3548,39 +3556,39 @@ static void revoke_at_finish_message(struct ceph_connection *con)
        WARN_ON(!iov_iter_is_kvec(&con->v2.out_iter));
        resid = iov_iter_count(&con->v2.out_iter);
 
-       if (!front_len(con->out_msg) && !middle_len(con->out_msg) &&
-           !data_len(con->out_msg)) {
+       if (!front_len(msg) && !middle_len(msg) &&
+           !data_len(msg)) {
                WARN_ON(!resid || resid > MESSAGE_HEAD_PLAIN_LEN);
                dout("%s con %p was sending head (empty message) - noop\n",
                     __func__, con);
                return;
        }
 
-       boundary = front_len(con->out_msg) + middle_len(con->out_msg) +
+       boundary = front_len(msg) + middle_len(msg) +
                   CEPH_EPILOGUE_PLAIN_LEN;
        if (resid > boundary) {
                resid -= boundary;
                WARN_ON(resid > MESSAGE_HEAD_PLAIN_LEN);
                dout("%s con %p was sending head\n", __func__, con);
-               if (front_len(con->out_msg))
-                       prepare_zero_front(con, front_len(con->out_msg));
-               if (middle_len(con->out_msg))
-                       prepare_zero_middle(con, middle_len(con->out_msg));
+               if (front_len(msg))
+                       prepare_zero_front(con, msg, front_len(msg));
+               if (middle_len(msg))
+                       prepare_zero_middle(con, msg, middle_len(msg));
                con->v2.out_iter.count -= CEPH_EPILOGUE_PLAIN_LEN;
                WARN_ON(iov_iter_count(&con->v2.out_iter) != resid);
                con->v2.out_state = OUT_S_QUEUE_ZEROS;
                return;
        }
 
-       boundary = middle_len(con->out_msg) + CEPH_EPILOGUE_PLAIN_LEN;
+       boundary = middle_len(msg) + CEPH_EPILOGUE_PLAIN_LEN;
        if (resid > boundary) {
                resid -= boundary;
                dout("%s con %p was sending front\n", __func__, con);
-               prepare_zero_front(con, resid);
-               if (middle_len(con->out_msg))
-                       prepare_zero_middle(con, middle_len(con->out_msg));
+               prepare_zero_front(con, msg, resid);
+               if (middle_len(msg))
+                       prepare_zero_middle(con, msg, middle_len(msg));
                con->v2.out_iter.count -= CEPH_EPILOGUE_PLAIN_LEN;
-               queue_zeros(con);
+               queue_zeros(con, msg);
                return;
        }
 
@@ -3588,9 +3596,9 @@ static void revoke_at_finish_message(struct ceph_connection *con)
        if (resid > boundary) {
                resid -= boundary;
                dout("%s con %p was sending middle\n", __func__, con);
-               prepare_zero_middle(con, resid);
+               prepare_zero_middle(con, msg, resid);
                con->v2.out_iter.count -= CEPH_EPILOGUE_PLAIN_LEN;
-               queue_zeros(con);
+               queue_zeros(con, msg);
                return;
        }
 
@@ -3598,7 +3606,7 @@ static void revoke_at_finish_message(struct ceph_connection *con)
        dout("%s con %p was sending epilogue - noop\n", __func__, con);
 }
 
-void ceph_con_v2_revoke(struct ceph_connection *con)
+void ceph_con_v2_revoke(struct ceph_connection *con, struct ceph_msg *msg)
 {
        WARN_ON(con->v2.out_zero);
 
@@ -3611,13 +3619,13 @@ void ceph_con_v2_revoke(struct ceph_connection *con)
 
        switch (con->v2.out_state) {
        case OUT_S_QUEUE_DATA:
-               revoke_at_queue_data(con);
+               revoke_at_queue_data(con, msg);
                break;
        case OUT_S_QUEUE_DATA_CONT:
-               revoke_at_queue_data_cont(con);
+               revoke_at_queue_data_cont(con, msg);
                break;
        case OUT_S_FINISH_MESSAGE:
-               revoke_at_finish_message(con);
+               revoke_at_finish_message(con, msg);
                break;
        default:
                WARN(1, "bad out_state %d", con->v2.out_state);