[PATCH 0/8] libceph: implement cursor for outgoing data items

[PATCH 0/8] libceph: implement cursor for outgoing data items

[Alex Elder]

This series makes the four distinct data items in a message (the
pages array, the pagelist, the bio, and the trail) take a generic
form, and then arranges for them all to be handled using a common
set of cursor routines that determine what data needs to be sent
next for a message.  There is more consolidation of this code
coming, but for now I'm leaving this series at 8 patches, which
includes the definition of cursor code for the three types of
data item.

					-Alex

[PATCH 1/8] libceph: define ceph_msg_has_*() data macros
[PATCH 2/8] libceph: be explicit about message data representation
[PATCH 3/8] libceph: abstract message data
[PATCH 4/8] libceph: start defining message data cursor
[PATCH 5/8] libceph: prepare for other message data item types
[PATCH 6/8] libceph: use data cursor for message pagelist
[PATCH 7/8] libceph: implement bio message data item cursor
[PATCH 8/8] libceph: implement pages array cursor

[PATCH 1/8] libceph: define ceph_msg_has_*() data macros

[Alex Elder]

Define and use macros ceph_msg_has_*() to determine whether to
operate on the pages, pagelist, bio, and trail fields of a message.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |    7 +++++++
 net/ceph/messenger.c           |   44
++++++++++++++++++++++++----------------
 2 files changed, 34 insertions(+), 17 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 1991a6f..889fe47 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -64,6 +64,13 @@ struct ceph_messenger {
 	u32 required_features;
 };

+#define ceph_msg_has_pages(m)		((m)->pages != NULL)
+#define ceph_msg_has_pagelist(m)	((m)->pagelist != NULL)
+#ifdef CONFIG_BLOCK
+#define ceph_msg_has_bio(m)		((m)->bio != NULL)
+#endif /* CONFIG_BLOCK */
+#define ceph_msg_has_trail(m)		((m)->trail != NULL)
+
 /*
  * a single message.  it contains a header (src, dest, message type, etc.),
  * footer (crc values, mainly), a "front" message body, and possibly a
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index f70bc92..c74b528 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -746,12 +746,12 @@ static void prepare_message_data(struct ceph_msg *msg,

 	/* initialize page iterator */
 	msg_pos->page = 0;
-	if (msg->pages)
+	if (ceph_msg_has_pages(msg))
 		msg_pos->page_pos = msg->page_alignment;
 	else
 		msg_pos->page_pos = 0;
 #ifdef CONFIG_BLOCK
-	if (msg->bio)
+	if (ceph_msg_has_bio(msg))
 		init_bio_iter(msg->bio, &msg->bio_iter, &msg->bio_seg);
 #endif
 	msg_pos->data_pos = 0;
@@ -1052,14 +1052,16 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 	msg_pos->did_page_crc = false;
-	if (in_trail)
+	if (in_trail) {
+		BUG_ON(!ceph_msg_has_trail(msg));
 		list_rotate_left(&msg->trail->head);
-	else if (msg->pagelist)
+	} else if (ceph_msg_has_pagelist(msg)) {
 		list_rotate_left(&msg->pagelist->head);
 #ifdef CONFIG_BLOCK
-	else if (msg->bio)
+	} else if (ceph_msg_has_bio(msg)) {
 		iter_bio_next(&msg->bio_iter, &msg->bio_seg);
 #endif
+	}
 }

 static void in_msg_pos_next(struct ceph_connection *con, size_t len,
@@ -1114,8 +1116,13 @@ static int write_partial_message_data(struct
ceph_connection *con)
 	int ret;
 	int total_max_write;
 	bool in_trail = false;
-	const size_t trail_len = (msg->trail ? msg->trail->length : 0);
-	const size_t trail_off = data_len - trail_len;
+	size_t trail_len = 0;
+	size_t trail_off = data_len;
+
+	if (ceph_msg_has_trail(msg)) {
+		trail_len = msg->trail->length;
+		trail_off -= trail_len;
+	}

 	dout("%s %p msg %p page %d offset %d\n", __func__,
 	     con, msg, msg_pos->page, msg_pos->page_pos);
@@ -1140,17 +1147,17 @@ static int write_partial_message_data(struct
ceph_connection *con)
 			total_max_write = trail_off - msg_pos->data_pos;

 		if (in_trail) {
+			BUG_ON(!ceph_msg_has_trail(msg));
 			total_max_write = data_len - msg_pos->data_pos;
-
 			page = list_first_entry(&msg->trail->head,
 						struct page, lru);
-		} else if (msg->pages) {
+		} else if (ceph_msg_has_pages(msg)) {
 			page = msg->pages[msg_pos->page];
-		} else if (msg->pagelist) {
+		} else if (ceph_msg_has_pagelist(msg)) {
 			page = list_first_entry(&msg->pagelist->head,
 						struct page, lru);
 #ifdef CONFIG_BLOCK
-		} else if (msg->bio) {
+		} else if (ceph_msg_has_bio(msg)) {
 			struct bio_vec *bv;

 			bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
@@ -1908,13 +1915,13 @@ 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 (msg->pages) {
+		if (ceph_msg_has_pages(msg)) {
 			ret = read_partial_message_pages(con, msg->pages,
 						 data_len, do_datacrc);
 			if (ret <= 0)
 				return ret;
 #ifdef CONFIG_BLOCK
-		} else if (msg->bio) {
+		} else if (ceph_msg_has_bio(msg)) {
 			ret = read_partial_message_bio(con,
 						 data_len, do_datacrc);
 			if (ret <= 0)
@@ -2946,16 +2953,19 @@ void ceph_msg_last_put(struct kref *kref)
 		ceph_buffer_put(m->middle);
 		m->middle = NULL;
 	}
-	m->length = 0;
-	m->pages = NULL;
+	if (ceph_msg_has_pages(m)) {
+		m->length = 0;
+		m->pages = NULL;
+	}

-	if (m->pagelist) {
+	if (ceph_msg_has_pagelist(m)) {
 		ceph_pagelist_release(m->pagelist);
 		kfree(m->pagelist);
 		m->pagelist = NULL;
 	}

-	m->trail = NULL;
+	if (ceph_msg_has_trail(m))
+		m->trail = NULL;

 	if (m->pool)
 		ceph_msgpool_put(m->pool, m);
-- 
1.7.9.5

[PATCH 2/8] libceph: be explicit about message data representation

[Alex Elder]

A ceph message has a data payload portion.  The memory for that data
(either the source of data to send or the location to place data
that is received) is specified in several ways.  The ceph_msg
structure includes fields for all of those ways, but this
mispresents the fact that not all of them are used at a time.

Specifically, the data in a message can be in:
    - an array of pages
    - a list of pages
    - a list of Linux bios
    - a second list of pages (the "trail")
(The two page lists are currently only ever used for outgoing data.)

Impose more structure on the ceph message, making the grouping of
some of these fields explicit.  Shorten the name of the
"page_alignment" field.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |   33 ++++++++++++-------
 net/ceph/messenger.c           |   68
++++++++++++++++++++--------------------
 2 files changed, 55 insertions(+), 46 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 889fe47..fb2b18a 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -64,12 +64,12 @@ struct ceph_messenger {
 	u32 required_features;
 };

-#define ceph_msg_has_pages(m)		((m)->pages != NULL)
-#define ceph_msg_has_pagelist(m)	((m)->pagelist != NULL)
+#define ceph_msg_has_pages(m)		((m)->p.pages != NULL)
+#define ceph_msg_has_pagelist(m)	((m)->l.pagelist != NULL)
 #ifdef CONFIG_BLOCK
-#define ceph_msg_has_bio(m)		((m)->bio != NULL)
+#define ceph_msg_has_bio(m)		((m)->b.bio != NULL)
 #endif /* CONFIG_BLOCK */
-#define ceph_msg_has_trail(m)		((m)->trail != NULL)
+#define ceph_msg_has_trail(m)		((m)->t.trail != NULL)

 /*
  * a single message.  it contains a header (src, dest, message type, etc.),
@@ -82,16 +82,25 @@ struct ceph_msg {
 	struct kvec front;              /* unaligned blobs of message */
 	struct ceph_buffer *middle;

-	struct page **pages;		/* data payload.  NOT OWNER. */
-	unsigned int page_alignment;	/* io offset in first page */
-	size_t length;			/* # data bytes in array or list */
-	struct ceph_pagelist *pagelist; /* instead of pages */
+	/* data payload */
+	struct {
+		struct page	**pages;	/* NOT OWNER. */
+		size_t		length;		/* # data bytes in array */
+		unsigned int	alignment;	/* first page */
+	} p;
+	struct {
+		struct ceph_pagelist	*pagelist;
+	} l;
 #ifdef CONFIG_BLOCK
-	unsigned int bio_seg;		/* current bio segment */
-	struct bio  *bio;		/* instead of pages/pagelist */
-	struct bio  *bio_iter;		/* bio iterator */
+	struct {
+		struct bio	*bio_iter;	/* iterator */
+		struct bio	*bio;
+		unsigned int	bio_seg;	/* current seg in bio */
+	} b;
 #endif /* CONFIG_BLOCK */
-	struct ceph_pagelist *trail;	/* the trailing part of the data */
+	struct {
+		struct ceph_pagelist *trail;	/* trailing part of data */
+	} t;

 	struct ceph_connection *con;
 	struct list_head list_head;	/* links for connection lists */
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index c74b528..f485455 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -747,12 +747,12 @@ static void prepare_message_data(struct ceph_msg *msg,
 	/* initialize page iterator */
 	msg_pos->page = 0;
 	if (ceph_msg_has_pages(msg))
-		msg_pos->page_pos = msg->page_alignment;
+		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->bio, &msg->bio_iter, &msg->bio_seg);
+		init_bio_iter(msg->b.bio, &msg->b.bio_iter, &msg->b.bio_seg);
 #endif
 	msg_pos->data_pos = 0;
 	msg_pos->did_page_crc = false;
@@ -822,7 +822,7 @@ static void prepare_write_message(struct
ceph_connection *con)
 	dout("prepare_write_message %p seq %lld type %d len %d+%d+%d (%zd)\n",
 	     m, con->out_seq, le16_to_cpu(m->hdr.type),
 	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
-	     le32_to_cpu(m->hdr.data_len), m->length);
+	     le32_to_cpu(m->hdr.data_len), m->p.length);
 	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

 	/* tag + hdr + front + middle */
@@ -1054,12 +1054,12 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->did_page_crc = false;
 	if (in_trail) {
 		BUG_ON(!ceph_msg_has_trail(msg));
-		list_rotate_left(&msg->trail->head);
+		list_rotate_left(&msg->t.trail->head);
 	} else if (ceph_msg_has_pagelist(msg)) {
-		list_rotate_left(&msg->pagelist->head);
+		list_rotate_left(&msg->l.pagelist->head);
 #ifdef CONFIG_BLOCK
 	} else if (ceph_msg_has_bio(msg)) {
-		iter_bio_next(&msg->bio_iter, &msg->bio_seg);
+		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
 #endif
 	}
 }
@@ -1082,8 +1082,8 @@ static void in_msg_pos_next(struct ceph_connection
*con, size_t len,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 #ifdef CONFIG_BLOCK
-	if (msg->bio)
-		iter_bio_next(&msg->bio_iter, &msg->bio_seg);
+	if (msg->b.bio)
+		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
 #endif /* CONFIG_BLOCK */
 }

@@ -1120,7 +1120,7 @@ static int write_partial_message_data(struct
ceph_connection *con)
 	size_t trail_off = data_len;

 	if (ceph_msg_has_trail(msg)) {
-		trail_len = msg->trail->length;
+		trail_len = msg->t.trail->length;
 		trail_off -= trail_len;
 	}

@@ -1149,18 +1149,18 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		if (in_trail) {
 			BUG_ON(!ceph_msg_has_trail(msg));
 			total_max_write = data_len - msg_pos->data_pos;
-			page = list_first_entry(&msg->trail->head,
+			page = list_first_entry(&msg->t.trail->head,
 						struct page, lru);
 		} else if (ceph_msg_has_pages(msg)) {
-			page = msg->pages[msg_pos->page];
+			page = msg->p.pages[msg_pos->page];
 		} else if (ceph_msg_has_pagelist(msg)) {
-			page = list_first_entry(&msg->pagelist->head,
+			page = list_first_entry(&msg->l.pagelist->head,
 						struct page, lru);
 #ifdef CONFIG_BLOCK
 		} else if (ceph_msg_has_bio(msg)) {
 			struct bio_vec *bv;

-			bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
+			bv = bio_iovec_idx(msg->b.bio_iter, msg->b.bio_seg);
 			page = bv->bv_page;
 			bio_offset = bv->bv_offset;
 			max_write = bv->bv_len;
@@ -1880,8 +1880,8 @@ static int read_partial_message_bio(struct
ceph_connection *con,
 	int ret;

 	BUG_ON(!msg);
-	BUG_ON(!msg->bio_iter);
-	bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
+	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);
@@ -1916,7 +1916,7 @@ 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, msg->pages,
+			ret = read_partial_message_pages(con, msg->p.pages,
 						 data_len, do_datacrc);
 			if (ret <= 0)
 				return ret;
@@ -2741,12 +2741,12 @@ void ceph_msg_data_set_pages(struct ceph_msg
*msg, struct page **pages,
 {
 	BUG_ON(!pages);
 	BUG_ON(!length);
-	BUG_ON(msg->pages);
-	BUG_ON(msg->length);
+	BUG_ON(msg->p.pages);
+	BUG_ON(msg->p.length);

-	msg->pages = pages;
-	msg->length = length;
-	msg->page_alignment = alignment & ~PAGE_MASK;
+	msg->p.pages = pages;
+	msg->p.length = length;
+	msg->p.alignment = alignment & ~PAGE_MASK;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pages);

@@ -2755,18 +2755,18 @@ void ceph_msg_data_set_pagelist(struct ceph_msg
*msg,
 {
 	BUG_ON(!pagelist);
 	BUG_ON(!pagelist->length);
-	BUG_ON(msg->pagelist);
+	BUG_ON(msg->l.pagelist);

-	msg->pagelist = pagelist;
+	msg->l.pagelist = pagelist;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pagelist);

 void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
 {
 	BUG_ON(!bio);
-	BUG_ON(msg->bio);
+	BUG_ON(msg->b.bio);

-	msg->bio = bio;
+	msg->b.bio = bio;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_bio);

@@ -2774,9 +2774,9 @@ void ceph_msg_data_set_trail(struct ceph_msg *msg,
struct ceph_pagelist *trail)
 {
 	BUG_ON(!trail);
 	BUG_ON(!trail->length);
-	BUG_ON(msg->trail);
+	BUG_ON(msg->t.trail);

-	msg->trail = trail;
+	msg->t.trail = trail;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_trail);

@@ -2954,18 +2954,18 @@ void ceph_msg_last_put(struct kref *kref)
 		m->middle = NULL;
 	}
 	if (ceph_msg_has_pages(m)) {
-		m->length = 0;
-		m->pages = NULL;
+		m->p.length = 0;
+		m->p.pages = NULL;
 	}

 	if (ceph_msg_has_pagelist(m)) {
-		ceph_pagelist_release(m->pagelist);
-		kfree(m->pagelist);
-		m->pagelist = NULL;
+		ceph_pagelist_release(m->l.pagelist);
+		kfree(m->l.pagelist);
+		m->l.pagelist = NULL;
 	}

 	if (ceph_msg_has_trail(m))
-		m->trail = NULL;
+		m->t.trail = NULL;

 	if (m->pool)
 		ceph_msgpool_put(m->pool, m);
@@ -2977,7 +2977,7 @@ EXPORT_SYMBOL(ceph_msg_last_put);
 void ceph_msg_dump(struct ceph_msg *msg)
 {
 	pr_debug("msg_dump %p (front_max %d length %zd)\n", msg,
-		 msg->front_max, msg->length);
+		 msg->front_max, msg->p.length);
 	print_hex_dump(KERN_DEBUG, "header: ",
 		       DUMP_PREFIX_OFFSET, 16, 1,
 		       &msg->hdr, sizeof(msg->hdr), true);
-- 
1.7.9.5

[PATCH 3/8] libceph: abstract message data

[Alex Elder]

A ceph message has a data payload portion.  The memory for that data
(either the source of data to send or the location to place data
that is received) is specified in several ways.  The ceph_msg
structure includes fields for all of those ways, but this
mispresents the fact that not all of them are used at a time.

Specifically, the data in a message can be in:
    - an array of pages
    - a list of pages
    - a list of Linux bios
    - a second list of pages (the "trail")
(The two page lists are currently only ever used for outgoing data.)

Impose more structure on the ceph message, making the grouping of
some of these fields explicit.  Shorten the name of the
"page_alignment" field.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |   33 ++++++++++++-------
 net/ceph/messenger.c           |   68
++++++++++++++++++++--------------------
 2 files changed, 55 insertions(+), 46 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 889fe47..fb2b18a 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -64,12 +64,12 @@ struct ceph_messenger {
 	u32 required_features;
 };

-#define ceph_msg_has_pages(m)		((m)->pages != NULL)
-#define ceph_msg_has_pagelist(m)	((m)->pagelist != NULL)
+#define ceph_msg_has_pages(m)		((m)->p.pages != NULL)
+#define ceph_msg_has_pagelist(m)	((m)->l.pagelist != NULL)
 #ifdef CONFIG_BLOCK
-#define ceph_msg_has_bio(m)		((m)->bio != NULL)
+#define ceph_msg_has_bio(m)		((m)->b.bio != NULL)
 #endif /* CONFIG_BLOCK */
-#define ceph_msg_has_trail(m)		((m)->trail != NULL)
+#define ceph_msg_has_trail(m)		((m)->t.trail != NULL)

 /*
  * a single message.  it contains a header (src, dest, message type, etc.),
@@ -82,16 +82,25 @@ struct ceph_msg {
 	struct kvec front;              /* unaligned blobs of message */
 	struct ceph_buffer *middle;

-	struct page **pages;		/* data payload.  NOT OWNER. */
-	unsigned int page_alignment;	/* io offset in first page */
-	size_t length;			/* # data bytes in array or list */
-	struct ceph_pagelist *pagelist; /* instead of pages */
+	/* data payload */
+	struct {
+		struct page	**pages;	/* NOT OWNER. */
+		size_t		length;		/* # data bytes in array */
+		unsigned int	alignment;	/* first page */
+	} p;
+	struct {
+		struct ceph_pagelist	*pagelist;
+	} l;
 #ifdef CONFIG_BLOCK
-	unsigned int bio_seg;		/* current bio segment */
-	struct bio  *bio;		/* instead of pages/pagelist */
-	struct bio  *bio_iter;		/* bio iterator */
+	struct {
+		struct bio	*bio_iter;	/* iterator */
+		struct bio	*bio;
+		unsigned int	bio_seg;	/* current seg in bio */
+	} b;
 #endif /* CONFIG_BLOCK */
-	struct ceph_pagelist *trail;	/* the trailing part of the data */
+	struct {
+		struct ceph_pagelist *trail;	/* trailing part of data */
+	} t;

 	struct ceph_connection *con;
 	struct list_head list_head;	/* links for connection lists */
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index c74b528..f485455 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -747,12 +747,12 @@ static void prepare_message_data(struct ceph_msg *msg,
 	/* initialize page iterator */
 	msg_pos->page = 0;
 	if (ceph_msg_has_pages(msg))
-		msg_pos->page_pos = msg->page_alignment;
+		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->bio, &msg->bio_iter, &msg->bio_seg);
+		init_bio_iter(msg->b.bio, &msg->b.bio_iter, &msg->b.bio_seg);
 #endif
 	msg_pos->data_pos = 0;
 	msg_pos->did_page_crc = false;
@@ -822,7 +822,7 @@ static void prepare_write_message(struct
ceph_connection *con)
 	dout("prepare_write_message %p seq %lld type %d len %d+%d+%d (%zd)\n",
 	     m, con->out_seq, le16_to_cpu(m->hdr.type),
 	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
-	     le32_to_cpu(m->hdr.data_len), m->length);
+	     le32_to_cpu(m->hdr.data_len), m->p.length);
 	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

 	/* tag + hdr + front + middle */
@@ -1054,12 +1054,12 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->did_page_crc = false;
 	if (in_trail) {
 		BUG_ON(!ceph_msg_has_trail(msg));
-		list_rotate_left(&msg->trail->head);
+		list_rotate_left(&msg->t.trail->head);
 	} else if (ceph_msg_has_pagelist(msg)) {
-		list_rotate_left(&msg->pagelist->head);
+		list_rotate_left(&msg->l.pagelist->head);
 #ifdef CONFIG_BLOCK
 	} else if (ceph_msg_has_bio(msg)) {
-		iter_bio_next(&msg->bio_iter, &msg->bio_seg);
+		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
 #endif
 	}
 }
@@ -1082,8 +1082,8 @@ static void in_msg_pos_next(struct ceph_connection
*con, size_t len,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 #ifdef CONFIG_BLOCK
-	if (msg->bio)
-		iter_bio_next(&msg->bio_iter, &msg->bio_seg);
+	if (msg->b.bio)
+		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
 #endif /* CONFIG_BLOCK */
 }

@@ -1120,7 +1120,7 @@ static int write_partial_message_data(struct
ceph_connection *con)
 	size_t trail_off = data_len;

 	if (ceph_msg_has_trail(msg)) {
-		trail_len = msg->trail->length;
+		trail_len = msg->t.trail->length;
 		trail_off -= trail_len;
 	}

@@ -1149,18 +1149,18 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		if (in_trail) {
 			BUG_ON(!ceph_msg_has_trail(msg));
 			total_max_write = data_len - msg_pos->data_pos;
-			page = list_first_entry(&msg->trail->head,
+			page = list_first_entry(&msg->t.trail->head,
 						struct page, lru);
 		} else if (ceph_msg_has_pages(msg)) {
-			page = msg->pages[msg_pos->page];
+			page = msg->p.pages[msg_pos->page];
 		} else if (ceph_msg_has_pagelist(msg)) {
-			page = list_first_entry(&msg->pagelist->head,
+			page = list_first_entry(&msg->l.pagelist->head,
 						struct page, lru);
 #ifdef CONFIG_BLOCK
 		} else if (ceph_msg_has_bio(msg)) {
 			struct bio_vec *bv;

-			bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
+			bv = bio_iovec_idx(msg->b.bio_iter, msg->b.bio_seg);
 			page = bv->bv_page;
 			bio_offset = bv->bv_offset;
 			max_write = bv->bv_len;
@@ -1880,8 +1880,8 @@ static int read_partial_message_bio(struct
ceph_connection *con,
 	int ret;

 	BUG_ON(!msg);
-	BUG_ON(!msg->bio_iter);
-	bv = bio_iovec_idx(msg->bio_iter, msg->bio_seg);
+	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);
@@ -1916,7 +1916,7 @@ 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, msg->pages,
+			ret = read_partial_message_pages(con, msg->p.pages,
 						 data_len, do_datacrc);
 			if (ret <= 0)
 				return ret;
@@ -2741,12 +2741,12 @@ void ceph_msg_data_set_pages(struct ceph_msg
*msg, struct page **pages,
 {
 	BUG_ON(!pages);
 	BUG_ON(!length);
-	BUG_ON(msg->pages);
-	BUG_ON(msg->length);
+	BUG_ON(msg->p.pages);
+	BUG_ON(msg->p.length);

-	msg->pages = pages;
-	msg->length = length;
-	msg->page_alignment = alignment & ~PAGE_MASK;
+	msg->p.pages = pages;
+	msg->p.length = length;
+	msg->p.alignment = alignment & ~PAGE_MASK;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pages);

@@ -2755,18 +2755,18 @@ void ceph_msg_data_set_pagelist(struct ceph_msg
*msg,
 {
 	BUG_ON(!pagelist);
 	BUG_ON(!pagelist->length);
-	BUG_ON(msg->pagelist);
+	BUG_ON(msg->l.pagelist);

-	msg->pagelist = pagelist;
+	msg->l.pagelist = pagelist;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pagelist);

 void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
 {
 	BUG_ON(!bio);
-	BUG_ON(msg->bio);
+	BUG_ON(msg->b.bio);

-	msg->bio = bio;
+	msg->b.bio = bio;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_bio);

@@ -2774,9 +2774,9 @@ void ceph_msg_data_set_trail(struct ceph_msg *msg,
struct ceph_pagelist *trail)
 {
 	BUG_ON(!trail);
 	BUG_ON(!trail->length);
-	BUG_ON(msg->trail);
+	BUG_ON(msg->t.trail);

-	msg->trail = trail;
+	msg->t.trail = trail;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_trail);

@@ -2954,18 +2954,18 @@ void ceph_msg_last_put(struct kref *kref)
 		m->middle = NULL;
 	}
 	if (ceph_msg_has_pages(m)) {
-		m->length = 0;
-		m->pages = NULL;
+		m->p.length = 0;
+		m->p.pages = NULL;
 	}

 	if (ceph_msg_has_pagelist(m)) {
-		ceph_pagelist_release(m->pagelist);
-		kfree(m->pagelist);
-		m->pagelist = NULL;
+		ceph_pagelist_release(m->l.pagelist);
+		kfree(m->l.pagelist);
+		m->l.pagelist = NULL;
 	}

 	if (ceph_msg_has_trail(m))
-		m->trail = NULL;
+		m->t.trail = NULL;

 	if (m->pool)
 		ceph_msgpool_put(m->pool, m);
@@ -2977,7 +2977,7 @@ EXPORT_SYMBOL(ceph_msg_last_put);
 void ceph_msg_dump(struct ceph_msg *msg)
 {
 	pr_debug("msg_dump %p (front_max %d length %zd)\n", msg,
-		 msg->front_max, msg->length);
+		 msg->front_max, msg->p.length);
 	print_hex_dump(KERN_DEBUG, "header: ",
 		       DUMP_PREFIX_OFFSET, 16, 1,
 		       &msg->hdr, sizeof(msg->hdr), true);
-- 
1.7.9.5

[PATCH 4/8] libceph: start defining message data cursor

[Alex Elder]

This patch lays out the foundation for using generic routines to
manage processing items of message data.

For simplicity, we'll start with just the trail portion of a
message, because it stands alone and is only present for outgoing
data.

First some basic concepts.  We'll use the term "data item" to
represent one of the ceph_msg_data structures associated with a
message.  There are currently four of those, with single-letter
field names p, l, b, and t.  A data item is further broken into
"pieces" which always lie in a single page.  A data item will
include a "cursor" that will track state as the memory defined by
the item is consumed by sending data from or receiving data into it.

We define three routines to manipulate a data item's cursor: the
"init" routine; the "next" routine; and the "advance" routine.  The
"init" routine initializes the cursor so it points at the beginning
of the first piece in the item.  The "next" routine returns the
page, page offset, and length (limited by both the page and item
size) of the next unconsumed piece in the item.  It also indicates
to the caller whether the piece being returned is the last one in
the data item.

The "advance" routine consumes the requested number of bytes in the
item (advancing the cursor).  This is used to record the number of
bytes from the current piece that were actually sent or received by
the network code.  It returns an indication of whether the result
means the current piece has been fully consumed.  This is used by
the message send code to determine whether it should calculate the
CRC for the next piece processed.

The trail of a message is implemented as a ceph pagelist.  The
routines defined for it will be usable for non-trail pagelist data
as well.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |    7 ++
 net/ceph/messenger.c           |  138
+++++++++++++++++++++++++++++++++++++---
 2 files changed, 135 insertions(+), 10 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 5860dd0..1486243 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -95,6 +95,12 @@ static __inline__ bool ceph_msg_data_type_valid(enum
ceph_msg_data_type type)
 	}
 }

+struct ceph_msg_data_cursor {
+	bool		last_piece;	/* now at last piece of data item */
+	struct page	*page;		/* current page in pagelist */
+	size_t		offset;		/* pagelist bytes consumed */
+};
+
 struct ceph_msg_data {
 	enum ceph_msg_data_type		type;
 	union {
@@ -112,6 +118,7 @@ struct ceph_msg_data {
 		};
 		struct ceph_pagelist	*pagelist;
 	};
+	struct ceph_msg_data_cursor	cursor;		/* pagelist only */
 };

 /*
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index f256b4b..b978cf8 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -21,6 +21,9 @@
 #include <linux/ceph/pagelist.h>
 #include <linux/export.h>

+#define list_entry_next(pos, member)					\
+	list_entry(pos->member.next, typeof(*pos), member)
+
 /*
  * Ceph uses the messenger to exchange ceph_msg messages with other
  * hosts in the system.  The messenger provides ordered and reliable
@@ -738,6 +741,109 @@ static void iter_bio_next(struct bio **bio_iter,
unsigned int *seg)
 }
 #endif

+/*
+ * Message data is handled (sent or received) in pieces, where each
+ * piece resides on a single page.  The network layer might not
+ * consume an entire piece at once.  A data item's cursor keeps
+ * track of which piece is next to process and how much remains to
+ * be processed in that piece.  It also tracks whether the current
+ * piece is the last one in the data item.
+ */
+static void ceph_msg_data_cursor_init(struct ceph_msg_data *data)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct ceph_pagelist *pagelist;
+	struct page *page;
+
+	if (data->type != CEPH_MSG_DATA_PAGELIST)
+		return;
+
+	pagelist = data->pagelist;
+	BUG_ON(!pagelist);
+	if (!pagelist->length)
+		return;		/* pagelist can be assigned but empty */
+
+	BUG_ON(list_empty(&pagelist->head));
+	page = list_first_entry(&pagelist->head, struct page, lru);
+
+	cursor->page = page;
+	cursor->offset = 0;
+	cursor->last_piece = pagelist->length <= PAGE_SIZE;
+}
+
+/*
+ * Return the page containing the next piece to process for a given
+ * data item, and supply the page offset and length of that piece.
+ * Indicate whether this is the last piece in this data item.
+ */
+static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
+						size_t *page_offset,
+						size_t *length,
+						bool *last_piece)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct ceph_pagelist *pagelist;
+	size_t piece_end;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
+
+	pagelist = data->pagelist;
+	BUG_ON(!pagelist);
+
+	BUG_ON(!cursor->page);
+	BUG_ON(cursor->offset >= pagelist->length);
+
+	*last_piece = cursor->last_piece;
+	if (*last_piece) {
+		/* pagelist offset is always 0 */
+		piece_end = pagelist->length & ~PAGE_MASK;
+		if (!piece_end)
+			piece_end = PAGE_SIZE;
+	} else {
+		piece_end = PAGE_SIZE;
+	}
+	*page_offset = cursor->offset & ~PAGE_MASK;
+	*length = piece_end - *page_offset;
+
+	return data->cursor.page;
+}
+
+/*
+ * Returns true if the result moves the cursor on to the next piece
+ * (the next page) of the pagelist.
+ */
+static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct ceph_pagelist *pagelist;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
+
+	pagelist = data->pagelist;
+	BUG_ON(!pagelist);
+	BUG_ON(!cursor->page);
+	BUG_ON(cursor->offset + bytes > pagelist->length);
+	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);
+
+	/* Advance the cursor offset */
+
+	cursor->offset += bytes;
+	/* pagelist offset is always 0 */
+	if (!bytes || cursor->offset & ~PAGE_MASK)
+		return false;	/* more bytes to process in the current page */
+
+	/* Move on to the next page */
+
+	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
+	cursor->page = list_entry_next(cursor->page, lru);
+
+	/* cursor offset is at page boundary; pagelist offset is always 0 */
+	if (pagelist->length - cursor->offset <= PAGE_SIZE)
+		cursor->last_piece = true;
+
+	return true;
+}
+
 static void prepare_message_data(struct ceph_msg *msg,
 				struct ceph_msg_pos *msg_pos)
 {
@@ -755,6 +861,12 @@ static void prepare_message_data(struct ceph_msg *msg,
 		init_bio_iter(msg->b.bio, &msg->b.bio_iter, &msg->b.bio_seg);
 #endif
 	msg_pos->data_pos = 0;
+
+	/* If there's a trail, initialize its cursor */
+
+	if (ceph_msg_has_trail(msg))
+		ceph_msg_data_cursor_init(&msg->t);
+
 	msg_pos->did_page_crc = false;
 }

@@ -1045,6 +1157,12 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,

 	msg_pos->data_pos += sent;
 	msg_pos->page_pos += sent;
+	if (in_trail) {
+		bool need_crc;
+
+		need_crc = ceph_msg_data_advance(&msg->t, sent);
+		BUG_ON(need_crc && sent != len);
+	}
 	if (sent < len)
 		return;

@@ -1052,10 +1170,7 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 	msg_pos->did_page_crc = false;
-	if (in_trail) {
-		BUG_ON(!ceph_msg_has_trail(msg));
-		list_rotate_left(&msg->t.pagelist->head);
-	} else if (ceph_msg_has_pagelist(msg)) {
+	if (ceph_msg_has_pagelist(msg)) {
 		list_rotate_left(&msg->l.pagelist->head);
 #ifdef CONFIG_BLOCK
 	} else if (ceph_msg_has_bio(msg)) {
@@ -1141,6 +1256,8 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		size_t length;
 		int max_write = PAGE_SIZE;
 		int bio_offset = 0;
+		bool use_cursor = false;
+		bool last_piece = true;	/* preserve existing behavior */

 		in_trail = in_trail || msg_pos->data_pos >= trail_off;
 		if (!in_trail)
@@ -1148,9 +1265,9 @@ static int write_partial_message_data(struct
ceph_connection *con)

 		if (in_trail) {
 			BUG_ON(!ceph_msg_has_trail(msg));
-			total_max_write = data_len - msg_pos->data_pos;
-			page = list_first_entry(&msg->t.pagelist->head,
-						struct page, lru);
+			use_cursor = true;
+			page = ceph_msg_data_next(&msg->t, &page_offset,
+							&length, &last_piece);
 		} else if (ceph_msg_has_pages(msg)) {
 			page = msg->p.pages[msg_pos->page];
 		} else if (ceph_msg_has_pagelist(msg)) {
@@ -1168,8 +1285,9 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		} else {
 			page = zero_page;
 		}
-		length = min_t(int, max_write - msg_pos->page_pos,
-			    total_max_write);
+		if (!use_cursor)
+			length = min_t(int, max_write - msg_pos->page_pos,
+					    total_max_write);

 		page_offset = msg_pos->page_pos + bio_offset;
 		if (do_datacrc && !msg_pos->did_page_crc) {
@@ -1180,7 +1298,7 @@ static int write_partial_message_data(struct
ceph_connection *con)
 			msg_pos->did_page_crc = true;
 		}
 		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
-				      length, true);
+				      length, last_piece);
 		if (ret <= 0)
 			goto out;

-- 
1.7.9.5

[PATCH 5/8] libceph: prepare for other message data item types

[Alex Elder]

This just inserts some infrastructure in preparation for handling
other types of ceph message data items.  No functional changes,
just trying to simplify review by separating out some noise.

I will fold this into the previous patch before committing.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |    8 +++-
 net/ceph/messenger.c           |   99
++++++++++++++++++++++++++++++++--------
 2 files changed, 85 insertions(+), 22 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 1486243..716c3fd 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -97,8 +97,12 @@ static __inline__ bool ceph_msg_data_type_valid(enum
ceph_msg_data_type type)

 struct ceph_msg_data_cursor {
 	bool		last_piece;	/* now at last piece of data item */
-	struct page	*page;		/* current page in pagelist */
-	size_t		offset;		/* pagelist bytes consumed */
+	union {
+		struct {				/* pagelist */
+			struct page	*page;		/* page from list */
+			size_t		offset;		/* bytes from list */
+		};
+	};
 };

 struct ceph_msg_data {
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index b978cf8..ce81164 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -742,21 +742,16 @@ static void iter_bio_next(struct bio **bio_iter,
unsigned int *seg)
 #endif

 /*
- * Message data is handled (sent or received) in pieces, where each
- * piece resides on a single page.  The network layer might not
- * consume an entire piece at once.  A data item's cursor keeps
- * track of which piece is next to process and how much remains to
- * be processed in that piece.  It also tracks whether the current
- * piece is the last one in the data item.
+ * For a pagelist, a piece is whatever remains to be consumed in the
+ * first page in the list, or the front of the next page.
  */
-static void ceph_msg_data_cursor_init(struct ceph_msg_data *data)
+static void ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data *data)
 {
 	struct ceph_msg_data_cursor *cursor = &data->cursor;
 	struct ceph_pagelist *pagelist;
 	struct page *page;

-	if (data->type != CEPH_MSG_DATA_PAGELIST)
-		return;
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

 	pagelist = data->pagelist;
 	BUG_ON(!pagelist);
@@ -771,12 +766,7 @@ static void ceph_msg_data_cursor_init(struct
ceph_msg_data *data)
 	cursor->last_piece = pagelist->length <= PAGE_SIZE;
 }

-/*
- * Return the page containing the next piece to process for a given
- * data item, and supply the page offset and length of that piece.
- * Indicate whether this is the last piece in this data item.
- */
-static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
+static struct page *ceph_msg_data_pagelist_next(struct ceph_msg_data *data,
 						size_t *page_offset,
 						size_t *length,
 						bool *last_piece)
@@ -808,11 +798,8 @@ static struct page *ceph_msg_data_next(struct
ceph_msg_data *data,
 	return data->cursor.page;
 }

-/*
- * Returns true if the result moves the cursor on to the next piece
- * (the next page) of the pagelist.
- */
-static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
+static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data *data,
+						size_t bytes)
 {
 	struct ceph_msg_data_cursor *cursor = &data->cursor;
 	struct ceph_pagelist *pagelist;
@@ -844,6 +831,78 @@ static bool ceph_msg_data_advance(struct
ceph_msg_data *data, size_t bytes)
 	return true;
 }

+/*
+ * Message data is handled (sent or received) in pieces, where each
+ * piece resides on a single page.  The network layer might not
+ * consume an entire piece at once.  A data item's cursor keeps
+ * track of which piece is next to process and how much remains to
+ * be processed in that piece.  It also tracks whether the current
+ * piece is the last one in the data item.
+ */
+static void ceph_msg_data_cursor_init(struct ceph_msg_data *data)
+{
+	switch (data->type) {
+	case CEPH_MSG_DATA_NONE:
+	case CEPH_MSG_DATA_PAGES:
+		break;
+	case CEPH_MSG_DATA_PAGELIST:
+		ceph_msg_data_pagelist_cursor_init(data);
+		break;
+#ifdef CONFIG_BLOCK
+	case CEPH_MSG_DATA_BIO:
+#endif /* CONFIG_BLOCK */
+	default:
+	    break;
+	}
+}
+
+/*
+ * Return the page containing the next piece to process for a given
+ * data item, and supply the page offset and length of that piece.
+ * Indicate whether this is the last piece in this data item.
+ */
+static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
+					size_t *page_offset,
+					size_t *length,
+					bool *last_piece)
+{
+	switch (data->type) {
+	case CEPH_MSG_DATA_NONE:
+	case CEPH_MSG_DATA_PAGES:
+		break;
+	case CEPH_MSG_DATA_PAGELIST:
+		return ceph_msg_data_pagelist_next(data, page_offset, length,
+						last_piece);
+#ifdef CONFIG_BLOCK
+	case CEPH_MSG_DATA_BIO:
+		break;
+#endif /* CONFIG_BLOCK */
+	}
+	BUG();
+	return NULL;
+}
+
+/*
+ * Returns true if the result moves the cursor on to the next piece
+ * of the data item.
+ */
+static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
+{
+	switch (data->type) {
+	case CEPH_MSG_DATA_NONE:
+	case CEPH_MSG_DATA_PAGES:
+		break;
+	case CEPH_MSG_DATA_PAGELIST:
+		return ceph_msg_data_pagelist_advance(data, bytes);
+#ifdef CONFIG_BLOCK
+	case CEPH_MSG_DATA_BIO:
+		break;
+#endif /* CONFIG_BLOCK */
+	}
+	BUG();
+	return false;
+}
+
 static void prepare_message_data(struct ceph_msg *msg,
 				struct ceph_msg_pos *msg_pos)
 {
-- 
1.7.9.5

[PATCH 6/8] libceph: use data cursor for message pagelist

[Alex Elder]

Switch to using the message cursor for the (non-trail) outgoing
pagelist data item in a message if present.

Notes on the logic changes in out_msg_pos_next():
    - only the mds client uses a ceph pagelist for message data;
    - if the mds client ever uses a pagelist, it never uses a page
      array (or anything else, for that matter) for data in the same
      message;
    - only the osd client uses the trail portion of a message data,
      and when it does, it never uses any other data fields for
      outgoing data in the same message; and finally
    - only the rbd client uses bio message data (never pagelist).

Therefore out_msg_pos_next() can assume:
    - if we're in the trail portion of a message, the message data
      pagelist, data, and bio can be ignored; and
    - if there is a page list, there will never be any a bio or page
      array data, and vice-versa.

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

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index ce81164..c0f89c1 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -921,8 +921,10 @@ static void prepare_message_data(struct ceph_msg *msg,
 #endif
 	msg_pos->data_pos = 0;

-	/* If there's a trail, initialize its cursor */
+	/* Initialize data cursors */

+	if (ceph_msg_has_pagelist(msg))
+		ceph_msg_data_cursor_init(&msg->l);
 	if (ceph_msg_has_trail(msg))
 		ceph_msg_data_cursor_init(&msg->t);

@@ -1210,18 +1212,19 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 {
 	struct ceph_msg *msg = con->out_msg;
 	struct ceph_msg_pos *msg_pos = &con->out_msg_pos;
+	bool need_crc = false;

 	BUG_ON(!msg);
 	BUG_ON(!sent);

 	msg_pos->data_pos += sent;
 	msg_pos->page_pos += sent;
-	if (in_trail) {
-		bool need_crc;
-
+	if (in_trail)
 		need_crc = ceph_msg_data_advance(&msg->t, sent);
-		BUG_ON(need_crc && sent != len);
-	}
+	else if (ceph_msg_has_pagelist(msg))
+		need_crc = ceph_msg_data_advance(&msg->l, sent);
+	BUG_ON(need_crc && sent != len);
+
 	if (sent < len)
 		return;

@@ -1229,13 +1232,10 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 	msg_pos->did_page_crc = false;
-	if (ceph_msg_has_pagelist(msg)) {
-		list_rotate_left(&msg->l.pagelist->head);
 #ifdef CONFIG_BLOCK
-	} else if (ceph_msg_has_bio(msg)) {
+	if (ceph_msg_has_bio(msg))
 		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
 #endif
-	}
 }

 static void in_msg_pos_next(struct ceph_connection *con, size_t len,
@@ -1330,8 +1330,9 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		} else if (ceph_msg_has_pages(msg)) {
 			page = msg->p.pages[msg_pos->page];
 		} else if (ceph_msg_has_pagelist(msg)) {
-			page = list_first_entry(&msg->l.pagelist->head,
-						struct page, lru);
+			use_cursor = true;
+			page = ceph_msg_data_next(&msg->l, &page_offset,
+							&length, &last_piece);
 #ifdef CONFIG_BLOCK
 		} else if (ceph_msg_has_bio(msg)) {
 			struct bio_vec *bv;
-- 
1.7.9.5

[PATCH 7/8] libceph: implement bio message data item cursor

[Alex Elder]

Implement and use cursor routines for bio message data items for
outbound message data.

(See the previous commit for reasoning in support of the changes
in out_msg_pos_next().)

Signed-off-by: Alex Elder <elder@inktank.com>
---
 include/linux/ceph/messenger.h |    7 +++
 net/ceph/messenger.c           |  128
++++++++++++++++++++++++++++++++++------
 2 files changed, 118 insertions(+), 17 deletions(-)

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 716c3fd..76b4645 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -98,6 +98,13 @@ static __inline__ bool ceph_msg_data_type_valid(enum
ceph_msg_data_type type)
 struct ceph_msg_data_cursor {
 	bool		last_piece;	/* now at last piece of data item */
 	union {
+#ifdef CONFIG_BLOCK
+		struct {				/* bio */
+			struct bio	*bio;		/* bio from list */
+			unsigned int	vector_index;	/* vector from bio */
+			unsigned int	vector_offset;	/* bytes from vector */
+		};
+#endif /* CONFIG_BLOCK */
 		struct {				/* pagelist */
 			struct page	*page;		/* page from list */
 			size_t		offset;		/* bytes from list */
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index c0f89c1..5ddbe5d 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -739,6 +739,97 @@ static void iter_bio_next(struct bio **bio_iter,
unsigned int *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
+ * entry in the current bio iovec, or the first entry in the next
+ * bio in the list.
+ */
+static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data *data)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct bio *bio;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_BIO);
+
+	bio = data->bio;
+	BUG_ON(!bio);
+	BUG_ON(!bio->bi_vcnt);
+	/* resid = bio->bi_size */
+
+	cursor->bio = bio;
+	cursor->vector_index = 0;
+	cursor->vector_offset = 0;
+	cursor->last_piece = !bio->bi_next && bio->bi_vcnt == 1;
+}
+
+static struct page *ceph_msg_data_bio_next(struct ceph_msg_data *data,
+						size_t *page_offset,
+						size_t *length,
+						bool *last_piece)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct bio *bio;
+	struct bio_vec *bio_vec;
+	unsigned int index;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_BIO);
+
+	bio = cursor->bio;
+	BUG_ON(!bio);
+
+	index = cursor->vector_index;
+	BUG_ON(index >= (unsigned int) bio->bi_vcnt);
+
+	bio_vec = &bio->bi_io_vec[index];
+	BUG_ON(cursor->vector_offset >= bio_vec->bv_len);
+	*page_offset = (size_t) (bio_vec->bv_offset + cursor->vector_offset);
+	BUG_ON(*page_offset >= PAGE_SIZE);
+	*length = (size_t) (bio_vec->bv_len - cursor->vector_offset);
+	BUG_ON(*length > PAGE_SIZE);
+	*last_piece = cursor->last_piece;
+
+	return bio_vec->bv_page;
+}
+
+static bool ceph_msg_data_bio_advance(struct ceph_msg_data *data,
size_t bytes)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	struct bio *bio;
+	struct bio_vec *bio_vec;
+	unsigned int index;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_BIO);
+
+	bio = cursor->bio;
+	BUG_ON(!bio);
+
+	index = cursor->vector_index;
+	BUG_ON(index >= (unsigned int) bio->bi_vcnt);
+	bio_vec = &bio->bi_io_vec[index];
+	BUG_ON(cursor->vector_offset + bytes > bio_vec->bv_len);
+
+	/* Advance the cursor offset */
+
+	cursor->vector_offset += bytes;
+	if (cursor->vector_offset < bio_vec->bv_len)
+		return false;	/* more bytes to process in this segment */
+
+	/* Move on to the next segment, and possibly the next bio */
+
+	if (++cursor->vector_index == (unsigned int) bio->bi_vcnt) {
+		bio = bio->bi_next;
+		cursor->bio = bio;
+		cursor->vector_index = 0;
+	}
+	cursor->vector_offset = 0;
+
+	if (!cursor->last_piece && bio && !bio->bi_next)
+		if (cursor->vector_index == (unsigned int) bio->bi_vcnt - 1)
+			cursor->last_piece = true;
+
+	return true;
+}
 #endif

 /*
@@ -850,6 +941,8 @@ static void ceph_msg_data_cursor_init(struct
ceph_msg_data *data)
 		break;
 #ifdef CONFIG_BLOCK
 	case CEPH_MSG_DATA_BIO:
+		ceph_msg_data_bio_cursor_init(data);
+		break;
 #endif /* CONFIG_BLOCK */
 	default:
 	    break;
@@ -875,7 +968,8 @@ static struct page *ceph_msg_data_next(struct
ceph_msg_data *data,
 						last_piece);
 #ifdef CONFIG_BLOCK
 	case CEPH_MSG_DATA_BIO:
-		break;
+		return ceph_msg_data_bio_next(data, page_offset, length,
+						last_piece);
 #endif /* CONFIG_BLOCK */
 	}
 	BUG();
@@ -896,7 +990,7 @@ static bool ceph_msg_data_advance(struct
ceph_msg_data *data, size_t bytes)
 		return ceph_msg_data_pagelist_advance(data, bytes);
 #ifdef CONFIG_BLOCK
 	case CEPH_MSG_DATA_BIO:
-		break;
+		return ceph_msg_data_bio_advance(data, bytes);
 #endif /* CONFIG_BLOCK */
 	}
 	BUG();
@@ -923,6 +1017,10 @@ static void prepare_message_data(struct ceph_msg *msg,

 	/* Initialize data cursors */

+#ifdef CONFIG_BLOCK
+	if (ceph_msg_has_bio(msg))
+		ceph_msg_data_cursor_init(&msg->b);
+#endif /* CONFIG_BLOCK */
 	if (ceph_msg_has_pagelist(msg))
 		ceph_msg_data_cursor_init(&msg->l);
 	if (ceph_msg_has_trail(msg))
@@ -1223,6 +1321,10 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 		need_crc = ceph_msg_data_advance(&msg->t, sent);
 	else if (ceph_msg_has_pagelist(msg))
 		need_crc = ceph_msg_data_advance(&msg->l, sent);
+#ifdef CONFIG_BLOCK
+	else if (ceph_msg_has_bio(msg))
+		need_crc = ceph_msg_data_advance(&msg->b, sent);
+#endif /* CONFIG_BLOCK */
 	BUG_ON(need_crc && sent != len);

 	if (sent < len)
@@ -1232,10 +1334,6 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->page_pos = 0;
 	msg_pos->page++;
 	msg_pos->did_page_crc = false;
-#ifdef CONFIG_BLOCK
-	if (ceph_msg_has_bio(msg))
-		iter_bio_next(&msg->b.bio_iter, &msg->b.bio_seg);
-#endif
 }

 static void in_msg_pos_next(struct ceph_connection *con, size_t len,
@@ -1313,8 +1411,6 @@ static int write_partial_message_data(struct
ceph_connection *con)
 		struct page *page = NULL;
 		size_t page_offset;
 		size_t length;
-		int max_write = PAGE_SIZE;
-		int bio_offset = 0;
 		bool use_cursor = false;
 		bool last_piece = true;	/* preserve existing behavior */

@@ -1335,21 +1431,19 @@ static int write_partial_message_data(struct
ceph_connection *con)
 							&length, &last_piece);
 #ifdef CONFIG_BLOCK
 		} else if (ceph_msg_has_bio(msg)) {
-			struct bio_vec *bv;
-
-			bv = bio_iovec_idx(msg->b.bio_iter, msg->b.bio_seg);
-			page = bv->bv_page;
-			bio_offset = bv->bv_offset;
-			max_write = bv->bv_len;
+			use_cursor = true;
+			page = ceph_msg_data_next(&msg->b, &page_offset,
+							&length, &last_piece);
 #endif
 		} else {
 			page = zero_page;
 		}
-		if (!use_cursor)
-			length = min_t(int, max_write - msg_pos->page_pos,
+		if (!use_cursor) {
+			length = min_t(int, PAGE_SIZE - msg_pos->page_pos,
 					    total_max_write);

-		page_offset = msg_pos->page_pos + bio_offset;
+			page_offset = msg_pos->page_pos;
+		}
 		if (do_datacrc && !msg_pos->did_page_crc) {
 			u32 crc = le32_to_cpu(msg->footer.data_crc);

-- 
1.7.9.5

[PATCH 8/8] libceph: implement pages array cursor

[Alex Elder]

Implement and use cursor routines for page array message data items
for outbound message data.

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

diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
index 76b4645..b53b9ef 100644
--- a/include/linux/ceph/messenger.h
+++ b/include/linux/ceph/messenger.h
@@ -105,6 +105,12 @@ struct ceph_msg_data_cursor {
 			unsigned int	vector_offset;	/* bytes from vector */
 		};
 #endif /* CONFIG_BLOCK */
+		struct {				/* pages */
+			size_t		resid;		/* bytes from array */
+			unsigned int	page_offset;	/* offset in page */
+			unsigned short	page_index;	/* index in array */
+			unsigned short	page_count;	/* pages in array */
+		};
 		struct {				/* pagelist */
 			struct page	*page;		/* page from list */
 			size_t		offset;		/* bytes from list */
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 5ddbe5d..12d416f 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -833,6 +833,81 @@ static bool ceph_msg_data_bio_advance(struct
ceph_msg_data *data, size_t bytes)
 #endif

 /*
+ * For a page array, a piece comes from the first page in the array
+ * that has not already been fully consumed.
+ */
+static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data *data)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+	int page_count;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
+
+	BUG_ON(!data->pages);
+	BUG_ON(!data->length);
+
+	page_count = calc_pages_for(data->alignment, (u64)data->length);
+	BUG_ON(page_count > (int) USHRT_MAX);
+	cursor->resid = data->length;
+	cursor->page_offset = data->alignment & ~PAGE_MASK;
+	cursor->page_index = 0;
+	cursor->page_count = (unsigned short) page_count;
+	cursor->last_piece = cursor->page_count == 1;
+}
+
+static struct page *ceph_msg_data_pages_next(struct ceph_msg_data *data,
+					size_t *page_offset,
+					size_t *length,
+					bool *last_piece)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
+
+	BUG_ON(cursor->page_index >= cursor->page_count);
+	BUG_ON(cursor->page_offset >= PAGE_SIZE);
+	BUG_ON(!cursor->resid);
+
+	*page_offset = cursor->page_offset;
+	*last_piece = cursor->last_piece;
+	if (*last_piece) {
+		BUG_ON(*page_offset + cursor->resid > PAGE_SIZE);
+		*length = cursor->resid;
+	} else {
+		*length = PAGE_SIZE - *page_offset;
+	}
+
+	return data->pages[cursor->page_index];
+}
+
+static bool ceph_msg_data_pages_advance(struct ceph_msg_data *data,
+						size_t bytes)
+{
+	struct ceph_msg_data_cursor *cursor = &data->cursor;
+
+	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
+
+	BUG_ON(cursor->page_offset + bytes > PAGE_SIZE);
+	BUG_ON(bytes > cursor->resid);
+
+	/* Advance the cursor page offset */
+
+	cursor->resid -= bytes;
+	cursor->page_offset += bytes;
+	if (!bytes || cursor->page_offset & ~PAGE_MASK)
+		return false;	/* more bytes to process in the current page */
+
+	/* Move on to the next page */
+
+	BUG_ON(cursor->page_index >= cursor->page_count);
+	cursor->page_offset = 0;
+	cursor->page_index++;
+	cursor->last_piece = cursor->page_index == cursor->page_count - 1;
+
+	return true;
+}
+
+/*
  * For a pagelist, a piece is whatever remains to be consumed in the
  * first page in the list, or the front of the next page.
  */
@@ -934,7 +1009,9 @@ static void ceph_msg_data_cursor_init(struct
ceph_msg_data *data)
 {
 	switch (data->type) {
 	case CEPH_MSG_DATA_NONE:
+		break;
 	case CEPH_MSG_DATA_PAGES:
+		ceph_msg_data_pages_cursor_init(data);
 		break;
 	case CEPH_MSG_DATA_PAGELIST:
 		ceph_msg_data_pagelist_cursor_init(data);
@@ -961,8 +1038,10 @@ static struct page *ceph_msg_data_next(struct
ceph_msg_data *data,
 {
 	switch (data->type) {
 	case CEPH_MSG_DATA_NONE:
-	case CEPH_MSG_DATA_PAGES:
 		break;
+	case CEPH_MSG_DATA_PAGES:
+		return ceph_msg_data_pages_next(data, page_offset, length,
+						last_piece);
 	case CEPH_MSG_DATA_PAGELIST:
 		return ceph_msg_data_pagelist_next(data, page_offset, length,
 						last_piece);
@@ -984,8 +1063,9 @@ static bool ceph_msg_data_advance(struct
ceph_msg_data *data, size_t bytes)
 {
 	switch (data->type) {
 	case CEPH_MSG_DATA_NONE:
-	case CEPH_MSG_DATA_PAGES:
 		break;
+	case CEPH_MSG_DATA_PAGES:
+		return ceph_msg_data_pages_advance(data, bytes);
 	case CEPH_MSG_DATA_PAGELIST:
 		return ceph_msg_data_pagelist_advance(data, bytes);
 #ifdef CONFIG_BLOCK
@@ -1021,6 +1101,8 @@ static void prepare_message_data(struct ceph_msg *msg,
 	if (ceph_msg_has_bio(msg))
 		ceph_msg_data_cursor_init(&msg->b);
 #endif /* CONFIG_BLOCK */
+	if (ceph_msg_has_pages(msg))
+		ceph_msg_data_cursor_init(&msg->p);
 	if (ceph_msg_has_pagelist(msg))
 		ceph_msg_data_cursor_init(&msg->l);
 	if (ceph_msg_has_trail(msg))
@@ -1319,6 +1401,8 @@ static void out_msg_pos_next(struct
ceph_connection *con, struct page *page,
 	msg_pos->page_pos += sent;
 	if (in_trail)
 		need_crc = ceph_msg_data_advance(&msg->t, sent);
+	else if (ceph_msg_has_pages(msg))
+		need_crc = ceph_msg_data_advance(&msg->p, sent);
 	else if (ceph_msg_has_pagelist(msg))
 		need_crc = ceph_msg_data_advance(&msg->l, sent);
 #ifdef CONFIG_BLOCK
@@ -1424,7 +1508,9 @@ static int write_partial_message_data(struct
ceph_connection *con)
 			page = ceph_msg_data_next(&msg->t, &page_offset,
 							&length, &last_piece);
 		} else if (ceph_msg_has_pages(msg)) {
-			page = msg->p.pages[msg_pos->page];
+			use_cursor = true;
+			page = ceph_msg_data_next(&msg->p, &page_offset,
+							&length, &last_piece);
 		} else if (ceph_msg_has_pagelist(msg)) {
 			use_cursor = true;
 			page = ceph_msg_data_next(&msg->l, &page_offset,
-- 
1.7.9.5

Re: [PATCH 0/8] libceph: implement cursor for outgoing data items

[Alex Elder]

On 03/10/2013 02:06 PM, Alex Elder wrote:
> This series makes the four distinct data items in a message (the
> pages array, the pagelist, the bio, and the trail) take a generic
> form, and then arranges for them all to be handled using a common
> set of cursor routines that determine what data needs to be sent
> next for a message.  There is more consolidation of this code
> coming, but for now I'm leaving this series at 8 patches, which
> includes the definition of cursor code for the three types of
> data item.
> 
> 					-Alex

These patches are available in branch "review/wip-cursor" on the
ceph-client git repository.  That branch is based on the branch
"review/wip-msgr-refactor".

Someday I will remember to say that every time I send out one
of these things...

					-Alex

> [PATCH 1/8] libceph: define ceph_msg_has_*() data macros
> [PATCH 2/8] libceph: be explicit about message data representation
> [PATCH 3/8] libceph: abstract message data
> [PATCH 4/8] libceph: start defining message data cursor
> [PATCH 5/8] libceph: prepare for other message data item types
> [PATCH 6/8] libceph: use data cursor for message pagelist
> [PATCH 7/8] libceph: implement bio message data item cursor
> [PATCH 8/8] libceph: implement pages array cursor
> 

Re: [PATCH 0/8] libceph: implement cursor for outgoing data items

[Josh Durgin]

On 03/10/2013 12:06 PM, Alex Elder wrote:
> This series makes the four distinct data items in a message (the
> pages array, the pagelist, the bio, and the trail) take a generic
> form, and then arranges for them all to be handled using a common
> set of cursor routines that determine what data needs to be sent
> next for a message.  There is more consolidation of this code
> coming, but for now I'm leaving this series at 8 patches, which
> includes the definition of cursor code for the three types of
> data item.
>
> 					-Alex
>
> [PATCH 1/8] libceph: define ceph_msg_has_*() data macros
> [PATCH 2/8] libceph: be explicit about message data representation
> [PATCH 3/8] libceph: abstract message data
> [PATCH 4/8] libceph: start defining message data cursor
> [PATCH 5/8] libceph: prepare for other message data item types
> [PATCH 6/8] libceph: use data cursor for message pagelist
> [PATCH 7/8] libceph: implement bio message data item cursor
> [PATCH 8/8] libceph: implement pages array cursor

These look good. I assume you're going to remove the single character
struct members in a later series.

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

Re: [PATCH 0/8] libceph: implement cursor for outgoing data items

[Alex Elder]

On 03/11/2013 05:24 PM, Josh Durgin wrote:
> On 03/10/2013 12:06 PM, Alex Elder wrote:
>> This series makes the four distinct data items in a message (the
>> pages array, the pagelist, the bio, and the trail) take a generic
>> form, and then arranges for them all to be handled using a common
>> set of cursor routines that determine what data needs to be sent
>> next for a message.  There is more consolidation of this code
>> coming, but for now I'm leaving this series at 8 patches, which
>> includes the definition of cursor code for the three types of
>> data item.
>>
>>                     -Alex
>>
>> [PATCH 1/8] libceph: define ceph_msg_has_*() data macros
>> [PATCH 2/8] libceph: be explicit about message data representation
>> [PATCH 3/8] libceph: abstract message data
>> [PATCH 4/8] libceph: start defining message data cursor
>> [PATCH 5/8] libceph: prepare for other message data item types
>> [PATCH 6/8] libceph: use data cursor for message pagelist
>> [PATCH 7/8] libceph: implement bio message data item cursor
>> [PATCH 8/8] libceph: implement pages array cursor
> 
> These look good. I assume you're going to remove the single character
> struct members in a later series.

Yes, I did the single characters for brevity, knowing
they'd be going away.

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