librbd: make read methods target buffers or bufferlists

This will let us transition much of the read code to move bufferlists
around instead of copying data between buffers.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/librbd/AioCompletion.cc b/src/librbd/AioCompletion.cc
index de924e614b4ab8544952a7728283b2c04d4a0def..39be4ecab8f4e96ad90079f740d1bd1e3db9389d 100644 (file)
--- a/src/librbd/AioCompletion.cc
+++ b/src/librbd/AioCompletion.cc
@@ -31,15 +31,24 @@ namespace librbd {
     assert(pending_count);
     int count = --pending_count;
     if (!count) {
+      ldout(cct, 20) << "AioCompletion::complete_request() rval " << rval << " read_buf " << (void*)read_buf
+                    << " read_bl " << (void*)read_bl << dendl;
       if (rval >= 0 && aio_type == AIO_TYPE_READ) {
        // FIXME: make the destriper write directly into a buffer so
        // that we avoid shuffling pointers and copying zeros around.
        bufferlist bl;
        destriper.assemble_result(bl, true);
-       assert(bl.length() == read_buf_len);
-       bl.copy(0, read_buf_len, read_buf);
-       ldout(cct, 20) << "AioCompletion::complete_request() copied resulting " << bl.length()
-                      << " bytes to " << (void*)read_buf << dendl;
+       if (read_buf) {
+         assert(bl.length() == read_buf_len);
+         bl.copy(0, read_buf_len, read_buf);
+         ldout(cct, 20) << "AioCompletion::complete_request() copied resulting " << bl.length()
+                        << " bytes to " << (void*)read_buf << dendl;
+       }
+       if (read_bl) {
+         ldout(cct, 20) << "AioCompletion::complete_request() moving resulting " << bl.length()
+                        << " bytes to bl " << (void*)read_bl << dendl;
+         read_bl->claim(bl);
+       }
       }      
 
       complete();

diff --git a/src/librbd/AioCompletion.h b/src/librbd/AioCompletion.h
index 6b3c2e675f77359857da7491506639b03e3a063e..8db852db023116e8626e68ed40529716e111e7ea 100644 (file)
--- a/src/librbd/AioCompletion.h
+++ b/src/librbd/AioCompletion.h
@@ -55,6 +55,7 @@ namespace librbd {
     aio_type_t aio_type;
 
     Filer::StripedReadResult destriper;
+    bufferlist *read_bl;
     char *read_buf;
     size_t read_buf_len;
 
@@ -63,7 +64,7 @@ namespace librbd {
                      complete_arg(NULL), rbd_comp(NULL), pending_count(1),
                      ref(1), released(false), ictx(NULL),
                      aio_type(AIO_TYPE_NONE),
-                     read_buf(NULL), read_buf_len(0) {
+                     read_bl(NULL), read_buf(NULL), read_buf_len(0) {
     }
     ~AioCompletion() {
     }

diff --git a/src/librbd/AioRequest.cc b/src/librbd/AioRequest.cc
index cc55c9d10147c89b84f3f595bfafed61273bd581..0765b6cd9f46696f97b34bf58414d971789f13b8 100644 (file)
--- a/src/librbd/AioRequest.cc
+++ b/src/librbd/AioRequest.cc
@@ -54,7 +54,7 @@ namespace librbd {
     assert(m_ictx->parent_lock.is_locked());
 
     m_parent_completion = aio_create_completion_internal(this, rbd_req_cb);
-    aio_read(m_ictx->parent, image_extents, m_read_data.c_str(),
+    aio_read(m_ictx->parent, image_extents, m_read_data.c_str(), NULL,
             m_parent_completion);
   }
 

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index fd226c27f2de3a915c3a3ee0bafebf8691e01f02..9c35a4552c92a6969579f42a0eaf29c199a95c2c 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1898,7 +1898,7 @@ reprotect_and_return_err:
       uint64_t object_overlap = ictx->prune_parent_extents(objectx, overlap);
       assert(object_overlap <= object_size);
 
-      if ((r = read(ictx->parent, objectx, buf)) < 0) {
+      if ((r = read(ictx->parent, objectx, buf, NULL)) < 0) {
        lderr(ictx->cct) << "reading from parent failed" << dendl;
        goto err;
       }
@@ -2090,7 +2090,7 @@ reprotect_and_return_err:
 
       Context *ctx = new C_SafeCond(&mylock, &cond, &done, &ret);
       AioCompletion *c = aio_create_completion_internal(ctx, rbd_ctx_cb);
-      r = aio_read(ictx, off, read_len, bl.c_str(), c);
+      r = aio_read(ictx, off, read_len, bl.c_str(), NULL, c);
       if (r < 0) {
        c->release();
        delete ctx;
@@ -2138,7 +2138,7 @@ reprotect_and_return_err:
     return read_iterate(ictx, ofs, len, simple_read_cb, buf);
   }
 
-  ssize_t read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents, char *buf)
+  ssize_t read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents, char *buf, bufferlist *pbl)
   {
     Mutex mylock("IoCtxImpl::write::mylock");
     Cond cond;
@@ -2147,7 +2147,7 @@ reprotect_and_return_err:
 
     Context *ctx = new C_SafeCond(&mylock, &cond, &done, &ret);
     AioCompletion *c = aio_create_completion_internal(ctx, rbd_ctx_cb);
-    int r = aio_read(ictx, image_extents, buf, c);
+    int r = aio_read(ictx, image_extents, buf, pbl, c);
     if (r < 0) {
       c->release();
       delete ctx;
@@ -2524,17 +2524,16 @@ reprotect_and_return_err:
   }
 
   int aio_read(ImageCtx *ictx, uint64_t off, size_t len,
-              char *buf,
+              char *buf, bufferlist *bl,
               AioCompletion *c)
   {
     vector<pair<uint64_t,uint64_t> > image_extents(1);
     image_extents[0] = make_pair(off, len);
-    return aio_read(ictx, image_extents, buf, c);
+    return aio_read(ictx, image_extents, buf, bl, c);
   }
 
   int aio_read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents,
-              char *buf,
-              AioCompletion *c)
+              char *buf, bufferlist *pbl, AioCompletion *c)
   {
     ldout(ictx->cct, 20) << "aio_read " << ictx << " " << image_extents << dendl;
 
@@ -2566,6 +2565,7 @@ reprotect_and_return_err:
 
     c->read_buf = buf;
     c->read_buf_len = buffer_ofs;
+    c->read_bl = pbl;
 
     c->get();
     c->init_time(ictx, AIO_TYPE_READ);

diff --git a/src/librbd/internal.h b/src/librbd/internal.h
index 7c3b2c7d6ad4c51e2e0ec8dd12ca1b8380758dc9..51a76a4fa267bb280776822212aee3382ea7ce27 100644 (file)
--- a/src/librbd/internal.h
+++ b/src/librbd/internal.h
@@ -164,17 +164,17 @@ namespace librbd {
                       int (*cb)(uint64_t, size_t, const char *, void *),
                       void *arg);
   ssize_t read(ImageCtx *ictx, uint64_t off, size_t len, char *buf);
-  ssize_t read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents, char *buf);
+  ssize_t read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents,
+              char *buf, bufferlist *pbl);
   ssize_t write(ImageCtx *ictx, uint64_t off, size_t len, const char *buf);
   int discard(ImageCtx *ictx, uint64_t off, uint64_t len);
   int aio_write(ImageCtx *ictx, uint64_t off, size_t len, const char *buf,
                 AioCompletion *c);
   int aio_discard(ImageCtx *ictx, uint64_t off, uint64_t len, AioCompletion *c);
   int aio_read(ImageCtx *ictx, uint64_t off, size_t len,
-               char *buf, AioCompletion *c);
+               char *buf, bufferlist *pbl, AioCompletion *c);
   int aio_read(ImageCtx *ictx, const vector<pair<uint64_t,uint64_t> >& image_extents,
-              char *buf,
-              AioCompletion *c);
+              char *buf, bufferlist *pbl, AioCompletion *c);
   int flush(ImageCtx *ictx);
   int _flush(ImageCtx *ictx);
 

diff --git a/src/librbd/librbd.cc b/src/librbd/librbd.cc
index 2055bf967ede2d4f6c25366ef0679c6fbfe507cf..0bef213974d0d9df65456712d2b259d1b2635a55 100644 (file)
--- a/src/librbd/librbd.cc
+++ b/src/librbd/librbd.cc
@@ -432,7 +432,7 @@ namespace librbd {
     bl.push_back(ptr);
     ldout(ictx->cct, 10) << "Image::aio_read() buf=" << (void *)bl.c_str() << "~"
                         << (void *)(bl.c_str() + len - 1) << dendl;
-    return librbd::aio_read(ictx, off, len, bl.c_str(), (librbd::AioCompletion *)c->pc);
+    return librbd::aio_read(ictx, off, len, bl.c_str(), NULL, (librbd::AioCompletion *)c->pc);
   }
 
   int Image::flush()
@@ -973,7 +973,7 @@ extern "C" int rbd_aio_read(rbd_image_t image, uint64_t off, size_t len,
 {
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
-  return librbd::aio_read(ictx, off, len, buf,
+  return librbd::aio_read(ictx, off, len, buf, NULL,
                          (librbd::AioCompletion *)comp->pc);
 }