[PATCH 0/4] libceph: use cursor for incoming data

[PATCH 0/4] libceph: use cursor for incoming data

[Alex Elder]

This series changes the incoming data path for the messenger
to use the new data item cursors.

					-Alex

[PATCH 1/4] libceph: use cursor for bio reads
[PATCH 2/4] libceph: kill ceph message bio_iter, bio_seg
[PATCH 3/4] libceph: use cursor for inbound data pages
[PATCH 4/4] libceph: get rid of read helpers

[PATCH 1/4] libceph: use cursor for bio reads

[Alex Elder]

Replace the use of the information in con->in_msg_pos for incoming
bio data.  The old in_msg_pos and the new cursor mechanism do
basically the same thing, just slightly differently.

The main functional difference is that in_msg_pos keeps track of the
length of the complete bio list, and assumed it was fully consumed
when that many bytes had been transferred.  The cursor does not assume
a length, it simply consumes all bytes in the bio list.  Because the
only user of bio data is the rbd client, and because the length of a
bio list provided by rbd client always matches the number of bytes
in the list, both ways of tracking length are equivalent.

In addition, for in_msg_pos the initial bio vector is selected as
the initial value of the bio->bi_idx, while the cursor assumes this
is zero.  Again, the rbd client always passes 0 as the initial index
so the effect is the same.

Other than that, they basically match:
    in_msg_pos      cursor
    ----------      ------
    bio_iter        bio
    bio_seg         vec_index
    page_pos        page_offset

The in_msg_pos field is initialized by a call to init_bio_iter().
The bio cursor is initialized by ceph_msg_data_cursor_init().
Both now happen in the same spot, in prepare_message_data().

The in_msg_pos field is advanced by a call to in_msg_pos_next(),
which updates page_pos and calls iter_bio_next() to move to the next
bio vector, or to the next bio in the list.  The cursor is advanced
by ceph_msg_data_advance().  That isn't currently happening so
add a call to that in in_msg_pos_next().

Finally, the next piece of data to use for a read is determined
by a bunch of lines in read_partial_message_bio().  Those can be
replaced by an equivalent ceph_msg_data_bio_next() call.

Since the read path has no need to know whether the next piece of a
bio to use is the last, allow it to pass a null pointer as its
last_piece argument to ceph_msg_data_bio_next().

This partially resolves:
    http://tracker.ceph.com/issues/4428

Signed-off-by: Alex Elder <elder@inktank.com>
---
 net/ceph/messenger.c |   17 ++++++-----------
 1 file changed, 6 insertions(+), 11 deletions(-)

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 07ace1d..1b003b4 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -1450,6 +1450,10 @@ static void in_msg_pos_next(struct
ceph_connection *con, size_t len,

 	msg_pos->data_pos += received;
 	msg_pos->page_pos += received;
+#ifdef CONFIG_BLOCK
+	if (ceph_msg_has_bio(msg))
+		(void) ceph_msg_data_advance(&msg->b, received);
+#endif /* CONFIG_BLOCK */
 	if (received < len)
 		return;

@@ -2225,23 +2229,14 @@ static int read_partial_message_bio(struct
ceph_connection *con,
 				    unsigned int data_len, bool do_datacrc)
 {
 	struct ceph_msg *msg = con->in_msg;
-	struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
-	struct bio_vec *bv;
 	struct page *page;
 	size_t page_offset;
 	size_t length;
-	unsigned int left;
 	int ret;

 	BUG_ON(!msg);
-	BUG_ON(!msg->b.bio_iter);
-	bv = bio_iovec_idx(msg->b.bio_iter, msg->b.bio_seg);
-	page = bv->bv_page;
-	page_offset = bv->bv_offset + msg_pos->page_pos;
-	BUG_ON(msg_pos->data_pos >= data_len);
-	left = data_len - msg_pos->data_pos;
-	BUG_ON(msg_pos->page_pos >= bv->bv_len);
-	length = min_t(unsigned int, bv->bv_len - msg_pos->page_pos, left);
+
+	page = ceph_msg_data_next(&msg->b, &page_offset, &length, NULL);

 	ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
 	if (ret <= 0)
-- 
1.7.9.5

[PATCH 2/4] libceph: kill ceph message bio_iter, bio_seg

[Alex Elder]

The bio_iter and bio_seg fields in a message are no longer used, we
use the cursor instead.  So get rid of them and the functions that
operate on them them.

This is related to:
    http://tracker.ceph.com/issues/4428

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |    6 +-----
 net/ceph/messenger.c           |   31 -------------------------------
 2 files changed, 1 insertion(+), 36 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 459e552..252e01b 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -121,11 +121,7 @@ struct ceph_msg_data {
 	enum ceph_msg_data_type		type;
 	union {
 #ifdef CONFIG_BLOCK
-		struct {
-			struct bio	*bio_iter;	/* iterator */
-			struct bio	*bio;
-			unsigned int	bio_seg;	/* current seg in bio */
-		};
+		struct bio		*bio;
 #endif /* CONFIG_BLOCK */
 		struct {
 			struct page	**pages;	/* NOT OWNER. */
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 1b003b4..835447c 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -716,29 +716,6 @@ static void con_out_kvec_add(struct ceph_connection
*con,
 }

 #ifdef CONFIG_BLOCK
-static void init_bio_iter(struct bio *bio, struct bio **bio_iter,
-			unsigned int *bio_seg)
-{
-	if (!bio) {
-		*bio_iter = NULL;
-		*bio_seg = 0;
-		return;
-	}
-	*bio_iter = bio;
-	*bio_seg = (unsigned int) bio->bi_idx;
-}
-
-static void iter_bio_next(struct bio **bio_iter, unsigned int *seg)
-{
-	if (*bio_iter == NULL)
-		return;
-
-	BUG_ON(*seg >= (*bio_iter)->bi_vcnt);
-
-	(*seg)++;
-	if (*seg == (*bio_iter)->bi_vcnt)
-		init_bio_iter((*bio_iter)->bi_next, bio_iter, seg);
-}

 /*
  * For a bio data item, a piece is whatever remains of the next
@@ -1112,10 +1089,6 @@ static void prepare_message_data(struct ceph_msg
*msg,
 		msg_pos->page_pos = msg->p.alignment;
 	else
 		msg_pos->page_pos = 0;
-#ifdef CONFIG_BLOCK
-	if (ceph_msg_has_bio(msg))
-		init_bio_iter(msg->b.bio, &msg->b.bio_iter, &msg->b.bio_seg);
-#endif
 	msg_pos->data_pos = 0;

 	/* Initialize data cursors */
@@ -1460,10 +1433,6 @@ static void in_msg_pos_next(struct
ceph_connection *con, size_t len,
 	BUG_ON(received != len);
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
-#ifdef CONFIG_BLOCK
-	if (msg->b.bio)
-		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
-#endif /* CONFIG_BLOCK */
 }

 static u32 ceph_crc32c_page(u32 crc, struct page *page,
-- 
1.7.9.5

[PATCH 3/4] libceph: use cursor for inbound data pages

[Alex Elder]

The cursor code for a page array selects the right page, page
offset, and length to use for a ceph_tcp_recvpage() call, so
we can use it to replace a block in read_partial_message_pages().

This partially resolves:
    http://tracker.ceph.com/issues/4428

Signed-off-by: Alex Elder <elder@inktank.com>
---
 net/ceph/messenger.c |   17 ++++-------------
 1 file changed, 4 insertions(+), 13 deletions(-)

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 835447c..5507a12 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -1423,8 +1423,10 @@ static void in_msg_pos_next(struct
ceph_connection *con, size_t len,

 	msg_pos->data_pos += received;
 	msg_pos->page_pos += received;
+	if (ceph_msg_has_pages(msg))
+		(void) ceph_msg_data_advance(&msg->p, received);
 #ifdef CONFIG_BLOCK
-	if (ceph_msg_has_bio(msg))
+	else if (ceph_msg_has_bio(msg))
 		(void) ceph_msg_data_advance(&msg->b, received);
 #endif /* CONFIG_BLOCK */
 	if (received < len)
@@ -2162,23 +2164,12 @@ static int read_partial_message_pages(struct
ceph_connection *con,
 				      unsigned int data_len, bool do_datacrc)
 {
 	struct ceph_msg *msg = con->in_msg;
-	struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
-	struct page **pages;
 	struct page *page;
 	size_t page_offset;
 	size_t length;
-	unsigned int left;
 	int ret;

-	/* (page) data */
-	pages = msg->p.pages;
-	BUG_ON(pages == NULL);
-	page = pages[msg_pos->page];
-	page_offset = msg_pos->page_pos;
-	BUG_ON(msg_pos->data_pos >= data_len);
-	left = data_len - msg_pos->data_pos;
-	BUG_ON(page_offset >= PAGE_SIZE);
-	length = min_t(unsigned int, PAGE_SIZE - page_offset, left);
+	page = ceph_msg_data_next(&msg->p, &page_offset, &length, NULL);

 	ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
 	if (ret <= 0)
-- 
1.7.9.5

[PATCH 4/4] libceph: get rid of read helpers

[Alex Elder]

Now that read_partial_message_pages() and read_partial_message_bio()
are literally identical functions we can factor them out.  They're
pretty simple as well, so just move their relevant content into
read_partial_msg_data().

This is and previous patches together resolve:
    http://tracker.ceph.com/issues/4428

Signed-off-by: Alex Elder <elder@inktank.com>
---
 net/ceph/messenger.c |   80
++++++++++++--------------------------------------
 1 file changed, 18 insertions(+), 62 deletions(-)

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 5507a12..190b74b 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -2158,66 +2158,15 @@ static int read_partial_message_section(struct
ceph_connection *con,
 	return 1;
 }

-static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);
-
-static int read_partial_message_pages(struct ceph_connection *con,
-				      unsigned int data_len, bool do_datacrc)
-{
-	struct ceph_msg *msg = con->in_msg;
-	struct page *page;
-	size_t page_offset;
-	size_t length;
-	int ret;
-
-	page = ceph_msg_data_next(&msg->p, &page_offset, &length, NULL);
-
-	ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
-	if (ret <= 0)
-		return ret;
-
-	if (do_datacrc)
-		con->in_data_crc = ceph_crc32c_page(con->in_data_crc, page,
-							page_offset, ret);
-
-	in_msg_pos_next(con, length, ret);
-
-	return ret;
-}
-
-#ifdef CONFIG_BLOCK
-static int read_partial_message_bio(struct ceph_connection *con,
-				    unsigned int data_len, bool do_datacrc)
-{
-	struct ceph_msg *msg = con->in_msg;
-	struct page *page;
-	size_t page_offset;
-	size_t length;
-	int ret;
-
-	BUG_ON(!msg);
-
-	page = ceph_msg_data_next(&msg->b, &page_offset, &length, NULL);
-
-	ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
-	if (ret <= 0)
-		return ret;
-
-	if (do_datacrc)
-		con->in_data_crc = ceph_crc32c_page(con->in_data_crc, page,
-							page_offset, ret);
-
-	in_msg_pos_next(con, length, ret);
-
-	return ret;
-}
-#endif
-
 static int read_partial_msg_data(struct ceph_connection *con)
 {
 	struct ceph_msg *msg = con->in_msg;
 	struct ceph_msg_pos *msg_pos = &con->in_msg_pos;
 	const bool do_datacrc = !con->msgr->nocrc;
 	unsigned int data_len;
+	struct page *page;
+	size_t page_offset;
+	size_t length;
 	int ret;

 	BUG_ON(!msg);
@@ -2225,20 +2174,25 @@ static int read_partial_msg_data(struct
ceph_connection *con)
 	data_len = le32_to_cpu(con->in_hdr.data_len);
 	while (msg_pos->data_pos < data_len) {
 		if (ceph_msg_has_pages(msg)) {
-			ret = read_partial_message_pages(con, data_len,
-							do_datacrc);
-			if (ret <= 0)
-				return ret;
+			page = ceph_msg_data_next(&msg->p, &page_offset,
+					&length, NULL);
 #ifdef CONFIG_BLOCK
 		} else if (ceph_msg_has_bio(msg)) {
-			ret = read_partial_message_bio(con,
-						 data_len, do_datacrc);
-			if (ret <= 0)
-				return ret;
+			page = ceph_msg_data_next(&msg->b, &page_offset,
+					&length, NULL);
 #endif
 		} else {
 			BUG_ON(1);
 		}
+		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
+		if (ret <= 0)
+			return ret;
+
+		if (do_datacrc)
+			con->in_data_crc = ceph_crc32c_page(con->in_data_crc,
+							page, page_offset, ret);
+
+		in_msg_pos_next(con, length, ret);
 	}

 	return 1;	/* must return > 0 to indicate success */
@@ -2247,6 +2201,8 @@ static int read_partial_msg_data(struct
ceph_connection *con)
 /*
  * read (part of) a message.
  */
+static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);
+
 static int read_partial_message(struct ceph_connection *con)
 {
 	struct ceph_msg *m = con->in_msg;
-- 
1.7.9.5

Re: [PATCH 0/4] libceph: use cursor for incoming data

[Josh Durgin]

On 03/12/2013 06:04 PM, Alex Elder wrote:
> This series changes the incoming data path for the messenger
> to use the new data item cursors.
>
> 					-Alex
>
> [PATCH 1/4] libceph: use cursor for bio reads
> [PATCH 2/4] libceph: kill ceph message bio_iter, bio_seg
> [PATCH 3/4] libceph: use cursor for inbound data pages
> [PATCH 4/4] libceph: get rid of read helpers

These all look good.

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>