ceph/src/librbd/ImageWatcher.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/ImageWatcher.h"
   5 #include "librbd/ExclusiveLock.h"
   6 #include "librbd/ImageCtx.h"
   7 #include "librbd/ImageState.h"
   8 #include "librbd/internal.h"
   9 #include "librbd/Operations.h"
  10 #include "librbd/TaskFinisher.h"
  11 #include "librbd/Types.h"
  12 #include "librbd/Utils.h"
  13 #include "librbd/exclusive_lock/Policy.h"
  14 #include "librbd/image_watcher/NotifyLockOwner.h"
  15 #include "librbd/io/AioCompletion.h"
  16 #include "librbd/watcher/Utils.h"
  17 #include "include/encoding.h"
  18 #include "common/errno.h"
  19 #include "common/WorkQueue.h"
  20 #include <boost/bind.hpp>
  21
  22 #define dout_subsys ceph_subsys_rbd
  23 #undef dout_prefix
  24 #define dout_prefix *_dout << "librbd::ImageWatcher: "
  25
  26 namespace librbd {
  27
  28 using namespace image_watcher;
  29 using namespace watch_notify;
  30 using util::create_async_context_callback;
  31 using util::create_context_callback;
  32 using util::create_rados_callback;
  33 using librbd::watcher::util::HandlePayloadVisitor;
  34
  35 using ceph::encode;
  36 using ceph::decode;
  37
  38 static const double     RETRY_DELAY_SECONDS = 1.0;
  39
  40 template <typename I>
  41 struct ImageWatcher<I>::C_ProcessPayload : public Context {
  42   ImageWatcher *image_watcher;
  43   uint64_t notify_id;
  44   uint64_t handle;
  45   watch_notify::Payload payload;
  46
  47   C_ProcessPayload(ImageWatcher *image_watcher_, uint64_t notify_id_,
  48                    uint64_t handle_, const watch_notify::Payload &payload)
  49     : image_watcher(image_watcher_), notify_id(notify_id_), handle(handle_),
  50       payload(payload) {
  51   }
  52
  53   void finish(int r) override {
  54     image_watcher->m_async_op_tracker.start_op();
  55     if (image_watcher->notifications_blocked()) {
  56       // requests are blocked -- just ack the notification
  57       bufferlist bl;
  58       image_watcher->acknowledge_notify(notify_id, handle, bl);
  59     } else {
  60       image_watcher->process_payload(notify_id, handle, payload);
  61     }
  62     image_watcher->m_async_op_tracker.finish_op();
  63   }
  64 };
  65
  66 template <typename I>
  67 ImageWatcher<I>::ImageWatcher(I &image_ctx)
  68   : Watcher(image_ctx.md_ctx, image_ctx.op_work_queue, image_ctx.header_oid),
  69     m_image_ctx(image_ctx),
  70     m_task_finisher(new TaskFinisher<Task>(*m_image_ctx.cct)),
  71     m_async_request_lock(ceph::make_shared_mutex(
  72       util::unique_lock_name("librbd::ImageWatcher::m_async_request_lock", this))),
  73     m_owner_client_id_lock(ceph::make_mutex(
  74       util::unique_lock_name("librbd::ImageWatcher::m_owner_client_id_lock", this)))
  75 {
  76 }
  77
  78 template <typename I>
  79 ImageWatcher<I>::~ImageWatcher()
  80 {
  81   delete m_task_finisher;
  82 }
  83
  84 template <typename I>
  85 void ImageWatcher<I>::unregister_watch(Context *on_finish) {
  86   CephContext *cct = m_image_ctx.cct;
  87   ldout(cct, 10) << this << " unregistering image watcher" << dendl;
  88
  89   cancel_async_requests();
  90
  91   on_finish = new LambdaContext([this, on_finish](int r) {
  92     m_async_op_tracker.wait_for_ops(on_finish);
  93   });
  94   auto ctx = new LambdaContext([this, on_finish](int r) {
  95     m_task_finisher->cancel_all(on_finish);
  96   });
  97   Watcher::unregister_watch(ctx);
  98 }
  99
 100 template <typename I>
 101 void ImageWatcher<I>::block_notifies(Context *on_finish) {
 102   CephContext *cct = m_image_ctx.cct;
 103   ldout(cct, 10) << this << " "  << __func__ << dendl;
 104
 105   on_finish = new LambdaContext([this, on_finish](int r) {
 106       cancel_async_requests();
 107       on_finish->complete(r);
 108     });
 109   Watcher::block_notifies(on_finish);
 110 }
 111
 112 template <typename I>
 113 void ImageWatcher<I>::schedule_async_progress(const AsyncRequestId &request,
 114                                               uint64_t offset, uint64_t total) {
 115   auto ctx = new LambdaContext(
 116     boost::bind(&ImageWatcher<I>::notify_async_progress, this, request, offset,
 117                 total));
 118   m_task_finisher->queue(Task(TASK_CODE_ASYNC_PROGRESS, request), ctx);
 119 }
 120
 121 template <typename I>
 122 int ImageWatcher<I>::notify_async_progress(const AsyncRequestId &request,
 123                                            uint64_t offset, uint64_t total) {
 124   ldout(m_image_ctx.cct, 20) << this << " remote async request progress: "
 125                              << request << " @ " << offset
 126                              << "/" << total << dendl;
 127
 128   send_notify(AsyncProgressPayload(request, offset, total));
 129   return 0;
 130 }
 131
 132 template <typename I>
 133 void ImageWatcher<I>::schedule_async_complete(const AsyncRequestId &request,
 134                                               int r) {
 135   m_async_op_tracker.start_op();
 136   auto ctx = new LambdaContext(
 137     boost::bind(&ImageWatcher<I>::notify_async_complete, this, request, r));
 138   m_task_finisher->queue(ctx);
 139 }
 140
 141 template <typename I>
 142 void ImageWatcher<I>::notify_async_complete(const AsyncRequestId &request,
 143                                             int r) {
 144   ldout(m_image_ctx.cct, 20) << this << " remote async request finished: "
 145                              << request << " = " << r << dendl;
 146
 147   send_notify(AsyncCompletePayload(request, r),
 148     new LambdaContext(boost::bind(&ImageWatcher<I>::handle_async_complete,
 149                         this, request, r, _1)));
 150 }
 151
 152 template <typename I>
 153 void ImageWatcher<I>::handle_async_complete(const AsyncRequestId &request,
 154                                             int r, int ret_val) {
 155   ldout(m_image_ctx.cct, 20) << this << " " << __func__ << ": "
 156                              << "request=" << request << ", r=" << ret_val
 157                              << dendl;
 158   if (ret_val < 0) {
 159     lderr(m_image_ctx.cct) << this << " failed to notify async complete: "
 160                            << cpp_strerror(ret_val) << dendl;
 161     if (ret_val == -ETIMEDOUT && !is_unregistered()) {
 162       schedule_async_complete(request, r);
 163       m_async_op_tracker.finish_op();
 164       return;
 165     }
 166   }
 167
 168   std::unique_lock async_request_locker{m_async_request_lock};
 169   m_async_pending.erase(request);
 170   m_async_op_tracker.finish_op();
 171 }
 172
 173 template <typename I>
 174 void ImageWatcher<I>::notify_flatten(uint64_t request_id,
 175                                      ProgressContext &prog_ctx,
 176                                      Context *on_finish) {
 177   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 178   ceph_assert(m_image_ctx.exclusive_lock &&
 179               !m_image_ctx.exclusive_lock->is_lock_owner());
 180
 181   AsyncRequestId async_request_id(get_client_id(), request_id);
 182
 183   notify_async_request(async_request_id, FlattenPayload(async_request_id),
 184                        prog_ctx, on_finish);
 185 }
 186
 187 template <typename I>
 188 void ImageWatcher<I>::notify_resize(uint64_t request_id, uint64_t size,
 189                                     bool allow_shrink,
 190                                     ProgressContext &prog_ctx,
 191                                     Context *on_finish) {
 192   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 193   ceph_assert(m_image_ctx.exclusive_lock &&
 194               !m_image_ctx.exclusive_lock->is_lock_owner());
 195
 196   AsyncRequestId async_request_id(get_client_id(), request_id);
 197
 198   notify_async_request(async_request_id,
 199                        ResizePayload(size, allow_shrink, async_request_id),
 200                        prog_ctx, on_finish);
 201 }
 202
 203 template <typename I>
 204 void ImageWatcher<I>::notify_snap_create(const cls::rbd::SnapshotNamespace &snap_namespace,
 205                                          const std::string &snap_name,
 206                                          Context *on_finish) {
 207   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 208   ceph_assert(m_image_ctx.exclusive_lock &&
 209               !m_image_ctx.exclusive_lock->is_lock_owner());
 210
 211   notify_lock_owner(SnapCreatePayload(snap_namespace, snap_name), on_finish);
 212 }
 213
 214 template <typename I>
 215 void ImageWatcher<I>::notify_snap_rename(const snapid_t &src_snap_id,
 216                                          const std::string &dst_snap_name,
 217                                          Context *on_finish) {
 218   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 219   ceph_assert(m_image_ctx.exclusive_lock &&
 220               !m_image_ctx.exclusive_lock->is_lock_owner());
 221
 222   notify_lock_owner(SnapRenamePayload(src_snap_id, dst_snap_name), on_finish);
 223 }
 224
 225 template <typename I>
 226 void ImageWatcher<I>::notify_snap_remove(const cls::rbd::SnapshotNamespace &snap_namespace,
 227                                          const std::string &snap_name,
 228                                          Context *on_finish) {
 229   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 230   ceph_assert(m_image_ctx.exclusive_lock &&
 231               !m_image_ctx.exclusive_lock->is_lock_owner());
 232
 233   notify_lock_owner(SnapRemovePayload(snap_namespace, snap_name), on_finish);
 234 }
 235
 236 template <typename I>
 237 void ImageWatcher<I>::notify_snap_protect(const cls::rbd::SnapshotNamespace &snap_namespace,
 238                                           const std::string &snap_name,
 239                                           Context *on_finish) {
 240   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 241   ceph_assert(m_image_ctx.exclusive_lock &&
 242               !m_image_ctx.exclusive_lock->is_lock_owner());
 243
 244   notify_lock_owner(SnapProtectPayload(snap_namespace, snap_name), on_finish);
 245 }
 246
 247 template <typename I>
 248 void ImageWatcher<I>::notify_snap_unprotect(const cls::rbd::SnapshotNamespace &snap_namespace,
 249                                             const std::string &snap_name,
 250                                             Context *on_finish) {
 251   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 252   ceph_assert(m_image_ctx.exclusive_lock &&
 253               !m_image_ctx.exclusive_lock->is_lock_owner());
 254
 255   notify_lock_owner(SnapUnprotectPayload(snap_namespace, snap_name), on_finish);
 256 }
 257
 258 template <typename I>
 259 void ImageWatcher<I>::notify_rebuild_object_map(uint64_t request_id,
 260                                                 ProgressContext &prog_ctx,
 261                                                 Context *on_finish) {
 262   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 263   ceph_assert(m_image_ctx.exclusive_lock &&
 264               !m_image_ctx.exclusive_lock->is_lock_owner());
 265
 266   AsyncRequestId async_request_id(get_client_id(), request_id);
 267
 268   notify_async_request(async_request_id,
 269                        RebuildObjectMapPayload(async_request_id),
 270                        prog_ctx, on_finish);
 271 }
 272
 273 template <typename I>
 274 void ImageWatcher<I>::notify_rename(const std::string &image_name,
 275                                     Context *on_finish) {
 276   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 277   ceph_assert(m_image_ctx.exclusive_lock &&
 278               !m_image_ctx.exclusive_lock->is_lock_owner());
 279
 280   notify_lock_owner(RenamePayload(image_name), on_finish);
 281 }
 282
 283 template <typename I>
 284 void ImageWatcher<I>::notify_update_features(uint64_t features, bool enabled,
 285                                              Context *on_finish) {
 286   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 287   ceph_assert(m_image_ctx.exclusive_lock &&
 288               !m_image_ctx.exclusive_lock->is_lock_owner());
 289
 290   notify_lock_owner(UpdateFeaturesPayload(features, enabled), on_finish);
 291 }
 292
 293 template <typename I>
 294 void ImageWatcher<I>::notify_migrate(uint64_t request_id,
 295                                      ProgressContext &prog_ctx,
 296                                      Context *on_finish) {
 297   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 298   ceph_assert(m_image_ctx.exclusive_lock &&
 299               !m_image_ctx.exclusive_lock->is_lock_owner());
 300
 301   AsyncRequestId async_request_id(get_client_id(), request_id);
 302
 303   notify_async_request(async_request_id, MigratePayload(async_request_id),
 304                        prog_ctx, on_finish);
 305 }
 306
 307 template <typename I>
 308 void ImageWatcher<I>::notify_sparsify(uint64_t request_id, size_t sparse_size,
 309                                       ProgressContext &prog_ctx,
 310                                       Context *on_finish) {
 311   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 312   ceph_assert(m_image_ctx.exclusive_lock &&
 313               !m_image_ctx.exclusive_lock->is_lock_owner());
 314
 315   AsyncRequestId async_request_id(get_client_id(), request_id);
 316
 317   notify_async_request(async_request_id,
 318                        SparsifyPayload(async_request_id, sparse_size), prog_ctx,
 319                        on_finish);
 320 }
 321
 322 template <typename I>
 323 void ImageWatcher<I>::notify_header_update(Context *on_finish) {
 324   ldout(m_image_ctx.cct, 10) << this << ": " << __func__ << dendl;
 325
 326   // supports legacy (empty buffer) clients
 327   send_notify(HeaderUpdatePayload(), on_finish);
 328 }
 329
 330 template <typename I>
 331 void ImageWatcher<I>::notify_header_update(librados::IoCtx &io_ctx,
 332                                            const std::string &oid) {
 333   // supports legacy (empty buffer) clients
 334   bufferlist bl;
 335   encode(NotifyMessage(HeaderUpdatePayload()), bl);
 336   io_ctx.notify2(oid, bl, watcher::Notifier::NOTIFY_TIMEOUT, nullptr);
 337 }
 338
 339 template <typename I>
 340 void ImageWatcher<I>::schedule_cancel_async_requests() {
 341   auto ctx = new LambdaContext(
 342     boost::bind(&ImageWatcher<I>::cancel_async_requests, this));
 343   m_task_finisher->queue(TASK_CODE_CANCEL_ASYNC_REQUESTS, ctx);
 344 }
 345
 346 template <typename I>
 347 void ImageWatcher<I>::cancel_async_requests() {
 348   std::unique_lock l{m_async_request_lock};
 349   for (std::map<AsyncRequestId, AsyncRequest>::iterator iter =
 350          m_async_requests.begin();
 351        iter != m_async_requests.end(); ++iter) {
 352     iter->second.first->complete(-ERESTART);
 353   }
 354   m_async_requests.clear();
 355 }
 356
 357 template <typename I>
 358 void ImageWatcher<I>::set_owner_client_id(const ClientId& client_id) {
 359   ceph_assert(ceph_mutex_is_locked(m_owner_client_id_lock));
 360   m_owner_client_id = client_id;
 361   ldout(m_image_ctx.cct, 10) << this << " current lock owner: "
 362                              << m_owner_client_id << dendl;
 363 }
 364
 365 template <typename I>
 366 ClientId ImageWatcher<I>::get_client_id() {
 367   std::shared_lock l{this->m_watch_lock};
 368   return ClientId(m_image_ctx.md_ctx.get_instance_id(), this->m_watch_handle);
 369 }
 370
 371 template <typename I>
 372 void ImageWatcher<I>::notify_acquired_lock() {
 373   ldout(m_image_ctx.cct, 10) << this << " notify acquired lock" << dendl;
 374
 375   ClientId client_id = get_client_id();
 376   {
 377     std::lock_guard owner_client_id_locker{m_owner_client_id_lock};
 378     set_owner_client_id(client_id);
 379   }
 380
 381   send_notify(AcquiredLockPayload(client_id));
 382 }
 383
 384 template <typename I>
 385 void ImageWatcher<I>::notify_released_lock() {
 386   ldout(m_image_ctx.cct, 10) << this << " notify released lock" << dendl;
 387
 388   {
 389     std::lock_guard owner_client_id_locker{m_owner_client_id_lock};
 390     set_owner_client_id(ClientId());
 391   }
 392
 393   send_notify(ReleasedLockPayload(get_client_id()));
 394 }
 395
 396 template <typename I>
 397 void ImageWatcher<I>::schedule_request_lock(bool use_timer, int timer_delay) {
 398   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 399
 400   if (m_image_ctx.exclusive_lock == nullptr) {
 401     // exclusive lock dynamically disabled via image refresh
 402     return;
 403   }
 404   ceph_assert(m_image_ctx.exclusive_lock &&
 405               !m_image_ctx.exclusive_lock->is_lock_owner());
 406
 407   std::shared_lock watch_locker{this->m_watch_lock};
 408   if (this->is_registered(this->m_watch_lock)) {
 409     ldout(m_image_ctx.cct, 15) << this << " requesting exclusive lock" << dendl;
 410
 411     auto ctx = new LambdaContext(
 412       boost::bind(&ImageWatcher<I>::notify_request_lock, this));
 413     if (use_timer) {
 414       if (timer_delay < 0) {
 415         timer_delay = RETRY_DELAY_SECONDS;
 416       }
 417       m_task_finisher->add_event_after(TASK_CODE_REQUEST_LOCK,
 418                                        timer_delay, ctx);
 419     } else {
 420       m_task_finisher->queue(TASK_CODE_REQUEST_LOCK, ctx);
 421     }
 422   }
 423 }
 424
 425 template <typename I>
 426 void ImageWatcher<I>::notify_request_lock() {
 427   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 428   std::shared_lock image_locker{m_image_ctx.image_lock};
 429
 430   // ExclusiveLock state machine can be dynamically disabled or
 431   // race with task cancel
 432   if (m_image_ctx.exclusive_lock == nullptr ||
 433       m_image_ctx.exclusive_lock->is_lock_owner()) {
 434     return;
 435   }
 436
 437   ldout(m_image_ctx.cct, 10) << this << " notify request lock" << dendl;
 438
 439   notify_lock_owner(RequestLockPayload(get_client_id(), false),
 440       create_context_callback<
 441         ImageWatcher, &ImageWatcher<I>::handle_request_lock>(this));
 442 }
 443
 444 template <typename I>
 445 void ImageWatcher<I>::handle_request_lock(int r) {
 446   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 447   std::shared_lock image_locker{m_image_ctx.image_lock};
 448
 449   // ExclusiveLock state machine cannot transition -- but can be
 450   // dynamically disabled
 451   if (m_image_ctx.exclusive_lock == nullptr) {
 452     return;
 453   }
 454
 455   if (r == -ETIMEDOUT) {
 456     ldout(m_image_ctx.cct, 5) << this << " timed out requesting lock: retrying"
 457                               << dendl;
 458
 459     // treat this is a dead client -- so retest acquiring the lock
 460     m_image_ctx.exclusive_lock->handle_peer_notification(0);
 461   } else if (r == -EROFS) {
 462     ldout(m_image_ctx.cct, 5) << this << " peer will not release lock" << dendl;
 463     m_image_ctx.exclusive_lock->handle_peer_notification(r);
 464   } else if (r < 0) {
 465     lderr(m_image_ctx.cct) << this << " error requesting lock: "
 466                            << cpp_strerror(r) << dendl;
 467     schedule_request_lock(true);
 468   } else {
 469     // lock owner acked -- but resend if we don't see them release the lock
 470     int retry_timeout = m_image_ctx.cct->_conf.template get_val<int64_t>(
 471       "client_notify_timeout");
 472     ldout(m_image_ctx.cct, 15) << this << " will retry in " << retry_timeout
 473                                << " seconds" << dendl;
 474     schedule_request_lock(true, retry_timeout);
 475   }
 476 }
 477
 478 template <typename I>
 479 void ImageWatcher<I>::notify_lock_owner(const Payload& payload,
 480                                         Context *on_finish) {
 481   ceph_assert(on_finish != nullptr);
 482   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 483
 484   bufferlist bl;
 485   encode(NotifyMessage(payload), bl);
 486
 487   NotifyLockOwner *notify_lock_owner = NotifyLockOwner::create(
 488     m_image_ctx, this->m_notifier, std::move(bl), on_finish);
 489   notify_lock_owner->send();
 490 }
 491
 492 template <typename I>
 493 Context *ImageWatcher<I>::remove_async_request(const AsyncRequestId &id) {
 494   std::unique_lock async_request_locker{m_async_request_lock};
 495   auto it = m_async_requests.find(id);
 496   if (it != m_async_requests.end()) {
 497     Context *on_complete = it->second.first;
 498     m_async_requests.erase(it);
 499     return on_complete;
 500   }
 501   return nullptr;
 502 }
 503
 504 template <typename I>
 505 void ImageWatcher<I>::schedule_async_request_timed_out(const AsyncRequestId &id) {
 506   ldout(m_image_ctx.cct, 20) << "scheduling async request time out: " << id
 507                              << dendl;
 508
 509   Context *ctx = new LambdaContext(boost::bind(
 510     &ImageWatcher<I>::async_request_timed_out, this, id));
 511
 512   Task task(TASK_CODE_ASYNC_REQUEST, id);
 513   m_task_finisher->cancel(task);
 514
 515   m_task_finisher->add_event_after(
 516     task, m_image_ctx.config.template get_val<uint64_t>("rbd_request_timed_out_seconds"),
 517     ctx);
 518 }
 519
 520 template <typename I>
 521 void ImageWatcher<I>::async_request_timed_out(const AsyncRequestId &id) {
 522   Context *on_complete = remove_async_request(id);
 523   if (on_complete != nullptr) {
 524     ldout(m_image_ctx.cct, 5) << "async request timed out: " << id << dendl;
 525     m_image_ctx.op_work_queue->queue(on_complete, -ETIMEDOUT);
 526   }
 527 }
 528
 529 template <typename I>
 530 void ImageWatcher<I>::notify_async_request(const AsyncRequestId &async_request_id,
 531                                            const Payload& payload,
 532                                            ProgressContext& prog_ctx,
 533                                            Context *on_finish) {
 534   ceph_assert(on_finish != nullptr);
 535   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
 536
 537   ldout(m_image_ctx.cct, 10) << this << " async request: " << async_request_id
 538                              << dendl;
 539
 540   Context *on_notify = new LambdaContext([this, async_request_id](int r) {
 541     if (r < 0) {
 542       // notification failed -- don't expect updates
 543       Context *on_complete = remove_async_request(async_request_id);
 544       if (on_complete != nullptr) {
 545         on_complete->complete(r);
 546       }
 547     }
 548   });
 549
 550   Context *on_complete = new LambdaContext(
 551     [this, async_request_id, on_finish](int r) {
 552       m_task_finisher->cancel(Task(TASK_CODE_ASYNC_REQUEST, async_request_id));
 553       on_finish->complete(r);
 554     });
 555
 556   {
 557     std::unique_lock async_request_locker{m_async_request_lock};
 558     m_async_requests[async_request_id] = AsyncRequest(on_complete, &prog_ctx);
 559   }
 560
 561   schedule_async_request_timed_out(async_request_id);
 562   notify_lock_owner(payload, on_notify);
 563 }
 564
 565 template <typename I>
 566 int ImageWatcher<I>::prepare_async_request(const AsyncRequestId& async_request_id,
 567                                            bool* new_request, Context** ctx,
 568                                            ProgressContext** prog_ctx) {
 569   if (async_request_id.client_id == get_client_id()) {
 570     return -ERESTART;
 571   } else {
 572     std::unique_lock l{m_async_request_lock};
 573     if (m_async_pending.count(async_request_id) == 0) {
 574       m_async_pending.insert(async_request_id);
 575       *new_request = true;
 576       *prog_ctx = new RemoteProgressContext(*this, async_request_id);
 577       *ctx = new RemoteContext(*this, async_request_id, *prog_ctx);
 578     } else {
 579       *new_request = false;
 580     }
 581   }
 582   return 0;
 583 }
 584
 585 template <typename I>
 586 bool ImageWatcher<I>::handle_payload(const HeaderUpdatePayload &payload,
 587                                      C_NotifyAck *ack_ctx) {
 588   ldout(m_image_ctx.cct, 10) << this << " image header updated" << dendl;
 589
 590   m_image_ctx.state->handle_update_notification();
 591   m_image_ctx.perfcounter->inc(l_librbd_notify);
 592   if (ack_ctx != nullptr) {
 593     m_image_ctx.state->flush_update_watchers(new C_ResponseMessage(ack_ctx));
 594     return false;
 595   }
 596   return true;
 597 }
 598
 599 template <typename I>
 600 bool ImageWatcher<I>::handle_payload(const AcquiredLockPayload &payload,
 601                                      C_NotifyAck *ack_ctx) {
 602   ldout(m_image_ctx.cct, 10) << this << " image exclusively locked announcement"
 603                              << dendl;
 604
 605   bool cancel_async_requests = true;
 606   if (payload.client_id.is_valid()) {
 607     std::lock_guard owner_client_id_locker{m_owner_client_id_lock};
 608     if (payload.client_id == m_owner_client_id) {
 609       cancel_async_requests = false;
 610     }
 611     set_owner_client_id(payload.client_id);
 612   }
 613
 614   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 615   if (m_image_ctx.exclusive_lock != nullptr) {
 616     // potentially wake up the exclusive lock state machine now that
 617     // a lock owner has advertised itself
 618     m_image_ctx.exclusive_lock->handle_peer_notification(0);
 619   }
 620   if (cancel_async_requests &&
 621       (m_image_ctx.exclusive_lock == nullptr ||
 622        !m_image_ctx.exclusive_lock->is_lock_owner())) {
 623     schedule_cancel_async_requests();
 624   }
 625   return true;
 626 }
 627
 628 template <typename I>
 629 bool ImageWatcher<I>::handle_payload(const ReleasedLockPayload &payload,
 630                                      C_NotifyAck *ack_ctx) {
 631   ldout(m_image_ctx.cct, 10) << this << " exclusive lock released" << dendl;
 632
 633   bool cancel_async_requests = true;
 634   if (payload.client_id.is_valid()) {
 635     std::lock_guard l{m_owner_client_id_lock};
 636     if (payload.client_id != m_owner_client_id) {
 637       ldout(m_image_ctx.cct, 10) << this << " unexpected owner: "
 638                                  << payload.client_id << " != "
 639                                  << m_owner_client_id << dendl;
 640       cancel_async_requests = false;
 641     } else {
 642       set_owner_client_id(ClientId());
 643     }
 644   }
 645
 646   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 647   if (cancel_async_requests &&
 648       (m_image_ctx.exclusive_lock == nullptr ||
 649        !m_image_ctx.exclusive_lock->is_lock_owner())) {
 650     schedule_cancel_async_requests();
 651   }
 652
 653   // alert the exclusive lock state machine that the lock is available
 654   if (m_image_ctx.exclusive_lock != nullptr &&
 655       !m_image_ctx.exclusive_lock->is_lock_owner()) {
 656     m_task_finisher->cancel(TASK_CODE_REQUEST_LOCK);
 657     m_image_ctx.exclusive_lock->handle_peer_notification(0);
 658   }
 659   return true;
 660 }
 661
 662 template <typename I>
 663 bool ImageWatcher<I>::handle_payload(const RequestLockPayload &payload,
 664                                      C_NotifyAck *ack_ctx) {
 665   ldout(m_image_ctx.cct, 10) << this << " exclusive lock requested" << dendl;
 666   if (payload.client_id == get_client_id()) {
 667     return true;
 668   }
 669
 670   std::shared_lock l{m_image_ctx.owner_lock};
 671   if (m_image_ctx.exclusive_lock != nullptr &&
 672       m_image_ctx.exclusive_lock->is_lock_owner()) {
 673     int r = 0;
 674     bool accept_request = m_image_ctx.exclusive_lock->accept_request(
 675       exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r);
 676
 677     if (accept_request) {
 678       ceph_assert(r == 0);
 679       std::lock_guard owner_client_id_locker{m_owner_client_id_lock};
 680       if (!m_owner_client_id.is_valid()) {
 681         return true;
 682       }
 683
 684       ldout(m_image_ctx.cct, 10) << this << " queuing release of exclusive lock"
 685                                  << dendl;
 686       r = m_image_ctx.get_exclusive_lock_policy()->lock_requested(
 687         payload.force);
 688     }
 689     encode(ResponseMessage(r), ack_ctx->out);
 690   }
 691   return true;
 692 }
 693
 694 template <typename I>
 695 bool ImageWatcher<I>::handle_payload(const AsyncProgressPayload &payload,
 696                                      C_NotifyAck *ack_ctx) {
 697   std::shared_lock l{m_async_request_lock};
 698   std::map<AsyncRequestId, AsyncRequest>::iterator req_it =
 699     m_async_requests.find(payload.async_request_id);
 700   if (req_it != m_async_requests.end()) {
 701     ldout(m_image_ctx.cct, 20) << this << " request progress: "
 702                                << payload.async_request_id << " @ "
 703                                << payload.offset << "/" << payload.total
 704                                << dendl;
 705     schedule_async_request_timed_out(payload.async_request_id);
 706     req_it->second.second->update_progress(payload.offset, payload.total);
 707   }
 708   return true;
 709 }
 710
 711 template <typename I>
 712 bool ImageWatcher<I>::handle_payload(const AsyncCompletePayload &payload,
 713                                      C_NotifyAck *ack_ctx) {
 714   Context *on_complete = remove_async_request(payload.async_request_id);
 715   if (on_complete != nullptr) {
 716     ldout(m_image_ctx.cct, 10) << this << " request finished: "
 717                                << payload.async_request_id << "="
 718                                << payload.result << dendl;
 719     on_complete->complete(payload.result);
 720   }
 721   return true;
 722 }
 723
 724 template <typename I>
 725 bool ImageWatcher<I>::handle_payload(const FlattenPayload &payload,
 726                                      C_NotifyAck *ack_ctx) {
 727
 728   std::shared_lock l{m_image_ctx.owner_lock};
 729   if (m_image_ctx.exclusive_lock != nullptr) {
 730     int r;
 731     if (m_image_ctx.exclusive_lock->accept_request(
 732           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 733       bool new_request;
 734       Context *ctx;
 735       ProgressContext *prog_ctx;
 736       r = prepare_async_request(payload.async_request_id, &new_request,
 737                                 &ctx, &prog_ctx);
 738       if (r == 0 && new_request) {
 739         ldout(m_image_ctx.cct, 10) << this << " remote flatten request: "
 740                                    << payload.async_request_id << dendl;
 741         m_image_ctx.operations->execute_flatten(*prog_ctx, ctx);
 742       }
 743
 744       encode(ResponseMessage(r), ack_ctx->out);
 745     } else if (r < 0) {
 746       encode(ResponseMessage(r), ack_ctx->out);
 747     }
 748   }
 749   return true;
 750 }
 751
 752 template <typename I>
 753 bool ImageWatcher<I>::handle_payload(const ResizePayload &payload,
 754                                      C_NotifyAck *ack_ctx) {
 755   std::shared_lock l{m_image_ctx.owner_lock};
 756   if (m_image_ctx.exclusive_lock != nullptr) {
 757     int r;
 758     if (m_image_ctx.exclusive_lock->accept_request(
 759           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 760       bool new_request;
 761       Context *ctx;
 762       ProgressContext *prog_ctx;
 763       r = prepare_async_request(payload.async_request_id, &new_request,
 764                                 &ctx, &prog_ctx);
 765       if (r == 0 && new_request) {
 766         ldout(m_image_ctx.cct, 10) << this << " remote resize request: "
 767                                    << payload.async_request_id << " "
 768                                    << payload.size << " "
 769                                    << payload.allow_shrink << dendl;
 770         m_image_ctx.operations->execute_resize(payload.size, payload.allow_shrink, *prog_ctx, ctx, 0);
 771       }
 772
 773       encode(ResponseMessage(r), ack_ctx->out);
 774     } else if (r < 0) {
 775       encode(ResponseMessage(r), ack_ctx->out);
 776     }
 777   }
 778   return true;
 779 }
 780
 781 template <typename I>
 782 bool ImageWatcher<I>::handle_payload(const SnapCreatePayload &payload,
 783                                      C_NotifyAck *ack_ctx) {
 784   std::shared_lock l{m_image_ctx.owner_lock};
 785   if (m_image_ctx.exclusive_lock != nullptr) {
 786     int r;
 787     auto request_type = exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL;
 788
 789     // rbd-mirror needs to accept forced promotion orphan snap create requests
 790     auto mirror_ns = boost::get<cls::rbd::MirrorSnapshotNamespace>(
 791       &payload.snap_namespace);
 792     if (mirror_ns != nullptr && mirror_ns->is_orphan()) {
 793       request_type = exclusive_lock::OPERATION_REQUEST_TYPE_FORCE_PROMOTION;
 794     }
 795
 796     if (m_image_ctx.exclusive_lock->accept_request(request_type, &r)) {
 797       ldout(m_image_ctx.cct, 10) << this << " remote snap_create request: "
 798                                  << payload.snap_name << dendl;
 799
 800       m_image_ctx.operations->execute_snap_create(payload.snap_namespace,
 801                                                   payload.snap_name,
 802                                                   new C_ResponseMessage(ack_ctx),
 803                                                   0, false);
 804       return false;
 805     } else if (r < 0) {
 806       encode(ResponseMessage(r), ack_ctx->out);
 807     }
 808   }
 809   return true;
 810 }
 811
 812 template <typename I>
 813 bool ImageWatcher<I>::handle_payload(const SnapRenamePayload &payload,
 814                                      C_NotifyAck *ack_ctx) {
 815   std::shared_lock l{m_image_ctx.owner_lock};
 816   if (m_image_ctx.exclusive_lock != nullptr) {
 817     int r;
 818     if (m_image_ctx.exclusive_lock->accept_request(
 819           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 820       ldout(m_image_ctx.cct, 10) << this << " remote snap_rename request: "
 821                                  << payload.snap_id << " to "
 822                                  << payload.snap_name << dendl;
 823
 824       m_image_ctx.operations->execute_snap_rename(payload.snap_id,
 825                                                   payload.snap_name,
 826                                                   new C_ResponseMessage(ack_ctx));
 827       return false;
 828     } else if (r < 0) {
 829       encode(ResponseMessage(r), ack_ctx->out);
 830     }
 831   }
 832   return true;
 833 }
 834
 835 template <typename I>
 836 bool ImageWatcher<I>::handle_payload(const SnapRemovePayload &payload,
 837                                      C_NotifyAck *ack_ctx) {
 838   std::shared_lock l{m_image_ctx.owner_lock};
 839   if (m_image_ctx.exclusive_lock != nullptr) {
 840     auto request_type = exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL;
 841     if (cls::rbd::get_snap_namespace_type(payload.snap_namespace) ==
 842         cls::rbd::SNAPSHOT_NAMESPACE_TYPE_TRASH) {
 843       request_type = exclusive_lock::OPERATION_REQUEST_TYPE_TRASH_SNAP_REMOVE;
 844     }
 845     int r;
 846     if (m_image_ctx.exclusive_lock->accept_request(request_type, &r)) {
 847       ldout(m_image_ctx.cct, 10) << this << " remote snap_remove request: "
 848                                  << payload.snap_name << dendl;
 849
 850       m_image_ctx.operations->execute_snap_remove(payload.snap_namespace,
 851                                                   payload.snap_name,
 852                                                   new C_ResponseMessage(ack_ctx));
 853       return false;
 854     } else if (r < 0) {
 855       encode(ResponseMessage(r), ack_ctx->out);
 856     }
 857   }
 858   return true;
 859 }
 860
 861 template <typename I>
 862 bool ImageWatcher<I>::handle_payload(const SnapProtectPayload& payload,
 863                                      C_NotifyAck *ack_ctx) {
 864   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 865   if (m_image_ctx.exclusive_lock != nullptr) {
 866     int r;
 867     if (m_image_ctx.exclusive_lock->accept_request(
 868           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 869       ldout(m_image_ctx.cct, 10) << this << " remote snap_protect request: "
 870                                  << payload.snap_name << dendl;
 871
 872       m_image_ctx.operations->execute_snap_protect(payload.snap_namespace,
 873                                                    payload.snap_name,
 874                                                    new C_ResponseMessage(ack_ctx));
 875       return false;
 876     } else if (r < 0) {
 877       encode(ResponseMessage(r), ack_ctx->out);
 878     }
 879   }
 880   return true;
 881 }
 882
 883 template <typename I>
 884 bool ImageWatcher<I>::handle_payload(const SnapUnprotectPayload& payload,
 885                                      C_NotifyAck *ack_ctx) {
 886   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 887   if (m_image_ctx.exclusive_lock != nullptr) {
 888     int r;
 889     if (m_image_ctx.exclusive_lock->accept_request(
 890           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 891       ldout(m_image_ctx.cct, 10) << this << " remote snap_unprotect request: "
 892                                  << payload.snap_name << dendl;
 893
 894       m_image_ctx.operations->execute_snap_unprotect(payload.snap_namespace,
 895                                                      payload.snap_name,
 896                                                      new C_ResponseMessage(ack_ctx));
 897       return false;
 898     } else if (r < 0) {
 899       encode(ResponseMessage(r), ack_ctx->out);
 900     }
 901   }
 902   return true;
 903 }
 904
 905 template <typename I>
 906 bool ImageWatcher<I>::handle_payload(const RebuildObjectMapPayload& payload,
 907                                      C_NotifyAck *ack_ctx) {
 908   std::shared_lock l{m_image_ctx.owner_lock};
 909   if (m_image_ctx.exclusive_lock != nullptr) {
 910     int r;
 911     if (m_image_ctx.exclusive_lock->accept_request(
 912           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 913       bool new_request;
 914       Context *ctx;
 915       ProgressContext *prog_ctx;
 916       r = prepare_async_request(payload.async_request_id, &new_request,
 917                                 &ctx, &prog_ctx);
 918       if (r == 0 && new_request) {
 919         ldout(m_image_ctx.cct, 10) << this
 920                                    << " remote rebuild object map request: "
 921                                    << payload.async_request_id << dendl;
 922         m_image_ctx.operations->execute_rebuild_object_map(*prog_ctx, ctx);
 923       }
 924
 925       encode(ResponseMessage(r), ack_ctx->out);
 926     } else if (r < 0) {
 927       encode(ResponseMessage(r), ack_ctx->out);
 928     }
 929   }
 930   return true;
 931 }
 932
 933 template <typename I>
 934 bool ImageWatcher<I>::handle_payload(const RenamePayload& payload,
 935                                      C_NotifyAck *ack_ctx) {
 936   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 937   if (m_image_ctx.exclusive_lock != nullptr) {
 938     int r;
 939     if (m_image_ctx.exclusive_lock->accept_request(
 940           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 941       ldout(m_image_ctx.cct, 10) << this << " remote rename request: "
 942                                  << payload.image_name << dendl;
 943
 944       m_image_ctx.operations->execute_rename(payload.image_name,
 945                                              new C_ResponseMessage(ack_ctx));
 946       return false;
 947     } else if (r < 0) {
 948       encode(ResponseMessage(r), ack_ctx->out);
 949     }
 950   }
 951   return true;
 952 }
 953
 954 template <typename I>
 955 bool ImageWatcher<I>::handle_payload(const UpdateFeaturesPayload& payload,
 956                                      C_NotifyAck *ack_ctx) {
 957   std::shared_lock owner_locker{m_image_ctx.owner_lock};
 958   if (m_image_ctx.exclusive_lock != nullptr) {
 959     int r;
 960     if (m_image_ctx.exclusive_lock->accept_request(
 961           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 962       ldout(m_image_ctx.cct, 10) << this << " remote update_features request: "
 963                                  << payload.features << " "
 964                                  << (payload.enabled ? "enabled" : "disabled")
 965                                  << dendl;
 966
 967       m_image_ctx.operations->execute_update_features(
 968         payload.features, payload.enabled, new C_ResponseMessage(ack_ctx), 0);
 969       return false;
 970     } else if (r < 0) {
 971       encode(ResponseMessage(r), ack_ctx->out);
 972     }
 973   }
 974   return true;
 975 }
 976
 977 template <typename I>
 978 bool ImageWatcher<I>::handle_payload(const MigratePayload &payload,
 979                                      C_NotifyAck *ack_ctx) {
 980
 981   std::shared_lock l{m_image_ctx.owner_lock};
 982   if (m_image_ctx.exclusive_lock != nullptr) {
 983     int r;
 984     if (m_image_ctx.exclusive_lock->accept_request(
 985           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
 986       bool new_request;
 987       Context *ctx;
 988       ProgressContext *prog_ctx;
 989       r = prepare_async_request(payload.async_request_id, &new_request,
 990                                 &ctx, &prog_ctx);
 991       if (r == 0 && new_request) {
 992         ldout(m_image_ctx.cct, 10) << this << " remote migrate request: "
 993                                    << payload.async_request_id << dendl;
 994         m_image_ctx.operations->execute_migrate(*prog_ctx, ctx);
 995       }
 996
 997       encode(ResponseMessage(r), ack_ctx->out);
 998     } else if (r < 0) {
 999       encode(ResponseMessage(r), ack_ctx->out);
1000     }
1001   }
1002   return true;
1003 }
1004
1005 template <typename I>
1006 bool ImageWatcher<I>::handle_payload(const SparsifyPayload &payload,
1007                                      C_NotifyAck *ack_ctx) {
1008   std::shared_lock l{m_image_ctx.owner_lock};
1009   if (m_image_ctx.exclusive_lock != nullptr) {
1010     int r;
1011     if (m_image_ctx.exclusive_lock->accept_request(
1012           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r)) {
1013       bool new_request;
1014       Context *ctx;
1015       ProgressContext *prog_ctx;
1016       r = prepare_async_request(payload.async_request_id, &new_request,
1017                                 &ctx, &prog_ctx);
1018       if (r == 0 && new_request) {
1019         ldout(m_image_ctx.cct, 10) << this << " remote sparsify request: "
1020                                    << payload.async_request_id << dendl;
1021         m_image_ctx.operations->execute_sparsify(payload.sparse_size, *prog_ctx,
1022                                                  ctx);
1023       }
1024
1025       encode(ResponseMessage(r), ack_ctx->out);
1026     } else if (r < 0) {
1027       encode(ResponseMessage(r), ack_ctx->out);
1028     }
1029   }
1030   return true;
1031 }
1032
1033 template <typename I>
1034 bool ImageWatcher<I>::handle_payload(const UnknownPayload &payload,
1035                                      C_NotifyAck *ack_ctx) {
1036   std::shared_lock l{m_image_ctx.owner_lock};
1037   if (m_image_ctx.exclusive_lock != nullptr) {
1038     int r;
1039     if (m_image_ctx.exclusive_lock->accept_request(
1040           exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, &r) || r < 0) {
1041       encode(ResponseMessage(-EOPNOTSUPP), ack_ctx->out);
1042     }
1043   }
1044   return true;
1045 }
1046
1047 template <typename I>
1048 void ImageWatcher<I>::process_payload(uint64_t notify_id, uint64_t handle,
1049                                       const Payload &payload) {
1050   apply_visitor(HandlePayloadVisitor<ImageWatcher<I>>(this, notify_id, handle),
1051                 payload);
1052 }
1053
1054 template <typename I>
1055 void ImageWatcher<I>::handle_notify(uint64_t notify_id, uint64_t handle,
1056                                     uint64_t notifier_id, bufferlist &bl) {
1057   NotifyMessage notify_message;
1058   if (bl.length() == 0) {
1059     // legacy notification for header updates
1060     notify_message = NotifyMessage(HeaderUpdatePayload());
1061   } else {
1062     try {
1063       auto iter = bl.cbegin();
1064       decode(notify_message, iter);
1065     } catch (const buffer::error &err) {
1066       lderr(m_image_ctx.cct) << this << " error decoding image notification: "
1067                              << err.what() << dendl;
1068       return;
1069     }
1070   }
1071
1072   // if an image refresh is required, refresh before processing the request
1073   if (notify_message.check_for_refresh() &&
1074       m_image_ctx.state->is_refresh_required()) {
1075     m_image_ctx.state->refresh(new C_ProcessPayload(this, notify_id, handle,
1076                                                     notify_message.payload));
1077   } else {
1078     process_payload(notify_id, handle, notify_message.payload);
1079   }
1080 }
1081
1082 template <typename I>
1083 void ImageWatcher<I>::handle_error(uint64_t handle, int err) {
1084   lderr(m_image_ctx.cct) << this << " image watch failed: " << handle << ", "
1085                          << cpp_strerror(err) << dendl;
1086
1087   {
1088     std::lock_guard l{m_owner_client_id_lock};
1089     set_owner_client_id(ClientId());
1090   }
1091
1092   Watcher::handle_error(handle, err);
1093 }
1094
1095 template <typename I>
1096 void ImageWatcher<I>::handle_rewatch_complete(int r) {
1097   CephContext *cct = m_image_ctx.cct;
1098   ldout(cct, 10) << this << " " << __func__ << ": r=" << r << dendl;
1099
1100   {
1101     std::shared_lock owner_locker{m_image_ctx.owner_lock};
1102     if (m_image_ctx.exclusive_lock != nullptr) {
1103       // update the lock cookie with the new watch handle
1104       m_image_ctx.exclusive_lock->reacquire_lock(nullptr);
1105     }
1106   }
1107
1108   // image might have been updated while we didn't have active watch
1109   handle_payload(HeaderUpdatePayload(), nullptr);
1110 }
1111
1112 template <typename I>
1113 void ImageWatcher<I>::send_notify(const Payload &payload, Context *ctx) {
1114   bufferlist bl;
1115
1116   encode(NotifyMessage(payload), bl);
1117   Watcher::send_notify(bl, nullptr, ctx);
1118 }
1119
1120 template <typename I>
1121 void ImageWatcher<I>::RemoteContext::finish(int r) {
1122   m_image_watcher.schedule_async_complete(m_async_request_id, r);
1123 }
1124
1125 template <typename I>
1126 void ImageWatcher<I>::C_ResponseMessage::finish(int r) {
1127   CephContext *cct = notify_ack->cct;
1128   ldout(cct, 10) << this << " C_ResponseMessage: r=" << r << dendl;
1129
1130   encode(ResponseMessage(r), notify_ack->out);
1131   notify_ack->complete(0);
1132 }
1133
1134 } // namespace librbd
1135
1136 template class librbd::ImageWatcher<librbd::ImageCtx>;