Skip to content

Commit

Permalink
---
Browse files Browse the repository at this point in the history 
yaml
---
r: 373382
b: refs/heads/master
c: 4c59b4a
h: refs/heads/master
v: v3
  • Loading branch information
Alex Elder authored and Sage Weil committed May 2, 2013
1 parent c1ca030 commit fe17d7a
Show file tree
Hide file tree
Showing 3 changed files with 41 additions and 96 deletions.
2 changes: 1 addition & 1 deletion [refs]
View file
Original file line number Diff line number Diff line change
@@ -1,2 +1,2 @@
---
refs/heads/master: 686be20875db63c6103573565c63db20153ee6e1
refs/heads/master: 4c59b4a278f9b7a418ad8af933fd7b341df64393
12 changes: 2 additions & 10 deletions trunk/include/linux/ceph/messenger.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,11 +64,7 @@ struct ceph_messenger {
u32 required_features;
};

#define ceph_msg_has_pages(m) ((m)->p.type == CEPH_MSG_DATA_PAGES)
#define ceph_msg_has_pagelist(m) ((m)->l.type == CEPH_MSG_DATA_PAGELIST)
#ifdef CONFIG_BLOCK
#define ceph_msg_has_bio(m) ((m)->b.type == CEPH_MSG_DATA_BIO)
#endif /* CONFIG_BLOCK */
#define ceph_msg_has_data(m) ((m)->data.type != CEPH_MSG_DATA_NONE)

enum ceph_msg_data_type {
CEPH_MSG_DATA_NONE, /* message contains no data payload */
Expand Down Expand Up @@ -145,11 +141,7 @@ struct ceph_msg {
struct ceph_buffer *middle;

/* data payload */
struct ceph_msg_data p; /* pages */
struct ceph_msg_data l; /* pagelist */
#ifdef CONFIG_BLOCK
struct ceph_msg_data b; /* bio */
#endif /* CONFIG_BLOCK */
struct ceph_msg_data data;

struct ceph_connection *con;
struct list_head list_head; /* links for connection lists */
Expand Down
123 changes: 38 additions & 85 deletions trunk/net/ceph/messenger.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1085,22 +1085,15 @@ 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->p.alignment;
if (ceph_msg_has_data(msg))
msg_pos->page_pos = msg->data.alignment;
else
msg_pos->page_pos = 0;
msg_pos->data_pos = 0;

/* Initialize data cursors */
/* Initialize data cursor */

#ifdef CONFIG_BLOCK
if (ceph_msg_has_bio(msg))
ceph_msg_data_cursor_init(&msg->b, data_len);
#endif /* CONFIG_BLOCK */
if (ceph_msg_has_pages(msg))
ceph_msg_data_cursor_init(&msg->p, data_len);
if (ceph_msg_has_pagelist(msg))
ceph_msg_data_cursor_init(&msg->l, data_len);
ceph_msg_data_cursor_init(&msg->data, data_len);

msg_pos->did_page_crc = false;
}
Expand Down Expand Up @@ -1166,10 +1159,10 @@ static void prepare_write_message(struct ceph_connection *con)
m->needs_out_seq = false;
}

dout("prepare_write_message %p seq %lld type %d len %d+%d+%d (%zd)\n",
dout("prepare_write_message %p seq %lld type %d len %d+%d+%d\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->p.length);
le32_to_cpu(m->hdr.data_len));
BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

/* tag + hdr + front + middle */
Expand Down Expand Up @@ -1411,14 +1404,7 @@ static void out_msg_pos_next(struct ceph_connection *con, struct page *page,

msg_pos->data_pos += sent;
msg_pos->page_pos += sent;
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
else if (ceph_msg_has_bio(msg))
need_crc = ceph_msg_data_advance(&msg->b, sent);
#endif /* CONFIG_BLOCK */
need_crc = ceph_msg_data_advance(&msg->data, sent);
BUG_ON(need_crc && sent != len);

if (sent < len)
Expand All @@ -1441,12 +1427,8 @@ 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
else if (ceph_msg_has_bio(msg))
(void) ceph_msg_data_advance(&msg->b, received);
#endif /* CONFIG_BLOCK */
(void) ceph_msg_data_advance(&msg->data, received);

if (received < len)
return;

Expand Down Expand Up @@ -1486,6 +1468,9 @@ static int write_partial_message_data(struct ceph_connection *con)
dout("%s %p msg %p page %d offset %d\n", __func__,
con, msg, msg_pos->page, msg_pos->page_pos);

if (WARN_ON(!ceph_msg_has_data(msg)))
return -EINVAL;

/*
* Iterate through each page that contains data to be
* written, and send as much as possible for each.
Expand All @@ -1500,23 +1485,8 @@ static int write_partial_message_data(struct ceph_connection *con)
size_t length;
bool last_piece;

if (ceph_msg_has_pages(msg)) {
page = ceph_msg_data_next(&msg->p, &page_offset,
&length, &last_piece);
} else if (ceph_msg_has_pagelist(msg)) {
page = ceph_msg_data_next(&msg->l, &page_offset,
&length, &last_piece);
#ifdef CONFIG_BLOCK
} else if (ceph_msg_has_bio(msg)) {
page = ceph_msg_data_next(&msg->b, &page_offset,
&length, &last_piece);
#endif
} else {
WARN(1, "con %p data_len %u but no outbound data\n",
con, data_len);
ret = -EINVAL;
goto out;
}
page = ceph_msg_data_next(&msg->data, &page_offset, &length,
&last_piece);
if (do_datacrc && !msg_pos->did_page_crc) {
u32 crc = le32_to_cpu(msg->footer.data_crc);

Expand Down Expand Up @@ -2197,28 +2167,20 @@ static int read_partial_msg_data(struct ceph_connection *con)
int ret;

BUG_ON(!msg);
if (WARN_ON(!ceph_msg_has_data(msg)))
return -EIO;

data_len = le32_to_cpu(con->in_hdr.data_len);
while (msg_pos->data_pos < data_len) {
if (ceph_msg_has_pages(msg)) {
page = ceph_msg_data_next(&msg->p, &page_offset,
&length, NULL);
#ifdef CONFIG_BLOCK
} else if (ceph_msg_has_bio(msg)) {
page = ceph_msg_data_next(&msg->b, &page_offset,
&length, NULL);
#endif
} else {
BUG_ON(1);
}
page = ceph_msg_data_next(&msg->data, &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);
}

Expand Down Expand Up @@ -3043,12 +3005,12 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
{
BUG_ON(!pages);
BUG_ON(!length);
BUG_ON(msg->p.type != CEPH_MSG_DATA_NONE);
BUG_ON(msg->data.type != CEPH_MSG_DATA_NONE);

msg->p.type = CEPH_MSG_DATA_PAGES;
msg->p.pages = pages;
msg->p.length = length;
msg->p.alignment = alignment & ~PAGE_MASK;
msg->data.type = CEPH_MSG_DATA_PAGES;
msg->data.pages = pages;
msg->data.length = length;
msg->data.alignment = alignment & ~PAGE_MASK;
}
EXPORT_SYMBOL(ceph_msg_data_set_pages);

Expand All @@ -3057,20 +3019,20 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
{
BUG_ON(!pagelist);
BUG_ON(!pagelist->length);
BUG_ON(msg->l.type != CEPH_MSG_DATA_NONE);
BUG_ON(msg->data.type != CEPH_MSG_DATA_NONE);

msg->l.type = CEPH_MSG_DATA_PAGELIST;
msg->l.pagelist = pagelist;
msg->data.type = CEPH_MSG_DATA_PAGELIST;
msg->data.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->b.type != CEPH_MSG_DATA_NONE);
BUG_ON(msg->data.type != CEPH_MSG_DATA_NONE);

msg->b.type = CEPH_MSG_DATA_BIO;
msg->b.bio = bio;
msg->data.type = CEPH_MSG_DATA_BIO;
msg->data.bio = bio;
}
EXPORT_SYMBOL(ceph_msg_data_set_bio);

Expand All @@ -3094,9 +3056,7 @@ struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
INIT_LIST_HEAD(&m->list_head);
kref_init(&m->kref);

ceph_msg_data_init(&m->p);
ceph_msg_data_init(&m->l);
ceph_msg_data_init(&m->b);
ceph_msg_data_init(&m->data);

/* front */
m->front_max = front_len;
Expand Down Expand Up @@ -3251,20 +3211,13 @@ void ceph_msg_last_put(struct kref *kref)
ceph_buffer_put(m->middle);
m->middle = NULL;
}
if (ceph_msg_has_pages(m)) {
m->p.length = 0;
m->p.pages = NULL;
m->p.type = CEPH_OSD_DATA_TYPE_NONE;
}
if (ceph_msg_has_pagelist(m)) {
ceph_pagelist_release(m->l.pagelist);
kfree(m->l.pagelist);
m->l.pagelist = NULL;
m->l.type = CEPH_OSD_DATA_TYPE_NONE;
}
if (ceph_msg_has_bio(m)) {
m->b.bio = NULL;
m->b.type = CEPH_OSD_DATA_TYPE_NONE;
if (ceph_msg_has_data(m)) {
if (m->data.type == CEPH_MSG_DATA_PAGELIST) {
ceph_pagelist_release(m->data.pagelist);
kfree(m->data.pagelist);
}
memset(&m->data, 0, sizeof m->data);
ceph_msg_data_init(&m->data);
}

if (m->pool)
Expand All @@ -3277,7 +3230,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->p.length);
msg->front_max, msg->data.length);
print_hex_dump(KERN_DEBUG, "header: ",
DUMP_PREFIX_OFFSET, 16, 1,
&msg->hdr, sizeof(msg->hdr), true);
Expand Down

0 comments on commit fe17d7a

Please sign in to comment.