ceph/src/librbd/ImageCtx.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include <errno.h>
   4 #include <boost/assign/list_of.hpp>
   5 #include <stddef.h>
   6
   7 #include "common/ceph_context.h"
   8 #include "common/dout.h"
   9 #include "common/errno.h"
  10 #include "common/perf_counters.h"
  11 #include "common/WorkQueue.h"
  12 #include "common/Timer.h"
  13
  14 #include "librbd/AsyncRequest.h"
  15 #include "librbd/ExclusiveLock.h"
  16 #include "librbd/internal.h"
  17 #include "librbd/ImageCtx.h"
  18 #include "librbd/ImageState.h"
  19 #include "librbd/ImageWatcher.h"
  20 #include "librbd/Journal.h"
  21 #include "librbd/LibrbdAdminSocketHook.h"
  22 #include "librbd/ObjectMap.h"
  23 #include "librbd/Operations.h"
  24 #include "librbd/operation/ResizeRequest.h"
  25 #include "librbd/Types.h"
  26 #include "librbd/Utils.h"
  27 #include "librbd/LibrbdWriteback.h"
  28 #include "librbd/exclusive_lock/AutomaticPolicy.h"
  29 #include "librbd/exclusive_lock/StandardPolicy.h"
  30 #include "librbd/io/AioCompletion.h"
  31 #include "librbd/io/AsyncOperation.h"
  32 #include "librbd/io/ImageRequestWQ.h"
  33 #include "librbd/io/ObjectDispatcher.h"
  34 #include "librbd/journal/StandardPolicy.h"
  35
  36 #include "osdc/Striper.h"
  37 #include <boost/bind.hpp>
  38 #include <boost/algorithm/string/predicate.hpp>
  39
  40 #define dout_subsys ceph_subsys_rbd
  41 #undef dout_prefix
  42 #define dout_prefix *_dout << "librbd::ImageCtx: "
  43
  44 using std::map;
  45 using std::pair;
  46 using std::set;
  47 using std::string;
  48 using std::vector;
  49
  50 using ceph::bufferlist;
  51 using librados::snap_t;
  52 using librados::IoCtx;
  53
  54 namespace librbd {
  55
  56 namespace {
  57
  58 class ThreadPoolSingleton : public ThreadPool {
  59 public:
  60   ContextWQ *op_work_queue;
  61
  62   explicit ThreadPoolSingleton(CephContext *cct)
  63     : ThreadPool(cct, "librbd::thread_pool", "tp_librbd", 1,
  64                  "rbd_op_threads"),
  65       op_work_queue(new ContextWQ("librbd::op_work_queue",
  66                                   cct->_conf.get_val<uint64_t>("rbd_op_thread_timeout"),
  67                                   this)) {
  68     start();
  69   }
  70   ~ThreadPoolSingleton() override {
  71     op_work_queue->drain();
  72     delete op_work_queue;
  73
  74     stop();
  75   }
  76 };
  77
  78 class SafeTimerSingleton : public SafeTimer {
  79 public:
  80   Mutex lock;
  81
  82   explicit SafeTimerSingleton(CephContext *cct)
  83       : SafeTimer(cct, lock, true),
  84         lock("librbd::Journal::SafeTimerSingleton::lock") {
  85     init();
  86   }
  87   ~SafeTimerSingleton() {
  88     Mutex::Locker locker(lock);
  89     shutdown();
  90   }
  91 };
  92
  93 } // anonymous namespace
  94
  95   const string ImageCtx::METADATA_CONF_PREFIX = "conf_";
  96
  97   ImageCtx::ImageCtx(const string &image_name, const string &image_id,
  98                      const char *snap, IoCtx& p, bool ro)
  99     : cct((CephContext*)p.cct()),
 100       config(cct->_conf),
 101       perfcounter(NULL),
 102       snap_id(CEPH_NOSNAP),
 103       snap_exists(true),
 104       read_only(ro),
 105       exclusive_locked(false),
 106       name(image_name),
 107       image_watcher(NULL),
 108       journal(NULL),
 109       owner_lock(util::unique_lock_name("librbd::ImageCtx::owner_lock", this)),
 110       md_lock(util::unique_lock_name("librbd::ImageCtx::md_lock", this)),
 111       snap_lock(util::unique_lock_name("librbd::ImageCtx::snap_lock", this)),
 112       timestamp_lock(util::unique_lock_name("librbd::ImageCtx::timestamp_lock", this)),
 113       parent_lock(util::unique_lock_name("librbd::ImageCtx::parent_lock", this)),
 114       object_map_lock(util::unique_lock_name("librbd::ImageCtx::object_map_lock", this)),
 115       async_ops_lock(util::unique_lock_name("librbd::ImageCtx::async_ops_lock", this)),
 116       copyup_list_lock(util::unique_lock_name("librbd::ImageCtx::copyup_list_lock", this)),
 117       completed_reqs_lock(util::unique_lock_name("librbd::ImageCtx::completed_reqs_lock", this)),
 118       extra_read_flags(0),
 119       old_format(false),
 120       order(0), size(0), features(0),
 121       format_string(NULL),
 122       id(image_id), parent(NULL),
 123       stripe_unit(0), stripe_count(0), flags(0),
 124       readahead(),
 125       total_bytes_read(0),
 126       state(new ImageState<>(this)),
 127       operations(new Operations<>(*this)),
 128       exclusive_lock(nullptr), object_map(nullptr),
 129       io_work_queue(nullptr), op_work_queue(nullptr),
 130       asok_hook(nullptr),
 131       trace_endpoint("librbd")
 132   {
 133     md_ctx.dup(p);
 134     data_ctx.dup(p);
 135     if (snap)
 136       snap_name = snap;
 137
 138     memset(&header, 0, sizeof(header));
 139
 140     ThreadPool *thread_pool;
 141     get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 142     io_work_queue = new io::ImageRequestWQ<>(
 143       this, "librbd::io_work_queue",
 144       cct->_conf.get_val<uint64_t>("rbd_op_thread_timeout"),
 145       thread_pool);
 146     io_object_dispatcher = new io::ObjectDispatcher<>(this);
 147
 148     if (cct->_conf.get_val<bool>("rbd_auto_exclusive_lock_until_manual_request")) {
 149       exclusive_lock_policy = new exclusive_lock::AutomaticPolicy(this);
 150     } else {
 151       exclusive_lock_policy = new exclusive_lock::StandardPolicy(this);
 152     }
 153     journal_policy = new journal::StandardPolicy<ImageCtx>(this);
 154   }
 155
 156   ImageCtx::ImageCtx(const string &image_name, const string &image_id,
 157                      uint64_t snap_id, IoCtx& p, bool ro)
 158     : ImageCtx(image_name, image_id, "", p, ro) {
 159     open_snap_id = snap_id;
 160   }
 161
 162   ImageCtx::~ImageCtx() {
 163     ceph_assert(image_watcher == NULL);
 164     ceph_assert(exclusive_lock == NULL);
 165     ceph_assert(object_map == NULL);
 166     ceph_assert(journal == NULL);
 167     ceph_assert(asok_hook == NULL);
 168
 169     if (perfcounter) {
 170       perf_stop();
 171     }
 172     delete[] format_string;
 173
 174     md_ctx.aio_flush();
 175     data_ctx.aio_flush();
 176     io_work_queue->drain();
 177
 178     delete io_object_dispatcher;
 179
 180     delete journal_policy;
 181     delete exclusive_lock_policy;
 182     delete io_work_queue;
 183     delete operations;
 184     delete state;
 185   }
 186
 187   void ImageCtx::init() {
 188     ceph_assert(!header_oid.empty());
 189     ceph_assert(old_format || !id.empty());
 190
 191     asok_hook = new LibrbdAdminSocketHook(this);
 192
 193     string pname = string("librbd-") + id + string("-") +
 194       data_ctx.get_pool_name() + string("-") + name;
 195     if (!snap_name.empty()) {
 196       pname += "-";
 197       pname += snap_name;
 198     }
 199
 200     trace_endpoint.copy_name(pname);
 201     perf_start(pname);
 202
 203     ceph_assert(image_watcher == NULL);
 204     image_watcher = new ImageWatcher<>(*this);
 205   }
 206
 207   void ImageCtx::shutdown() {
 208     delete image_watcher;
 209     image_watcher = nullptr;
 210
 211     delete asok_hook;
 212     asok_hook = nullptr;
 213   }
 214
 215   void ImageCtx::init_layout()
 216   {
 217     if (stripe_unit == 0 || stripe_count == 0) {
 218       stripe_unit = 1ull << order;
 219       stripe_count = 1;
 220     }
 221
 222     vector<uint64_t> alignments;
 223     alignments.push_back(stripe_count << order); // object set (in file striping terminology)
 224     alignments.push_back(stripe_unit * stripe_count); // stripe
 225     alignments.push_back(stripe_unit); // stripe unit
 226     readahead.set_alignments(alignments);
 227
 228     layout = file_layout_t();
 229     layout.stripe_unit = stripe_unit;
 230     layout.stripe_count = stripe_count;
 231     layout.object_size = 1ull << order;
 232     layout.pool_id = data_ctx.get_id();  // FIXME: pool id overflow?
 233
 234     delete[] format_string;
 235     size_t len = object_prefix.length() + 16;
 236     format_string = new char[len];
 237     if (old_format) {
 238       snprintf(format_string, len, "%s.%%012llx", object_prefix.c_str());
 239     } else {
 240       snprintf(format_string, len, "%s.%%016llx", object_prefix.c_str());
 241     }
 242
 243     ldout(cct, 10) << "init_layout stripe_unit " << stripe_unit
 244                    << " stripe_count " << stripe_count
 245                    << " object_size " << layout.object_size
 246                    << " prefix " << object_prefix
 247                    << " format " << format_string
 248                    << dendl;
 249   }
 250
 251   void ImageCtx::perf_start(string name) {
 252     auto perf_prio = PerfCountersBuilder::PRIO_DEBUGONLY;
 253     if (child == nullptr) {
 254       // ensure top-level IO stats are exported for librbd daemons
 255       perf_prio = PerfCountersBuilder::PRIO_USEFUL;
 256     }
 257
 258     PerfCountersBuilder plb(cct, name, l_librbd_first, l_librbd_last);
 259
 260     plb.add_u64_counter(l_librbd_rd, "rd", "Reads", "r", perf_prio);
 261     plb.add_u64_counter(l_librbd_rd_bytes, "rd_bytes", "Data size in reads",
 262                         "rb", perf_prio, unit_t(UNIT_BYTES));
 263     plb.add_time_avg(l_librbd_rd_latency, "rd_latency", "Latency of reads",
 264                      "rl", perf_prio);
 265     plb.add_u64_counter(l_librbd_wr, "wr", "Writes", "w", perf_prio);
 266     plb.add_u64_counter(l_librbd_wr_bytes, "wr_bytes", "Written data",
 267                         "wb", perf_prio, unit_t(UNIT_BYTES));
 268     plb.add_time_avg(l_librbd_wr_latency, "wr_latency", "Write latency",
 269                      "wl", perf_prio);
 270     plb.add_u64_counter(l_librbd_discard, "discard", "Discards");
 271     plb.add_u64_counter(l_librbd_discard_bytes, "discard_bytes", "Discarded data", NULL, 0, unit_t(UNIT_BYTES));
 272     plb.add_time_avg(l_librbd_discard_latency, "discard_latency", "Discard latency");
 273     plb.add_u64_counter(l_librbd_flush, "flush", "Flushes");
 274     plb.add_time_avg(l_librbd_flush_latency, "flush_latency", "Latency of flushes");
 275     plb.add_u64_counter(l_librbd_ws, "ws", "WriteSames");
 276     plb.add_u64_counter(l_librbd_ws_bytes, "ws_bytes", "WriteSame data", NULL, 0, unit_t(UNIT_BYTES));
 277     plb.add_time_avg(l_librbd_ws_latency, "ws_latency", "WriteSame latency");
 278     plb.add_u64_counter(l_librbd_cmp, "cmp", "CompareAndWrites");
 279     plb.add_u64_counter(l_librbd_cmp_bytes, "cmp_bytes", "Data size in cmps", NULL, 0, unit_t(UNIT_BYTES));
 280     plb.add_time_avg(l_librbd_cmp_latency, "cmp_latency", "Latency of cmps");
 281     plb.add_u64_counter(l_librbd_snap_create, "snap_create", "Snap creations");
 282     plb.add_u64_counter(l_librbd_snap_remove, "snap_remove", "Snap removals");
 283     plb.add_u64_counter(l_librbd_snap_rollback, "snap_rollback", "Snap rollbacks");
 284     plb.add_u64_counter(l_librbd_snap_rename, "snap_rename", "Snap rename");
 285     plb.add_u64_counter(l_librbd_notify, "notify", "Updated header notifications");
 286     plb.add_u64_counter(l_librbd_resize, "resize", "Resizes");
 287     plb.add_u64_counter(l_librbd_readahead, "readahead", "Read ahead");
 288     plb.add_u64_counter(l_librbd_readahead_bytes, "readahead_bytes", "Data size in read ahead", NULL, 0, unit_t(UNIT_BYTES));
 289     plb.add_u64_counter(l_librbd_invalidate_cache, "invalidate_cache", "Cache invalidates");
 290
 291     plb.add_time(l_librbd_opened_time, "opened_time", "Opened time",
 292                  "ots", perf_prio);
 293     plb.add_time(l_librbd_lock_acquired_time, "lock_acquired_time",
 294                  "Lock acquired time", "lats", perf_prio);
 295
 296     perfcounter = plb.create_perf_counters();
 297     cct->get_perfcounters_collection()->add(perfcounter);
 298
 299     perfcounter->tset(l_librbd_opened_time, ceph_clock_now());
 300   }
 301
 302   void ImageCtx::perf_stop() {
 303     ceph_assert(perfcounter);
 304     cct->get_perfcounters_collection()->remove(perfcounter);
 305     delete perfcounter;
 306   }
 307
 308   void ImageCtx::set_read_flag(unsigned flag) {
 309     extra_read_flags |= flag;
 310   }
 311
 312   int ImageCtx::get_read_flags(snap_t snap_id) {
 313     int flags = librados::OPERATION_NOFLAG | extra_read_flags;
 314     if (snap_id == LIBRADOS_SNAP_HEAD)
 315       return flags;
 316
 317     if (config.get_val<bool>("rbd_balance_snap_reads"))
 318       flags |= librados::OPERATION_BALANCE_READS;
 319     else if (config.get_val<bool>("rbd_localize_snap_reads"))
 320       flags |= librados::OPERATION_LOCALIZE_READS;
 321     return flags;
 322   }
 323
 324   int ImageCtx::snap_set(uint64_t in_snap_id) {
 325     ceph_assert(snap_lock.is_wlocked());
 326     auto it = snap_info.find(in_snap_id);
 327     if (in_snap_id != CEPH_NOSNAP && it != snap_info.end()) {
 328       snap_id = in_snap_id;
 329       snap_namespace = it->second.snap_namespace;
 330       snap_name = it->second.name;
 331       snap_exists = true;
 332       data_ctx.snap_set_read(snap_id);
 333       return 0;
 334     }
 335     return -ENOENT;
 336   }
 337
 338   void ImageCtx::snap_unset()
 339   {
 340     ceph_assert(snap_lock.is_wlocked());
 341     snap_id = CEPH_NOSNAP;
 342     snap_namespace = {};
 343     snap_name = "";
 344     snap_exists = true;
 345     data_ctx.snap_set_read(snap_id);
 346   }
 347
 348   snap_t ImageCtx::get_snap_id(const cls::rbd::SnapshotNamespace& in_snap_namespace,
 349                                const string& in_snap_name) const
 350   {
 351     ceph_assert(snap_lock.is_locked());
 352     auto it = snap_ids.find({in_snap_namespace, in_snap_name});
 353     if (it != snap_ids.end()) {
 354       return it->second;
 355     }
 356     return CEPH_NOSNAP;
 357   }
 358
 359   const SnapInfo* ImageCtx::get_snap_info(snap_t in_snap_id) const
 360   {
 361     ceph_assert(snap_lock.is_locked());
 362     map<snap_t, SnapInfo>::const_iterator it =
 363       snap_info.find(in_snap_id);
 364     if (it != snap_info.end())
 365       return &it->second;
 366     return nullptr;
 367   }
 368
 369   int ImageCtx::get_snap_name(snap_t in_snap_id,
 370                               string *out_snap_name) const
 371   {
 372     ceph_assert(snap_lock.is_locked());
 373     const SnapInfo *info = get_snap_info(in_snap_id);
 374     if (info) {
 375       *out_snap_name = info->name;
 376       return 0;
 377     }
 378     return -ENOENT;
 379   }
 380
 381   int ImageCtx::get_snap_namespace(snap_t in_snap_id,
 382                                    cls::rbd::SnapshotNamespace *out_snap_namespace) const
 383   {
 384     ceph_assert(snap_lock.is_locked());
 385     const SnapInfo *info = get_snap_info(in_snap_id);
 386     if (info) {
 387       *out_snap_namespace = info->snap_namespace;
 388       return 0;
 389     }
 390     return -ENOENT;
 391   }
 392
 393   int ImageCtx::get_parent_spec(snap_t in_snap_id,
 394                                 cls::rbd::ParentImageSpec *out_pspec) const
 395   {
 396     const SnapInfo *info = get_snap_info(in_snap_id);
 397     if (info) {
 398       *out_pspec = info->parent.spec;
 399       return 0;
 400     }
 401     return -ENOENT;
 402   }
 403
 404   uint64_t ImageCtx::get_current_size() const
 405   {
 406     ceph_assert(snap_lock.is_locked());
 407     return size;
 408   }
 409
 410   uint64_t ImageCtx::get_object_size() const
 411   {
 412     return 1ull << order;
 413   }
 414
 415   string ImageCtx::get_object_name(uint64_t num) const {
 416     char buf[object_prefix.length() + 32];
 417     snprintf(buf, sizeof(buf), format_string, num);
 418     return string(buf);
 419   }
 420
 421   uint64_t ImageCtx::get_stripe_unit() const
 422   {
 423     return stripe_unit;
 424   }
 425
 426   uint64_t ImageCtx::get_stripe_count() const
 427   {
 428     return stripe_count;
 429   }
 430
 431   uint64_t ImageCtx::get_stripe_period() const
 432   {
 433     return stripe_count * (1ull << order);
 434   }
 435
 436   utime_t ImageCtx::get_create_timestamp() const
 437   {
 438     return create_timestamp;
 439   }
 440
 441   utime_t ImageCtx::get_access_timestamp() const
 442   {
 443     return access_timestamp;
 444   }
 445
 446   utime_t ImageCtx::get_modify_timestamp() const
 447   {
 448     return modify_timestamp;
 449   }
 450
 451   void ImageCtx::set_access_timestamp(utime_t at)
 452   {
 453     ceph_assert(timestamp_lock.is_wlocked());
 454     access_timestamp = at;
 455   }
 456
 457   void ImageCtx::set_modify_timestamp(utime_t mt)
 458   {
 459     ceph_assert(timestamp_lock.is_locked());
 460     modify_timestamp = mt;
 461   }
 462
 463   int ImageCtx::is_snap_protected(snap_t in_snap_id,
 464                                   bool *is_protected) const
 465   {
 466     ceph_assert(snap_lock.is_locked());
 467     const SnapInfo *info = get_snap_info(in_snap_id);
 468     if (info) {
 469       *is_protected =
 470         (info->protection_status == RBD_PROTECTION_STATUS_PROTECTED);
 471       return 0;
 472     }
 473     return -ENOENT;
 474   }
 475
 476   int ImageCtx::is_snap_unprotected(snap_t in_snap_id,
 477                                     bool *is_unprotected) const
 478   {
 479     ceph_assert(snap_lock.is_locked());
 480     const SnapInfo *info = get_snap_info(in_snap_id);
 481     if (info) {
 482       *is_unprotected =
 483         (info->protection_status == RBD_PROTECTION_STATUS_UNPROTECTED);
 484       return 0;
 485     }
 486     return -ENOENT;
 487   }
 488
 489   void ImageCtx::add_snap(cls::rbd::SnapshotNamespace in_snap_namespace,
 490                           string in_snap_name,
 491                           snap_t id, uint64_t in_size,
 492                           const ParentImageInfo &parent,
 493                           uint8_t protection_status, uint64_t flags,
 494                           utime_t timestamp)
 495   {
 496     ceph_assert(snap_lock.is_wlocked());
 497     snaps.push_back(id);
 498     SnapInfo info(in_snap_name, in_snap_namespace,
 499                   in_size, parent, protection_status, flags, timestamp);
 500     snap_info.insert({id, info});
 501     snap_ids.insert({{in_snap_namespace, in_snap_name}, id});
 502   }
 503
 504   void ImageCtx::rm_snap(cls::rbd::SnapshotNamespace in_snap_namespace,
 505                          string in_snap_name,
 506                          snap_t id)
 507   {
 508     ceph_assert(snap_lock.is_wlocked());
 509     snaps.erase(std::remove(snaps.begin(), snaps.end(), id), snaps.end());
 510     snap_info.erase(id);
 511     snap_ids.erase({in_snap_namespace, in_snap_name});
 512   }
 513
 514   uint64_t ImageCtx::get_image_size(snap_t in_snap_id) const
 515   {
 516     ceph_assert(snap_lock.is_locked());
 517     if (in_snap_id == CEPH_NOSNAP) {
 518       if (!resize_reqs.empty() &&
 519           resize_reqs.front()->shrinking()) {
 520         return resize_reqs.front()->get_image_size();
 521       }
 522       return size;
 523     }
 524
 525     const SnapInfo *info = get_snap_info(in_snap_id);
 526     if (info) {
 527       return info->size;
 528     }
 529     return 0;
 530   }
 531
 532   uint64_t ImageCtx::get_object_count(snap_t in_snap_id) const {
 533     ceph_assert(snap_lock.is_locked());
 534     uint64_t image_size = get_image_size(in_snap_id);
 535     return Striper::get_num_objects(layout, image_size);
 536   }
 537
 538   bool ImageCtx::test_features(uint64_t features) const
 539   {
 540     RWLock::RLocker l(snap_lock);
 541     return test_features(features, snap_lock);
 542   }
 543
 544   bool ImageCtx::test_features(uint64_t in_features,
 545                                const RWLock &in_snap_lock) const
 546   {
 547     ceph_assert(snap_lock.is_locked());
 548     return ((features & in_features) == in_features);
 549   }
 550
 551   bool ImageCtx::test_op_features(uint64_t in_op_features) const
 552   {
 553     RWLock::RLocker snap_locker(snap_lock);
 554     return test_op_features(in_op_features, snap_lock);
 555   }
 556
 557   bool ImageCtx::test_op_features(uint64_t in_op_features,
 558                                   const RWLock &in_snap_lock) const
 559   {
 560     ceph_assert(snap_lock.is_locked());
 561     return ((op_features & in_op_features) == in_op_features);
 562   }
 563
 564   int ImageCtx::get_flags(librados::snap_t _snap_id, uint64_t *_flags) const
 565   {
 566     ceph_assert(snap_lock.is_locked());
 567     if (_snap_id == CEPH_NOSNAP) {
 568       *_flags = flags;
 569       return 0;
 570     }
 571     const SnapInfo *info = get_snap_info(_snap_id);
 572     if (info) {
 573       *_flags = info->flags;
 574       return 0;
 575     }
 576     return -ENOENT;
 577   }
 578
 579   int ImageCtx::test_flags(librados::snap_t in_snap_id,
 580                            uint64_t flags, bool *flags_set) const
 581   {
 582     RWLock::RLocker l(snap_lock);
 583     return test_flags(in_snap_id, flags, snap_lock, flags_set);
 584   }
 585
 586   int ImageCtx::test_flags(librados::snap_t in_snap_id,
 587                            uint64_t flags, const RWLock &in_snap_lock,
 588                            bool *flags_set) const
 589   {
 590     ceph_assert(snap_lock.is_locked());
 591     uint64_t snap_flags;
 592     int r = get_flags(in_snap_id, &snap_flags);
 593     if (r < 0) {
 594       return r;
 595     }
 596     *flags_set = ((snap_flags & flags) == flags);
 597     return 0;
 598   }
 599
 600   int ImageCtx::update_flags(snap_t in_snap_id, uint64_t flag, bool enabled)
 601   {
 602     ceph_assert(snap_lock.is_wlocked());
 603     uint64_t *_flags;
 604     if (in_snap_id == CEPH_NOSNAP) {
 605       _flags = &flags;
 606     } else {
 607       map<snap_t, SnapInfo>::iterator it = snap_info.find(in_snap_id);
 608       if (it == snap_info.end()) {
 609         return -ENOENT;
 610       }
 611       _flags = &it->second.flags;
 612     }
 613
 614     if (enabled) {
 615       (*_flags) |= flag;
 616     } else {
 617       (*_flags) &= ~flag;
 618     }
 619     return 0;
 620   }
 621
 622   const ParentImageInfo* ImageCtx::get_parent_info(snap_t in_snap_id) const
 623   {
 624     ceph_assert(snap_lock.is_locked());
 625     ceph_assert(parent_lock.is_locked());
 626     if (in_snap_id == CEPH_NOSNAP)
 627       return &parent_md;
 628     const SnapInfo *info = get_snap_info(in_snap_id);
 629     if (info)
 630       return &info->parent;
 631     return NULL;
 632   }
 633
 634   int64_t ImageCtx::get_parent_pool_id(snap_t in_snap_id) const
 635   {
 636     const auto info = get_parent_info(in_snap_id);
 637     if (info)
 638       return info->spec.pool_id;
 639     return -1;
 640   }
 641
 642   string ImageCtx::get_parent_image_id(snap_t in_snap_id) const
 643   {
 644     const auto info = get_parent_info(in_snap_id);
 645     if (info)
 646       return info->spec.image_id;
 647     return "";
 648   }
 649
 650   uint64_t ImageCtx::get_parent_snap_id(snap_t in_snap_id) const
 651   {
 652     const auto info = get_parent_info(in_snap_id);
 653     if (info)
 654       return info->spec.snap_id;
 655     return CEPH_NOSNAP;
 656   }
 657
 658   int ImageCtx::get_parent_overlap(snap_t in_snap_id, uint64_t *overlap) const
 659   {
 660     ceph_assert(snap_lock.is_locked());
 661     const auto info = get_parent_info(in_snap_id);
 662     if (info) {
 663       *overlap = info->overlap;
 664       return 0;
 665     }
 666     return -ENOENT;
 667   }
 668
 669   void ImageCtx::register_watch(Context *on_finish) {
 670     ceph_assert(image_watcher != NULL);
 671     image_watcher->register_watch(on_finish);
 672   }
 673
 674   uint64_t ImageCtx::prune_parent_extents(vector<pair<uint64_t,uint64_t> >& objectx,
 675                                           uint64_t overlap)
 676   {
 677     // drop extents completely beyond the overlap
 678     while (!objectx.empty() && objectx.back().first >= overlap)
 679       objectx.pop_back();
 680
 681     // trim final overlapping extent
 682     if (!objectx.empty() && objectx.back().first + objectx.back().second > overlap)
 683       objectx.back().second = overlap - objectx.back().first;
 684
 685     uint64_t len = 0;
 686     for (vector<pair<uint64_t,uint64_t> >::iterator p = objectx.begin();
 687          p != objectx.end();
 688          ++p)
 689       len += p->second;
 690     ldout(cct, 10) << "prune_parent_extents image overlap " << overlap
 691                    << ", object overlap " << len
 692                    << " from image extents " << objectx << dendl;
 693     return len;
 694   }
 695
 696   void ImageCtx::cancel_async_requests() {
 697     C_SaferCond ctx;
 698     cancel_async_requests(&ctx);
 699     ctx.wait();
 700   }
 701
 702   void ImageCtx::cancel_async_requests(Context *on_finish) {
 703     {
 704       Mutex::Locker async_ops_locker(async_ops_lock);
 705       if (!async_requests.empty()) {
 706         ldout(cct, 10) << "canceling async requests: count="
 707                        << async_requests.size() << dendl;
 708         for (auto req : async_requests) {
 709           ldout(cct, 10) << "canceling async request: " << req << dendl;
 710           req->cancel();
 711         }
 712         async_requests_waiters.push_back(on_finish);
 713         return;
 714       }
 715     }
 716
 717     on_finish->complete(0);
 718   }
 719
 720   void ImageCtx::clear_pending_completions() {
 721     Mutex::Locker l(completed_reqs_lock);
 722     ldout(cct, 10) << "clear pending AioCompletion: count="
 723                    << completed_reqs.size() << dendl;
 724     completed_reqs.clear();
 725   }
 726
 727   void ImageCtx::apply_metadata(const std::map<std::string, bufferlist> &meta,
 728                                 bool thread_safe) {
 729     ldout(cct, 20) << __func__ << dendl;
 730
 731     // reset settings back to global defaults
 732     for (auto& key : config_overrides) {
 733       std::string value;
 734       int r = cct->_conf.get_val(key, &value);
 735       ceph_assert(r == 0);
 736
 737       config.set_val(key, value);
 738     }
 739     config_overrides.clear();
 740
 741     // extract config overrides
 742     for (auto meta_pair : meta) {
 743       if (!boost::starts_with(meta_pair.first, METADATA_CONF_PREFIX)) {
 744         continue;
 745       }
 746
 747       std::string key = meta_pair.first.substr(METADATA_CONF_PREFIX.size());
 748       if (!boost::starts_with(key, "rbd_")) {
 749         // ignore non-RBD configuration keys
 750         // TODO use option schema to determine applicable subsystem
 751         ldout(cct, 0) << __func__ << ": ignoring config " << key << dendl;
 752         continue;
 753       }
 754
 755       if (config.find_option(key) != nullptr) {
 756         std::string val(meta_pair.second.c_str(), meta_pair.second.length());
 757         int r = config.set_val(key, val);
 758         if (r >= 0) {
 759           ldout(cct, 20) << __func__ << ": " << key << "=" << val << dendl;
 760           config_overrides.insert(key);
 761         } else {
 762           lderr(cct) << __func__ << ": failed to set config " << key << " "
 763                      << "with value " << val << ": " << cpp_strerror(r)
 764                      << dendl;
 765         }
 766       }
 767     }
 768
 769 #define ASSIGN_OPTION(param, type)              \
 770     param = config.get_val<type>("rbd_"#param)
 771
 772     bool skip_partial_discard = true;
 773     ASSIGN_OPTION(non_blocking_aio, bool);
 774     ASSIGN_OPTION(cache, bool);
 775     ASSIGN_OPTION(cache_writethrough_until_flush, bool);
 776     ASSIGN_OPTION(cache_max_dirty, Option::size_t);
 777     ASSIGN_OPTION(sparse_read_threshold_bytes, Option::size_t);
 778     ASSIGN_OPTION(readahead_max_bytes, Option::size_t);
 779     ASSIGN_OPTION(readahead_disable_after_bytes, Option::size_t);
 780     ASSIGN_OPTION(clone_copy_on_read, bool);
 781     ASSIGN_OPTION(enable_alloc_hint, bool);
 782     ASSIGN_OPTION(mirroring_replay_delay, uint64_t);
 783     ASSIGN_OPTION(mtime_update_interval, uint64_t);
 784     ASSIGN_OPTION(atime_update_interval, uint64_t);
 785     ASSIGN_OPTION(skip_partial_discard, bool);
 786     ASSIGN_OPTION(discard_granularity_bytes, uint64_t);
 787     ASSIGN_OPTION(blkin_trace_all, bool);
 788
 789 #undef ASSIGN_OPTION
 790
 791     if (sparse_read_threshold_bytes == 0) {
 792       sparse_read_threshold_bytes = get_object_size();
 793     }
 794     if (!skip_partial_discard) {
 795       discard_granularity_bytes = 0;
 796     }
 797
 798     io_work_queue->apply_qos_schedule_tick_min(
 799       config.get_val<uint64_t>("rbd_qos_schedule_tick_min"));
 800
 801     io_work_queue->apply_qos_limit(
 802       RBD_QOS_IOPS_THROTTLE,
 803       config.get_val<uint64_t>("rbd_qos_iops_limit"),
 804       config.get_val<uint64_t>("rbd_qos_iops_burst"));
 805     io_work_queue->apply_qos_limit(
 806       RBD_QOS_BPS_THROTTLE,
 807       config.get_val<uint64_t>("rbd_qos_bps_limit"),
 808       config.get_val<uint64_t>("rbd_qos_bps_burst"));
 809     io_work_queue->apply_qos_limit(
 810       RBD_QOS_READ_IOPS_THROTTLE,
 811       config.get_val<uint64_t>("rbd_qos_read_iops_limit"),
 812       config.get_val<uint64_t>("rbd_qos_read_iops_burst"));
 813     io_work_queue->apply_qos_limit(
 814       RBD_QOS_WRITE_IOPS_THROTTLE,
 815       config.get_val<uint64_t>("rbd_qos_write_iops_limit"),
 816       config.get_val<uint64_t>("rbd_qos_write_iops_burst"));
 817     io_work_queue->apply_qos_limit(
 818       RBD_QOS_READ_BPS_THROTTLE,
 819       config.get_val<uint64_t>("rbd_qos_read_bps_limit"),
 820       config.get_val<uint64_t>("rbd_qos_read_bps_burst"));
 821     io_work_queue->apply_qos_limit(
 822       RBD_QOS_WRITE_BPS_THROTTLE,
 823       config.get_val<uint64_t>("rbd_qos_write_bps_limit"),
 824       config.get_val<uint64_t>("rbd_qos_write_bps_burst"));
 825   }
 826
 827   ExclusiveLock<ImageCtx> *ImageCtx::create_exclusive_lock() {
 828     return new ExclusiveLock<ImageCtx>(*this);
 829   }
 830
 831   ObjectMap<ImageCtx> *ImageCtx::create_object_map(uint64_t snap_id) {
 832     return new ObjectMap<ImageCtx>(*this, snap_id);
 833   }
 834
 835   Journal<ImageCtx> *ImageCtx::create_journal() {
 836     return new Journal<ImageCtx>(*this);
 837   }
 838
 839   void ImageCtx::set_image_name(const std::string &image_name) {
 840     // update the name so rename can be invoked repeatedly
 841     RWLock::RLocker owner_locker(owner_lock);
 842     RWLock::WLocker snap_locker(snap_lock);
 843     name = image_name;
 844     if (old_format) {
 845       header_oid = util::old_header_name(image_name);
 846     }
 847   }
 848
 849   void ImageCtx::notify_update() {
 850     state->handle_update_notification();
 851     ImageWatcher<>::notify_header_update(md_ctx, header_oid);
 852   }
 853
 854   void ImageCtx::notify_update(Context *on_finish) {
 855     state->handle_update_notification();
 856     image_watcher->notify_header_update(on_finish);
 857   }
 858
 859   exclusive_lock::Policy *ImageCtx::get_exclusive_lock_policy() const {
 860     ceph_assert(owner_lock.is_locked());
 861     ceph_assert(exclusive_lock_policy != nullptr);
 862     return exclusive_lock_policy;
 863   }
 864
 865   void ImageCtx::set_exclusive_lock_policy(exclusive_lock::Policy *policy) {
 866     ceph_assert(owner_lock.is_wlocked());
 867     ceph_assert(policy != nullptr);
 868     delete exclusive_lock_policy;
 869     exclusive_lock_policy = policy;
 870   }
 871
 872   journal::Policy *ImageCtx::get_journal_policy() const {
 873     ceph_assert(snap_lock.is_locked());
 874     ceph_assert(journal_policy != nullptr);
 875     return journal_policy;
 876   }
 877
 878   void ImageCtx::set_journal_policy(journal::Policy *policy) {
 879     ceph_assert(snap_lock.is_wlocked());
 880     ceph_assert(policy != nullptr);
 881     delete journal_policy;
 882     journal_policy = policy;
 883   }
 884
 885   bool ImageCtx::is_writeback_cache_enabled() const {
 886     return (cache && cache_max_dirty > 0);
 887   }
 888
 889   void ImageCtx::get_thread_pool_instance(CephContext *cct,
 890                                           ThreadPool **thread_pool,
 891                                           ContextWQ **op_work_queue) {
 892     auto thread_pool_singleton =
 893       &cct->lookup_or_create_singleton_object<ThreadPoolSingleton>(
 894         "librbd::thread_pool", false, cct);
 895     *thread_pool = thread_pool_singleton;
 896     *op_work_queue = thread_pool_singleton->op_work_queue;
 897   }
 898
 899   void ImageCtx::get_timer_instance(CephContext *cct, SafeTimer **timer,
 900                                     Mutex **timer_lock) {
 901     auto safe_timer_singleton =
 902       &cct->lookup_or_create_singleton_object<SafeTimerSingleton>(
 903         "librbd::journal::safe_timer", false, cct);
 904     *timer = safe_timer_singleton;
 905     *timer_lock = &safe_timer_singleton->lock;
 906   }
 907 }