ceph/src/tools/rbd_mirror/ImageSync.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 "ImageSync.h"
   5 #include "InstanceWatcher.h"
   6 #include "ProgressContext.h"
   7 #include "common/debug.h"
   8 #include "common/Timer.h"
   9 #include "common/errno.h"
  10 #include "librbd/DeepCopyRequest.h"
  11 #include "librbd/ImageCtx.h"
  12 #include "librbd/ImageState.h"
  13 #include "librbd/Utils.h"
  14 #include "librbd/internal.h"
  15 #include "librbd/asio/ContextWQ.h"
  16 #include "librbd/deep_copy/Handler.h"
  17 #include "tools/rbd_mirror/Threads.h"
  18 #include "tools/rbd_mirror/image_sync/SyncPointCreateRequest.h"
  19 #include "tools/rbd_mirror/image_sync/SyncPointPruneRequest.h"
  20 #include "tools/rbd_mirror/image_sync/Types.h"
  21
  22 #define dout_context g_ceph_context
  23 #define dout_subsys ceph_subsys_rbd_mirror
  24 #undef dout_prefix
  25 #define dout_prefix *_dout << "rbd::mirror::ImageSync: " \
  26                            << this << " " << __func__
  27
  28 namespace rbd {
  29 namespace mirror {
  30
  31 using namespace image_sync;
  32 using librbd::util::create_async_context_callback;
  33 using librbd::util::create_context_callback;
  34 using librbd::util::unique_lock_name;
  35
  36 template <typename I>
  37 class ImageSync<I>::ImageCopyProgressHandler
  38   : public librbd::deep_copy::NoOpHandler {
  39 public:
  40   ImageCopyProgressHandler(ImageSync *image_sync) : image_sync(image_sync) {
  41   }
  42
  43   int update_progress(uint64_t object_no, uint64_t object_count) override {
  44     image_sync->handle_copy_image_update_progress(object_no, object_count);
  45     return 0;
  46   }
  47
  48   ImageSync *image_sync;
  49 };
  50
  51 template <typename I>
  52 ImageSync<I>::ImageSync(
  53     Threads<I>* threads,
  54     I *local_image_ctx,
  55     I *remote_image_ctx,
  56     const std::string &local_mirror_uuid,
  57     image_sync::SyncPointHandler* sync_point_handler,
  58     InstanceWatcher<I> *instance_watcher,
  59     ProgressContext *progress_ctx,
  60     Context *on_finish)
  61   : CancelableRequest("rbd::mirror::ImageSync", local_image_ctx->cct,
  62                       on_finish),
  63     m_threads(threads),
  64     m_local_image_ctx(local_image_ctx),
  65     m_remote_image_ctx(remote_image_ctx),
  66     m_local_mirror_uuid(local_mirror_uuid),
  67     m_sync_point_handler(sync_point_handler),
  68     m_instance_watcher(instance_watcher),
  69     m_progress_ctx(progress_ctx),
  70     m_lock(ceph::make_mutex(unique_lock_name("ImageSync::m_lock", this))),
  71     m_update_sync_point_interval(
  72       m_local_image_ctx->cct->_conf.template get_val<double>(
  73         "rbd_mirror_sync_point_update_age")) {
  74 }
  75
  76 template <typename I>
  77 ImageSync<I>::~ImageSync() {
  78   ceph_assert(m_image_copy_request == nullptr);
  79   ceph_assert(m_image_copy_prog_handler == nullptr);
  80   ceph_assert(m_update_sync_ctx == nullptr);
  81 }
  82
  83 template <typename I>
  84 void ImageSync<I>::send() {
  85   send_notify_sync_request();
  86 }
  87
  88 template <typename I>
  89 void ImageSync<I>::cancel() {
  90   std::lock_guard locker{m_lock};
  91
  92   dout(10) << dendl;
  93
  94   m_canceled = true;
  95
  96   if (m_instance_watcher->cancel_sync_request(m_local_image_ctx->id)) {
  97     return;
  98   }
  99
 100   if (m_image_copy_request != nullptr) {
 101     m_image_copy_request->cancel();
 102   }
 103 }
 104
 105 template <typename I>
 106 void ImageSync<I>::send_notify_sync_request() {
 107   update_progress("NOTIFY_SYNC_REQUEST");
 108
 109   dout(10) << dendl;
 110
 111   m_lock.lock();
 112   if (m_canceled) {
 113     m_lock.unlock();
 114     CancelableRequest::finish(-ECANCELED);
 115     return;
 116   }
 117
 118   Context *ctx = create_async_context_callback(
 119     m_threads->work_queue, create_context_callback<
 120       ImageSync<I>, &ImageSync<I>::handle_notify_sync_request>(this));
 121   m_instance_watcher->notify_sync_request(m_local_image_ctx->id, ctx);
 122   m_lock.unlock();
 123 }
 124
 125 template <typename I>
 126 void ImageSync<I>::handle_notify_sync_request(int r) {
 127   dout(10) << ": r=" << r << dendl;
 128
 129   m_lock.lock();
 130   if (r == 0 && m_canceled) {
 131     r = -ECANCELED;
 132   }
 133   m_lock.unlock();
 134
 135   if (r < 0) {
 136     CancelableRequest::finish(r);
 137     return;
 138   }
 139
 140   send_prune_catch_up_sync_point();
 141 }
 142
 143 template <typename I>
 144 void ImageSync<I>::send_prune_catch_up_sync_point() {
 145   update_progress("PRUNE_CATCH_UP_SYNC_POINT");
 146
 147   if (m_sync_point_handler->get_sync_points().empty()) {
 148     send_create_sync_point();
 149     return;
 150   }
 151
 152   dout(10) << dendl;
 153
 154   // prune will remove sync points with missing snapshots and
 155   // ensure we have a maximum of one sync point (in case we
 156   // restarted)
 157   Context *ctx = create_context_callback<
 158     ImageSync<I>, &ImageSync<I>::handle_prune_catch_up_sync_point>(this);
 159   SyncPointPruneRequest<I> *request = SyncPointPruneRequest<I>::create(
 160     m_remote_image_ctx, false, m_sync_point_handler, ctx);
 161   request->send();
 162 }
 163
 164 template <typename I>
 165 void ImageSync<I>::handle_prune_catch_up_sync_point(int r) {
 166   dout(10) << ": r=" << r << dendl;
 167
 168   if (r < 0) {
 169     derr << ": failed to prune catch-up sync point: "
 170          << cpp_strerror(r) << dendl;
 171     finish(r);
 172     return;
 173   }
 174
 175   send_create_sync_point();
 176 }
 177
 178 template <typename I>
 179 void ImageSync<I>::send_create_sync_point() {
 180   update_progress("CREATE_SYNC_POINT");
 181
 182   // TODO: when support for disconnecting laggy clients is added,
 183   //       re-connect and create catch-up sync point
 184   if (!m_sync_point_handler->get_sync_points().empty()) {
 185     send_copy_image();
 186     return;
 187   }
 188
 189   dout(10) << dendl;
 190
 191   Context *ctx = create_context_callback<
 192     ImageSync<I>, &ImageSync<I>::handle_create_sync_point>(this);
 193   SyncPointCreateRequest<I> *request = SyncPointCreateRequest<I>::create(
 194     m_remote_image_ctx, m_local_mirror_uuid, m_sync_point_handler, ctx);
 195   request->send();
 196 }
 197
 198 template <typename I>
 199 void ImageSync<I>::handle_create_sync_point(int r) {
 200   dout(10) << ": r=" << r << dendl;
 201
 202   if (r < 0) {
 203     derr << ": failed to create sync point: " << cpp_strerror(r)
 204          << dendl;
 205     finish(r);
 206     return;
 207   }
 208
 209   send_copy_image();
 210 }
 211
 212 template <typename I>
 213 void ImageSync<I>::send_copy_image() {
 214   librados::snap_t snap_id_start = 0;
 215   librados::snap_t snap_id_end;
 216   librbd::deep_copy::ObjectNumber object_number;
 217   int r = 0;
 218
 219   m_snap_seqs_copy = m_sync_point_handler->get_snap_seqs();
 220   m_sync_points_copy = m_sync_point_handler->get_sync_points();
 221   ceph_assert(!m_sync_points_copy.empty());
 222   auto &sync_point = m_sync_points_copy.front();
 223
 224   {
 225     std::shared_lock image_locker{m_remote_image_ctx->image_lock};
 226     snap_id_end = m_remote_image_ctx->get_snap_id(
 227         cls::rbd::UserSnapshotNamespace(), sync_point.snap_name);
 228     if (snap_id_end == CEPH_NOSNAP) {
 229       derr << ": failed to locate snapshot: " << sync_point.snap_name << dendl;
 230       r = -ENOENT;
 231     } else if (!sync_point.from_snap_name.empty()) {
 232       snap_id_start = m_remote_image_ctx->get_snap_id(
 233         cls::rbd::UserSnapshotNamespace(), sync_point.from_snap_name);
 234       if (snap_id_start == CEPH_NOSNAP) {
 235         derr << ": failed to locate from snapshot: "
 236              << sync_point.from_snap_name << dendl;
 237         r = -ENOENT;
 238       }
 239     }
 240     object_number = sync_point.object_number;
 241   }
 242   if (r < 0) {
 243     finish(r);
 244     return;
 245   }
 246
 247   m_lock.lock();
 248   if (m_canceled) {
 249     m_lock.unlock();
 250     finish(-ECANCELED);
 251     return;
 252   }
 253
 254   dout(10) << dendl;
 255
 256   Context *ctx = create_context_callback<
 257     ImageSync<I>, &ImageSync<I>::handle_copy_image>(this);
 258   m_image_copy_prog_handler = new ImageCopyProgressHandler(this);
 259   m_image_copy_request = librbd::DeepCopyRequest<I>::create(
 260       m_remote_image_ctx, m_local_image_ctx, snap_id_start, snap_id_end,
 261       0, false, object_number, m_threads->work_queue, &m_snap_seqs_copy,
 262       m_image_copy_prog_handler, ctx);
 263   m_image_copy_request->get();
 264   m_lock.unlock();
 265
 266   update_progress("COPY_IMAGE");
 267
 268   m_image_copy_request->send();
 269 }
 270
 271 template <typename I>
 272 void ImageSync<I>::handle_copy_image(int r) {
 273   dout(10) << ": r=" << r << dendl;
 274
 275   {
 276     std::scoped_lock locker{m_threads->timer_lock, m_lock};
 277     m_image_copy_request->put();
 278     m_image_copy_request = nullptr;
 279     delete m_image_copy_prog_handler;
 280     m_image_copy_prog_handler = nullptr;
 281     if (r == 0 && m_canceled) {
 282       r = -ECANCELED;
 283     }
 284
 285     if (m_update_sync_ctx != nullptr) {
 286       m_threads->timer->cancel_event(m_update_sync_ctx);
 287       m_update_sync_ctx = nullptr;
 288     }
 289
 290     if (m_updating_sync_point) {
 291       m_ret_val = r;
 292       return;
 293     }
 294   }
 295
 296   if (r == -ECANCELED) {
 297     dout(10) << ": image copy canceled" << dendl;
 298     finish(r);
 299     return;
 300   } else if (r < 0) {
 301     derr << ": failed to copy image: " << cpp_strerror(r) << dendl;
 302     finish(r);
 303     return;
 304   }
 305
 306   send_flush_sync_point();
 307 }
 308
 309 template <typename I>
 310 void ImageSync<I>::handle_copy_image_update_progress(uint64_t object_no,
 311                                                      uint64_t object_count) {
 312   int percent = 100 * object_no / object_count;
 313   update_progress("COPY_IMAGE " + stringify(percent) + "%");
 314
 315   std::lock_guard locker{m_lock};
 316   m_image_copy_object_no = object_no;
 317   m_image_copy_object_count = object_count;
 318
 319   if (m_update_sync_ctx == nullptr && !m_updating_sync_point) {
 320     send_update_sync_point();
 321   }
 322 }
 323
 324 template <typename I>
 325 void ImageSync<I>::send_update_sync_point() {
 326   ceph_assert(ceph_mutex_is_locked(m_lock));
 327
 328   m_update_sync_ctx = nullptr;
 329
 330   if (m_canceled) {
 331     return;
 332   }
 333
 334   ceph_assert(!m_sync_points_copy.empty());
 335   auto sync_point = &m_sync_points_copy.front();
 336
 337   if (sync_point->object_number &&
 338       (m_image_copy_object_no - 1) == sync_point->object_number.get()) {
 339     // update sync point did not progress since last sync
 340     return;
 341   }
 342
 343   m_updating_sync_point = true;
 344
 345   if (m_image_copy_object_no > 0) {
 346     sync_point->object_number = m_image_copy_object_no - 1;
 347   }
 348
 349   auto ctx = create_context_callback<
 350     ImageSync<I>, &ImageSync<I>::handle_update_sync_point>(this);
 351   m_sync_point_handler->update_sync_points(m_snap_seqs_copy,
 352                                            m_sync_points_copy, false, ctx);
 353 }
 354
 355 template <typename I>
 356 void ImageSync<I>::handle_update_sync_point(int r) {
 357   CephContext *cct = m_local_image_ctx->cct;
 358   ldout(cct, 20) << ": r=" << r << dendl;
 359
 360   {
 361     std::scoped_lock locker{m_threads->timer_lock, m_lock};
 362     m_updating_sync_point = false;
 363
 364     if (m_image_copy_request != nullptr) {
 365       m_update_sync_ctx = new LambdaContext(
 366         [this](int r) {
 367           std::lock_guard locker{m_lock};
 368           this->send_update_sync_point();
 369         });
 370       m_threads->timer->add_event_after(
 371         m_update_sync_point_interval, m_update_sync_ctx);
 372       return;
 373     }
 374   }
 375
 376   send_flush_sync_point();
 377 }
 378
 379 template <typename I>
 380 void ImageSync<I>::send_flush_sync_point() {
 381   if (m_ret_val < 0) {
 382     finish(m_ret_val);
 383     return;
 384   }
 385
 386   update_progress("FLUSH_SYNC_POINT");
 387
 388   ceph_assert(!m_sync_points_copy.empty());
 389   auto sync_point = &m_sync_points_copy.front();
 390
 391   if (m_image_copy_object_no > 0) {
 392     sync_point->object_number = m_image_copy_object_no - 1;
 393   } else {
 394     sync_point->object_number = boost::none;
 395   }
 396
 397   auto ctx = create_context_callback<
 398     ImageSync<I>, &ImageSync<I>::handle_flush_sync_point>(this);
 399   m_sync_point_handler->update_sync_points(m_snap_seqs_copy,
 400                                            m_sync_points_copy, false, ctx);
 401 }
 402
 403 template <typename I>
 404 void ImageSync<I>::handle_flush_sync_point(int r) {
 405   dout(10) << ": r=" << r << dendl;
 406
 407   if (r < 0) {
 408     derr << ": failed to update client data: " << cpp_strerror(r)
 409          << dendl;
 410     finish(r);
 411     return;
 412   }
 413
 414   send_prune_sync_points();
 415 }
 416
 417 template <typename I>
 418 void ImageSync<I>::send_prune_sync_points() {
 419   dout(10) << dendl;
 420
 421   update_progress("PRUNE_SYNC_POINTS");
 422
 423   Context *ctx = create_context_callback<
 424     ImageSync<I>, &ImageSync<I>::handle_prune_sync_points>(this);
 425   SyncPointPruneRequest<I> *request = SyncPointPruneRequest<I>::create(
 426     m_remote_image_ctx, true, m_sync_point_handler, ctx);
 427   request->send();
 428 }
 429
 430 template <typename I>
 431 void ImageSync<I>::handle_prune_sync_points(int r) {
 432   dout(10) << ": r=" << r << dendl;
 433
 434   if (r < 0) {
 435     derr << ": failed to prune sync point: "
 436          << cpp_strerror(r) << dendl;
 437     finish(r);
 438     return;
 439   }
 440
 441   if (!m_sync_point_handler->get_sync_points().empty()) {
 442     send_copy_image();
 443     return;
 444   }
 445
 446   finish(0);
 447 }
 448
 449 template <typename I>
 450 void ImageSync<I>::update_progress(const std::string &description) {
 451   dout(20) << ": " << description << dendl;
 452
 453   if (m_progress_ctx) {
 454     m_progress_ctx->update_progress("IMAGE_SYNC/" + description);
 455   }
 456 }
 457
 458 template <typename I>
 459 void ImageSync<I>::finish(int r) {
 460   dout(20) << ": r=" << r << dendl;
 461
 462   m_instance_watcher->notify_sync_complete(m_local_image_ctx->id);
 463   CancelableRequest::finish(r);
 464 }
 465
 466 } // namespace mirror
 467 } // namespace rbd
 468
 469 template class rbd::mirror::ImageSync<librbd::ImageCtx>;