Commit ca8b3a69 authored by Alex Elder's avatar Alex Elder Committed by Sage Weil

libceph: implement multiple data items in a message

This patch adds support to the messenger for more than one data item
in its data list.

A message data cursor has two more fields to support this:
    - a count of the number of bytes left to be consumed across
      all data items in the list, "total_resid"
    - a pointer to the head of the list (for validation only)

The cursor initialization routine has been split into two parts: the
outer one, which initializes the cursor for traversing the entire
list of data items; and the inner one, which initializes the cursor
to start processing a single data item.

When a message cursor is first initialized, the outer initialization
routine sets total_resid to the length provided.  The data pointer
is initialized to the first data item on the list.  From there, the
inner initialization routine finishes by setting up to process the
data item the cursor points to.

Advancing the cursor consumes bytes in total_resid.  If the resid
field reaches zero, it means the current data item is fully
consumed.  If total_resid indicates there is more data, the cursor
is advanced to point to the next data item, and then the inner
initialization routine prepares for using that.  (A check is made at
this point to make sure we don't wrap around the front of the list.)

The type-specific init routines are modified so they can be given a
length that's larger than what the data item can support.  The resid
field is initialized to the smaller of the provided length and the
length of the entire data item.

When total_resid reaches zero, we're done.

This resolves:
    http://tracker.ceph.com/issues/3761Signed-off-by: default avatarAlex Elder <elder@inktank.com>
Reviewed-by: default avatarJosh Durgin <josh.durgin@inktank.com>
parent 5240d9f9
...@@ -108,7 +108,10 @@ struct ceph_msg_data { ...@@ -108,7 +108,10 @@ struct ceph_msg_data {
}; };
struct ceph_msg_data_cursor { struct ceph_msg_data_cursor {
struct ceph_msg_data *data; /* data item this describes */ size_t total_resid; /* across all data items */
struct list_head *data_head; /* = &ceph_msg->data */
struct ceph_msg_data *data; /* current data item */
size_t resid; /* bytes not yet consumed */ size_t resid; /* bytes not yet consumed */
bool last_piece; /* current is last piece */ bool last_piece; /* current is last piece */
bool need_crc; /* crc update needed */ bool need_crc; /* crc update needed */
......
...@@ -734,7 +734,7 @@ static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor, ...@@ -734,7 +734,7 @@ static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
BUG_ON(!bio); BUG_ON(!bio);
BUG_ON(!bio->bi_vcnt); BUG_ON(!bio->bi_vcnt);
cursor->resid = length; cursor->resid = min(length, data->bio_length);
cursor->bio = bio; cursor->bio = bio;
cursor->vector_index = 0; cursor->vector_index = 0;
cursor->vector_offset = 0; cursor->vector_offset = 0;
...@@ -833,9 +833,8 @@ static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor, ...@@ -833,9 +833,8 @@ static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor,
BUG_ON(!data->pages); BUG_ON(!data->pages);
BUG_ON(!data->length); BUG_ON(!data->length);
BUG_ON(length > data->length); /* short reads are OK */
cursor->resid = length; cursor->resid = min(length, data->length);
page_count = calc_pages_for(data->alignment, (u64)data->length); page_count = calc_pages_for(data->alignment, (u64)data->length);
cursor->page_offset = data->alignment & ~PAGE_MASK; cursor->page_offset = data->alignment & ~PAGE_MASK;
cursor->page_index = 0; cursor->page_index = 0;
...@@ -904,7 +903,6 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor, ...@@ -904,7 +903,6 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
pagelist = data->pagelist; pagelist = data->pagelist;
BUG_ON(!pagelist); BUG_ON(!pagelist);
BUG_ON(length > pagelist->length); /* short reads are OK */
if (!length) if (!length)
return; /* pagelist can be assigned but empty */ return; /* pagelist can be assigned but empty */
...@@ -912,7 +910,7 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor, ...@@ -912,7 +910,7 @@ ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
BUG_ON(list_empty(&pagelist->head)); BUG_ON(list_empty(&pagelist->head));
page = list_first_entry(&pagelist->head, struct page, lru); page = list_first_entry(&pagelist->head, struct page, lru);
cursor->resid = length; cursor->resid = min(length, pagelist->length);
cursor->page = page; cursor->page = page;
cursor->offset = 0; cursor->offset = 0;
cursor->last_piece = length <= PAGE_SIZE; cursor->last_piece = length <= PAGE_SIZE;
...@@ -982,13 +980,10 @@ static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor, ...@@ -982,13 +980,10 @@ static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor,
* be processed in that piece. It also tracks whether the current * be processed in that piece. It also tracks whether the current
* piece is the last one in the data item. * piece is the last one in the data item.
*/ */
static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length) static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
{ {
struct ceph_msg_data_cursor *cursor = &msg->cursor; size_t length = cursor->total_resid;
struct ceph_msg_data *data;
data = list_first_entry(&msg->data, struct ceph_msg_data, links);
cursor->data = data;
switch (cursor->data->type) { switch (cursor->data->type) {
case CEPH_MSG_DATA_PAGELIST: case CEPH_MSG_DATA_PAGELIST:
ceph_msg_data_pagelist_cursor_init(cursor, length); ceph_msg_data_pagelist_cursor_init(cursor, length);
...@@ -1009,6 +1004,25 @@ static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length) ...@@ -1009,6 +1004,25 @@ static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
cursor->need_crc = true; cursor->need_crc = true;
} }
static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
{
struct ceph_msg_data_cursor *cursor = &msg->cursor;
struct ceph_msg_data *data;
BUG_ON(!length);
BUG_ON(length > msg->data_length);
BUG_ON(list_empty(&msg->data));
data = list_first_entry(&msg->data, struct ceph_msg_data, links);
cursor->data_head = &msg->data;
cursor->total_resid = length;
data = list_first_entry(&msg->data, struct ceph_msg_data, links);
cursor->data = data;
__ceph_msg_data_cursor_init(cursor);
}
/* /*
* Return the page containing the next piece to process for a given * Return the page containing the next piece to process for a given
* data item, and supply the page offset and length of that piece. * data item, and supply the page offset and length of that piece.
...@@ -1073,8 +1087,16 @@ static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor, ...@@ -1073,8 +1087,16 @@ static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor,
BUG(); BUG();
break; break;
} }
cursor->total_resid -= bytes;
cursor->need_crc = new_piece; cursor->need_crc = new_piece;
if (!cursor->resid && cursor->total_resid) {
WARN_ON(!cursor->last_piece);
BUG_ON(list_is_last(&cursor->data->links, cursor->data_head));
cursor->data = list_entry_next(cursor->data, links);
__ceph_msg_data_cursor_init(cursor);
}
return new_piece; return new_piece;
} }
...@@ -2990,8 +3012,6 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages, ...@@ -2990,8 +3012,6 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
BUG_ON(!pages); BUG_ON(!pages);
BUG_ON(!length); BUG_ON(!length);
BUG_ON(msg->data_length);
BUG_ON(!list_empty(&msg->data));
data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES); data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
BUG_ON(!data); BUG_ON(!data);
...@@ -3012,8 +3032,6 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg, ...@@ -3012,8 +3032,6 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
BUG_ON(!pagelist); BUG_ON(!pagelist);
BUG_ON(!pagelist->length); BUG_ON(!pagelist->length);
BUG_ON(msg->data_length);
BUG_ON(!list_empty(&msg->data));
data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST); data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
BUG_ON(!data); BUG_ON(!data);
...@@ -3031,8 +3049,6 @@ void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio, ...@@ -3031,8 +3049,6 @@ void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio,
struct ceph_msg_data *data; struct ceph_msg_data *data;
BUG_ON(!bio); BUG_ON(!bio);
BUG_ON(msg->data_length);
BUG_ON(!list_empty(&msg->data));
data = ceph_msg_data_create(CEPH_MSG_DATA_BIO); data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
BUG_ON(!data); BUG_ON(!data);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment