Commit 7399212d authored by Max Kellermann's avatar Max Kellermann Committed by Ilya Dryomov
Browse files

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: default avatarMax Kellermann <max.kellermann@ionos.com>
Reviewed-by: default avatarViacheslav Dubeyko <Slava.Dubeyko@ibm.com>
Signed-off-by: default avatarIlya Dryomov <idryomov@gmail.com>
parent 59699a5a

include/linux/ceph/messenger.h

+2 −2
Original line number Diff line number Diff line
@@ -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);

net/ceph/messenger.c

+2 −2
Original line number Diff line number Diff line
@@ -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 {

net/ceph/messenger_v1.c

+22 −23
Original line number Diff line number Diff line
@@ -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 @@ static int write_partial_kvec(struct ceph_connection *con) 
 *  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 @@ int ceph_con_v1_try_read(struct ceph_connection *con) 
 */

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 @@ int ceph_con_v1_try_write(struct ceph_connection *con) 
	}



	/* 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 @@ int ceph_con_v1_try_write(struct ceph_connection *con) 
	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) {

net/ceph/messenger_v2.c

+88 −80
Original line number Diff line number Diff line
@@ -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 @@ static int prepare_message_secure(struct ceph_connection *con) 
	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);