ceph/src/librbd/io/ReadResult.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "librbd/io/ReadResult.h"
   5 #include "include/buffer.h"
   6 #include "common/dout.h"
   7 #include "librbd/io/AioCompletion.h"
   8 #include "librbd/io/Utils.h"
   9 #include <boost/variant/apply_visitor.hpp>
  10 #include <boost/variant/static_visitor.hpp>
  11
  12 #define dout_subsys ceph_subsys_rbd
  13 #undef dout_prefix
  14 #define dout_prefix *_dout << "librbd::io::ReadResult: " << this \
  15                            << " " << __func__ << ": "
  16
  17 namespace librbd {
  18 namespace io {
  19
  20 struct ReadResult::SetImageExtentsVisitor : public boost::static_visitor<void> {
  21   Extents image_extents;
  22
  23   explicit SetImageExtentsVisitor(const Extents& image_extents)
  24     : image_extents(image_extents) {
  25   }
  26
  27   void operator()(Linear &linear) const {
  28     uint64_t length = util::get_extents_length(image_extents);
  29
  30     ceph_assert(length <= linear.buf_len);
  31     linear.buf_len = length;
  32   }
  33
  34   void operator()(SparseBufferlist &sbl) const {
  35     sbl.image_extents = image_extents;
  36   }
  37
  38   template <typename T>
  39   void operator()(T &t) const {
  40   }
  41 };
  42
  43 struct ReadResult::AssembleResultVisitor : public boost::static_visitor<void> {
  44   CephContext *cct;
  45   Striper::StripedReadResult &destriper;
  46
  47   AssembleResultVisitor(CephContext *cct, Striper::StripedReadResult &destriper)
  48     : cct(cct), destriper(destriper) {
  49   }
  50
  51   void operator()(Empty &empty) const {
  52     ldout(cct, 20) << "dropping read result" << dendl;
  53   }
  54
  55   void operator()(Linear &linear) const {
  56     ldout(cct, 20) << "copying resulting bytes to "
  57                    << reinterpret_cast<void*>(linear.buf) << dendl;
  58     destriper.assemble_result(cct, linear.buf, linear.buf_len);
  59   }
  60
  61   void operator()(Vector &vector) const {
  62     bufferlist bl;
  63     destriper.assemble_result(cct, bl, true);
  64
  65     ldout(cct, 20) << "copying resulting " << bl.length() << " bytes to iovec "
  66                    << reinterpret_cast<const void*>(vector.iov) << dendl;
  67
  68     bufferlist::iterator it = bl.begin();
  69     size_t length = bl.length();
  70     size_t offset = 0;
  71     int idx = 0;
  72     for (; offset < length && idx < vector.iov_count; idx++) {
  73       size_t len = std::min(vector.iov[idx].iov_len, length - offset);
  74       it.copy(len, static_cast<char *>(vector.iov[idx].iov_base));
  75       offset += len;
  76     }
  77     ceph_assert(offset == bl.length());
  78   }
  79
  80   void operator()(Bufferlist &bufferlist) const {
  81     bufferlist.bl->clear();
  82     destriper.assemble_result(cct, *bufferlist.bl, true);
  83
  84     ldout(cct, 20) << "moved resulting " << bufferlist.bl->length() << " "
  85                    << "bytes to bl " << reinterpret_cast<void*>(bufferlist.bl)
  86                    << dendl;
  87   }
  88
  89   void operator()(SparseBufferlist &sparse_bufferlist) const {
  90     sparse_bufferlist.bl->clear();
  91
  92     ExtentMap buffer_extent_map;
  93     auto buffer_extents_length = destriper.assemble_result(
  94       cct, &buffer_extent_map, sparse_bufferlist.bl);
  95
  96     ldout(cct, 20) << "image_extents="
  97                    << sparse_bufferlist.image_extents << ", "
  98                    << "buffer_extent_map=" << buffer_extent_map << dendl;
  99
 100     sparse_bufferlist.extent_map->clear();
 101     sparse_bufferlist.extent_map->reserve(buffer_extent_map.size());
 102
 103     // The extent-map is logically addressed by buffer-extents not image- or
 104     // object-extents. Translate this address mapping to image-extent
 105     // logical addressing since it's tied to an image-extent read
 106     uint64_t buffer_offset = 0;
 107     auto bem_it = buffer_extent_map.begin();
 108     for (auto [image_offset, image_length] : sparse_bufferlist.image_extents) {
 109       while (bem_it != buffer_extent_map.end()) {
 110         auto [buffer_extent_offset, buffer_extent_length] = *bem_it;
 111
 112         if (buffer_offset + image_length <= buffer_extent_offset) {
 113           // skip any image extent that is not included in the results
 114           break;
 115         }
 116
 117         // current buffer-extent should be within the current image-extent
 118         ceph_assert(buffer_offset <= buffer_extent_offset &&
 119                     buffer_offset + image_length >=
 120                       buffer_extent_offset + buffer_extent_length);
 121         auto image_extent_offset =
 122           image_offset + (buffer_extent_offset - buffer_offset);
 123         ldout(cct, 20) << "mapping buffer extent " << buffer_extent_offset
 124                        << "~" << buffer_extent_length << " to image extent "
 125                        << image_extent_offset << "~" << buffer_extent_length
 126                        << dendl;
 127         sparse_bufferlist.extent_map->emplace_back(
 128           image_extent_offset, buffer_extent_length);
 129         ++bem_it;
 130       }
 131
 132       buffer_offset += image_length;
 133     }
 134     ceph_assert(buffer_offset == buffer_extents_length);
 135     ceph_assert(bem_it == buffer_extent_map.end());
 136
 137     ldout(cct, 20) << "moved resulting " << *sparse_bufferlist.extent_map
 138                    << " extents of total " << sparse_bufferlist.bl->length()
 139                    << " bytes to bl "
 140                    << reinterpret_cast<void*>(sparse_bufferlist.bl) << dendl;
 141   }
 142 };
 143
 144 ReadResult::C_ImageReadRequest::C_ImageReadRequest(
 145     AioCompletion *aio_completion, uint64_t buffer_offset,
 146     const Extents image_extents)
 147   : aio_completion(aio_completion), buffer_offset(buffer_offset),
 148     image_extents(image_extents) {
 149   aio_completion->add_request();
 150 }
 151
 152 void ReadResult::C_ImageReadRequest::finish(int r) {
 153   CephContext *cct = aio_completion->ictx->cct;
 154   ldout(cct, 10) << "C_ImageReadRequest: r=" << r
 155                  << dendl;
 156   if (r >= 0 || (ignore_enoent && r == -ENOENT)) {
 157     striper::LightweightBufferExtents buffer_extents;
 158     size_t length = 0;
 159     for (auto &image_extent : image_extents) {
 160       buffer_extents.emplace_back(buffer_offset + length, image_extent.second);
 161       length += image_extent.second;
 162     }
 163     ceph_assert(r == -ENOENT || length == bl.length());
 164
 165     aio_completion->lock.lock();
 166     aio_completion->read_result.m_destriper.add_partial_result(
 167       cct, std::move(bl), buffer_extents);
 168     aio_completion->lock.unlock();
 169     r = length;
 170   }
 171
 172   aio_completion->complete_request(r);
 173 }
 174
 175 ReadResult::C_ObjectReadRequest::C_ObjectReadRequest(
 176     AioCompletion *aio_completion, ReadExtents&& extents)
 177   : aio_completion(aio_completion), extents(std::move(extents)) {
 178   aio_completion->add_request();
 179 }
 180
 181 void ReadResult::C_ObjectReadRequest::finish(int r) {
 182   CephContext *cct = aio_completion->ictx->cct;
 183   ldout(cct, 10) << "C_ObjectReadRequest: r=" << r
 184                  << dendl;
 185
 186   if (r == -ENOENT) {
 187     r = 0;
 188   }
 189   if (r >= 0) {
 190     uint64_t object_len = 0;
 191     aio_completion->lock.lock();
 192     for (auto& extent: extents) {
 193       ldout(cct, 10) << " got " << extent.extent_map
 194                      << " for " << extent.buffer_extents
 195                      << " bl " << extent.bl.length() << dendl;
 196
 197       aio_completion->read_result.m_destriper.add_partial_sparse_result(
 198               cct, std::move(extent.bl), extent.extent_map, extent.offset,
 199               extent.buffer_extents);
 200
 201       object_len += extent.length;
 202     }
 203     aio_completion->lock.unlock();
 204     r = object_len;
 205   }
 206
 207   aio_completion->complete_request(r);
 208 }
 209
 210 ReadResult::C_ObjectReadMergedExtents::C_ObjectReadMergedExtents(
 211         CephContext* cct, ReadExtents* extents, Context* on_finish)
 212         : cct(cct), extents(extents), on_finish(on_finish) {
 213 }
 214
 215 void ReadResult::C_ObjectReadMergedExtents::finish(int r) {
 216   if (r >= 0) {
 217     for (auto& extent: *extents) {
 218       if (bl.length() < extent.length) {
 219         lderr(cct) << "Merged extents length is less than expected" << dendl;
 220         r = -EIO;
 221         break;
 222       }
 223       bl.splice(0, extent.length, &extent.bl);
 224     }
 225     if (bl.length() != 0) {
 226       lderr(cct) << "Merged extents length is greater than expected" << dendl;
 227       r = -EIO;
 228     }
 229   }
 230   on_finish->complete(r);
 231 }
 232
 233 ReadResult::ReadResult() : m_buffer(Empty()) {
 234 }
 235
 236 ReadResult::ReadResult(char *buf, size_t buf_len)
 237   : m_buffer(Linear(buf, buf_len)) {
 238 }
 239
 240 ReadResult::ReadResult(const struct iovec *iov, int iov_count)
 241   : m_buffer(Vector(iov, iov_count)) {
 242 }
 243
 244 ReadResult::ReadResult(ceph::bufferlist *bl)
 245   : m_buffer(Bufferlist(bl)) {
 246 }
 247
 248 ReadResult::ReadResult(Extents* extent_map, ceph::bufferlist* bl)
 249   : m_buffer(SparseBufferlist(extent_map, bl)) {
 250 }
 251
 252 void ReadResult::set_image_extents(const Extents& image_extents) {
 253   boost::apply_visitor(SetImageExtentsVisitor(image_extents), m_buffer);
 254 }
 255
 256 void ReadResult::assemble_result(CephContext *cct) {
 257   boost::apply_visitor(AssembleResultVisitor(cct, m_destriper), m_buffer);
 258 }
 259
 260 } // namespace io
 261 } // namespace librbd
 262