ceph/src/librbd/internal.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include "include/int_types.h"
   4
   5 #include <errno.h>
   6 #include <limits.h>
   7
   8 #include "include/types.h"
   9 #include "include/uuid.h"
  10 #include "common/ceph_context.h"
  11 #include "common/dout.h"
  12 #include "common/errno.h"
  13 #include "common/Throttle.h"
  14 #include "common/event_socket.h"
  15 #include "cls/lock/cls_lock_client.h"
  16 #include "include/stringify.h"
  17
  18 #include "cls/rbd/cls_rbd.h"
  19 #include "cls/rbd/cls_rbd_types.h"
  20 #include "cls/rbd/cls_rbd_client.h"
  21 #include "cls/journal/cls_journal_types.h"
  22 #include "cls/journal/cls_journal_client.h"
  23
  24 #include "librbd/ExclusiveLock.h"
  25 #include "librbd/ImageCtx.h"
  26 #include "librbd/ImageState.h"
  27 #include "librbd/internal.h"
  28 #include "librbd/Journal.h"
  29 #include "librbd/ObjectMap.h"
  30 #include "librbd/Operations.h"
  31 #include "librbd/Types.h"
  32 #include "librbd/Utils.h"
  33 #include "librbd/api/Image.h"
  34 #include "librbd/exclusive_lock/AutomaticPolicy.h"
  35 #include "librbd/exclusive_lock/StandardPolicy.h"
  36 #include "librbd/image/CloneRequest.h"
  37 #include "librbd/image/CreateRequest.h"
  38 #include "librbd/image/RemoveRequest.h"
  39 #include "librbd/io/AioCompletion.h"
  40 #include "librbd/io/ImageRequest.h"
  41 #include "librbd/io/ImageRequestWQ.h"
  42 #include "librbd/io/ObjectRequest.h"
  43 #include "librbd/io/ReadResult.h"
  44 #include "librbd/journal/Types.h"
  45 #include "librbd/managed_lock/Types.h"
  46 #include "librbd/mirror/EnableRequest.h"
  47 #include "librbd/operation/TrimRequest.h"
  48
  49 #include "journal/Journaler.h"
  50
  51 #include <boost/scope_exit.hpp>
  52 #include <boost/variant.hpp>
  53 #include "include/assert.h"
  54
  55 #define dout_subsys ceph_subsys_rbd
  56 #undef dout_prefix
  57 #define dout_prefix *_dout << "librbd: "
  58
  59 #define rbd_howmany(x, y)  (((x) + (y) - 1) / (y))
  60
  61 using std::map;
  62 using std::pair;
  63 using std::set;
  64 using std::string;
  65 using std::vector;
  66 // list binds to list() here, so std::list is explicitly used below
  67
  68 using ceph::bufferlist;
  69 using librados::snap_t;
  70 using librados::IoCtx;
  71 using librados::Rados;
  72
  73 namespace librbd {
  74
  75 namespace {
  76
  77 int validate_pool(IoCtx &io_ctx, CephContext *cct) {
  78   if (!cct->_conf->get_val<bool>("rbd_validate_pool")) {
  79     return 0;
  80   }
  81
  82   int r = io_ctx.stat(RBD_DIRECTORY, NULL, NULL);
  83   if (r == 0) {
  84     return 0;
  85   } else if (r < 0 && r != -ENOENT) {
  86     lderr(cct) << "failed to stat RBD directory: " << cpp_strerror(r) << dendl;
  87     return r;
  88   }
  89
  90   // allocate a self-managed snapshot id if this a new pool to force
  91   // self-managed snapshot mode
  92   uint64_t snap_id;
  93   r = io_ctx.selfmanaged_snap_create(&snap_id);
  94   if (r == -EINVAL) {
  95     lderr(cct) << "pool not configured for self-managed RBD snapshot support"
  96                << dendl;
  97     return r;
  98   } else if (r < 0) {
  99     lderr(cct) << "failed to allocate self-managed snapshot: "
 100                << cpp_strerror(r) << dendl;
 101     return r;
 102   }
 103
 104   r = io_ctx.selfmanaged_snap_remove(snap_id);
 105   if (r < 0) {
 106     lderr(cct) << "failed to release self-managed snapshot " << snap_id
 107                << ": " << cpp_strerror(r) << dendl;
 108   }
 109   return 0;
 110 }
 111
 112
 113 } // anonymous namespace
 114
 115   int detect_format(IoCtx &io_ctx, const string &name,
 116                     bool *old_format, uint64_t *size)
 117   {
 118     CephContext *cct = (CephContext *)io_ctx.cct();
 119     if (old_format)
 120       *old_format = true;
 121     int r = io_ctx.stat(util::old_header_name(name), size, NULL);
 122     if (r == -ENOENT) {
 123       if (old_format)
 124         *old_format = false;
 125       r = io_ctx.stat(util::id_obj_name(name), size, NULL);
 126       if (r < 0)
 127         return r;
 128     } else if (r < 0) {
 129       return r;
 130     }
 131
 132     ldout(cct, 20) << "detect format of " << name << " : "
 133                    << (old_format ? (*old_format ? "old" : "new") :
 134                        "don't care")  << dendl;
 135     return 0;
 136   }
 137
 138   bool has_parent(int64_t parent_pool_id, uint64_t off, uint64_t overlap)
 139   {
 140     return (parent_pool_id != -1 && off <= overlap);
 141   }
 142
 143   void init_rbd_header(struct rbd_obj_header_ondisk& ondisk,
 144                        uint64_t size, int order, uint64_t bid)
 145   {
 146     uint32_t hi = bid >> 32;
 147     uint32_t lo = bid & 0xFFFFFFFF;
 148     uint32_t extra = rand() % 0xFFFFFFFF;
 149     memset(&ondisk, 0, sizeof(ondisk));
 150
 151     memcpy(&ondisk.text, RBD_HEADER_TEXT, sizeof(RBD_HEADER_TEXT));
 152     memcpy(&ondisk.signature, RBD_HEADER_SIGNATURE,
 153            sizeof(RBD_HEADER_SIGNATURE));
 154     memcpy(&ondisk.version, RBD_HEADER_VERSION, sizeof(RBD_HEADER_VERSION));
 155
 156     snprintf(ondisk.block_name, sizeof(ondisk.block_name), "rb.%x.%x.%x",
 157              hi, lo, extra);
 158
 159     ondisk.image_size = size;
 160     ondisk.options.order = order;
 161     ondisk.options.crypt_type = RBD_CRYPT_NONE;
 162     ondisk.options.comp_type = RBD_COMP_NONE;
 163     ondisk.snap_seq = 0;
 164     ondisk.snap_count = 0;
 165     ondisk.reserved = 0;
 166     ondisk.snap_names_len = 0;
 167   }
 168
 169   void image_info(ImageCtx *ictx, image_info_t& info, size_t infosize)
 170   {
 171     int obj_order = ictx->order;
 172     ictx->snap_lock.get_read();
 173     info.size = ictx->get_image_size(ictx->snap_id);
 174     ictx->snap_lock.put_read();
 175     info.obj_size = 1ULL << obj_order;
 176     info.num_objs = Striper::get_num_objects(ictx->layout, info.size);
 177     info.order = obj_order;
 178     strncpy(info.block_name_prefix, ictx->object_prefix.c_str(),
 179             RBD_MAX_BLOCK_NAME_SIZE);
 180     info.block_name_prefix[RBD_MAX_BLOCK_NAME_SIZE - 1] = '\0';
 181
 182     // clear deprecated fields
 183     info.parent_pool = -1L;
 184     info.parent_name[0] = '\0';
 185   }
 186
 187   uint64_t oid_to_object_no(const string& oid, const string& object_prefix)
 188   {
 189     istringstream iss(oid);
 190     // skip object prefix and separator
 191     iss.ignore(object_prefix.length() + 1);
 192     uint64_t num;
 193     iss >> std::hex >> num;
 194     return num;
 195   }
 196
 197   void trim_image(ImageCtx *ictx, uint64_t newsize, ProgressContext& prog_ctx)
 198   {
 199     assert(ictx->owner_lock.is_locked());
 200     assert(ictx->exclusive_lock == nullptr ||
 201            ictx->exclusive_lock->is_lock_owner());
 202
 203     C_SaferCond ctx;
 204     ictx->snap_lock.get_read();
 205     operation::TrimRequest<> *req = operation::TrimRequest<>::create(
 206       *ictx, &ctx, ictx->size, newsize, prog_ctx);
 207     ictx->snap_lock.put_read();
 208     req->send();
 209
 210     int r = ctx.wait();
 211     if (r < 0) {
 212       lderr(ictx->cct) << "warning: failed to remove some object(s): "
 213                        << cpp_strerror(r) << dendl;
 214     }
 215   }
 216
 217   int read_header_bl(IoCtx& io_ctx, const string& header_oid,
 218                      bufferlist& header, uint64_t *ver)
 219   {
 220     int r;
 221     uint64_t off = 0;
 222 #define READ_SIZE 4096
 223     do {
 224       bufferlist bl;
 225       r = io_ctx.read(header_oid, bl, READ_SIZE, off);
 226       if (r < 0)
 227         return r;
 228       header.claim_append(bl);
 229       off += r;
 230     } while (r == READ_SIZE);
 231
 232     if (header.length() < sizeof(RBD_HEADER_TEXT) ||
 233         memcmp(RBD_HEADER_TEXT, header.c_str(), sizeof(RBD_HEADER_TEXT))) {
 234       CephContext *cct = (CephContext *)io_ctx.cct();
 235       lderr(cct) << "unrecognized header format" << dendl;
 236       return -ENXIO;
 237     }
 238
 239     if (ver)
 240       *ver = io_ctx.get_last_version();
 241
 242     return 0;
 243   }
 244
 245   int read_header(IoCtx& io_ctx, const string& header_oid,
 246                   struct rbd_obj_header_ondisk *header, uint64_t *ver)
 247   {
 248     bufferlist header_bl;
 249     int r = read_header_bl(io_ctx, header_oid, header_bl, ver);
 250     if (r < 0)
 251       return r;
 252     if (header_bl.length() < (int)sizeof(*header))
 253       return -EIO;
 254     memcpy(header, header_bl.c_str(), sizeof(*header));
 255
 256     return 0;
 257   }
 258
 259   int tmap_set(IoCtx& io_ctx, const string& imgname)
 260   {
 261     bufferlist cmdbl, emptybl;
 262     __u8 c = CEPH_OSD_TMAP_SET;
 263     ::encode(c, cmdbl);
 264     ::encode(imgname, cmdbl);
 265     ::encode(emptybl, cmdbl);
 266     return io_ctx.tmap_update(RBD_DIRECTORY, cmdbl);
 267   }
 268
 269   int tmap_rm(IoCtx& io_ctx, const string& imgname)
 270   {
 271     bufferlist cmdbl;
 272     __u8 c = CEPH_OSD_TMAP_RM;
 273     ::encode(c, cmdbl);
 274     ::encode(imgname, cmdbl);
 275     return io_ctx.tmap_update(RBD_DIRECTORY, cmdbl);
 276   }
 277
 278   typedef boost::variant<std::string,uint64_t> image_option_value_t;
 279   typedef std::map<int,image_option_value_t> image_options_t;
 280   typedef std::shared_ptr<image_options_t> image_options_ref;
 281
 282   enum image_option_type_t {
 283     STR,
 284     UINT64,
 285   };
 286
 287   const std::map<int, image_option_type_t> IMAGE_OPTIONS_TYPE_MAPPING = {
 288     {RBD_IMAGE_OPTION_FORMAT, UINT64},
 289     {RBD_IMAGE_OPTION_FEATURES, UINT64},
 290     {RBD_IMAGE_OPTION_ORDER, UINT64},
 291     {RBD_IMAGE_OPTION_STRIPE_UNIT, UINT64},
 292     {RBD_IMAGE_OPTION_STRIPE_COUNT, UINT64},
 293     {RBD_IMAGE_OPTION_JOURNAL_ORDER, UINT64},
 294     {RBD_IMAGE_OPTION_JOURNAL_SPLAY_WIDTH, UINT64},
 295     {RBD_IMAGE_OPTION_JOURNAL_POOL, STR},
 296     {RBD_IMAGE_OPTION_FEATURES_SET, UINT64},
 297     {RBD_IMAGE_OPTION_FEATURES_CLEAR, UINT64},
 298     {RBD_IMAGE_OPTION_DATA_POOL, STR},
 299   };
 300
 301   std::string image_option_name(int optname) {
 302     switch (optname) {
 303     case RBD_IMAGE_OPTION_FORMAT:
 304       return "format";
 305     case RBD_IMAGE_OPTION_FEATURES:
 306       return "features";
 307     case RBD_IMAGE_OPTION_ORDER:
 308       return "order";
 309     case RBD_IMAGE_OPTION_STRIPE_UNIT:
 310       return "stripe_unit";
 311     case RBD_IMAGE_OPTION_STRIPE_COUNT:
 312       return "stripe_count";
 313     case RBD_IMAGE_OPTION_JOURNAL_ORDER:
 314       return "journal_order";
 315     case RBD_IMAGE_OPTION_JOURNAL_SPLAY_WIDTH:
 316       return "journal_splay_width";
 317     case RBD_IMAGE_OPTION_JOURNAL_POOL:
 318       return "journal_pool";
 319     case RBD_IMAGE_OPTION_FEATURES_SET:
 320       return "features_set";
 321     case RBD_IMAGE_OPTION_FEATURES_CLEAR:
 322       return "features_clear";
 323     case RBD_IMAGE_OPTION_DATA_POOL:
 324       return "data_pool";
 325     default:
 326       return "unknown (" + stringify(optname) + ")";
 327     }
 328   }
 329
 330   std::ostream &operator<<(std::ostream &os, const ImageOptions &opts) {
 331     os << "[";
 332
 333     const char *delimiter = "";
 334     for (auto &i : IMAGE_OPTIONS_TYPE_MAPPING) {
 335       if (i.second == STR) {
 336         std::string val;
 337         if (opts.get(i.first, &val) == 0) {
 338           os << delimiter << image_option_name(i.first) << "=" << val;
 339           delimiter = ", ";
 340         }
 341       } else if (i.second == UINT64) {
 342         uint64_t val;
 343         if (opts.get(i.first, &val) == 0) {
 344           os << delimiter << image_option_name(i.first) << "=" << val;
 345           delimiter = ", ";
 346         }
 347       }
 348     }
 349
 350     os << "]";
 351
 352     return os;
 353   }
 354
 355   void image_options_create(rbd_image_options_t* opts)
 356   {
 357     image_options_ref* opts_ = new image_options_ref(new image_options_t());
 358
 359     *opts = static_cast<rbd_image_options_t>(opts_);
 360   }
 361
 362   void image_options_create_ref(rbd_image_options_t* opts,
 363                                 rbd_image_options_t orig)
 364   {
 365     image_options_ref* orig_ = static_cast<image_options_ref*>(orig);
 366     image_options_ref* opts_ = new image_options_ref(*orig_);
 367
 368     *opts = static_cast<rbd_image_options_t>(opts_);
 369   }
 370
 371   void image_options_copy(rbd_image_options_t* opts,
 372                           const ImageOptions &orig)
 373   {
 374     image_options_ref* opts_ = new image_options_ref(new image_options_t());
 375
 376     *opts = static_cast<rbd_image_options_t>(opts_);
 377
 378     std::string str_val;
 379     uint64_t uint64_val;
 380     for (auto &i : IMAGE_OPTIONS_TYPE_MAPPING) {
 381       switch (i.second) {
 382       case STR:
 383         if (orig.get(i.first, &str_val) == 0) {
 384           image_options_set(*opts, i.first, str_val);
 385         }
 386         continue;
 387       case UINT64:
 388         if (orig.get(i.first, &uint64_val) == 0) {
 389           image_options_set(*opts, i.first, uint64_val);
 390         }
 391         continue;
 392       }
 393     }
 394   }
 395
 396   void image_options_destroy(rbd_image_options_t opts)
 397   {
 398     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 399
 400     delete opts_;
 401   }
 402
 403   int image_options_set(rbd_image_options_t opts, int optname,
 404                         const std::string& optval)
 405   {
 406     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 407
 408     std::map<int, image_option_type_t>::const_iterator i =
 409       IMAGE_OPTIONS_TYPE_MAPPING.find(optname);
 410
 411     if (i == IMAGE_OPTIONS_TYPE_MAPPING.end() || i->second != STR) {
 412       return -EINVAL;
 413     }
 414
 415     (*opts_->get())[optname] = optval;
 416     return 0;
 417   }
 418
 419   int image_options_set(rbd_image_options_t opts, int optname, uint64_t optval)
 420   {
 421     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 422
 423     std::map<int, image_option_type_t>::const_iterator i =
 424       IMAGE_OPTIONS_TYPE_MAPPING.find(optname);
 425
 426     if (i == IMAGE_OPTIONS_TYPE_MAPPING.end() || i->second != UINT64) {
 427       return -EINVAL;
 428     }
 429
 430     (*opts_->get())[optname] = optval;
 431     return 0;
 432   }
 433
 434   int image_options_get(rbd_image_options_t opts, int optname,
 435                         std::string* optval)
 436   {
 437     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 438
 439     std::map<int, image_option_type_t>::const_iterator i =
 440       IMAGE_OPTIONS_TYPE_MAPPING.find(optname);
 441
 442     if (i == IMAGE_OPTIONS_TYPE_MAPPING.end() || i->second != STR) {
 443       return -EINVAL;
 444     }
 445
 446     image_options_t::const_iterator j = (*opts_)->find(optname);
 447
 448     if (j == (*opts_)->end()) {
 449       return -ENOENT;
 450     }
 451
 452     *optval = boost::get<std::string>(j->second);
 453     return 0;
 454   }
 455
 456   int image_options_get(rbd_image_options_t opts, int optname, uint64_t* optval)
 457   {
 458     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 459
 460     std::map<int, image_option_type_t>::const_iterator i =
 461       IMAGE_OPTIONS_TYPE_MAPPING.find(optname);
 462
 463     if (i == IMAGE_OPTIONS_TYPE_MAPPING.end() || i->second != UINT64) {
 464       return -EINVAL;
 465     }
 466
 467     image_options_t::const_iterator j = (*opts_)->find(optname);
 468
 469     if (j == (*opts_)->end()) {
 470       return -ENOENT;
 471     }
 472
 473     *optval = boost::get<uint64_t>(j->second);
 474     return 0;
 475   }
 476
 477   int image_options_is_set(rbd_image_options_t opts, int optname,
 478                            bool* is_set)
 479   {
 480     if (IMAGE_OPTIONS_TYPE_MAPPING.find(optname) ==
 481           IMAGE_OPTIONS_TYPE_MAPPING.end()) {
 482       return -EINVAL;
 483     }
 484
 485     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 486     *is_set = ((*opts_)->find(optname) != (*opts_)->end());
 487     return 0;
 488   }
 489
 490   int image_options_unset(rbd_image_options_t opts, int optname)
 491   {
 492     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 493
 494     std::map<int, image_option_type_t>::const_iterator i =
 495       IMAGE_OPTIONS_TYPE_MAPPING.find(optname);
 496
 497     if (i == IMAGE_OPTIONS_TYPE_MAPPING.end()) {
 498       assert((*opts_)->find(optname) == (*opts_)->end());
 499       return -EINVAL;
 500     }
 501
 502     image_options_t::const_iterator j = (*opts_)->find(optname);
 503
 504     if (j == (*opts_)->end()) {
 505       return -ENOENT;
 506     }
 507
 508     (*opts_)->erase(j);
 509     return 0;
 510   }
 511
 512   void image_options_clear(rbd_image_options_t opts)
 513   {
 514     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 515
 516     (*opts_)->clear();
 517   }
 518
 519   bool image_options_is_empty(rbd_image_options_t opts)
 520   {
 521     image_options_ref* opts_ = static_cast<image_options_ref*>(opts);
 522
 523     return (*opts_)->empty();
 524   }
 525
 526   int list(IoCtx& io_ctx, vector<string>& names)
 527   {
 528     CephContext *cct = (CephContext *)io_ctx.cct();
 529     ldout(cct, 20) << "list " << &io_ctx << dendl;
 530
 531     bufferlist bl;
 532     int r = io_ctx.read(RBD_DIRECTORY, bl, 0, 0);
 533     if (r < 0) {
 534       if (r == -ENOENT) {
 535         r = 0;
 536       }
 537       return r;
 538     }
 539
 540     // old format images are in a tmap
 541     if (bl.length()) {
 542       bufferlist::iterator p = bl.begin();
 543       bufferlist header;
 544       map<string,bufferlist> m;
 545       ::decode(header, p);
 546       ::decode(m, p);
 547       for (map<string,bufferlist>::iterator q = m.begin(); q != m.end(); ++q) {
 548         names.push_back(q->first);
 549       }
 550     }
 551
 552     map<string, string> images;
 553     r = api::Image<>::list_images(io_ctx, &images);
 554     if (r < 0) {
 555       lderr(cct) << "error listing v2 images: " << cpp_strerror(r) << dendl;
 556       return r;
 557     }
 558     for (const auto& img_pair : images) {
 559       names.push_back(img_pair.first);
 560     }
 561
 562     return 0;
 563   }
 564
 565   int flatten_children(ImageCtx *ictx, const char* snap_name,
 566                        ProgressContext& pctx)
 567   {
 568     CephContext *cct = ictx->cct;
 569     ldout(cct, 20) << "children flatten " << ictx->name << dendl;
 570
 571     int r = ictx->state->refresh_if_required();
 572     if (r < 0) {
 573       return r;
 574     }
 575
 576     RWLock::RLocker l(ictx->snap_lock);
 577     snap_t snap_id = ictx->get_snap_id(cls::rbd::UserSnapshotNamespace(), snap_name);
 578     ParentSpec parent_spec(ictx->md_ctx.get_id(), ictx->id, snap_id);
 579     map< pair<int64_t, string>, set<string> > image_info;
 580
 581     r = api::Image<>::list_children(ictx, parent_spec, &image_info);
 582     if (r < 0) {
 583       return r;
 584     }
 585
 586     size_t size = image_info.size();
 587     if (size == 0)
 588       return 0;
 589
 590     size_t i = 0;
 591     Rados rados(ictx->md_ctx);
 592     for ( auto &info : image_info){
 593       string pool = info.first.second;
 594       IoCtx ioctx;
 595       r = rados.ioctx_create2(info.first.first, ioctx);
 596       if (r < 0) {
 597         lderr(cct) << "Error accessing child image pool " << pool
 598                    << dendl;
 599         return r;
 600       }
 601
 602       for (auto &id_it : info.second) {
 603         ImageCtx *imctx = new ImageCtx("", id_it, NULL, ioctx, false);
 604         int r = imctx->state->open(false);
 605         if (r < 0) {
 606           lderr(cct) << "error opening image: "
 607                      << cpp_strerror(r) << dendl;
 608           return r;
 609         }
 610
 611         if ((imctx->features & RBD_FEATURE_DEEP_FLATTEN) == 0 &&
 612             !imctx->snaps.empty()) {
 613           lderr(cct) << "snapshot in-use by " << pool << "/" << imctx->name
 614                      << dendl;
 615           imctx->state->close();
 616           return -EBUSY;
 617         }
 618
 619         librbd::NoOpProgressContext prog_ctx;
 620         r = imctx->operations->flatten(prog_ctx);
 621         if (r < 0) {
 622           lderr(cct) << "error flattening image: " << pool << "/" << id_it
 623                      << cpp_strerror(r) << dendl;
 624           imctx->state->close();
 625           return r;
 626         }
 627
 628         r = imctx->state->close();
 629         if (r < 0) {
 630           lderr(cct) << "failed to close image: " << cpp_strerror(r) << dendl;
 631           return r;
 632         }
 633       }
 634       pctx.update_progress(++i, size);
 635       assert(i <= size);
 636     }
 637
 638     return 0;
 639   }
 640
 641   int list_children(ImageCtx *ictx, set<pair<string, string> >& names)
 642   {
 643     CephContext *cct = ictx->cct;
 644     ldout(cct, 20) << "children list " << ictx->name << dendl;
 645
 646     int r = ictx->state->refresh_if_required();
 647     if (r < 0) {
 648       return r;
 649     }
 650
 651     RWLock::RLocker l(ictx->snap_lock);
 652     ParentSpec parent_spec(ictx->md_ctx.get_id(), ictx->id, ictx->snap_id);
 653     map< pair<int64_t, string>, set<string> > image_info;
 654
 655     r = api::Image<>::list_children(ictx, parent_spec, &image_info);
 656     if (r < 0) {
 657       return r;
 658     }
 659
 660     Rados rados(ictx->md_ctx);
 661     for ( auto &info : image_info){
 662       IoCtx ioctx;
 663       r = rados.ioctx_create2(info.first.first, ioctx);
 664       if (r < 0) {
 665         lderr(cct) << "Error accessing child image pool " << info.first.second
 666                    << dendl;
 667         return r;
 668       }
 669
 670       for (auto &id_it : info.second) {
 671         string name;
 672         r = cls_client::dir_get_name(&ioctx, RBD_DIRECTORY, id_it, &name);
 673         if (r < 0) {
 674           lderr(cct) << "Error looking up name for image id " << id_it
 675                      << " in pool " << info.first.second << dendl;
 676           return r;
 677         }
 678         names.insert(make_pair(info.first.second, name));
 679       }
 680     }
 681
 682     return 0;
 683   }
 684
 685   int get_snap_namespace(ImageCtx *ictx,
 686                          const char *snap_name,
 687                          cls::rbd::SnapshotNamespace *snap_namespace) {
 688     ldout(ictx->cct, 20) << "get_snap_namespace " << ictx << " " << snap_name
 689                          << dendl;
 690
 691     int r = ictx->state->refresh_if_required();
 692     if (r < 0)
 693       return r;
 694     RWLock::RLocker l(ictx->snap_lock);
 695     snap_t snap_id = ictx->get_snap_id(*snap_namespace, snap_name);
 696     if (snap_id == CEPH_NOSNAP)
 697       return -ENOENT;
 698     r = ictx->get_snap_namespace(snap_id, snap_namespace);
 699     return r;
 700   }
 701
 702   int snap_is_protected(ImageCtx *ictx, const char *snap_name, bool *is_protected)
 703   {
 704     ldout(ictx->cct, 20) << "snap_is_protected " << ictx << " " << snap_name
 705                          << dendl;
 706
 707     int r = ictx->state->refresh_if_required();
 708     if (r < 0)
 709       return r;
 710
 711     RWLock::RLocker l(ictx->snap_lock);
 712     snap_t snap_id = ictx->get_snap_id(cls::rbd::UserSnapshotNamespace(), snap_name);
 713     if (snap_id == CEPH_NOSNAP)
 714       return -ENOENT;
 715     bool is_unprotected;
 716     r = ictx->is_snap_unprotected(snap_id, &is_unprotected);
 717     // consider both PROTECTED or UNPROTECTING to be 'protected',
 718     // since in either state they can't be deleted
 719     *is_protected = !is_unprotected;
 720     return r;
 721   }
 722
 723   int create_v1(IoCtx& io_ctx, const char *imgname, uint64_t size, int order)
 724   {
 725     CephContext *cct = (CephContext *)io_ctx.cct();
 726
 727     ldout(cct, 20) << __func__ << " "  << &io_ctx << " name = " << imgname
 728                    << " size = " << size << " order = " << order << dendl;
 729     int r = validate_pool(io_ctx, cct);
 730     if (r < 0) {
 731       return r;
 732     }
 733
 734     ldout(cct, 2) << "adding rbd image to directory..." << dendl;
 735     r = tmap_set(io_ctx, imgname);
 736     if (r < 0) {
 737       lderr(cct) << "error adding image to directory: " << cpp_strerror(r)
 738                  << dendl;
 739       return r;
 740     }
 741
 742     Rados rados(io_ctx);
 743     uint64_t bid = rados.get_instance_id();
 744
 745     ldout(cct, 2) << "creating rbd image..." << dendl;
 746     struct rbd_obj_header_ondisk header;
 747     init_rbd_header(header, size, order, bid);
 748
 749     bufferlist bl;
 750     bl.append((const char *)&header, sizeof(header));
 751
 752     string header_oid = util::old_header_name(imgname);
 753     r = io_ctx.write(header_oid, bl, bl.length(), 0);
 754     if (r < 0) {
 755       lderr(cct) << "Error writing image header: " << cpp_strerror(r)
 756                  << dendl;
 757       int remove_r = tmap_rm(io_ctx, imgname);
 758       if (remove_r < 0) {
 759         lderr(cct) << "Could not remove image from directory after "
 760                    << "header creation failed: "
 761                    << cpp_strerror(remove_r) << dendl;
 762       }
 763       return r;
 764     }
 765
 766     ldout(cct, 2) << "done." << dendl;
 767     return 0;
 768   }
 769
 770   int create(librados::IoCtx& io_ctx, const char *imgname, uint64_t size,
 771              int *order)
 772   {
 773     uint64_t order_ = *order;
 774     ImageOptions opts;
 775
 776     int r = opts.set(RBD_IMAGE_OPTION_ORDER, order_);
 777     assert(r == 0);
 778
 779     r = create(io_ctx, imgname, "", size, opts, "", "", false);
 780
 781     int r1 = opts.get(RBD_IMAGE_OPTION_ORDER, &order_);
 782     assert(r1 == 0);
 783     *order = order_;
 784
 785     return r;
 786   }
 787
 788   int create(IoCtx& io_ctx, const char *imgname, uint64_t size,
 789              bool old_format, uint64_t features, int *order,
 790              uint64_t stripe_unit, uint64_t stripe_count)
 791   {
 792     if (!order)
 793       return -EINVAL;
 794
 795     uint64_t order_ = *order;
 796     uint64_t format = old_format ? 1 : 2;
 797     ImageOptions opts;
 798     int r;
 799
 800     r = opts.set(RBD_IMAGE_OPTION_FORMAT, format);
 801     assert(r == 0);
 802     r = opts.set(RBD_IMAGE_OPTION_FEATURES, features);
 803     assert(r == 0);
 804     r = opts.set(RBD_IMAGE_OPTION_ORDER, order_);
 805     assert(r == 0);
 806     r = opts.set(RBD_IMAGE_OPTION_STRIPE_UNIT, stripe_unit);
 807     assert(r == 0);
 808     r = opts.set(RBD_IMAGE_OPTION_STRIPE_COUNT, stripe_count);
 809     assert(r == 0);
 810
 811     r = create(io_ctx, imgname, "", size, opts, "", "", false);
 812
 813     int r1 = opts.get(RBD_IMAGE_OPTION_ORDER, &order_);
 814     assert(r1 == 0);
 815     *order = order_;
 816
 817     return r;
 818   }
 819
 820   int create(IoCtx& io_ctx, const std::string &image_name,
 821              const std::string &image_id, uint64_t size,
 822              ImageOptions& opts,
 823              const std::string &non_primary_global_image_id,
 824              const std::string &primary_mirror_uuid,
 825              bool skip_mirror_enable)
 826   {
 827     std::string id(image_id);
 828     if (id.empty()) {
 829       id = util::generate_image_id(io_ctx);
 830     }
 831
 832     CephContext *cct = (CephContext *)io_ctx.cct();
 833     ldout(cct, 10) << __func__ << " name=" << image_name << ", "
 834                    << "id= " << id << ", "
 835                    << "size=" << size << ", opts=" << opts << dendl;
 836
 837     uint64_t format;
 838     if (opts.get(RBD_IMAGE_OPTION_FORMAT, &format) != 0)
 839       format = cct->_conf->get_val<int64_t>("rbd_default_format");
 840     bool old_format = format == 1;
 841
 842     // make sure it doesn't already exist, in either format
 843     int r = detect_format(io_ctx, image_name, NULL, NULL);
 844     if (r != -ENOENT) {
 845       if (r) {
 846         lderr(cct) << "Could not tell if " << image_name << " already exists"
 847                    << dendl;
 848         return r;
 849       }
 850       lderr(cct) << "rbd image " << image_name << " already exists" << dendl;
 851       return -EEXIST;
 852     }
 853
 854     uint64_t order = 0;
 855     if (opts.get(RBD_IMAGE_OPTION_ORDER, &order) != 0 || order == 0) {
 856       order = cct->_conf->get_val<int64_t>("rbd_default_order");
 857     }
 858     r = image::CreateRequest<>::validate_order(cct, order);
 859     if (r < 0) {
 860       return r;
 861     }
 862
 863     if (old_format) {
 864       r = create_v1(io_ctx, image_name.c_str(), size, order);
 865     } else {
 866       ThreadPool *thread_pool;
 867       ContextWQ *op_work_queue;
 868       ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 869
 870       C_SaferCond cond;
 871       image::CreateRequest<> *req = image::CreateRequest<>::create(
 872         io_ctx, image_name, id, size, opts, non_primary_global_image_id,
 873         primary_mirror_uuid, skip_mirror_enable, op_work_queue, &cond);
 874       req->send();
 875
 876       r = cond.wait();
 877     }
 878
 879     int r1 = opts.set(RBD_IMAGE_OPTION_ORDER, order);
 880     assert(r1 == 0);
 881
 882     return r;
 883   }
 884
 885   /*
 886    * Parent may be in different pool, hence different IoCtx
 887    */
 888   int clone(IoCtx& p_ioctx, const char *p_name, const char *p_snap_name,
 889             IoCtx& c_ioctx, const char *c_name,
 890             uint64_t features, int *c_order,
 891             uint64_t stripe_unit, int stripe_count)
 892   {
 893     uint64_t order = *c_order;
 894
 895     ImageOptions opts;
 896     opts.set(RBD_IMAGE_OPTION_FEATURES, features);
 897     opts.set(RBD_IMAGE_OPTION_ORDER, order);
 898     opts.set(RBD_IMAGE_OPTION_STRIPE_UNIT, stripe_unit);
 899     opts.set(RBD_IMAGE_OPTION_STRIPE_COUNT, stripe_count);
 900
 901     int r = clone(p_ioctx, p_name, p_snap_name, c_ioctx, c_name, opts);
 902     opts.get(RBD_IMAGE_OPTION_ORDER, &order);
 903     *c_order = order;
 904     return r;
 905   }
 906
 907   int clone(IoCtx& p_ioctx, const char *p_name, const char *p_snap_name,
 908             IoCtx& c_ioctx, const char *c_name, ImageOptions& c_opts)
 909   {
 910     CephContext *cct = (CephContext *)p_ioctx.cct();
 911     if (p_snap_name == NULL) {
 912       lderr(cct) << "image to be cloned must be a snapshot" << dendl;
 913       return -EINVAL;
 914     }
 915
 916     // make sure parent snapshot exists
 917     ImageCtx *p_imctx = new ImageCtx(p_name, "", p_snap_name, p_ioctx, true);
 918     int r = p_imctx->state->open(false);
 919     if (r < 0) {
 920       lderr(cct) << "error opening parent image: "
 921                  << cpp_strerror(r) << dendl;
 922       return r;
 923     }
 924
 925     r = clone(p_imctx, c_ioctx, c_name, "", c_opts, "", "");
 926
 927     int close_r = p_imctx->state->close();
 928     if (r == 0 && close_r < 0) {
 929       r = close_r;
 930     }
 931
 932     if (r < 0) {
 933       return r;
 934     }
 935     return 0;
 936   }
 937
 938   int clone(ImageCtx *p_imctx, IoCtx& c_ioctx, const std::string &c_name,
 939             const std::string &c_id, ImageOptions& c_opts,
 940             const std::string &non_primary_global_image_id,
 941             const std::string &primary_mirror_uuid)
 942   {
 943     std::string id(c_id);
 944     if (id.empty()) {
 945       id = util::generate_image_id(c_ioctx);
 946     }
 947
 948     CephContext *cct = (CephContext *)c_ioctx.cct();
 949     ldout(cct, 10) << __func__ << " "
 950                    << "c_name=" << c_name << ", "
 951                    << "c_id= " << c_id << ", "
 952                    << "c_opts=" << c_opts << dendl;
 953
 954     ThreadPool *thread_pool;
 955     ContextWQ *op_work_queue;
 956     ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 957
 958     C_SaferCond cond;
 959     auto *req = image::CloneRequest<>::create(
 960       p_imctx, c_ioctx, c_name, id, c_opts,
 961       non_primary_global_image_id, primary_mirror_uuid, op_work_queue, &cond);
 962     req->send();
 963
 964     return cond.wait();
 965   }
 966
 967   int rename(IoCtx& io_ctx, const char *srcname, const char *dstname)
 968   {
 969     CephContext *cct = (CephContext *)io_ctx.cct();
 970     ldout(cct, 20) << "rename " << &io_ctx << " " << srcname << " -> "
 971                    << dstname << dendl;
 972
 973     ImageCtx *ictx = new ImageCtx(srcname, "", "", io_ctx, false);
 974     int r = ictx->state->open(false);
 975     if (r < 0) {
 976       lderr(cct) << "error opening source image: " << cpp_strerror(r) << dendl;
 977       return r;
 978     }
 979     BOOST_SCOPE_EXIT((ictx)) {
 980       ictx->state->close();
 981     } BOOST_SCOPE_EXIT_END
 982
 983     return ictx->operations->rename(dstname);
 984   }
 985
 986   int info(ImageCtx *ictx, image_info_t& info, size_t infosize)
 987   {
 988     ldout(ictx->cct, 20) << "info " << ictx << dendl;
 989
 990     int r = ictx->state->refresh_if_required();
 991     if (r < 0)
 992       return r;
 993
 994     image_info(ictx, info, infosize);
 995     return 0;
 996   }
 997
 998   int get_old_format(ImageCtx *ictx, uint8_t *old)
 999   {
1000     int r = ictx->state->refresh_if_required();
1001     if (r < 0)
1002       return r;
1003     *old = ictx->old_format;
1004     return 0;
1005   }
1006
1007   int get_size(ImageCtx *ictx, uint64_t *size)
1008   {
1009     int r = ictx->state->refresh_if_required();
1010     if (r < 0)
1011       return r;
1012     RWLock::RLocker l2(ictx->snap_lock);
1013     *size = ictx->get_image_size(ictx->snap_id);
1014     return 0;
1015   }
1016
1017   int get_features(ImageCtx *ictx, uint64_t *features)
1018   {
1019     int r = ictx->state->refresh_if_required();
1020     if (r < 0)
1021       return r;
1022     RWLock::RLocker l(ictx->snap_lock);
1023     *features = ictx->features;
1024     return 0;
1025   }
1026
1027   int get_overlap(ImageCtx *ictx, uint64_t *overlap)
1028   {
1029     int r = ictx->state->refresh_if_required();
1030     if (r < 0)
1031       return r;
1032     RWLock::RLocker l(ictx->snap_lock);
1033     RWLock::RLocker l2(ictx->parent_lock);
1034     return ictx->get_parent_overlap(ictx->snap_id, overlap);
1035   }
1036
1037   int get_parent_info(ImageCtx *ictx, string *parent_pool_name,
1038                       string *parent_name, string *parent_id,
1039                       string *parent_snap_name)
1040   {
1041     int r = ictx->state->refresh_if_required();
1042     if (r < 0)
1043       return r;
1044
1045     RWLock::RLocker l(ictx->snap_lock);
1046     RWLock::RLocker l2(ictx->parent_lock);
1047     if (ictx->parent == NULL) {
1048       return -ENOENT;
1049     }
1050
1051     ParentSpec parent_spec;
1052
1053     if (ictx->snap_id == CEPH_NOSNAP) {
1054       parent_spec = ictx->parent_md.spec;
1055     } else {
1056       r = ictx->get_parent_spec(ictx->snap_id, &parent_spec);
1057       if (r < 0) {
1058         lderr(ictx->cct) << "Can't find snapshot id = " << ictx->snap_id
1059                          << dendl;
1060         return r;
1061       }
1062       if (parent_spec.pool_id == -1)
1063         return -ENOENT;
1064     }
1065     if (parent_pool_name) {
1066       Rados rados(ictx->md_ctx);
1067       r = rados.pool_reverse_lookup(parent_spec.pool_id,
1068                                     parent_pool_name);
1069       if (r < 0) {
1070         lderr(ictx->cct) << "error looking up pool name: " << cpp_strerror(r)
1071                          << dendl;
1072         return r;
1073       }
1074     }
1075
1076     if (parent_snap_name) {
1077       RWLock::RLocker l(ictx->parent->snap_lock);
1078       r = ictx->parent->get_snap_name(parent_spec.snap_id,
1079                                       parent_snap_name);
1080       if (r < 0) {
1081         lderr(ictx->cct) << "error finding parent snap name: "
1082                          << cpp_strerror(r) << dendl;
1083         return r;
1084       }
1085     }
1086
1087     if (parent_name) {
1088       RWLock::RLocker snap_locker(ictx->parent->snap_lock);
1089       *parent_name = ictx->parent->name;
1090     }
1091     if (parent_id) {
1092       *parent_id = ictx->parent->id;
1093     }
1094
1095     return 0;
1096   }
1097
1098   int get_flags(ImageCtx *ictx, uint64_t *flags)
1099   {
1100     int r = ictx->state->refresh_if_required();
1101     if (r < 0) {
1102       return r;
1103     }
1104
1105     RWLock::RLocker l2(ictx->snap_lock);
1106     return ictx->get_flags(ictx->snap_id, flags);
1107   }
1108
1109   int set_image_notification(ImageCtx *ictx, int fd, int type)
1110   {
1111     CephContext *cct = ictx->cct;
1112     ldout(cct, 20) << __func__ << " " << ictx << " fd " << fd << " type" << type << dendl;
1113
1114     int r = ictx->state->refresh_if_required();
1115     if (r < 0) {
1116       return r;
1117     }
1118
1119     if (ictx->event_socket.is_valid())
1120       return -EINVAL;
1121     return ictx->event_socket.init(fd, type);
1122   }
1123
1124   int is_exclusive_lock_owner(ImageCtx *ictx, bool *is_owner)
1125   {
1126     CephContext *cct = ictx->cct;
1127     ldout(cct, 20) << __func__ << ": ictx=" << ictx << dendl;
1128     *is_owner = false;
1129
1130     RWLock::RLocker owner_locker(ictx->owner_lock);
1131     if (ictx->exclusive_lock == nullptr) {
1132       return 0;
1133     }
1134
1135     // might have been blacklisted by peer -- ensure we still own
1136     // the lock by pinging the OSD
1137     int r = ictx->exclusive_lock->assert_header_locked();
1138     if (r == -EBUSY || r == -ENOENT) {
1139       return 0;
1140     } else if (r < 0) {
1141       return r;
1142     }
1143
1144     *is_owner = true;
1145     return 0;
1146   }
1147
1148   int lock_acquire(ImageCtx *ictx, rbd_lock_mode_t lock_mode)
1149   {
1150     CephContext *cct = ictx->cct;
1151     ldout(cct, 20) << __func__ << ": ictx=" << ictx << ", "
1152                    << "lock_mode=" << lock_mode << dendl;
1153
1154     if (lock_mode != RBD_LOCK_MODE_EXCLUSIVE) {
1155       return -EOPNOTSUPP;
1156     }
1157
1158     C_SaferCond lock_ctx;
1159     {
1160       RWLock::WLocker l(ictx->owner_lock);
1161
1162       if (ictx->exclusive_lock == nullptr) {
1163         lderr(cct) << "exclusive-lock feature is not enabled" << dendl;
1164         return -EINVAL;
1165       }
1166
1167       if (ictx->get_exclusive_lock_policy()->may_auto_request_lock()) {
1168         ictx->set_exclusive_lock_policy(
1169           new exclusive_lock::StandardPolicy(ictx));
1170       }
1171
1172       if (ictx->exclusive_lock->is_lock_owner()) {
1173         return 0;
1174       }
1175
1176       ictx->exclusive_lock->acquire_lock(&lock_ctx);
1177     }
1178
1179     int r = lock_ctx.wait();
1180     if (r < 0) {
1181       lderr(cct) << "failed to request exclusive lock: " << cpp_strerror(r)
1182                  << dendl;
1183       return r;
1184     }
1185
1186     RWLock::RLocker l(ictx->owner_lock);
1187     if (ictx->exclusive_lock == nullptr) {
1188       return -EINVAL;
1189     } else if (!ictx->exclusive_lock->is_lock_owner()) {
1190       lderr(cct) << "failed to acquire exclusive lock" << dendl;
1191       return ictx->exclusive_lock->get_unlocked_op_error();
1192     }
1193
1194     return 0;
1195   }
1196
1197   int lock_release(ImageCtx *ictx)
1198   {
1199     CephContext *cct = ictx->cct;
1200     ldout(cct, 20) << __func__ << ": ictx=" << ictx << dendl;
1201
1202     C_SaferCond lock_ctx;
1203     {
1204       RWLock::WLocker l(ictx->owner_lock);
1205
1206       if (ictx->exclusive_lock == nullptr ||
1207           !ictx->exclusive_lock->is_lock_owner()) {
1208         lderr(cct) << "not exclusive lock owner" << dendl;
1209         return -EINVAL;
1210       }
1211
1212       ictx->exclusive_lock->release_lock(&lock_ctx);
1213     }
1214
1215     int r = lock_ctx.wait();
1216     if (r < 0) {
1217       lderr(cct) << "failed to release exclusive lock: " << cpp_strerror(r)
1218                  << dendl;
1219       return r;
1220     }
1221     return 0;
1222   }
1223
1224   int lock_get_owners(ImageCtx *ictx, rbd_lock_mode_t *lock_mode,
1225                       std::list<std::string> *lock_owners)
1226   {
1227     CephContext *cct = ictx->cct;
1228     ldout(cct, 20) << __func__ << ": ictx=" << ictx << dendl;
1229
1230     if (!ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
1231       lderr(cct) << "exclusive-lock feature is not enabled" << dendl;
1232       return -EINVAL;
1233     }
1234
1235     managed_lock::Locker locker;
1236     C_SaferCond get_owner_ctx;
1237     ExclusiveLock<>(*ictx).get_locker(&locker, &get_owner_ctx);
1238     int r = get_owner_ctx.wait();
1239     if (r == -ENOENT) {
1240       return r;
1241     } else if (r < 0) {
1242       lderr(cct) << "failed to determine current lock owner: "
1243                  << cpp_strerror(r) << dendl;
1244       return r;
1245     }
1246
1247     *lock_mode = RBD_LOCK_MODE_EXCLUSIVE;
1248     lock_owners->clear();
1249     lock_owners->emplace_back(locker.address);
1250     return 0;
1251   }
1252
1253   int lock_break(ImageCtx *ictx, rbd_lock_mode_t lock_mode,
1254                  const std::string &lock_owner)
1255   {
1256     CephContext *cct = ictx->cct;
1257     ldout(cct, 20) << __func__ << ": ictx=" << ictx << ", "
1258                    << "lock_mode=" << lock_mode << ", "
1259                    << "lock_owner=" << lock_owner << dendl;
1260
1261     if (lock_mode != RBD_LOCK_MODE_EXCLUSIVE) {
1262       return -EOPNOTSUPP;
1263     }
1264
1265     if (ictx->read_only) {
1266       return -EROFS;
1267     }
1268
1269     managed_lock::Locker locker;
1270     C_SaferCond get_owner_ctx;
1271     {
1272       RWLock::RLocker l(ictx->owner_lock);
1273
1274       if (ictx->exclusive_lock == nullptr) {
1275         lderr(cct) << "exclusive-lock feature is not enabled" << dendl;
1276         return -EINVAL;
1277       }
1278
1279       ictx->exclusive_lock->get_locker(&locker, &get_owner_ctx);
1280     }
1281     int r = get_owner_ctx.wait();
1282     if (r == -ENOENT) {
1283       return r;
1284     } else if (r < 0) {
1285       lderr(cct) << "failed to determine current lock owner: "
1286                  << cpp_strerror(r) << dendl;
1287       return r;
1288     }
1289
1290     if (locker.address != lock_owner) {
1291       return -EBUSY;
1292     }
1293
1294     C_SaferCond break_ctx;
1295     {
1296       RWLock::RLocker l(ictx->owner_lock);
1297
1298       if (ictx->exclusive_lock == nullptr) {
1299         lderr(cct) << "exclusive-lock feature is not enabled" << dendl;
1300         return -EINVAL;
1301       }
1302
1303       ictx->exclusive_lock->break_lock(locker, true, &break_ctx);
1304     }
1305     r = break_ctx.wait();
1306     if (r == -ENOENT) {
1307       return r;
1308     } else if (r < 0) {
1309       lderr(cct) << "failed to break lock: " << cpp_strerror(r) << dendl;
1310       return r;
1311     }
1312     return 0;
1313   }
1314
1315   int remove(IoCtx& io_ctx, const std::string &image_name,
1316              const std::string &image_id, ProgressContext& prog_ctx,
1317              bool force, bool from_trash_remove)
1318   {
1319     CephContext *cct((CephContext *)io_ctx.cct());
1320     ldout(cct, 20) << "remove " << &io_ctx << " "
1321                    << (image_id.empty() ? image_name : image_id) << dendl;
1322
1323     ThreadPool *thread_pool;
1324     ContextWQ *op_work_queue;
1325     ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
1326
1327     C_SaferCond cond;
1328     auto req = librbd::image::RemoveRequest<>::create(
1329       io_ctx, image_name, image_id, force, from_trash_remove, prog_ctx,
1330       op_work_queue, &cond);
1331     req->send();
1332
1333     return cond.wait();
1334   }
1335
1336   int trash_move(librados::IoCtx &io_ctx, rbd_trash_image_source_t source,
1337                  const std::string &image_name, uint64_t delay) {
1338     CephContext *cct((CephContext *)io_ctx.cct());
1339     ldout(cct, 20) << "trash_move " << &io_ctx << " " << image_name
1340                    << dendl;
1341
1342     std::string image_id;
1343     ImageCtx *ictx = new ImageCtx(image_name, "", nullptr, io_ctx, false);
1344     int r = ictx->state->open(true);
1345     if (r < 0) {
1346       ictx = nullptr;
1347
1348       if (r != -ENOENT) {
1349         ldout(cct, 2) << "error opening image: " << cpp_strerror(-r) << dendl;
1350         return r;
1351       }
1352
1353       // try to get image id from the directory
1354       r = cls_client::dir_get_id(&io_ctx, RBD_DIRECTORY, image_name, &image_id);
1355       if (r < 0) {
1356         if (r != -ENOENT) {
1357           ldout(cct, 2) << "error reading image id from dirctory: "
1358                         << cpp_strerror(-r) << dendl;
1359         }
1360         return r;
1361       }
1362     } else {
1363       if (ictx->old_format) {
1364         ictx->state->close();
1365         return -EOPNOTSUPP;
1366       }
1367
1368       image_id = ictx->id;
1369       ictx->owner_lock.get_read();
1370       if (ictx->exclusive_lock != nullptr) {
1371         r = ictx->operations->prepare_image_update(false);
1372         if (r < 0) {
1373           lderr(cct) << "cannot obtain exclusive lock - not removing" << dendl;
1374           ictx->owner_lock.put_read();
1375           ictx->state->close();
1376           return -EBUSY;
1377         }
1378       }
1379     }
1380
1381     BOOST_SCOPE_EXIT_ALL(ictx, cct) {
1382       if (ictx == nullptr)
1383         return;
1384
1385       bool is_locked = ictx->exclusive_lock != nullptr &&
1386                        ictx->exclusive_lock->is_lock_owner();
1387       if (is_locked) {
1388         C_SaferCond ctx;
1389         auto exclusive_lock = ictx->exclusive_lock;
1390         exclusive_lock->shut_down(&ctx);
1391         ictx->owner_lock.put_read();
1392         int r = ctx.wait();
1393         if (r < 0) {
1394           lderr(cct) << "error shutting down exclusive lock" << dendl;
1395         }
1396         delete exclusive_lock;
1397       } else {
1398         ictx->owner_lock.put_read();
1399       }
1400       ictx->state->close();
1401     };
1402
1403     ldout(cct, 2) << "adding image entry to rbd_trash" << dendl;
1404     utime_t ts = ceph_clock_now();
1405     utime_t deferment_end_time = ts;
1406     deferment_end_time += (double)delay;
1407     cls::rbd::TrashImageSource trash_source =
1408         static_cast<cls::rbd::TrashImageSource>(source);
1409     cls::rbd::TrashImageSpec trash_spec(trash_source, image_name, ts,
1410                                         deferment_end_time);
1411     r = cls_client::trash_add(&io_ctx, image_id, trash_spec);
1412     if (r < 0 && r != -EEXIST) {
1413       lderr(cct) << "error adding image " << image_name << " to rbd_trash"
1414                  << dendl;
1415       return r;
1416     } else if (r == -EEXIST) {
1417       ldout(cct, 10) << "found previous unfinished deferred remove for image:"
1418                      << image_id << dendl;
1419       // continue with removing image from directory
1420     }
1421
1422     ldout(cct, 2) << "removing id object..." << dendl;
1423     r = io_ctx.remove(util::id_obj_name(image_name));
1424     if (r < 0 && r != -ENOENT) {
1425       lderr(cct) << "error removing id object: " << cpp_strerror(r)
1426                  << dendl;
1427       return r;
1428     }
1429
1430     ldout(cct, 2) << "removing rbd image from v2 directory..." << dendl;
1431     r = cls_client::dir_remove_image(&io_ctx, RBD_DIRECTORY, image_name,
1432                                      image_id);
1433     if (r < 0) {
1434       if (r != -ENOENT) {
1435         lderr(cct) << "error removing image from v2 directory: "
1436                    << cpp_strerror(-r) << dendl;
1437       }
1438       return r;
1439     }
1440
1441     return 0;
1442   }
1443
1444   int trash_get(IoCtx &io_ctx, const std::string &id,
1445                 trash_image_info_t *info) {
1446     CephContext *cct((CephContext *)io_ctx.cct());
1447     ldout(cct, 20) << __func__ << " " << &io_ctx << dendl;
1448
1449     cls::rbd::TrashImageSpec spec;
1450     int r = cls_client::trash_get(&io_ctx, id, &spec);
1451     if (r == -ENOENT) {
1452       return r;
1453     } else if (r < 0) {
1454       lderr(cct) << "error retrieving trash entry: " << cpp_strerror(r)
1455                  << dendl;
1456       return r;
1457     }
1458
1459     rbd_trash_image_source_t source = static_cast<rbd_trash_image_source_t>(
1460       spec.source);
1461     *info = trash_image_info_t{id, spec.name, source, spec.deletion_time.sec(),
1462                                spec.deferment_end_time.sec()};
1463     return 0;
1464   }
1465
1466   int trash_list(IoCtx &io_ctx, vector<trash_image_info_t> &entries) {
1467     CephContext *cct((CephContext *)io_ctx.cct());
1468     ldout(cct, 20) << "trash_list " << &io_ctx << dendl;
1469
1470     bool more_entries;
1471     uint32_t max_read = 1024;
1472     std::string last_read = "";
1473     do {
1474       map<string, cls::rbd::TrashImageSpec> trash_entries;
1475       int r = cls_client::trash_list(&io_ctx, last_read, max_read,
1476                                      &trash_entries);
1477       if (r < 0 && r != -ENOENT) {
1478         lderr(cct) << "error listing rbd trash entries: " << cpp_strerror(r)
1479                    << dendl;
1480         return r;
1481       } else if (r == -ENOENT) {
1482         break;
1483       }
1484
1485       if (trash_entries.empty()) {
1486         break;
1487       }
1488
1489       for (const auto &entry : trash_entries) {
1490         rbd_trash_image_source_t source =
1491             static_cast<rbd_trash_image_source_t>(entry.second.source);
1492         entries.push_back({entry.first, entry.second.name, source,
1493                            entry.second.deletion_time.sec(),
1494                            entry.second.deferment_end_time.sec()});
1495       }
1496       last_read = trash_entries.rbegin()->first;
1497       more_entries = (trash_entries.size() >= max_read);
1498     } while (more_entries);
1499
1500     return 0;
1501   }
1502
1503   int trash_remove(IoCtx &io_ctx, const std::string &image_id, bool force,
1504                    ProgressContext& prog_ctx) {
1505     CephContext *cct((CephContext *)io_ctx.cct());
1506     ldout(cct, 20) << "trash_remove " << &io_ctx << " " << image_id
1507                    << " " << force << dendl;
1508
1509     cls::rbd::TrashImageSpec trash_spec;
1510     int r = cls_client::trash_get(&io_ctx, image_id, &trash_spec);
1511     if (r < 0) {
1512       lderr(cct) << "error getting image id " << image_id
1513                  << " info from trash: " << cpp_strerror(r) << dendl;
1514       return r;
1515     }
1516
1517     utime_t now = ceph_clock_now();
1518     if (now < trash_spec.deferment_end_time && !force) {
1519       lderr(cct) << "error: deferment time has not expired." << dendl;
1520       return -EPERM;
1521     }
1522
1523     r = remove(io_ctx, "", image_id, prog_ctx, false, true);
1524     if (r < 0) {
1525       lderr(cct) << "error removing image " << image_id
1526                  << ", which is pending deletion" << dendl;
1527       return r;
1528     }
1529     r = cls_client::trash_remove(&io_ctx, image_id);
1530     if (r < 0 && r != -ENOENT) {
1531       lderr(cct) << "error removing image " << image_id
1532                  << " from rbd_trash object" << dendl;
1533       return r;
1534     }
1535     return 0;
1536   }
1537
1538   int trash_restore(librados::IoCtx &io_ctx, const std::string &image_id,
1539                     const std::string &image_new_name) {
1540     CephContext *cct((CephContext *)io_ctx.cct());
1541     ldout(cct, 20) << "trash_restore " << &io_ctx << " " << image_id << " "
1542                    << image_new_name << dendl;
1543
1544     cls::rbd::TrashImageSpec trash_spec;
1545     int r = cls_client::trash_get(&io_ctx, image_id, &trash_spec);
1546     if (r < 0) {
1547       lderr(cct) << "error getting image id " << image_id
1548                  << " info from trash: " << cpp_strerror(r) << dendl;
1549       return r;
1550     }
1551
1552     std::string image_name = image_new_name;
1553     if (image_name.empty()) {
1554       // if user didn't specify a new name, let's try using the old name
1555       image_name = trash_spec.name;
1556       ldout(cct, 20) << "restoring image id " << image_id << " with name "
1557                      << image_name << dendl;
1558     }
1559
1560     // check if no image exists with the same name
1561     bool create_id_obj = true;
1562     std::string existing_id;
1563     r = cls_client::get_id(&io_ctx, util::id_obj_name(image_name), &existing_id);
1564     if (r < 0 && r != -ENOENT) {
1565       lderr(cct) << "error checking if image " << image_name << " exists: "
1566                  << cpp_strerror(r) << dendl;
1567       return r;
1568     } else if (r != -ENOENT){
1569       // checking if we are recovering from an incomplete restore
1570       if (existing_id != image_id) {
1571         ldout(cct, 2) << "an image with the same name already exists" << dendl;
1572         return -EEXIST;
1573       }
1574       create_id_obj = false;
1575     }
1576
1577     if (create_id_obj) {
1578       ldout(cct, 2) << "adding id object" << dendl;
1579       librados::ObjectWriteOperation op;
1580       op.create(true);
1581       cls_client::set_id(&op, image_id);
1582       r = io_ctx.operate(util::id_obj_name(image_name), &op);
1583       if (r < 0) {
1584         lderr(cct) << "error adding id object for image " << image_name
1585                    << ": " << cpp_strerror(r) << dendl;
1586         return r;
1587       }
1588     }
1589
1590     ldout(cct, 2) << "adding rbd image from v2 directory..." << dendl;
1591     r = cls_client::dir_add_image(&io_ctx, RBD_DIRECTORY, image_name,
1592                                   image_id);
1593     if (r < 0 && r != -EEXIST) {
1594       lderr(cct) << "error adding image to v2 directory: "
1595                  << cpp_strerror(r) << dendl;
1596       return r;
1597     }
1598
1599     ldout(cct, 2) << "removing image from trash..." << dendl;
1600     r = cls_client::trash_remove(&io_ctx, image_id);
1601     if (r < 0 && r != -ENOENT) {
1602       lderr(cct) << "error removing image id " << image_id << " from trash: "
1603                  << cpp_strerror(r) << dendl;
1604       return r;
1605     }
1606
1607     return 0;
1608   }
1609
1610   int snap_list(ImageCtx *ictx, vector<snap_info_t>& snaps)
1611   {
1612     ldout(ictx->cct, 20) << "snap_list " << ictx << dendl;
1613
1614     int r = ictx->state->refresh_if_required();
1615     if (r < 0)
1616       return r;
1617
1618     RWLock::RLocker l(ictx->snap_lock);
1619     for (map<snap_t, SnapInfo>::iterator it = ictx->snap_info.begin();
1620          it != ictx->snap_info.end(); ++it) {
1621       snap_info_t info;
1622       info.name = it->second.name;
1623       info.id = it->first;
1624       info.size = it->second.size;
1625       snaps.push_back(info);
1626     }
1627
1628     return 0;
1629   }
1630
1631   int snap_exists(ImageCtx *ictx, const cls::rbd::SnapshotNamespace& snap_namespace,
1632                   const char *snap_name, bool *exists)
1633   {
1634     ldout(ictx->cct, 20) << "snap_exists " << ictx << " " << snap_name << dendl;
1635
1636     int r = ictx->state->refresh_if_required();
1637     if (r < 0)
1638       return r;
1639
1640     RWLock::RLocker l(ictx->snap_lock);
1641     *exists = ictx->get_snap_id(snap_namespace, snap_name) != CEPH_NOSNAP;
1642     return 0;
1643   }
1644
1645   int snap_remove(ImageCtx *ictx, const char *snap_name, uint32_t flags,
1646                   ProgressContext& pctx)
1647   {
1648     ldout(ictx->cct, 20) << "snap_remove " << ictx << " " << snap_name << " flags: " << flags << dendl;
1649
1650     int r = 0;
1651
1652     r = ictx->state->refresh_if_required();
1653     if (r < 0)
1654       return r;
1655
1656     if (flags & RBD_SNAP_REMOVE_FLATTEN) {
1657         r = flatten_children(ictx, snap_name, pctx);
1658         if (r < 0) {
1659           return r;
1660         }
1661     }
1662
1663     bool is_protected;
1664     r = snap_is_protected(ictx, snap_name, &is_protected);
1665     if (r < 0) {
1666       return r;
1667     }
1668
1669     if (is_protected && flags & RBD_SNAP_REMOVE_UNPROTECT) {
1670       r = ictx->operations->snap_unprotect(cls::rbd::UserSnapshotNamespace(), snap_name);
1671       if (r < 0) {
1672         lderr(ictx->cct) << "failed to unprotect snapshot: " << snap_name << dendl;
1673         return r;
1674       }
1675
1676       r = snap_is_protected(ictx, snap_name, &is_protected);
1677       if (r < 0) {
1678         return r;
1679       }
1680       if (is_protected) {
1681         lderr(ictx->cct) << "snapshot is still protected after unprotection" << dendl;
1682         ceph_abort();
1683       }
1684     }
1685
1686     C_SaferCond ctx;
1687     ictx->operations->snap_remove(cls::rbd::UserSnapshotNamespace(), snap_name, &ctx);
1688
1689     r = ctx.wait();
1690     return r;
1691   }
1692
1693   int snap_get_timestamp(ImageCtx *ictx, uint64_t snap_id, struct timespec *timestamp)
1694   {
1695     std::map<librados::snap_t, SnapInfo>::iterator snap_it = ictx->snap_info.find(snap_id);
1696     assert(snap_it != ictx->snap_info.end());
1697     utime_t time = snap_it->second.timestamp;
1698     time.to_timespec(timestamp);
1699     return 0;
1700   }
1701
1702   int snap_get_limit(ImageCtx *ictx, uint64_t *limit)
1703   {
1704     int r = cls_client::snapshot_get_limit(&ictx->md_ctx, ictx->header_oid,
1705                                            limit);
1706     if (r == -EOPNOTSUPP) {
1707       *limit = UINT64_MAX;
1708       r = 0;
1709     }
1710     return r;
1711   }
1712
1713   int snap_set_limit(ImageCtx *ictx, uint64_t limit)
1714   {
1715     return ictx->operations->snap_set_limit(limit);
1716   }
1717
1718   struct CopyProgressCtx {
1719     explicit CopyProgressCtx(ProgressContext &p)
1720       : destictx(NULL), src_size(0), prog_ctx(p)
1721     { }
1722
1723     ImageCtx *destictx;
1724     uint64_t src_size;
1725     ProgressContext &prog_ctx;
1726   };
1727
1728   int copy(ImageCtx *src, IoCtx& dest_md_ctx, const char *destname,
1729            ImageOptions& opts, ProgressContext &prog_ctx, size_t sparse_size)
1730   {
1731     CephContext *cct = (CephContext *)dest_md_ctx.cct();
1732     ldout(cct, 20) << "copy " << src->name
1733                    << (src->snap_name.length() ? "@" + src->snap_name : "")
1734                    << " -> " << destname << " opts = " << opts << dendl;
1735
1736     src->snap_lock.get_read();
1737     uint64_t features = src->features;
1738     uint64_t src_size = src->get_image_size(src->snap_id);
1739     src->snap_lock.put_read();
1740     uint64_t format = src->old_format ? 1 : 2;
1741     if (opts.get(RBD_IMAGE_OPTION_FORMAT, &format) != 0) {
1742       opts.set(RBD_IMAGE_OPTION_FORMAT, format);
1743     }
1744     uint64_t stripe_unit = src->stripe_unit;
1745     if (opts.get(RBD_IMAGE_OPTION_STRIPE_UNIT, &stripe_unit) != 0) {
1746       opts.set(RBD_IMAGE_OPTION_STRIPE_UNIT, stripe_unit);
1747     }
1748     uint64_t stripe_count = src->stripe_count;
1749     if (opts.get(RBD_IMAGE_OPTION_STRIPE_COUNT, &stripe_count) != 0) {
1750       opts.set(RBD_IMAGE_OPTION_STRIPE_COUNT, stripe_count);
1751     }
1752     uint64_t order = src->order;
1753     if (opts.get(RBD_IMAGE_OPTION_ORDER, &order) != 0) {
1754       opts.set(RBD_IMAGE_OPTION_ORDER, order);
1755     }
1756     if (opts.get(RBD_IMAGE_OPTION_FEATURES, &features) != 0) {
1757       opts.set(RBD_IMAGE_OPTION_FEATURES, features);
1758     }
1759     if (features & ~RBD_FEATURES_ALL) {
1760       lderr(cct) << "librbd does not support requested features" << dendl;
1761       return -ENOSYS;
1762     }
1763
1764     int r = create(dest_md_ctx, destname, "", src_size, opts, "", "", false);
1765     if (r < 0) {
1766       lderr(cct) << "header creation failed" << dendl;
1767       return r;
1768     }
1769     opts.set(RBD_IMAGE_OPTION_ORDER, static_cast<uint64_t>(order));
1770
1771     ImageCtx *dest = new librbd::ImageCtx(destname, "", NULL,
1772                                           dest_md_ctx, false);
1773     r = dest->state->open(false);
1774     if (r < 0) {
1775       lderr(cct) << "failed to read newly created header" << dendl;
1776       return r;
1777     }
1778
1779     r = copy(src, dest, prog_ctx, sparse_size);
1780
1781     int close_r = dest->state->close();
1782     if (r == 0 && close_r < 0) {
1783       r = close_r;
1784     }
1785     return r;
1786   }
1787
1788   class C_CopyWrite : public Context {
1789   public:
1790     C_CopyWrite(bufferlist *bl, Context* ctx)
1791       : m_bl(bl), m_ctx(ctx) {}
1792     void finish(int r) override {
1793       delete m_bl;
1794       m_ctx->complete(r);
1795     }
1796   private:
1797     bufferlist *m_bl;
1798     Context *m_ctx;
1799   };
1800
1801   class C_CopyRead : public Context {
1802   public:
1803     C_CopyRead(SimpleThrottle *throttle, ImageCtx *dest, uint64_t offset,
1804                bufferlist *bl, size_t sparse_size)
1805       : m_throttle(throttle), m_dest(dest), m_offset(offset), m_bl(bl),
1806       m_sparse_size(sparse_size) {
1807       m_throttle->start_op();
1808     }
1809     void finish(int r) override {
1810       if (r < 0) {
1811         lderr(m_dest->cct) << "error reading from source image at offset "
1812                            << m_offset << ": " << cpp_strerror(r) << dendl;
1813         delete m_bl;
1814         m_throttle->end_op(r);
1815         return;
1816       }
1817       assert(m_bl->length() == (size_t)r);
1818
1819       if (m_bl->is_zero()) {
1820         delete m_bl;
1821         m_throttle->end_op(r);
1822         return;
1823       }
1824
1825       if (!m_sparse_size) {
1826         m_sparse_size = (1 << m_dest->order);
1827       }
1828
1829       auto *throttle = m_throttle;
1830       auto *end_op_ctx = new FunctionContext([throttle](int r) {
1831         throttle->end_op(r);
1832       });
1833       auto gather_ctx = new C_Gather(m_dest->cct, end_op_ctx);
1834
1835       bufferptr m_ptr(m_bl->length());
1836       m_bl->rebuild(m_ptr);
1837       size_t write_offset = 0;
1838       size_t write_length = 0;
1839       size_t offset = 0;
1840       size_t length = m_bl->length();
1841       while (offset < length) {
1842         if (util::calc_sparse_extent(m_ptr,
1843                                      m_sparse_size,
1844                                      length,
1845                                      &write_offset,
1846                                      &write_length,
1847                                      &offset)) {
1848           bufferptr write_ptr(m_ptr, write_offset, write_length);
1849           bufferlist *write_bl = new bufferlist();
1850           write_bl->push_back(write_ptr);
1851           Context *ctx = new C_CopyWrite(write_bl, gather_ctx->new_sub());
1852           auto comp = io::AioCompletion::create(ctx);
1853
1854           // coordinate through AIO WQ to ensure lock is acquired if needed
1855           m_dest->io_work_queue->aio_write(comp, m_offset + write_offset,
1856                                            write_length,
1857                                            std::move(*write_bl),
1858                                            LIBRADOS_OP_FLAG_FADVISE_DONTNEED,
1859                                            std::move(read_trace));
1860           write_offset = offset;
1861           write_length = 0;
1862         }
1863       }
1864       delete m_bl;
1865       assert(gather_ctx->get_sub_created_count() > 0);
1866       gather_ctx->activate();
1867     }
1868
1869     ZTracer::Trace read_trace;
1870
1871   private:
1872     SimpleThrottle *m_throttle;
1873     ImageCtx *m_dest;
1874     uint64_t m_offset;
1875     bufferlist *m_bl;
1876     size_t m_sparse_size;
1877   };
1878
1879   int copy(ImageCtx *src, ImageCtx *dest, ProgressContext &prog_ctx, size_t sparse_size)
1880   {
1881     src->snap_lock.get_read();
1882     uint64_t src_size = src->get_image_size(src->snap_id);
1883     src->snap_lock.put_read();
1884
1885     dest->snap_lock.get_read();
1886     uint64_t dest_size = dest->get_image_size(dest->snap_id);
1887     dest->snap_lock.put_read();
1888
1889     CephContext *cct = src->cct;
1890     if (dest_size < src_size) {
1891       lderr(cct) << " src size " << src_size << " > dest size "
1892                  << dest_size << dendl;
1893       return -EINVAL;
1894     }
1895     int r;
1896     const uint32_t MAX_KEYS = 64;
1897     map<string, bufferlist> pairs;
1898     std::string last_key = "";
1899     bool more_results = true;
1900
1901     while (more_results) {
1902       r = cls_client::metadata_list(&src->md_ctx, src->header_oid, last_key, 0, &pairs);
1903       if (r < 0 && r != -EOPNOTSUPP && r != -EIO) {
1904         lderr(cct) << "couldn't list metadata: " << cpp_strerror(r) << dendl;
1905         return r;
1906       } else if (r == 0 && !pairs.empty()) {
1907         r = cls_client::metadata_set(&dest->md_ctx, dest->header_oid, pairs);
1908         if (r < 0) {
1909           lderr(cct) << "couldn't set metadata: " << cpp_strerror(r) << dendl;
1910           return r;
1911         }
1912
1913         last_key = pairs.rbegin()->first;
1914       }
1915
1916       more_results = (pairs.size() == MAX_KEYS);
1917       pairs.clear();
1918     }
1919
1920     ZTracer::Trace trace;
1921     if (src->blkin_trace_all) {
1922       trace.init("copy", &src->trace_endpoint);
1923     }
1924
1925     RWLock::RLocker owner_lock(src->owner_lock);
1926     SimpleThrottle throttle(src->concurrent_management_ops, false);
1927     uint64_t period = src->get_stripe_period();
1928     unsigned fadvise_flags = LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL |
1929                              LIBRADOS_OP_FLAG_FADVISE_NOCACHE;
1930     for (uint64_t offset = 0; offset < src_size; offset += period) {
1931       if (throttle.pending_error()) {
1932         return throttle.wait_for_ret();
1933       }
1934
1935       uint64_t len = min(period, src_size - offset);
1936       bufferlist *bl = new bufferlist();
1937       auto ctx = new C_CopyRead(&throttle, dest, offset, bl, sparse_size);
1938       auto comp = io::AioCompletion::create_and_start<Context>(
1939         ctx, src, io::AIO_TYPE_READ);
1940
1941       io::ImageReadRequest<> req(*src, comp, {{offset, len}},
1942                                  io::ReadResult{bl}, fadvise_flags,
1943                                  std::move(trace));
1944       ctx->read_trace = req.get_trace();
1945
1946       req.send();
1947       prog_ctx.update_progress(offset, src_size);
1948     }
1949
1950     r = throttle.wait_for_ret();
1951     if (r >= 0)
1952       prog_ctx.update_progress(src_size, src_size);
1953     return r;
1954   }
1955
1956   int snap_set(ImageCtx *ictx, const cls::rbd::SnapshotNamespace &snap_namespace,
1957                const char *snap_name)
1958   {
1959     ldout(ictx->cct, 20) << "snap_set " << ictx << " snap = "
1960                          << (snap_name ? snap_name : "NULL") << dendl;
1961
1962     // ignore return value, since we may be set to a non-existent
1963     // snapshot and the user is trying to fix that
1964     ictx->state->refresh_if_required();
1965
1966     C_SaferCond ctx;
1967     std::string name(snap_name == nullptr ? "" : snap_name);
1968     ictx->state->snap_set(snap_namespace, name, &ctx);
1969
1970     int r = ctx.wait();
1971     if (r < 0) {
1972       if (r != -ENOENT) {
1973         lderr(ictx->cct) << "failed to " << (name.empty() ? "un" : "") << "set "
1974                          << "snapshot: " << cpp_strerror(r) << dendl;
1975       }
1976       return r;
1977     }
1978
1979     return 0;
1980   }
1981
1982   int list_lockers(ImageCtx *ictx,
1983                    std::list<locker_t> *lockers,
1984                    bool *exclusive,
1985                    string *tag)
1986   {
1987     ldout(ictx->cct, 20) << "list_locks on image " << ictx << dendl;
1988
1989     int r = ictx->state->refresh_if_required();
1990     if (r < 0)
1991       return r;
1992
1993     RWLock::RLocker locker(ictx->md_lock);
1994     if (exclusive)
1995       *exclusive = ictx->exclusive_locked;
1996     if (tag)
1997       *tag = ictx->lock_tag;
1998     if (lockers) {
1999       lockers->clear();
2000       map<rados::cls::lock::locker_id_t,
2001           rados::cls::lock::locker_info_t>::const_iterator it;
2002       for (it = ictx->lockers.begin(); it != ictx->lockers.end(); ++it) {
2003         locker_t locker;
2004         locker.client = stringify(it->first.locker);
2005         locker.cookie = it->first.cookie;
2006         locker.address = stringify(it->second.addr);
2007         lockers->push_back(locker);
2008       }
2009     }
2010
2011     return 0;
2012   }
2013
2014   int lock(ImageCtx *ictx, bool exclusive, const string& cookie,
2015            const string& tag)
2016   {
2017     ldout(ictx->cct, 20) << "lock image " << ictx << " exclusive=" << exclusive
2018                          << " cookie='" << cookie << "' tag='" << tag << "'"
2019                          << dendl;
2020
2021     int r = ictx->state->refresh_if_required();
2022     if (r < 0)
2023       return r;
2024
2025     /**
2026      * If we wanted we could do something more intelligent, like local
2027      * checks that we think we will succeed. But for now, let's not
2028      * duplicate that code.
2029      */
2030     {
2031       RWLock::RLocker locker(ictx->md_lock);
2032       r = rados::cls::lock::lock(&ictx->md_ctx, ictx->header_oid, RBD_LOCK_NAME,
2033                                  exclusive ? LOCK_EXCLUSIVE : LOCK_SHARED,
2034                                  cookie, tag, "", utime_t(), 0);
2035       if (r < 0) {
2036         return r;
2037       }
2038     }
2039
2040     ictx->notify_update();
2041     return 0;
2042   }
2043
2044   int unlock(ImageCtx *ictx, const string& cookie)
2045   {
2046     ldout(ictx->cct, 20) << "unlock image " << ictx
2047                          << " cookie='" << cookie << "'" << dendl;
2048
2049     int r = ictx->state->refresh_if_required();
2050     if (r < 0)
2051       return r;
2052
2053     {
2054       RWLock::RLocker locker(ictx->md_lock);
2055       r = rados::cls::lock::unlock(&ictx->md_ctx, ictx->header_oid,
2056                                    RBD_LOCK_NAME, cookie);
2057       if (r < 0) {
2058         return r;
2059       }
2060     }
2061
2062     ictx->notify_update();
2063     return 0;
2064   }
2065
2066   int break_lock(ImageCtx *ictx, const string& client,
2067                  const string& cookie)
2068   {
2069     ldout(ictx->cct, 20) << "break_lock image " << ictx << " client='" << client
2070                          << "' cookie='" << cookie << "'" << dendl;
2071
2072     int r = ictx->state->refresh_if_required();
2073     if (r < 0)
2074       return r;
2075
2076     entity_name_t lock_client;
2077     if (!lock_client.parse(client)) {
2078       lderr(ictx->cct) << "Unable to parse client '" << client
2079                        << "'" << dendl;
2080       return -EINVAL;
2081     }
2082
2083     if (ictx->blacklist_on_break_lock) {
2084       typedef std::map<rados::cls::lock::locker_id_t,
2085                        rados::cls::lock::locker_info_t> Lockers;
2086       Lockers lockers;
2087       ClsLockType lock_type;
2088       std::string lock_tag;
2089       r = rados::cls::lock::get_lock_info(&ictx->md_ctx, ictx->header_oid,
2090                                           RBD_LOCK_NAME, &lockers, &lock_type,
2091                                           &lock_tag);
2092       if (r < 0) {
2093         lderr(ictx->cct) << "unable to retrieve lock info: " << cpp_strerror(r)
2094                        << dendl;
2095         return r;
2096       }
2097
2098       std::string client_address;
2099       for (Lockers::iterator it = lockers.begin();
2100            it != lockers.end(); ++it) {
2101         if (it->first.locker == lock_client) {
2102           client_address = stringify(it->second.addr);
2103           break;
2104         }
2105       }
2106       if (client_address.empty()) {
2107         return -ENOENT;
2108       }
2109
2110       RWLock::RLocker locker(ictx->md_lock);
2111       librados::Rados rados(ictx->md_ctx);
2112       r = rados.blacklist_add(client_address,
2113                               ictx->blacklist_expire_seconds);
2114       if (r < 0) {
2115         lderr(ictx->cct) << "unable to blacklist client: " << cpp_strerror(r)
2116                        << dendl;
2117         return r;
2118       }
2119     }
2120
2121     r = rados::cls::lock::break_lock(&ictx->md_ctx, ictx->header_oid,
2122                                      RBD_LOCK_NAME, cookie, lock_client);
2123     if (r < 0)
2124       return r;
2125     ictx->notify_update();
2126     return 0;
2127   }
2128
2129   void rbd_ctx_cb(completion_t cb, void *arg)
2130   {
2131     Context *ctx = reinterpret_cast<Context *>(arg);
2132     auto comp = reinterpret_cast<io::AioCompletion *>(cb);
2133     ctx->complete(comp->get_return_value());
2134     comp->release();
2135   }
2136
2137   int64_t read_iterate(ImageCtx *ictx, uint64_t off, uint64_t len,
2138                        int (*cb)(uint64_t, size_t, const char *, void *),
2139                        void *arg)
2140   {
2141     utime_t start_time, elapsed;
2142
2143     ldout(ictx->cct, 20) << "read_iterate " << ictx << " off = " << off
2144                          << " len = " << len << dendl;
2145
2146     int r = ictx->state->refresh_if_required();
2147     if (r < 0)
2148       return r;
2149
2150     uint64_t mylen = len;
2151     ictx->snap_lock.get_read();
2152     r = clip_io(ictx, off, &mylen);
2153     ictx->snap_lock.put_read();
2154     if (r < 0)
2155       return r;
2156
2157     int64_t total_read = 0;
2158     uint64_t period = ictx->get_stripe_period();
2159     uint64_t left = mylen;
2160
2161     ZTracer::Trace trace;
2162     if (ictx->blkin_trace_all) {
2163       trace.init("read_iterate", &ictx->trace_endpoint);
2164     }
2165
2166     RWLock::RLocker owner_locker(ictx->owner_lock);
2167     start_time = ceph_clock_now();
2168     while (left > 0) {
2169       uint64_t period_off = off - (off % period);
2170       uint64_t read_len = min(period_off + period - off, left);
2171
2172       bufferlist bl;
2173
2174       C_SaferCond ctx;
2175       auto c = io::AioCompletion::create_and_start(&ctx, ictx,
2176                                                    io::AIO_TYPE_READ);
2177       io::ImageRequest<>::aio_read(ictx, c, {{off, read_len}},
2178                                    io::ReadResult{&bl}, 0, std::move(trace));
2179
2180       int ret = ctx.wait();
2181       if (ret < 0) {
2182         return ret;
2183       }
2184
2185       r = cb(total_read, ret, bl.c_str(), arg);
2186       if (r < 0) {
2187         return r;
2188       }
2189
2190       total_read += ret;
2191       left -= ret;
2192       off += ret;
2193     }
2194
2195     elapsed = ceph_clock_now() - start_time;
2196     ictx->perfcounter->tinc(l_librbd_rd_latency, elapsed);
2197     ictx->perfcounter->inc(l_librbd_rd);
2198     ictx->perfcounter->inc(l_librbd_rd_bytes, mylen);
2199     return total_read;
2200   }
2201
2202   // validate extent against image size; clip to image size if necessary
2203   int clip_io(ImageCtx *ictx, uint64_t off, uint64_t *len)
2204   {
2205     assert(ictx->snap_lock.is_locked());
2206     uint64_t image_size = ictx->get_image_size(ictx->snap_id);
2207     bool snap_exists = ictx->snap_exists;
2208
2209     if (!snap_exists)
2210       return -ENOENT;
2211
2212     // special-case "len == 0" requests: always valid
2213     if (*len == 0)
2214       return 0;
2215
2216     // can't start past end
2217     if (off >= image_size)
2218       return -EINVAL;
2219
2220     // clip requests that extend past end to just end
2221     if ((off + *len) > image_size)
2222       *len = (size_t)(image_size - off);
2223
2224     return 0;
2225   }
2226
2227   int flush(ImageCtx *ictx)
2228   {
2229     CephContext *cct = ictx->cct;
2230     ldout(cct, 20) << "flush " << ictx << dendl;
2231
2232     int r = ictx->state->refresh_if_required();
2233     if (r < 0) {
2234       return r;
2235     }
2236
2237     ictx->user_flushed();
2238     C_SaferCond ctx;
2239     {
2240       RWLock::RLocker owner_locker(ictx->owner_lock);
2241       ictx->flush(&ctx);
2242     }
2243     r = ctx.wait();
2244
2245     ictx->perfcounter->inc(l_librbd_flush);
2246     return r;
2247   }
2248
2249   int invalidate_cache(ImageCtx *ictx)
2250   {
2251     CephContext *cct = ictx->cct;
2252     ldout(cct, 20) << "invalidate_cache " << ictx << dendl;
2253
2254     int r = ictx->state->refresh_if_required();
2255     if (r < 0) {
2256       return r;
2257     }
2258
2259     RWLock::RLocker owner_locker(ictx->owner_lock);
2260     r = ictx->invalidate_cache(false);
2261     ictx->perfcounter->inc(l_librbd_invalidate_cache);
2262     return r;
2263   }
2264
2265   int poll_io_events(ImageCtx *ictx, io::AioCompletion **comps, int numcomp)
2266   {
2267     if (numcomp <= 0)
2268       return -EINVAL;
2269     CephContext *cct = ictx->cct;
2270     ldout(cct, 20) << __func__ << " " << ictx << " numcomp = " << numcomp
2271                    << dendl;
2272     int i = 0;
2273     Mutex::Locker l(ictx->completed_reqs_lock);
2274     while (i < numcomp) {
2275       if (ictx->completed_reqs.empty())
2276         break;
2277       comps[i++] = ictx->completed_reqs.front();
2278       ictx->completed_reqs.pop_front();
2279     }
2280     return i;
2281   }
2282
2283   int metadata_get(ImageCtx *ictx, const string &key, string *value)
2284   {
2285     CephContext *cct = ictx->cct;
2286     ldout(cct, 20) << "metadata_get " << ictx << " key=" << key << dendl;
2287
2288     int r = ictx->state->refresh_if_required();
2289     if (r < 0) {
2290       return r;
2291     }
2292
2293     return cls_client::metadata_get(&ictx->md_ctx, ictx->header_oid, key, value);
2294   }
2295
2296   int metadata_list(ImageCtx *ictx, const string &start, uint64_t max, map<string, bufferlist> *pairs)
2297   {
2298     CephContext *cct = ictx->cct;
2299     ldout(cct, 20) << "metadata_list " << ictx << dendl;
2300
2301     int r = ictx->state->refresh_if_required();
2302     if (r < 0) {
2303       return r;
2304     }
2305
2306     return cls_client::metadata_list(&ictx->md_ctx, ictx->header_oid, start, max, pairs);
2307   }
2308
2309   struct C_RBD_Readahead : public Context {
2310     ImageCtx *ictx;
2311     object_t oid;
2312     uint64_t offset;
2313     uint64_t length;
2314     C_RBD_Readahead(ImageCtx *ictx, object_t oid, uint64_t offset, uint64_t length)
2315       : ictx(ictx), oid(oid), offset(offset), length(length) { }
2316     void finish(int r) override {
2317       ldout(ictx->cct, 20) << "C_RBD_Readahead on " << oid << ": " << offset << "+" << length << dendl;
2318       ictx->readahead.dec_pending();
2319     }
2320   };
2321
2322   void readahead(ImageCtx *ictx,
2323                  const vector<pair<uint64_t,uint64_t> >& image_extents)
2324   {
2325     uint64_t total_bytes = 0;
2326     for (vector<pair<uint64_t,uint64_t> >::const_iterator p = image_extents.begin();
2327          p != image_extents.end();
2328          ++p) {
2329       total_bytes += p->second;
2330     }
2331
2332     ictx->md_lock.get_write();
2333     bool abort = ictx->readahead_disable_after_bytes != 0 &&
2334       ictx->total_bytes_read > ictx->readahead_disable_after_bytes;
2335     if (abort) {
2336       ictx->md_lock.put_write();
2337       return;
2338     }
2339     ictx->total_bytes_read += total_bytes;
2340     ictx->snap_lock.get_read();
2341     uint64_t image_size = ictx->get_image_size(ictx->snap_id);
2342     ictx->snap_lock.put_read();
2343     ictx->md_lock.put_write();
2344
2345     pair<uint64_t, uint64_t> readahead_extent = ictx->readahead.update(image_extents, image_size);
2346     uint64_t readahead_offset = readahead_extent.first;
2347     uint64_t readahead_length = readahead_extent.second;
2348
2349     if (readahead_length > 0) {
2350       ldout(ictx->cct, 20) << "(readahead logical) " << readahead_offset << "~" << readahead_length << dendl;
2351       map<object_t,vector<ObjectExtent> > readahead_object_extents;
2352       Striper::file_to_extents(ictx->cct, ictx->format_string, &ictx->layout,
2353                                readahead_offset, readahead_length, 0, readahead_object_extents);
2354       for (map<object_t,vector<ObjectExtent> >::iterator p = readahead_object_extents.begin(); p != readahead_object_extents.end(); ++p) {
2355         for (vector<ObjectExtent>::iterator q = p->second.begin(); q != p->second.end(); ++q) {
2356           ldout(ictx->cct, 20) << "(readahead) oid " << q->oid << " " << q->offset << "~" << q->length << dendl;
2357
2358           Context *req_comp = new C_RBD_Readahead(ictx, q->oid, q->offset, q->length);
2359           ictx->readahead.inc_pending();
2360           ictx->aio_read_from_cache(q->oid, q->objectno, NULL,
2361                                     q->length, q->offset,
2362                                     req_comp, 0, nullptr);
2363         }
2364       }
2365       ictx->perfcounter->inc(l_librbd_readahead);
2366       ictx->perfcounter->inc(l_librbd_readahead_bytes, readahead_length);
2367     }
2368   }
2369
2370
2371
2372 }