Skip to content

Commit

Permalink
libceph: move cursor into message
Browse files Browse the repository at this point in the history 
A message will only be processing a single data item at a time, so
there's no need for each data item to have its own cursor.

Move the cursor embedded in the message data structure into the
message itself.  To minimize the impact, keep the data->cursor
field, but make it be a pointer to the cursor in the message.

Move the definition of ceph_msg_data above ceph_msg_data_cursor so
the cursor can point to the data without a forward definition rather
than vice-versa.

This and the upcoming patches are part of:
    http://tracker.ceph.com/issues/3761

Signed-off-by: Alex Elder <elder@inktank.com>
Reviewed-by: Josh Durgin <josh.durgin@inktank.com>
  • Loading branch information
Alex Elder authored and Sage Weil committed May 2, 2013
1 parent c851c49 commit 36153ec
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 37 deletions.
43 changes: 22 additions & 21 deletions include/linux/ceph/messenger.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,25 @@ static __inline__ bool ceph_msg_data_type_valid(enum ceph_msg_data_type type)
}
}

struct ceph_msg_data {
enum ceph_msg_data_type type;
union {
#ifdef CONFIG_BLOCK
struct {
struct bio *bio;
size_t bio_length;
};
#endif /* CONFIG_BLOCK */
struct {
struct page **pages; /* NOT OWNER. */
size_t length; /* total # bytes */
unsigned int alignment; /* first page */
};
struct ceph_pagelist *pagelist;
};
struct ceph_msg_data_cursor *cursor;
};

struct ceph_msg_data_cursor {
size_t resid; /* bytes not yet consumed */
bool last_piece; /* now at last piece of data item */
Expand All @@ -112,25 +131,6 @@ struct ceph_msg_data_cursor {
};
};

struct ceph_msg_data {
enum ceph_msg_data_type type;
union {
#ifdef CONFIG_BLOCK
struct {
struct bio *bio;
size_t bio_length;
};
#endif /* CONFIG_BLOCK */
struct {
struct page **pages; /* NOT OWNER. */
size_t length; /* total # bytes */
unsigned int alignment; /* first page */
};
struct ceph_pagelist *pagelist;
};
struct ceph_msg_data_cursor cursor; /* pagelist only */
};

/*
* a single message. it contains a header (src, dest, message type, etc.),
* footer (crc values, mainly), a "front" message body, and possibly a
Expand All @@ -142,8 +142,9 @@ struct ceph_msg {
struct kvec front; /* unaligned blobs of message */
struct ceph_buffer *middle;

size_t data_length;
struct ceph_msg_data *data; /* data payload */
size_t data_length;
struct ceph_msg_data *data;
struct ceph_msg_data_cursor cursor;

struct ceph_connection *con;
struct list_head list_head; /* links for connection lists */
Expand Down
35 changes: 19 additions & 16 deletions net/ceph/messenger.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -725,7 +725,7 @@ static void con_out_kvec_add(struct ceph_connection *con,
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data *data,
size_t length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
struct bio *bio;

BUG_ON(data->type != CEPH_MSG_DATA_BIO);
Expand All @@ -745,7 +745,7 @@ static struct page *ceph_msg_data_bio_next(struct ceph_msg_data *data,
size_t *page_offset,
size_t *length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
struct bio *bio;
struct bio_vec *bio_vec;
unsigned int index;
Expand Down Expand Up @@ -774,7 +774,7 @@ static struct page *ceph_msg_data_bio_next(struct ceph_msg_data *data,

static bool ceph_msg_data_bio_advance(struct ceph_msg_data *data, size_t bytes)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
struct bio *bio;
struct bio_vec *bio_vec;
unsigned int index;
Expand Down Expand Up @@ -826,7 +826,7 @@ static bool ceph_msg_data_bio_advance(struct ceph_msg_data *data, size_t bytes)
static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data *data,
size_t length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
int page_count;

BUG_ON(data->type != CEPH_MSG_DATA_PAGES);
Expand All @@ -849,7 +849,7 @@ static struct page *ceph_msg_data_pages_next(struct ceph_msg_data *data,
size_t *page_offset,
size_t *length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;

BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

Expand All @@ -868,7 +868,7 @@ static struct page *ceph_msg_data_pages_next(struct ceph_msg_data *data,
static bool ceph_msg_data_pages_advance(struct ceph_msg_data *data,
size_t bytes)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;

BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

Expand Down Expand Up @@ -897,7 +897,7 @@ static bool ceph_msg_data_pages_advance(struct ceph_msg_data *data,
static void ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data *data,
size_t length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
struct ceph_pagelist *pagelist;
struct page *page;

Expand All @@ -923,7 +923,7 @@ static struct page *ceph_msg_data_pagelist_next(struct ceph_msg_data *data,
size_t *page_offset,
size_t *length)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
struct ceph_pagelist *pagelist;

BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
Expand All @@ -941,13 +941,13 @@ static struct page *ceph_msg_data_pagelist_next(struct ceph_msg_data *data,
else
*length = PAGE_SIZE - *page_offset;

return data->cursor.page;
return data->cursor->page;
}

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_msg_data_cursor *cursor = data->cursor;
struct ceph_pagelist *pagelist;

BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
Expand Down Expand Up @@ -1003,7 +1003,7 @@ static void ceph_msg_data_cursor_init(struct ceph_msg_data *data,
/* BUG(); */
break;
}
data->cursor.need_crc = true;
data->cursor->need_crc = true;
}

/*
Expand Down Expand Up @@ -1039,7 +1039,7 @@ static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
BUG_ON(*page_offset + *length > PAGE_SIZE);
BUG_ON(!*length);
if (last_piece)
*last_piece = data->cursor.last_piece;
*last_piece = data->cursor->last_piece;

return page;
}
Expand All @@ -1050,7 +1050,7 @@ static struct page *ceph_msg_data_next(struct ceph_msg_data *data,
*/
static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
{
struct ceph_msg_data_cursor *cursor = &data->cursor;
struct ceph_msg_data_cursor *cursor = data->cursor;
bool new_piece;

BUG_ON(bytes > cursor->resid);
Expand All @@ -1071,7 +1071,7 @@ static bool ceph_msg_data_advance(struct ceph_msg_data *data, size_t bytes)
BUG();
break;
}
data->cursor.need_crc = new_piece;
data->cursor->need_crc = new_piece;

return new_piece;
}
Expand Down Expand Up @@ -1404,7 +1404,7 @@ static u32 ceph_crc32c_page(u32 crc, struct page *page,
static int write_partial_message_data(struct ceph_connection *con)
{
struct ceph_msg *msg = con->out_msg;
struct ceph_msg_data_cursor *cursor = &msg->data->cursor;
struct ceph_msg_data_cursor *cursor = msg->data->cursor;
bool do_datacrc = !con->msgr->nocrc;
u32 crc;

Expand Down Expand Up @@ -2102,7 +2102,7 @@ static int read_partial_message_section(struct ceph_connection *con,
static int read_partial_msg_data(struct ceph_connection *con)
{
struct ceph_msg *msg = con->in_msg;
struct ceph_msg_data_cursor *cursor = &msg->data->cursor;
struct ceph_msg_data_cursor *cursor = msg->data->cursor;
const bool do_datacrc = !con->msgr->nocrc;
struct page *page;
size_t page_offset;
Expand Down Expand Up @@ -2991,6 +2991,7 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,

data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
BUG_ON(!data);
data->cursor = &msg->cursor;
data->pages = pages;
data->length = length;
data->alignment = alignment & ~PAGE_MASK;
Expand All @@ -3012,6 +3013,7 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg,

data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
BUG_ON(!data);
data->cursor = &msg->cursor;
data->pagelist = pagelist;

msg->data = data;
Expand All @@ -3031,6 +3033,7 @@ void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio,

data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
BUG_ON(!data);
data->cursor = &msg->cursor;
data->bio = bio;
data->bio_length = length;

Expand Down

0 comments on commit 36153ec

Please sign in to comment.