libceph: record message data length

Keep track of the length of the data portion for a message in a
separate field in the ceph_msg structure.  This information has
been maintained in wire byte order in the message header, but
that's going to change soon.

Signed-off-by: Alex Elder <elder@inktank.com>
Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

2 files changed, 10 insertions, 2 deletions

diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index ee160864e8ea..fa9b4d0243a0 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -2981,6 +2981,7 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
 
         BUG_ON(!pages);
         BUG_ON(!length);
+        BUG_ON(msg->data_length);
         BUG_ON(msg->data != NULL);
 
         data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
@@ -2990,6 +2991,7 @@ void ceph_msg_data_set_pages(struct ceph_msg *msg, struct page **pages,
         data->alignment = alignment & ~PAGE_MASK;
 
         msg->data = data;
+        msg->data_length = length;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pages);
 
@@ -3000,6 +3002,7 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
 
         BUG_ON(!pagelist);
         BUG_ON(!pagelist->length);
+        BUG_ON(msg->data_length);
         BUG_ON(msg->data != NULL);
 
         data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
@@ -3007,14 +3010,17 @@ void ceph_msg_data_set_pagelist(struct ceph_msg *msg,
         data->pagelist = pagelist;
 
         msg->data = data;
+        msg->data_length = pagelist->length;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_pagelist);
 
-void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
+void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio,
+                size_t length)
 {
         struct ceph_msg_data *data;
 
         BUG_ON(!bio);
+        BUG_ON(msg->data_length);
         BUG_ON(msg->data != NULL);
 
         data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
@@ -3022,6 +3028,7 @@ void ceph_msg_data_set_bio(struct ceph_msg *msg, struct bio *bio)
         data->bio = bio;
 
         msg->data = data;
+        msg->data_length = length;
 }
 EXPORT_SYMBOL(ceph_msg_data_set_bio);
 
@@ -3200,6 +3207,7 @@ void ceph_msg_last_put(struct kref *kref)
         }
         ceph_msg_data_destroy(m->data);
         m->data = NULL;
+        m->data_length = 0;
 
         if (m->pool)
                 ceph_msgpool_put(m->pool, m);
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index e0887923e5ab..0b4951e27532 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -1848,7 +1848,7 @@ static void ceph_osdc_msg_data_set(struct ceph_msg *msg,
                 ceph_msg_data_set_pagelist(msg, osd_data->pagelist);
 #ifdef CONFIG_BLOCK
         } else if (osd_data->type == CEPH_OSD_DATA_TYPE_BIO) {
-                ceph_msg_data_set_bio(msg, osd_data->bio);
+                ceph_msg_data_set_bio(msg, osd_data->bio, osd_data->bio_length);
 #endif
         } else {
                 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_NONE);