ceph/src/librbd/image/CreateRequest.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "librbd/image/CreateRequest.h"
   5 #include "include/ceph_assert.h"
   6 #include "common/dout.h"
   7 #include "common/errno.h"
   8 #include "common/ceph_context.h"
   9 #include "cls/rbd/cls_rbd_client.h"
  10 #include "osdc/Striper.h"
  11 #include "librbd/Features.h"
  12 #include "librbd/Journal.h"
  13 #include "librbd/ObjectMap.h"
  14 #include "librbd/Utils.h"
  15 #include "librbd/image/Types.h"
  16 #include "librbd/image/ValidatePoolRequest.h"
  17 #include "librbd/journal/CreateRequest.h"
  18 #include "librbd/journal/RemoveRequest.h"
  19 #include "librbd/mirror/EnableRequest.h"
  20 #include "journal/Journaler.h"
  21
  22
  23 #define dout_subsys ceph_subsys_rbd
  24 #undef dout_prefix
  25 #define dout_prefix *_dout << "librbd::image::CreateRequest: " << __func__ \
  26                            << ": "
  27
  28 namespace librbd {
  29 namespace image {
  30
  31 using util::create_rados_callback;
  32 using util::create_context_callback;
  33
  34 namespace {
  35
  36 int validate_features(CephContext *cct, uint64_t features) {
  37   if (features & ~RBD_FEATURES_ALL) {
  38     lderr(cct) << "librbd does not support requested features." << dendl;
  39     return -ENOSYS;
  40   }
  41   if ((features & RBD_FEATURE_OPERATIONS) != 0) {
  42     lderr(cct) << "cannot use internally controlled features" << dendl;
  43     return -EINVAL;
  44   }
  45   if ((features & RBD_FEATURE_FAST_DIFF) != 0 &&
  46       (features & RBD_FEATURE_OBJECT_MAP) == 0) {
  47     lderr(cct) << "cannot use fast diff without object map" << dendl;
  48     return -EINVAL;
  49   }
  50   if ((features & RBD_FEATURE_OBJECT_MAP) != 0 &&
  51       (features & RBD_FEATURE_EXCLUSIVE_LOCK) == 0) {
  52     lderr(cct) << "cannot use object map without exclusive lock" << dendl;
  53     return -EINVAL;
  54   }
  55   if ((features & RBD_FEATURE_JOURNALING) != 0 &&
  56       (features & RBD_FEATURE_EXCLUSIVE_LOCK) == 0) {
  57     lderr(cct) << "cannot use journaling without exclusive lock" << dendl;
  58     return -EINVAL;
  59   }
  60
  61   return 0;
  62 }
  63
  64 int validate_striping(CephContext *cct, uint8_t order, uint64_t stripe_unit,
  65                       uint64_t stripe_count) {
  66   if ((stripe_unit && !stripe_count) ||
  67       (!stripe_unit && stripe_count)) {
  68     lderr(cct) << "must specify both (or neither) of stripe-unit and "
  69                << "stripe-count" << dendl;
  70     return -EINVAL;
  71   } else if (stripe_unit && ((1ull << order) % stripe_unit || stripe_unit > (1ull << order))) {
  72     lderr(cct) << "stripe unit is not a factor of the object size" << dendl;
  73     return -EINVAL;
  74   }
  75   return 0;
  76 }
  77
  78 bool validate_layout(CephContext *cct, uint64_t size, file_layout_t &layout) {
  79   if (!librbd::ObjectMap<>::is_compatible(layout, size)) {
  80     lderr(cct) << "image size not compatible with object map" << dendl;
  81     return false;
  82   }
  83
  84   return true;
  85 }
  86
  87 int get_image_option(const ImageOptions &image_options, int option,
  88                      uint8_t *value) {
  89   uint64_t large_value;
  90   int r = image_options.get(option, &large_value);
  91   if (r < 0) {
  92     return r;
  93   }
  94   *value = static_cast<uint8_t>(large_value);
  95   return 0;
  96 }
  97
  98 } // anonymous namespace
  99
 100 template<typename I>
 101 int CreateRequest<I>::validate_order(CephContext *cct, uint8_t order) {
 102   if (order > 25 || order < 12) {
 103     lderr(cct) << "order must be in the range [12, 25]" << dendl;
 104     return -EDOM;
 105   }
 106   return 0;
 107 }
 108
 109 #undef dout_prefix
 110 #define dout_prefix *_dout << "librbd::image::CreateRequest: " << this << " " \
 111                            << __func__ << ": "
 112
 113 template<typename I>
 114 CreateRequest<I>::CreateRequest(const ConfigProxy& config, IoCtx &ioctx,
 115                                 const std::string &image_name,
 116                                 const std::string &image_id, uint64_t size,
 117                                 const ImageOptions &image_options,
 118                                 uint32_t create_flags,
 119                                 cls::rbd::MirrorImageMode mirror_image_mode,
 120                                 const std::string &non_primary_global_image_id,
 121                                 const std::string &primary_mirror_uuid,
 122                                 ContextWQ *op_work_queue, Context *on_finish)
 123   : m_config(config), m_image_name(image_name), m_image_id(image_id),
 124     m_size(size), m_create_flags(create_flags),
 125     m_mirror_image_mode(mirror_image_mode),
 126     m_non_primary_global_image_id(non_primary_global_image_id),
 127     m_primary_mirror_uuid(primary_mirror_uuid),
 128     m_op_work_queue(op_work_queue), m_on_finish(on_finish) {
 129
 130   m_io_ctx.dup(ioctx);
 131   m_cct = reinterpret_cast<CephContext *>(m_io_ctx.cct());
 132
 133   m_id_obj = util::id_obj_name(m_image_name);
 134   m_header_obj = util::header_name(m_image_id);
 135   m_objmap_name = ObjectMap<>::object_map_name(m_image_id, CEPH_NOSNAP);
 136   if (!non_primary_global_image_id.empty() &&
 137       (m_create_flags & CREATE_FLAG_MIRROR_ENABLE_MASK) == 0) {
 138     m_create_flags |= CREATE_FLAG_FORCE_MIRROR_ENABLE;
 139   }
 140
 141   if (image_options.get(RBD_IMAGE_OPTION_FEATURES, &m_features) != 0) {
 142     m_features = librbd::rbd_features_from_string(
 143       m_config.get_val<std::string>("rbd_default_features"), nullptr);
 144     m_negotiate_features = true;
 145   }
 146
 147   uint64_t features_clear = 0;
 148   uint64_t features_set = 0;
 149   image_options.get(RBD_IMAGE_OPTION_FEATURES_CLEAR, &features_clear);
 150   image_options.get(RBD_IMAGE_OPTION_FEATURES_SET, &features_set);
 151
 152   uint64_t features_conflict = features_clear & features_set;
 153   features_clear &= ~features_conflict;
 154   features_set &= ~features_conflict;
 155   m_features |= features_set;
 156   m_features &= ~features_clear;
 157
 158   if ((m_features & RBD_FEATURE_OBJECT_MAP) == RBD_FEATURE_OBJECT_MAP) {
 159       m_features |= RBD_FEATURE_FAST_DIFF;
 160   }
 161
 162   if (image_options.get(RBD_IMAGE_OPTION_STRIPE_UNIT, &m_stripe_unit) != 0 ||
 163       m_stripe_unit == 0) {
 164     m_stripe_unit = m_config.get_val<Option::size_t>("rbd_default_stripe_unit");
 165   }
 166   if (image_options.get(RBD_IMAGE_OPTION_STRIPE_COUNT, &m_stripe_count) != 0 ||
 167       m_stripe_count == 0) {
 168     m_stripe_count = m_config.get_val<uint64_t>("rbd_default_stripe_count");
 169   }
 170   if (get_image_option(image_options, RBD_IMAGE_OPTION_ORDER, &m_order) != 0 ||
 171       m_order == 0) {
 172     m_order = config.get_val<uint64_t>("rbd_default_order");
 173   }
 174   if (get_image_option(image_options, RBD_IMAGE_OPTION_JOURNAL_ORDER,
 175       &m_journal_order) != 0) {
 176     m_journal_order = m_config.get_val<uint64_t>("rbd_journal_order");
 177   }
 178   if (get_image_option(image_options, RBD_IMAGE_OPTION_JOURNAL_SPLAY_WIDTH,
 179                        &m_journal_splay_width) != 0) {
 180     m_journal_splay_width = m_config.get_val<uint64_t>(
 181       "rbd_journal_splay_width");
 182   }
 183   if (image_options.get(RBD_IMAGE_OPTION_JOURNAL_POOL, &m_journal_pool) != 0) {
 184     m_journal_pool = m_config.get_val<std::string>("rbd_journal_pool");
 185   }
 186   if (image_options.get(RBD_IMAGE_OPTION_DATA_POOL, &m_data_pool) != 0) {
 187     m_data_pool = m_config.get_val<std::string>("rbd_default_data_pool");
 188   }
 189
 190   m_layout.object_size = 1ull << m_order;
 191   if (m_stripe_unit == 0 || m_stripe_count == 0) {
 192     m_layout.stripe_unit = m_layout.object_size;
 193     m_layout.stripe_count = 1;
 194   } else {
 195     m_layout.stripe_unit = m_stripe_unit;
 196     m_layout.stripe_count = m_stripe_count;
 197   }
 198
 199   if (!m_data_pool.empty() && m_data_pool != ioctx.get_pool_name()) {
 200     m_features |= RBD_FEATURE_DATA_POOL;
 201   } else {
 202     m_data_pool.clear();
 203     m_features &= ~RBD_FEATURE_DATA_POOL;
 204   }
 205
 206   if ((m_stripe_unit != 0 && m_stripe_unit != (1ULL << m_order)) ||
 207       (m_stripe_count != 0 && m_stripe_count != 1)) {
 208     m_features |= RBD_FEATURE_STRIPINGV2;
 209   } else {
 210     m_features &= ~RBD_FEATURE_STRIPINGV2;
 211   }
 212
 213   ldout(m_cct, 10) << "name=" << m_image_name << ", "
 214                    << "id=" << m_image_id << ", "
 215                    << "size=" << m_size << ", "
 216                    << "features=" << m_features << ", "
 217                    << "order=" << (uint64_t)m_order << ", "
 218                    << "stripe_unit=" << m_stripe_unit << ", "
 219                    << "stripe_count=" << m_stripe_count << ", "
 220                    << "journal_order=" << (uint64_t)m_journal_order << ", "
 221                    << "journal_splay_width="
 222                    << (uint64_t)m_journal_splay_width << ", "
 223                    << "journal_pool=" << m_journal_pool << ", "
 224                    << "data_pool=" << m_data_pool << dendl;
 225 }
 226
 227 template<typename I>
 228 void CreateRequest<I>::send() {
 229   ldout(m_cct, 20) << dendl;
 230
 231   int r = validate_features(m_cct, m_features);
 232   if (r < 0) {
 233     complete(r);
 234     return;
 235   }
 236
 237   r = validate_order(m_cct, m_order);
 238   if (r < 0) {
 239     complete(r);
 240     return;
 241   }
 242
 243   r = validate_striping(m_cct, m_order, m_stripe_unit, m_stripe_count);
 244   if (r < 0) {
 245     complete(r);
 246     return;
 247   }
 248
 249   if (((m_features & RBD_FEATURE_OBJECT_MAP) != 0) &&
 250       (!validate_layout(m_cct, m_size, m_layout))) {
 251     complete(-EINVAL);
 252     return;
 253   }
 254
 255   validate_data_pool();
 256 }
 257
 258 template <typename I>
 259 void CreateRequest<I>::validate_data_pool() {
 260   m_data_io_ctx = m_io_ctx;
 261   if ((m_features & RBD_FEATURE_DATA_POOL) != 0) {
 262     librados::Rados rados(m_io_ctx);
 263     int r = rados.ioctx_create(m_data_pool.c_str(), m_data_io_ctx);
 264     if (r < 0) {
 265       lderr(m_cct) << "data pool " << m_data_pool << " does not exist" << dendl;
 266       complete(r);
 267       return;
 268     }
 269     m_data_pool_id = m_data_io_ctx.get_id();
 270     m_data_io_ctx.set_namespace(m_io_ctx.get_namespace());
 271   }
 272
 273   if (!m_config.get_val<bool>("rbd_validate_pool")) {
 274     add_image_to_directory();
 275     return;
 276   }
 277
 278   ldout(m_cct, 15) << dendl;
 279
 280   auto ctx = create_context_callback<
 281     CreateRequest<I>, &CreateRequest<I>::handle_validate_data_pool>(this);
 282   auto req = ValidatePoolRequest<I>::create(m_data_io_ctx, m_op_work_queue,
 283                                             ctx);
 284   req->send();
 285 }
 286
 287 template <typename I>
 288 void CreateRequest<I>::handle_validate_data_pool(int r) {
 289   ldout(m_cct, 15) << "r=" << r << dendl;
 290
 291   if (r == -EINVAL) {
 292     lderr(m_cct) << "pool does not support RBD images" << dendl;
 293     complete(r);
 294     return;
 295   } else if (r < 0) {
 296     lderr(m_cct) << "failed to validate pool: " << cpp_strerror(r) << dendl;
 297     complete(r);
 298     return;
 299   }
 300
 301   add_image_to_directory();
 302 }
 303
 304 template<typename I>
 305 void CreateRequest<I>::add_image_to_directory() {
 306   ldout(m_cct, 15) << dendl;
 307
 308   librados::ObjectWriteOperation op;
 309   if (!m_io_ctx.get_namespace().empty()) {
 310     cls_client::dir_state_assert(&op, cls::rbd::DIRECTORY_STATE_READY);
 311   }
 312   cls_client::dir_add_image(&op, m_image_name, m_image_id);
 313
 314   using klass = CreateRequest<I>;
 315   librados::AioCompletion *comp =
 316     create_rados_callback<klass, &klass::handle_add_image_to_directory>(this);
 317   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op);
 318   ceph_assert(r == 0);
 319   comp->release();
 320 }
 321
 322 template<typename I>
 323 void CreateRequest<I>::handle_add_image_to_directory(int r) {
 324   ldout(m_cct, 15) << "r=" << r << dendl;
 325
 326   if (r == -EEXIST) {
 327     ldout(m_cct, 5) << "directory entry for image " << m_image_name
 328                     << " already exists" << dendl;
 329     complete(r);
 330     return;
 331   } else if (!m_io_ctx.get_namespace().empty() && r == -ENOENT) {
 332     ldout(m_cct, 5) << "namespace " << m_io_ctx.get_namespace()
 333                     << " does not exist" << dendl;
 334     complete(r);
 335     return;
 336   } else if (r < 0) {
 337     lderr(m_cct) << "error adding image to directory: " << cpp_strerror(r)
 338                  << dendl;
 339     complete(r);
 340     return;
 341   }
 342
 343   create_id_object();
 344 }
 345
 346 template<typename I>
 347 void CreateRequest<I>::create_id_object() {
 348   ldout(m_cct, 15) << dendl;
 349
 350   librados::ObjectWriteOperation op;
 351   op.create(true);
 352   cls_client::set_id(&op, m_image_id);
 353
 354   using klass = CreateRequest<I>;
 355   librados::AioCompletion *comp =
 356     create_rados_callback<klass, &klass::handle_create_id_object>(this);
 357   int r = m_io_ctx.aio_operate(m_id_obj, comp, &op);
 358   ceph_assert(r == 0);
 359   comp->release();
 360 }
 361
 362 template<typename I>
 363 void CreateRequest<I>::handle_create_id_object(int r) {
 364   ldout(m_cct, 15) << "r=" << r << dendl;
 365
 366   if (r == -EEXIST) {
 367     ldout(m_cct, 5) << "id object for  " << m_image_name << " already exists"
 368                     << dendl;
 369     m_r_saved = r;
 370     remove_from_dir();
 371     return;
 372   } else if (r < 0) {
 373     lderr(m_cct) << "error creating RBD id object: " << cpp_strerror(r)
 374                  << dendl;
 375     m_r_saved = r;
 376     remove_from_dir();
 377     return;
 378   }
 379
 380   negotiate_features();
 381 }
 382
 383 template<typename I>
 384 void CreateRequest<I>::negotiate_features() {
 385   if (!m_negotiate_features) {
 386     create_image();
 387     return;
 388   }
 389
 390   ldout(m_cct, 15) << dendl;
 391
 392   librados::ObjectReadOperation op;
 393   cls_client::get_all_features_start(&op);
 394
 395   using klass = CreateRequest<I>;
 396   librados::AioCompletion *comp =
 397     create_rados_callback<klass, &klass::handle_negotiate_features>(this);
 398
 399   m_outbl.clear();
 400   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op, &m_outbl);
 401   ceph_assert(r == 0);
 402   comp->release();
 403 }
 404
 405 template<typename I>
 406 void CreateRequest<I>::handle_negotiate_features(int r) {
 407   ldout(m_cct, 15) << "r=" << r << dendl;
 408
 409   uint64_t all_features;
 410   if (r >= 0) {
 411     auto it = m_outbl.cbegin();
 412     r = cls_client::get_all_features_finish(&it, &all_features);
 413   }
 414   if (r < 0) {
 415     ldout(m_cct, 10) << "error retrieving server supported features set: "
 416                      << cpp_strerror(r) << dendl;
 417   } else if ((m_features & all_features) != m_features) {
 418     m_features &= all_features;
 419     ldout(m_cct, 10) << "limiting default features set to server supported: "
 420                      << m_features << dendl;
 421   }
 422
 423   create_image();
 424 }
 425
 426 template<typename I>
 427 void CreateRequest<I>::create_image() {
 428   ldout(m_cct, 15) << dendl;
 429   ceph_assert(m_data_pool.empty() || m_data_pool_id != -1);
 430
 431   ostringstream oss;
 432   oss << RBD_DATA_PREFIX;
 433   if (m_data_pool_id != -1) {
 434     oss << stringify(m_io_ctx.get_id()) << ".";
 435   }
 436   oss << m_image_id;
 437   if (oss.str().length() > RBD_MAX_BLOCK_NAME_PREFIX_LENGTH) {
 438     lderr(m_cct) << "object prefix '" << oss.str() << "' too large" << dendl;
 439     m_r_saved = -EINVAL;
 440     remove_id_object();
 441     return;
 442   }
 443
 444   librados::ObjectWriteOperation op;
 445   op.create(true);
 446   cls_client::create_image(&op, m_size, m_order, m_features, oss.str(),
 447                            m_data_pool_id);
 448
 449   using klass = CreateRequest<I>;
 450   librados::AioCompletion *comp =
 451     create_rados_callback<klass, &klass::handle_create_image>(this);
 452   int r = m_io_ctx.aio_operate(m_header_obj, comp, &op);
 453   ceph_assert(r == 0);
 454   comp->release();
 455 }
 456
 457 template<typename I>
 458 void CreateRequest<I>::handle_create_image(int r) {
 459   ldout(m_cct, 15) << "r=" << r << dendl;
 460
 461   if (r == -EEXIST) {
 462     ldout(m_cct, 5) << "image id already in-use" << dendl;
 463     complete(-EBADF);
 464     return;
 465   } else if (r < 0) {
 466     lderr(m_cct) << "error writing header: " << cpp_strerror(r) << dendl;
 467     m_r_saved = r;
 468     remove_id_object();
 469     return;
 470   }
 471
 472   set_stripe_unit_count();
 473 }
 474
 475 template<typename I>
 476 void CreateRequest<I>::set_stripe_unit_count() {
 477   if ((!m_stripe_unit && !m_stripe_count) ||
 478       ((m_stripe_count == 1) && (m_stripe_unit == (1ull << m_order)))) {
 479     object_map_resize();
 480     return;
 481   }
 482
 483   ldout(m_cct, 15) << dendl;
 484
 485   librados::ObjectWriteOperation op;
 486   cls_client::set_stripe_unit_count(&op, m_stripe_unit, m_stripe_count);
 487
 488   using klass = CreateRequest<I>;
 489   librados::AioCompletion *comp =
 490     create_rados_callback<klass, &klass::handle_set_stripe_unit_count>(this);
 491   int r = m_io_ctx.aio_operate(m_header_obj, comp, &op);
 492   ceph_assert(r == 0);
 493   comp->release();
 494 }
 495
 496 template<typename I>
 497 void CreateRequest<I>::handle_set_stripe_unit_count(int r) {
 498   ldout(m_cct, 15) << "r=" << r << dendl;
 499
 500   if (r < 0) {
 501     lderr(m_cct) << "error setting stripe unit/count: "
 502                  << cpp_strerror(r) << dendl;
 503     m_r_saved = r;
 504     remove_header_object();
 505     return;
 506   }
 507
 508   object_map_resize();
 509 }
 510
 511 template<typename I>
 512 void CreateRequest<I>::object_map_resize() {
 513   if ((m_features & RBD_FEATURE_OBJECT_MAP) == 0) {
 514     fetch_mirror_mode();
 515     return;
 516   }
 517
 518   ldout(m_cct, 15) << dendl;
 519
 520   librados::ObjectWriteOperation op;
 521   cls_client::object_map_resize(&op, Striper::get_num_objects(m_layout, m_size),
 522                                 OBJECT_NONEXISTENT);
 523
 524   using klass = CreateRequest<I>;
 525   librados::AioCompletion *comp =
 526     create_rados_callback<klass, &klass::handle_object_map_resize>(this);
 527   int r = m_io_ctx.aio_operate(m_objmap_name, comp, &op);
 528   ceph_assert(r == 0);
 529   comp->release();
 530 }
 531
 532 template<typename I>
 533 void CreateRequest<I>::handle_object_map_resize(int r) {
 534   ldout(m_cct, 15) << "r=" << r << dendl;
 535
 536   if (r < 0) {
 537     lderr(m_cct) << "error creating initial object map: "
 538                  << cpp_strerror(r) << dendl;
 539
 540     m_r_saved = r;
 541     remove_header_object();
 542     return;
 543   }
 544
 545   fetch_mirror_mode();
 546 }
 547
 548 template<typename I>
 549 void CreateRequest<I>::fetch_mirror_mode() {
 550   if ((m_features & RBD_FEATURE_JOURNALING) == 0) {
 551     mirror_image_enable();
 552     return;
 553   }
 554
 555   ldout(m_cct, 15) << dendl;
 556
 557   librados::ObjectReadOperation op;
 558   cls_client::mirror_mode_get_start(&op);
 559
 560   using klass = CreateRequest<I>;
 561   librados::AioCompletion *comp =
 562     create_rados_callback<klass, &klass::handle_fetch_mirror_mode>(this);
 563   m_outbl.clear();
 564   int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_outbl);
 565   ceph_assert(r == 0);
 566   comp->release();
 567 }
 568
 569 template<typename I>
 570 void CreateRequest<I>::handle_fetch_mirror_mode(int r) {
 571   ldout(m_cct, 15) << "r=" << r << dendl;
 572
 573   if ((r < 0) && (r != -ENOENT)) {
 574     lderr(m_cct) << "failed to retrieve mirror mode: " << cpp_strerror(r)
 575                  << dendl;
 576
 577     m_r_saved = r;
 578     remove_object_map();
 579     return;
 580   }
 581
 582   m_mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
 583   if (r == 0) {
 584     auto it = m_outbl.cbegin();
 585     r = cls_client::mirror_mode_get_finish(&it, &m_mirror_mode);
 586     if (r < 0) {
 587       lderr(m_cct) << "Failed to retrieve mirror mode" << dendl;
 588
 589       m_r_saved = r;
 590       remove_object_map();
 591       return;
 592     }
 593   }
 594
 595   journal_create();
 596 }
 597
 598 template<typename I>
 599 void CreateRequest<I>::journal_create() {
 600   ldout(m_cct, 15) << dendl;
 601
 602   using klass = CreateRequest<I>;
 603   Context *ctx = create_context_callback<klass, &klass::handle_journal_create>(
 604     this);
 605
 606   // only link to remote primary mirror uuid if in journal-based
 607   // mirroring mode
 608   bool use_primary_mirror_uuid = (
 609     !m_non_primary_global_image_id.empty() &&
 610     m_mirror_image_mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL);
 611
 612   librbd::journal::TagData tag_data;
 613   tag_data.mirror_uuid = (use_primary_mirror_uuid ? m_primary_mirror_uuid :
 614                           librbd::Journal<I>::LOCAL_MIRROR_UUID);
 615
 616   auto req = librbd::journal::CreateRequest<I>::create(
 617     m_io_ctx, m_image_id, m_journal_order, m_journal_splay_width,
 618     m_journal_pool, cls::journal::Tag::TAG_CLASS_NEW, tag_data,
 619     librbd::Journal<I>::IMAGE_CLIENT_ID, m_op_work_queue, ctx);
 620   req->send();
 621 }
 622
 623 template<typename I>
 624 void CreateRequest<I>::handle_journal_create(int r) {
 625   ldout(m_cct, 15) << "r=" << r << dendl;
 626
 627   if (r < 0) {
 628     lderr(m_cct) << "error creating journal: " << cpp_strerror(r)
 629                  << dendl;
 630
 631     m_r_saved = r;
 632     remove_object_map();
 633     return;
 634   }
 635
 636   mirror_image_enable();
 637 }
 638
 639 template<typename I>
 640 void CreateRequest<I>::mirror_image_enable() {
 641   auto mirror_enable_flag = (m_create_flags & CREATE_FLAG_MIRROR_ENABLE_MASK);
 642
 643   if ((m_mirror_mode != cls::rbd::MIRROR_MODE_POOL &&
 644        mirror_enable_flag != CREATE_FLAG_FORCE_MIRROR_ENABLE) ||
 645       (mirror_enable_flag == CREATE_FLAG_SKIP_MIRROR_ENABLE)) {
 646     complete(0);
 647     return;
 648   }
 649
 650   ldout(m_cct, 15) << dendl;
 651   auto ctx = create_context_callback<
 652     CreateRequest<I>, &CreateRequest<I>::handle_mirror_image_enable>(this);
 653
 654   auto req = mirror::EnableRequest<I>::create(
 655     m_io_ctx, m_image_id, m_mirror_image_mode,
 656     m_non_primary_global_image_id, true, m_op_work_queue, ctx);
 657   req->send();
 658 }
 659
 660 template<typename I>
 661 void CreateRequest<I>::handle_mirror_image_enable(int r) {
 662   ldout(m_cct, 15) << "r=" << r << dendl;
 663
 664   if (r < 0) {
 665     lderr(m_cct) << "cannot enable mirroring: " << cpp_strerror(r)
 666                  << dendl;
 667
 668     m_r_saved = r;
 669     journal_remove();
 670     return;
 671   }
 672
 673   complete(0);
 674 }
 675
 676 template<typename I>
 677 void CreateRequest<I>::complete(int r) {
 678   ldout(m_cct, 10) << "r=" << r << dendl;
 679
 680   m_data_io_ctx.close();
 681   auto on_finish = m_on_finish;
 682   delete this;
 683   on_finish->complete(r);
 684 }
 685
 686 // cleanup
 687 template<typename I>
 688 void CreateRequest<I>::journal_remove() {
 689   if ((m_features & RBD_FEATURE_JOURNALING) == 0) {
 690     remove_object_map();
 691     return;
 692   }
 693
 694   ldout(m_cct, 15) << dendl;
 695
 696   using klass = CreateRequest<I>;
 697   Context *ctx = create_context_callback<klass, &klass::handle_journal_remove>(
 698     this);
 699
 700   librbd::journal::RemoveRequest<I> *req =
 701     librbd::journal::RemoveRequest<I>::create(
 702       m_io_ctx, m_image_id, librbd::Journal<I>::IMAGE_CLIENT_ID, m_op_work_queue,
 703       ctx);
 704   req->send();
 705 }
 706
 707 template<typename I>
 708 void CreateRequest<I>::handle_journal_remove(int r) {
 709   ldout(m_cct, 15) << "r=" << r << dendl;
 710
 711   if (r < 0) {
 712     lderr(m_cct) << "error cleaning up journal after creation failed: "
 713                  << cpp_strerror(r) << dendl;
 714   }
 715
 716   remove_object_map();
 717 }
 718
 719 template<typename I>
 720 void CreateRequest<I>::remove_object_map() {
 721   if ((m_features & RBD_FEATURE_OBJECT_MAP) == 0) {
 722     remove_header_object();
 723     return;
 724   }
 725
 726   ldout(m_cct, 15) << dendl;
 727
 728   using klass = CreateRequest<I>;
 729   librados::AioCompletion *comp =
 730     create_rados_callback<klass, &klass::handle_remove_object_map>(this);
 731   int r = m_io_ctx.aio_remove(m_objmap_name, comp);
 732   ceph_assert(r == 0);
 733   comp->release();
 734 }
 735
 736 template<typename I>
 737 void CreateRequest<I>::handle_remove_object_map(int r) {
 738   ldout(m_cct, 15) << "r=" << r << dendl;
 739
 740   if (r < 0) {
 741     lderr(m_cct) << "error cleaning up object map after creation failed: "
 742                  << cpp_strerror(r) << dendl;
 743   }
 744
 745   remove_header_object();
 746 }
 747
 748 template<typename I>
 749 void CreateRequest<I>::remove_header_object() {
 750   ldout(m_cct, 15) << dendl;
 751
 752   using klass = CreateRequest<I>;
 753   librados::AioCompletion *comp =
 754     create_rados_callback<klass, &klass::handle_remove_header_object>(this);
 755   int r = m_io_ctx.aio_remove(m_header_obj, comp);
 756   ceph_assert(r == 0);
 757   comp->release();
 758 }
 759
 760 template<typename I>
 761 void CreateRequest<I>::handle_remove_header_object(int r) {
 762   ldout(m_cct, 15) << "r=" << r << dendl;
 763
 764   if (r < 0) {
 765     lderr(m_cct) << "error cleaning up image header after creation failed: "
 766                  << cpp_strerror(r) << dendl;
 767   }
 768
 769   remove_id_object();
 770 }
 771
 772 template<typename I>
 773 void CreateRequest<I>::remove_id_object() {
 774   ldout(m_cct, 15) << dendl;
 775
 776   using klass = CreateRequest<I>;
 777   librados::AioCompletion *comp =
 778     create_rados_callback<klass, &klass::handle_remove_id_object>(this);
 779   int r = m_io_ctx.aio_remove(m_id_obj, comp);
 780   ceph_assert(r == 0);
 781   comp->release();
 782 }
 783
 784 template<typename I>
 785 void CreateRequest<I>::handle_remove_id_object(int r) {
 786   ldout(m_cct, 15) << "r=" << r << dendl;
 787
 788   if (r < 0) {
 789     lderr(m_cct) << "error cleaning up id object after creation failed: "
 790                  << cpp_strerror(r) << dendl;
 791   }
 792
 793   remove_from_dir();
 794 }
 795
 796 template<typename I>
 797 void CreateRequest<I>::remove_from_dir() {
 798   ldout(m_cct, 15) << dendl;
 799
 800   librados::ObjectWriteOperation op;
 801   cls_client::dir_remove_image(&op, m_image_name, m_image_id);
 802
 803   using klass = CreateRequest<I>;
 804   librados::AioCompletion *comp =
 805     create_rados_callback<klass, &klass::handle_remove_from_dir>(this);
 806   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op);
 807   ceph_assert(r == 0);
 808   comp->release();
 809 }
 810
 811 template<typename I>
 812 void CreateRequest<I>::handle_remove_from_dir(int r) {
 813   ldout(m_cct, 15) << "r=" << r << dendl;
 814
 815   if (r < 0) {
 816     lderr(m_cct) << "error cleaning up image from rbd_directory object "
 817                  << "after creation failed: " << cpp_strerror(r) << dendl;
 818   }
 819
 820   complete(m_r_saved);
 821 }
 822
 823 } //namespace image
 824 } //namespace librbd
 825
 826 template class librbd::image::CreateRequest<librbd::ImageCtx>;