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