ceph/src/librbd/operation/ObjectMapIterate.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/operation/ObjectMapIterate.h"
   5 #include "common/dout.h"
   6 #include "common/errno.h"
   7 #include "osdc/Striper.h"
   8 #include "librbd/AsyncObjectThrottle.h"
   9 #include "librbd/ExclusiveLock.h"
  10 #include "librbd/ImageCtx.h"
  11 #include "librbd/ImageWatcher.h"
  12 #include "librbd/internal.h"
  13 #include "librbd/ObjectMap.h"
  14 #include "librbd/operation/ResizeRequest.h"
  15 #include "librbd/object_map/InvalidateRequest.h"
  16 #include "librbd/Utils.h"
  17 #include <boost/lambda/bind.hpp>
  18 #include <boost/lambda/construct.hpp>
  19
  20 #define dout_subsys ceph_subsys_rbd
  21 #undef dout_prefix
  22 #define dout_prefix *_dout << "librbd::ObjectMapIterateRequest: "
  23
  24 namespace librbd {
  25 namespace operation {
  26
  27 namespace {
  28
  29 template <typename I>
  30 class C_VerifyObjectCallback : public C_AsyncObjectThrottle<I> {
  31 public:
  32   C_VerifyObjectCallback(AsyncObjectThrottle<I> &throttle, I *image_ctx,
  33                          uint64_t snap_id, uint64_t object_no,
  34                          ObjectIterateWork<I> handle_mismatch,
  35                          std::atomic_flag *invalidate)
  36     : C_AsyncObjectThrottle<I>(throttle, *image_ctx),
  37     m_snap_id(snap_id), m_object_no(object_no),
  38     m_oid(image_ctx->get_object_name(m_object_no)),
  39     m_handle_mismatch(handle_mismatch),
  40     m_invalidate(invalidate)
  41   {
  42     m_io_ctx.dup(image_ctx->data_ctx);
  43     m_io_ctx.snap_set_read(CEPH_SNAPDIR);
  44   }
  45
  46   void complete(int r) override {
  47     I &image_ctx = this->m_image_ctx;
  48     if (should_complete(r)) {
  49       ldout(image_ctx.cct, 20) << m_oid << " C_VerifyObjectCallback completed "
  50                                << dendl;
  51       m_io_ctx.close();
  52
  53       this->finish(r);
  54       delete this;
  55     }
  56   }
  57
  58   int send() override {
  59     send_list_snaps();
  60     return 0;
  61   }
  62
  63 private:
  64   librados::IoCtx m_io_ctx;
  65   uint64_t m_snap_id;
  66   uint64_t m_object_no;
  67   std::string m_oid;
  68   ObjectIterateWork<I> m_handle_mismatch;
  69   std::atomic_flag *m_invalidate;
  70
  71   librados::snap_set_t m_snap_set;
  72   int m_snap_list_ret = 0;
  73
  74   bool should_complete(int r) {
  75     I &image_ctx = this->m_image_ctx;
  76     CephContext *cct = image_ctx.cct;
  77     if (r == 0) {
  78       r = m_snap_list_ret;
  79     }
  80     if (r < 0 && r != -ENOENT) {
  81       lderr(cct) << m_oid << " C_VerifyObjectCallback::should_complete: "
  82                  << "encountered an error: " << cpp_strerror(r) << dendl;
  83       return true;
  84     }
  85
  86     ldout(cct, 20) << m_oid << " C_VerifyObjectCallback::should_complete: "
  87                    << " r="
  88                    << r << dendl;
  89     return object_map_action(get_object_state());
  90   }
  91
  92   void send_list_snaps() {
  93     I &image_ctx = this->m_image_ctx;
  94     ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
  95     ldout(image_ctx.cct, 5) << m_oid
  96                             << " C_VerifyObjectCallback::send_list_snaps"
  97                             << dendl;
  98
  99     librados::ObjectReadOperation op;
 100     op.list_snaps(&m_snap_set, &m_snap_list_ret);
 101
 102     librados::AioCompletion *comp = util::create_rados_callback(this);
 103     int r = m_io_ctx.aio_operate(m_oid, comp, &op, NULL);
 104     ceph_assert(r == 0);
 105     comp->release();
 106   }
 107
 108   uint8_t get_object_state() {
 109     I &image_ctx = this->m_image_ctx;
 110     std::shared_lock image_locker{image_ctx.image_lock};
 111     for (std::vector<librados::clone_info_t>::const_iterator r =
 112            m_snap_set.clones.begin(); r != m_snap_set.clones.end(); ++r) {
 113       librados::snap_t from_snap_id;
 114       librados::snap_t to_snap_id;
 115       if (r->cloneid == librados::SNAP_HEAD) {
 116         from_snap_id = next_valid_snap_id(m_snap_set.seq + 1);
 117         to_snap_id = librados::SNAP_HEAD;
 118       } else {
 119         from_snap_id = next_valid_snap_id(r->snaps[0]);
 120         to_snap_id = r->snaps[r->snaps.size()-1];
 121       }
 122
 123       if (to_snap_id < m_snap_id) {
 124         continue;
 125       } else if (m_snap_id < from_snap_id) {
 126         break;
 127       }
 128
 129       if ((image_ctx.features & RBD_FEATURE_FAST_DIFF) != 0 &&
 130           from_snap_id != m_snap_id) {
 131         return OBJECT_EXISTS_CLEAN;
 132       }
 133       return OBJECT_EXISTS;
 134     }
 135     return OBJECT_NONEXISTENT;
 136   }
 137
 138   uint64_t next_valid_snap_id(uint64_t snap_id) {
 139     I &image_ctx = this->m_image_ctx;
 140     ceph_assert(ceph_mutex_is_locked(image_ctx.image_lock));
 141
 142     std::map<librados::snap_t, SnapInfo>::iterator it =
 143       image_ctx.snap_info.lower_bound(snap_id);
 144     if (it == image_ctx.snap_info.end()) {
 145       return CEPH_NOSNAP;
 146     }
 147     return it->first;
 148   }
 149
 150   bool object_map_action(uint8_t new_state) {
 151     I &image_ctx = this->m_image_ctx;
 152     CephContext *cct = image_ctx.cct;
 153     std::shared_lock owner_locker{image_ctx.owner_lock};
 154
 155     // should have been canceled prior to releasing lock
 156     ceph_assert(image_ctx.exclusive_lock == nullptr ||
 157                 image_ctx.exclusive_lock->is_lock_owner());
 158
 159     std::shared_lock image_locker{image_ctx.image_lock};
 160     ceph_assert(image_ctx.object_map != nullptr);
 161
 162     uint8_t state = (*image_ctx.object_map)[m_object_no];
 163     ldout(cct, 10) << "C_VerifyObjectCallback::object_map_action"
 164                    << " object " << image_ctx.get_object_name(m_object_no)
 165                    << " state " << (int)state
 166                    << " new_state " << (int)new_state << dendl;
 167
 168     if (state != new_state) {
 169       int r = 0;
 170
 171       ceph_assert(m_handle_mismatch);
 172       r = m_handle_mismatch(image_ctx, m_object_no, state, new_state);
 173       if (r) {
 174         lderr(cct) << "object map error: object "
 175                    << image_ctx.get_object_name(m_object_no)
 176                    << " marked as " << (int)state << ", but should be "
 177                    << (int)new_state << dendl;
 178         m_invalidate->test_and_set();
 179       } else {
 180         ldout(cct, 1) << "object map inconsistent: object "
 181                    << image_ctx.get_object_name(m_object_no)
 182                    << " marked as " << (int)state << ", but should be "
 183                    << (int)new_state << dendl;
 184       }
 185     }
 186
 187     return true;
 188   }
 189 };
 190
 191 } // anonymous namespace
 192
 193 template <typename I>
 194 void ObjectMapIterateRequest<I>::send() {
 195   if (!m_image_ctx.data_ctx.is_valid()) {
 196     this->async_complete(-ENODEV);
 197     return;
 198   }
 199
 200   send_verify_objects();
 201 }
 202
 203 template <typename I>
 204 bool ObjectMapIterateRequest<I>::should_complete(int r) {
 205   CephContext *cct = m_image_ctx.cct;
 206   ldout(cct, 5) << this << " should_complete: " << " r=" << r << dendl;
 207
 208   if (r == -ENODEV) {
 209     lderr(cct) << "missing data pool" << dendl;
 210     return true;
 211   }
 212
 213   if (r < 0) {
 214     lderr(cct) << "object map operation encountered an error: "
 215                << cpp_strerror(r) << dendl;
 216   }
 217
 218   std::shared_lock owner_lock{m_image_ctx.owner_lock};
 219   switch (m_state) {
 220   case STATE_VERIFY_OBJECTS:
 221     if (m_invalidate.test_and_set()) {
 222       send_invalidate_object_map();
 223       return false;
 224     } else if (r == 0) {
 225       return true;
 226     }
 227     break;
 228
 229   case STATE_INVALIDATE_OBJECT_MAP:
 230     if (r == 0) {
 231       return true;
 232     }
 233     break;
 234
 235   default:
 236     ceph_abort();
 237     break;
 238   }
 239
 240   if (r < 0) {
 241     return true;
 242   }
 243
 244   return false;
 245 }
 246
 247 template <typename I>
 248 void ObjectMapIterateRequest<I>::send_verify_objects() {
 249   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 250   CephContext *cct = m_image_ctx.cct;
 251
 252   uint64_t snap_id;
 253   uint64_t num_objects;
 254   {
 255     std::shared_lock l{m_image_ctx.image_lock};
 256     snap_id = m_image_ctx.snap_id;
 257     num_objects = Striper::get_num_objects(m_image_ctx.layout,
 258                                            m_image_ctx.get_image_size(snap_id));
 259   }
 260   ldout(cct, 5) << this << " send_verify_objects" << dendl;
 261
 262   m_state = STATE_VERIFY_OBJECTS;
 263
 264   typename AsyncObjectThrottle<I>::ContextFactory context_factory(
 265     boost::lambda::bind(boost::lambda::new_ptr<C_VerifyObjectCallback<I> >(),
 266                         boost::lambda::_1, &m_image_ctx, snap_id,
 267                         boost::lambda::_2, m_handle_mismatch, &m_invalidate));
 268   AsyncObjectThrottle<I> *throttle = new AsyncObjectThrottle<I>(
 269     this, m_image_ctx, context_factory, this->create_callback_context(),
 270     &m_prog_ctx, 0, num_objects);
 271   throttle->start_ops(
 272     m_image_ctx.config.template get_val<uint64_t>("rbd_concurrent_management_ops"));
 273 }
 274
 275 template <typename I>
 276 uint64_t ObjectMapIterateRequest<I>::get_image_size() const {
 277   ceph_assert(ceph_mutex_is_locked(m_image_ctx.image_lock));
 278   if (m_image_ctx.snap_id == CEPH_NOSNAP) {
 279     if (!m_image_ctx.resize_reqs.empty()) {
 280       return m_image_ctx.resize_reqs.front()->get_image_size();
 281     } else {
 282       return m_image_ctx.size;
 283     }
 284   }
 285   return  m_image_ctx.get_image_size(m_image_ctx.snap_id);
 286 }
 287
 288 template <typename I>
 289 void ObjectMapIterateRequest<I>::send_invalidate_object_map() {
 290   CephContext *cct = m_image_ctx.cct;
 291
 292   ldout(cct, 5) << this << " send_invalidate_object_map" << dendl;
 293   m_state = STATE_INVALIDATE_OBJECT_MAP;
 294
 295   object_map::InvalidateRequest<I>*req =
 296     object_map::InvalidateRequest<I>::create(m_image_ctx, m_image_ctx.snap_id,
 297                                              true,
 298                                              this->create_callback_context());
 299
 300   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 301   std::unique_lock image_locker{m_image_ctx.image_lock};
 302   req->send();
 303 }
 304
 305 } // namespace operation
 306 } // namespace librbd
 307
 308 template class librbd::operation::ObjectMapIterateRequest<librbd::ImageCtx>;