ceph/src/librbd/ImageState.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/ImageState.h"
   5 #include "include/rbd/librbd.hpp"
   6 #include "common/dout.h"
   7 #include "common/errno.h"
   8 #include "common/Cond.h"
   9 #include "common/WorkQueue.h"
  10 #include "librbd/AsioEngine.h"
  11 #include "librbd/ImageCtx.h"
  12 #include "librbd/TaskFinisher.h"
  13 #include "librbd/Utils.h"
  14 #include "librbd/asio/ContextWQ.h"
  15 #include "librbd/image/CloseRequest.h"
  16 #include "librbd/image/OpenRequest.h"
  17 #include "librbd/image/RefreshRequest.h"
  18 #include "librbd/image/SetSnapRequest.h"
  19
  20 #define dout_subsys ceph_subsys_rbd
  21 #undef dout_prefix
  22 #define dout_prefix *_dout << "librbd::ImageState: " << this << " "
  23
  24 namespace librbd {
  25
  26 using util::create_async_context_callback;
  27 using util::create_context_callback;
  28
  29 class ImageUpdateWatchers {
  30 public:
  31
  32   explicit ImageUpdateWatchers(CephContext *cct) : m_cct(cct),
  33     m_lock(ceph::make_mutex(util::unique_lock_name("librbd::ImageUpdateWatchers::m_lock", this))) {
  34   }
  35
  36   ~ImageUpdateWatchers() {
  37     ceph_assert(m_watchers.empty());
  38     ceph_assert(m_in_flight.empty());
  39     ceph_assert(m_pending_unregister.empty());
  40     ceph_assert(m_on_shut_down_finish == nullptr);
  41
  42     destroy_work_queue();
  43   }
  44
  45   void flush(Context *on_finish) {
  46     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << dendl;
  47     {
  48       std::lock_guard locker{m_lock};
  49       if (!m_in_flight.empty()) {
  50         Context *ctx = new LambdaContext(
  51           [this, on_finish](int r) {
  52             ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
  53                              << ": completing flush" << dendl;
  54             on_finish->complete(r);
  55           });
  56         m_work_queue->queue(ctx, 0);
  57         return;
  58       }
  59     }
  60     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
  61                      << ": completing flush" << dendl;
  62     on_finish->complete(0);
  63   }
  64
  65   void shut_down(Context *on_finish) {
  66     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << dendl;
  67     {
  68       std::lock_guard locker{m_lock};
  69       ceph_assert(m_on_shut_down_finish == nullptr);
  70       m_watchers.clear();
  71       if (!m_in_flight.empty()) {
  72         m_on_shut_down_finish = on_finish;
  73         return;
  74       }
  75     }
  76     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
  77                      << ": completing shut down" << dendl;
  78     on_finish->complete(0);
  79   }
  80
  81   void register_watcher(UpdateWatchCtx *watcher, uint64_t *handle) {
  82     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << ": watcher="
  83                      << watcher << dendl;
  84
  85     std::lock_guard locker{m_lock};
  86     ceph_assert(m_on_shut_down_finish == nullptr);
  87
  88     create_work_queue();
  89
  90     *handle = m_next_handle++;
  91     m_watchers.insert(std::make_pair(*handle, watcher));
  92   }
  93
  94   void unregister_watcher(uint64_t handle, Context *on_finish) {
  95     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << ": handle="
  96                      << handle << dendl;
  97     int r = 0;
  98     {
  99       std::lock_guard locker{m_lock};
 100       auto it = m_watchers.find(handle);
 101       if (it == m_watchers.end()) {
 102         r = -ENOENT;
 103       } else {
 104         if (m_in_flight.find(handle) != m_in_flight.end()) {
 105           ceph_assert(m_pending_unregister.find(handle) == m_pending_unregister.end());
 106           m_pending_unregister[handle] = on_finish;
 107           on_finish = nullptr;
 108         }
 109         m_watchers.erase(it);
 110       }
 111     }
 112
 113     if (on_finish) {
 114       ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
 115                        << ": completing unregister" << dendl;
 116       on_finish->complete(r);
 117     }
 118   }
 119
 120   void notify() {
 121     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << dendl;
 122
 123     std::lock_guard locker{m_lock};
 124     for (auto it : m_watchers) {
 125       send_notify(it.first, it.second);
 126     }
 127   }
 128
 129   void send_notify(uint64_t handle, UpdateWatchCtx *watcher) {
 130     ceph_assert(ceph_mutex_is_locked(m_lock));
 131
 132     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << ": handle="
 133                      << handle << ", watcher=" << watcher << dendl;
 134
 135     m_in_flight.insert(handle);
 136
 137     Context *ctx = new LambdaContext(
 138       [this, handle, watcher](int r) {
 139         handle_notify(handle, watcher);
 140       });
 141
 142     m_work_queue->queue(ctx, 0);
 143   }
 144
 145   void handle_notify(uint64_t handle, UpdateWatchCtx *watcher) {
 146
 147     ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__ << ": handle="
 148                      << handle << ", watcher=" << watcher << dendl;
 149
 150     watcher->handle_notify();
 151
 152     Context *on_unregister_finish = nullptr;
 153     Context *on_shut_down_finish = nullptr;
 154
 155     {
 156       std::lock_guard locker{m_lock};
 157
 158       auto in_flight_it = m_in_flight.find(handle);
 159       ceph_assert(in_flight_it != m_in_flight.end());
 160       m_in_flight.erase(in_flight_it);
 161
 162       // If there is no more in flight notifications for this watcher
 163       // and it is pending unregister, complete it now.
 164       if (m_in_flight.find(handle) == m_in_flight.end()) {
 165         auto it = m_pending_unregister.find(handle);
 166         if (it != m_pending_unregister.end()) {
 167           on_unregister_finish = it->second;
 168           m_pending_unregister.erase(it);
 169         }
 170       }
 171
 172       if (m_in_flight.empty()) {
 173         ceph_assert(m_pending_unregister.empty());
 174         if (m_on_shut_down_finish != nullptr) {
 175           std::swap(m_on_shut_down_finish, on_shut_down_finish);
 176         }
 177       }
 178     }
 179
 180     if (on_unregister_finish != nullptr) {
 181       ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
 182                        << ": completing unregister" << dendl;
 183       on_unregister_finish->complete(0);
 184     }
 185
 186     if (on_shut_down_finish != nullptr) {
 187       ldout(m_cct, 20) << "ImageUpdateWatchers::" << __func__
 188                        << ": completing shut down" << dendl;
 189       on_shut_down_finish->complete(0);
 190     }
 191   }
 192
 193 private:
 194   class ThreadPoolSingleton : public ThreadPool {
 195   public:
 196     explicit ThreadPoolSingleton(CephContext *cct)
 197       : ThreadPool(cct, "librbd::ImageUpdateWatchers::thread_pool", "tp_librbd",
 198                    1) {
 199       start();
 200     }
 201     ~ThreadPoolSingleton() override {
 202       stop();
 203     }
 204   };
 205
 206   CephContext *m_cct;
 207   ceph::mutex m_lock;
 208   ContextWQ *m_work_queue = nullptr;
 209   std::map<uint64_t, UpdateWatchCtx*> m_watchers;
 210   uint64_t m_next_handle = 0;
 211   std::multiset<uint64_t> m_in_flight;
 212   std::map<uint64_t, Context*> m_pending_unregister;
 213   Context *m_on_shut_down_finish = nullptr;
 214
 215   void create_work_queue() {
 216     if (m_work_queue != nullptr) {
 217       return;
 218     }
 219     auto& thread_pool = m_cct->lookup_or_create_singleton_object<
 220       ThreadPoolSingleton>("librbd::ImageUpdateWatchers::thread_pool",
 221                            false, m_cct);
 222     m_work_queue = new ContextWQ("librbd::ImageUpdateWatchers::work_queue",
 223                                  ceph::make_timespan(
 224                                    m_cct->_conf.get_val<uint64_t>("rbd_op_thread_timeout")),
 225                                  &thread_pool);
 226   }
 227
 228   void destroy_work_queue() {
 229     if (m_work_queue == nullptr) {
 230       return;
 231     }
 232     m_work_queue->drain();
 233     delete m_work_queue;
 234   }
 235 };
 236
 237 class QuiesceWatchers {
 238 public:
 239   explicit QuiesceWatchers(CephContext *cct, asio::ContextWQ* work_queue)
 240     : m_cct(cct),
 241       m_work_queue(work_queue),
 242       m_lock(ceph::make_mutex(util::unique_lock_name(
 243         "librbd::QuiesceWatchers::m_lock", this))) {
 244   }
 245
 246   ~QuiesceWatchers() {
 247     ceph_assert(m_pending_unregister.empty());
 248     ceph_assert(m_on_notify == nullptr);
 249   }
 250
 251   void register_watcher(QuiesceWatchCtx *watcher, uint64_t *handle) {
 252     ldout(m_cct, 20) << "QuiesceWatchers::" << __func__ << ": watcher="
 253                      << watcher << dendl;
 254
 255     std::lock_guard locker{m_lock};
 256
 257     *handle = m_next_handle++;
 258     m_watchers[*handle] = watcher;
 259   }
 260
 261   void unregister_watcher(uint64_t handle, Context *on_finish) {
 262     int r = 0;
 263     {
 264       std::lock_guard locker{m_lock};
 265       auto it = m_watchers.find(handle);
 266       if (it == m_watchers.end()) {
 267         r = -ENOENT;
 268       } else {
 269         if (m_on_notify != nullptr) {
 270           ceph_assert(!m_pending_unregister.count(handle));
 271           m_pending_unregister[handle] = on_finish;
 272           on_finish = nullptr;
 273         }
 274         m_watchers.erase(it);
 275       }
 276     }
 277
 278     if (on_finish) {
 279       ldout(m_cct, 20) << "QuiesceWatchers::" << __func__
 280                        << ": completing unregister " << handle << dendl;
 281       on_finish->complete(r);
 282     }
 283   }
 284
 285   void notify_quiesce(Context *on_finish) {
 286     std::lock_guard locker{m_lock};
 287     if (m_blocked) {
 288       ldout(m_cct, 20) << "QuiesceWatchers::" << __func__ << ": queue" << dendl;
 289       m_pending_notify.push_back(on_finish);
 290       return;
 291     }
 292
 293     notify(QUIESCE, on_finish);
 294   }
 295
 296   void notify_unquiesce(Context *on_finish) {
 297     std::lock_guard locker{m_lock};
 298
 299     notify(UNQUIESCE, on_finish);
 300   }
 301
 302   void quiesce_complete(uint64_t handle, int r) {
 303     Context *on_notify = nullptr;
 304     {
 305       std::lock_guard locker{m_lock};
 306       ceph_assert(m_on_notify != nullptr);
 307       ceph_assert(m_handle_quiesce_cnt > 0);
 308
 309       m_handle_quiesce_cnt--;
 310
 311       if (r < 0) {
 312         ldout(m_cct, 10) << "QuiesceWatchers::" << __func__ << ": watcher "
 313                          << handle << " failed" << dendl;
 314         m_failed_watchers.insert(handle);
 315         m_ret_val = r;
 316       }
 317
 318       if (m_handle_quiesce_cnt > 0) {
 319         return;
 320       }
 321
 322       std::swap(on_notify, m_on_notify);
 323       r = m_ret_val;
 324     }
 325
 326     on_notify->complete(r);
 327   }
 328
 329 private:
 330   enum EventType {QUIESCE, UNQUIESCE};
 331
 332   CephContext *m_cct;
 333   asio::ContextWQ *m_work_queue;
 334
 335   ceph::mutex m_lock;
 336   std::map<uint64_t, QuiesceWatchCtx*> m_watchers;
 337   uint64_t m_next_handle = 0;
 338   Context *m_on_notify = nullptr;
 339   std::list<Context *> m_pending_notify;
 340   std::map<uint64_t, Context*> m_pending_unregister;
 341   uint64_t m_handle_quiesce_cnt = 0;
 342   std::set<uint64_t> m_failed_watchers;
 343   bool m_blocked = false;
 344   int m_ret_val = 0;
 345
 346   void notify(EventType event_type, Context *on_finish) {
 347     ceph_assert(ceph_mutex_is_locked(m_lock));
 348
 349     if (m_watchers.empty()) {
 350       m_work_queue->queue(on_finish);
 351       return;
 352     }
 353
 354     ldout(m_cct, 20) << "QuiesceWatchers::" << __func__ << " event: "
 355                      << event_type << dendl;
 356
 357     Context *ctx = nullptr;
 358     if (event_type == QUIESCE) {
 359       ceph_assert(!m_blocked);
 360       ceph_assert(m_handle_quiesce_cnt == 0);
 361
 362       m_blocked = true;
 363       m_handle_quiesce_cnt = m_watchers.size();
 364       m_failed_watchers.clear();
 365       m_ret_val = 0;
 366     } else {
 367       ceph_assert(event_type == UNQUIESCE);
 368       ceph_assert(m_blocked);
 369
 370       ctx = create_async_context_callback(
 371         m_work_queue, create_context_callback<
 372           QuiesceWatchers, &QuiesceWatchers::handle_notify_unquiesce>(this));
 373     }
 374     auto gather_ctx = new C_Gather(m_cct, ctx);
 375
 376     ceph_assert(m_on_notify == nullptr);
 377
 378     m_on_notify = on_finish;
 379
 380     for (auto &[handle, watcher] : m_watchers) {
 381       send_notify(handle, watcher, event_type, gather_ctx->new_sub());
 382     }
 383
 384     gather_ctx->activate();
 385   }
 386
 387   void send_notify(uint64_t handle, QuiesceWatchCtx *watcher,
 388                    EventType event_type, Context *on_finish) {
 389     auto ctx = new LambdaContext(
 390       [this, handle, watcher, event_type, on_finish](int) {
 391         ldout(m_cct, 20) << "QuiesceWatchers::" << __func__ << ": handle="
 392                          << handle << ", event_type=" << event_type << dendl;
 393         switch (event_type) {
 394         case QUIESCE:
 395           watcher->handle_quiesce();
 396           break;
 397         case UNQUIESCE:
 398           {
 399             std::lock_guard locker{m_lock};
 400
 401             if (m_failed_watchers.count(handle)) {
 402               ldout(m_cct, 20) << "QuiesceWatchers::" << __func__
 403                                << ": skip for failed watcher" << dendl;
 404               break;
 405             }
 406           }
 407           watcher->handle_unquiesce();
 408           break;
 409         default:
 410           ceph_abort_msgf("invalid event_type %d", event_type);
 411         }
 412
 413         on_finish->complete(0);
 414       });
 415
 416     m_work_queue->queue(ctx);
 417   }
 418
 419   void handle_notify_unquiesce(int r) {
 420     ldout(m_cct, 20) << "QuiesceWatchers::" << __func__ << ": r=" << r
 421                      << dendl;
 422
 423     ceph_assert(r == 0);
 424
 425     std::unique_lock locker{m_lock};
 426
 427     if (!m_pending_unregister.empty()) {
 428       std::map<uint64_t, Context*> pending_unregister;
 429       std::swap(pending_unregister, m_pending_unregister);
 430       locker.unlock();
 431       for (auto &it : pending_unregister) {
 432         ldout(m_cct, 20) << "QuiesceWatchers::" << __func__
 433                          << ": completing unregister " << it.first << dendl;
 434         it.second->complete(0);
 435       }
 436       locker.lock();
 437     }
 438
 439     Context *on_notify = nullptr;
 440     std::swap(on_notify, m_on_notify);
 441
 442     ceph_assert(m_blocked);
 443     m_blocked = false;
 444
 445     if (!m_pending_notify.empty()) {
 446       auto on_finish = m_pending_notify.front();
 447       m_pending_notify.pop_front();
 448       notify(QUIESCE, on_finish);
 449     }
 450
 451     locker.unlock();
 452     on_notify->complete(0);
 453   }
 454 };
 455
 456 template <typename I>
 457 ImageState<I>::ImageState(I *image_ctx)
 458   : m_image_ctx(image_ctx), m_state(STATE_UNINITIALIZED),
 459     m_lock(ceph::make_mutex(util::unique_lock_name("librbd::ImageState::m_lock", this))),
 460     m_last_refresh(0), m_refresh_seq(0),
 461     m_update_watchers(new ImageUpdateWatchers(image_ctx->cct)),
 462     m_quiesce_watchers(new QuiesceWatchers(
 463       image_ctx->cct, image_ctx->asio_engine->get_work_queue())) {
 464 }
 465
 466 template <typename I>
 467 ImageState<I>::~ImageState() {
 468   ceph_assert(m_state == STATE_UNINITIALIZED || m_state == STATE_CLOSED);
 469   delete m_update_watchers;
 470   delete m_quiesce_watchers;
 471 }
 472
 473 template <typename I>
 474 int ImageState<I>::open(uint64_t flags) {
 475   C_SaferCond ctx;
 476   open(flags, &ctx);
 477
 478   int r = ctx.wait();
 479   return r;
 480 }
 481
 482 template <typename I>
 483 void ImageState<I>::open(uint64_t flags, Context *on_finish) {
 484   CephContext *cct = m_image_ctx->cct;
 485   ldout(cct, 20) << __func__ << dendl;
 486
 487   m_lock.lock();
 488   ceph_assert(m_state == STATE_UNINITIALIZED);
 489   m_open_flags = flags;
 490
 491   Action action(ACTION_TYPE_OPEN);
 492   action.refresh_seq = m_refresh_seq;
 493
 494   execute_action_unlock(action, on_finish);
 495 }
 496
 497 template <typename I>
 498 int ImageState<I>::close() {
 499   C_SaferCond ctx;
 500   close(&ctx);
 501
 502   int r = ctx.wait();
 503   return r;
 504 }
 505
 506 template <typename I>
 507 void ImageState<I>::close(Context *on_finish) {
 508   CephContext *cct = m_image_ctx->cct;
 509   ldout(cct, 20) << __func__ << dendl;
 510
 511   m_lock.lock();
 512   ceph_assert(!is_closed());
 513
 514   Action action(ACTION_TYPE_CLOSE);
 515   action.refresh_seq = m_refresh_seq;
 516   execute_action_unlock(action, on_finish);
 517 }
 518
 519 template <typename I>
 520 void ImageState<I>::handle_update_notification() {
 521   std::lock_guard locker{m_lock};
 522   ++m_refresh_seq;
 523
 524   CephContext *cct = m_image_ctx->cct;
 525   ldout(cct, 20) << __func__ << ": refresh_seq = " << m_refresh_seq << ", "
 526                  << "last_refresh = " << m_last_refresh << dendl;
 527
 528   switch (m_state) {
 529   case STATE_UNINITIALIZED:
 530   case STATE_CLOSED:
 531   case STATE_OPENING:
 532   case STATE_CLOSING:
 533     ldout(cct, 5) << "dropping update notification to watchers" << dendl;
 534     return;
 535   default:
 536     break;
 537   }
 538
 539   m_update_watchers->notify();
 540 }
 541
 542 template <typename I>
 543 bool ImageState<I>::is_refresh_required() const {
 544   std::lock_guard locker{m_lock};
 545   return (m_last_refresh != m_refresh_seq || find_pending_refresh() != nullptr);
 546 }
 547
 548 template <typename I>
 549 int ImageState<I>::refresh() {
 550   C_SaferCond refresh_ctx;
 551   refresh(&refresh_ctx);
 552   return refresh_ctx.wait();
 553 }
 554
 555 template <typename I>
 556 void ImageState<I>::refresh(Context *on_finish) {
 557   CephContext *cct = m_image_ctx->cct;
 558   ldout(cct, 20) << __func__ << dendl;
 559
 560   m_lock.lock();
 561   if (is_closed()) {
 562     m_lock.unlock();
 563     on_finish->complete(-ESHUTDOWN);
 564     return;
 565   }
 566
 567   Action action(ACTION_TYPE_REFRESH);
 568   action.refresh_seq = m_refresh_seq;
 569   execute_action_unlock(action, on_finish);
 570 }
 571
 572 template <typename I>
 573 int ImageState<I>::refresh_if_required() {
 574   C_SaferCond ctx;
 575   {
 576     m_lock.lock();
 577     Action action(ACTION_TYPE_REFRESH);
 578     action.refresh_seq = m_refresh_seq;
 579
 580     auto refresh_action = find_pending_refresh();
 581     if (refresh_action != nullptr) {
 582       // if a refresh is in-flight, delay until it is finished
 583       action = *refresh_action;
 584     } else if (m_last_refresh == m_refresh_seq) {
 585       m_lock.unlock();
 586       return 0;
 587     } else if (is_closed()) {
 588       m_lock.unlock();
 589       return -ESHUTDOWN;
 590     }
 591
 592     execute_action_unlock(action, &ctx);
 593   }
 594
 595   return ctx.wait();
 596 }
 597
 598 template <typename I>
 599 const typename ImageState<I>::Action *
 600 ImageState<I>::find_pending_refresh() const {
 601   ceph_assert(ceph_mutex_is_locked(m_lock));
 602
 603   auto it = std::find_if(m_actions_contexts.rbegin(),
 604                          m_actions_contexts.rend(),
 605                          [](const ActionContexts& action_contexts) {
 606       return (action_contexts.first == ACTION_TYPE_REFRESH);
 607     });
 608   if (it != m_actions_contexts.rend()) {
 609     return &it->first;
 610   }
 611   return nullptr;
 612 }
 613
 614 template <typename I>
 615 void ImageState<I>::snap_set(uint64_t snap_id, Context *on_finish) {
 616   CephContext *cct = m_image_ctx->cct;
 617   ldout(cct, 20) << __func__ << ": snap_id=" << snap_id << dendl;
 618
 619   Action action(ACTION_TYPE_SET_SNAP);
 620   action.snap_id = snap_id;
 621
 622   m_lock.lock();
 623   execute_action_unlock(action, on_finish);
 624 }
 625
 626 template <typename I>
 627 void ImageState<I>::prepare_lock(Context *on_ready) {
 628   CephContext *cct = m_image_ctx->cct;
 629   ldout(cct, 10) << __func__ << dendl;
 630
 631   m_lock.lock();
 632   if (is_closed()) {
 633     m_lock.unlock();
 634     on_ready->complete(-ESHUTDOWN);
 635     return;
 636   }
 637
 638   Action action(ACTION_TYPE_LOCK);
 639   action.on_ready = on_ready;
 640   execute_action_unlock(action, nullptr);
 641 }
 642
 643 template <typename I>
 644 void ImageState<I>::handle_prepare_lock_complete() {
 645   CephContext *cct = m_image_ctx->cct;
 646   ldout(cct, 10) << __func__ << dendl;
 647
 648   m_lock.lock();
 649   if (m_state != STATE_PREPARING_LOCK) {
 650     m_lock.unlock();
 651     return;
 652   }
 653
 654   complete_action_unlock(STATE_OPEN, 0);
 655 }
 656
 657 template <typename I>
 658 int ImageState<I>::register_update_watcher(UpdateWatchCtx *watcher,
 659                                          uint64_t *handle) {
 660   CephContext *cct = m_image_ctx->cct;
 661   ldout(cct, 20) << __func__ << dendl;
 662
 663   m_update_watchers->register_watcher(watcher, handle);
 664
 665   ldout(cct, 20) << __func__ << ": handle=" << *handle << dendl;
 666   return 0;
 667 }
 668
 669 template <typename I>
 670 void ImageState<I>::unregister_update_watcher(uint64_t handle,
 671                                               Context *on_finish) {
 672   CephContext *cct = m_image_ctx->cct;
 673   ldout(cct, 20) << __func__ << ": handle=" << handle << dendl;
 674
 675   m_update_watchers->unregister_watcher(handle, on_finish);
 676 }
 677
 678 template <typename I>
 679 int ImageState<I>::unregister_update_watcher(uint64_t handle) {
 680   C_SaferCond ctx;
 681   unregister_update_watcher(handle, &ctx);
 682   return ctx.wait();
 683 }
 684
 685 template <typename I>
 686 void ImageState<I>::flush_update_watchers(Context *on_finish) {
 687   CephContext *cct = m_image_ctx->cct;
 688   ldout(cct, 20) << __func__ << dendl;
 689
 690   m_update_watchers->flush(on_finish);
 691 }
 692
 693 template <typename I>
 694 void ImageState<I>::shut_down_update_watchers(Context *on_finish) {
 695   CephContext *cct = m_image_ctx->cct;
 696   ldout(cct, 20) << __func__ << dendl;
 697
 698   m_update_watchers->shut_down(on_finish);
 699 }
 700
 701 template <typename I>
 702 bool ImageState<I>::is_transition_state() const {
 703   switch (m_state) {
 704   case STATE_UNINITIALIZED:
 705   case STATE_OPEN:
 706   case STATE_CLOSED:
 707     return false;
 708   case STATE_OPENING:
 709   case STATE_CLOSING:
 710   case STATE_REFRESHING:
 711   case STATE_SETTING_SNAP:
 712   case STATE_PREPARING_LOCK:
 713     break;
 714   }
 715   return true;
 716 }
 717
 718 template <typename I>
 719 bool ImageState<I>::is_closed() const {
 720   ceph_assert(ceph_mutex_is_locked(m_lock));
 721
 722   return ((m_state == STATE_CLOSED) ||
 723           (!m_actions_contexts.empty() &&
 724            m_actions_contexts.back().first.action_type == ACTION_TYPE_CLOSE));
 725 }
 726
 727 template <typename I>
 728 void ImageState<I>::append_context(const Action &action, Context *context) {
 729   ceph_assert(ceph_mutex_is_locked(m_lock));
 730
 731   ActionContexts *action_contexts = nullptr;
 732   for (auto &action_ctxs : m_actions_contexts) {
 733     if (action == action_ctxs.first) {
 734       action_contexts = &action_ctxs;
 735       break;
 736     }
 737   }
 738
 739   if (action_contexts == nullptr) {
 740     m_actions_contexts.push_back({action, {}});
 741     action_contexts = &m_actions_contexts.back();
 742   }
 743
 744   if (context != nullptr) {
 745     action_contexts->second.push_back(context);
 746   }
 747 }
 748
 749 template <typename I>
 750 void ImageState<I>::execute_next_action_unlock() {
 751   ceph_assert(ceph_mutex_is_locked(m_lock));
 752   ceph_assert(!m_actions_contexts.empty());
 753   switch (m_actions_contexts.front().first.action_type) {
 754   case ACTION_TYPE_OPEN:
 755     send_open_unlock();
 756     return;
 757   case ACTION_TYPE_CLOSE:
 758     send_close_unlock();
 759     return;
 760   case ACTION_TYPE_REFRESH:
 761     send_refresh_unlock();
 762     return;
 763   case ACTION_TYPE_SET_SNAP:
 764     send_set_snap_unlock();
 765     return;
 766   case ACTION_TYPE_LOCK:
 767     send_prepare_lock_unlock();
 768     return;
 769   }
 770   ceph_abort();
 771 }
 772
 773 template <typename I>
 774 void ImageState<I>::execute_action_unlock(const Action &action,
 775                                           Context *on_finish) {
 776   ceph_assert(ceph_mutex_is_locked(m_lock));
 777
 778   append_context(action, on_finish);
 779   if (!is_transition_state()) {
 780     execute_next_action_unlock();
 781   } else {
 782     m_lock.unlock();
 783   }
 784 }
 785
 786 template <typename I>
 787 void ImageState<I>::complete_action_unlock(State next_state, int r) {
 788   ceph_assert(ceph_mutex_is_locked(m_lock));
 789   ceph_assert(!m_actions_contexts.empty());
 790
 791   ActionContexts action_contexts(std::move(m_actions_contexts.front()));
 792   m_actions_contexts.pop_front();
 793
 794   m_state = next_state;
 795   m_lock.unlock();
 796
 797   if (next_state == STATE_CLOSED ||
 798       (next_state == STATE_UNINITIALIZED && r < 0)) {
 799     // the ImageCtx must be deleted outside the scope of its callback threads
 800     auto ctx = new LambdaContext(
 801       [image_ctx=m_image_ctx, contexts=std::move(action_contexts.second)]
 802       (int r) {
 803         delete image_ctx;
 804         for (auto ctx : contexts) {
 805           ctx->complete(r);
 806         }
 807       });
 808     TaskFinisherSingleton::get_singleton(m_image_ctx->cct).queue(ctx, r);
 809   } else {
 810     for (auto ctx : action_contexts.second) {
 811       if (next_state == STATE_OPEN) {
 812         // we couldn't originally wrap the open callback w/ an async wrapper in
 813         // case the image failed to open
 814         ctx = create_async_context_callback(*m_image_ctx, ctx);
 815       }
 816       ctx->complete(r);
 817     }
 818
 819     m_lock.lock();
 820     if (!is_transition_state() && !m_actions_contexts.empty()) {
 821       execute_next_action_unlock();
 822     } else {
 823       m_lock.unlock();
 824     }
 825   }
 826 }
 827
 828 template <typename I>
 829 void ImageState<I>::send_open_unlock() {
 830   ceph_assert(ceph_mutex_is_locked(m_lock));
 831   CephContext *cct = m_image_ctx->cct;
 832   ldout(cct, 10) << this << " " << __func__ << dendl;
 833
 834   m_state = STATE_OPENING;
 835
 836   Context *ctx = create_context_callback<
 837     ImageState<I>, &ImageState<I>::handle_open>(this);
 838   image::OpenRequest<I> *req = image::OpenRequest<I>::create(
 839     m_image_ctx, m_open_flags, ctx);
 840
 841   m_lock.unlock();
 842   req->send();
 843 }
 844
 845 template <typename I>
 846 void ImageState<I>::handle_open(int r) {
 847   CephContext *cct = m_image_ctx->cct;
 848   ldout(cct, 10) << this << " " << __func__ << ": r=" << r << dendl;
 849
 850   if (r < 0 && r != -ENOENT) {
 851     lderr(cct) << "failed to open image: " << cpp_strerror(r) << dendl;
 852   }
 853
 854   m_lock.lock();
 855   complete_action_unlock(r < 0 ? STATE_UNINITIALIZED : STATE_OPEN, r);
 856 }
 857
 858 template <typename I>
 859 void ImageState<I>::send_close_unlock() {
 860   ceph_assert(ceph_mutex_is_locked(m_lock));
 861   CephContext *cct = m_image_ctx->cct;
 862   ldout(cct, 10) << this << " " << __func__ << dendl;
 863
 864   m_state = STATE_CLOSING;
 865
 866   Context *ctx = create_context_callback<
 867     ImageState<I>, &ImageState<I>::handle_close>(this);
 868   image::CloseRequest<I> *req = image::CloseRequest<I>::create(
 869     m_image_ctx, ctx);
 870
 871   m_lock.unlock();
 872   req->send();
 873 }
 874
 875 template <typename I>
 876 void ImageState<I>::handle_close(int r) {
 877   CephContext *cct = m_image_ctx->cct;
 878   ldout(cct, 10) << this << " " << __func__ << ": r=" << r << dendl;
 879
 880   if (r < 0) {
 881     lderr(cct) << "error occurred while closing image: " << cpp_strerror(r)
 882                << dendl;
 883   }
 884
 885   m_lock.lock();
 886   complete_action_unlock(STATE_CLOSED, r);
 887 }
 888
 889 template <typename I>
 890 void ImageState<I>::send_refresh_unlock() {
 891   ceph_assert(ceph_mutex_is_locked(m_lock));
 892   CephContext *cct = m_image_ctx->cct;
 893   ldout(cct, 10) << this << " " << __func__ << dendl;
 894
 895   m_state = STATE_REFRESHING;
 896   ceph_assert(!m_actions_contexts.empty());
 897   auto &action_context = m_actions_contexts.front().first;
 898   ceph_assert(action_context.action_type == ACTION_TYPE_REFRESH);
 899
 900   Context *ctx = create_async_context_callback(
 901     *m_image_ctx, create_context_callback<
 902       ImageState<I>, &ImageState<I>::handle_refresh>(this));
 903   image::RefreshRequest<I> *req = image::RefreshRequest<I>::create(
 904     *m_image_ctx, false, false, ctx);
 905
 906   m_lock.unlock();
 907   req->send();
 908 }
 909
 910 template <typename I>
 911 void ImageState<I>::handle_refresh(int r) {
 912   CephContext *cct = m_image_ctx->cct;
 913   ldout(cct, 10) << this << " " << __func__ << ": r=" << r << dendl;
 914
 915   m_lock.lock();
 916   ceph_assert(!m_actions_contexts.empty());
 917
 918   ActionContexts &action_contexts(m_actions_contexts.front());
 919   ceph_assert(action_contexts.first.action_type == ACTION_TYPE_REFRESH);
 920   ceph_assert(m_last_refresh <= action_contexts.first.refresh_seq);
 921
 922   if (r == -ERESTART) {
 923     ldout(cct, 5) << "incomplete refresh: not updating sequence" << dendl;
 924     r = 0;
 925   } else {
 926     m_last_refresh = action_contexts.first.refresh_seq;
 927   }
 928
 929   complete_action_unlock(STATE_OPEN, r);
 930 }
 931
 932 template <typename I>
 933 void ImageState<I>::send_set_snap_unlock() {
 934   ceph_assert(ceph_mutex_is_locked(m_lock));
 935
 936   m_state = STATE_SETTING_SNAP;
 937
 938   ceph_assert(!m_actions_contexts.empty());
 939   ActionContexts &action_contexts(m_actions_contexts.front());
 940   ceph_assert(action_contexts.first.action_type == ACTION_TYPE_SET_SNAP);
 941
 942   CephContext *cct = m_image_ctx->cct;
 943   ldout(cct, 10) << this << " " << __func__ << ": "
 944                  << "snap_id=" << action_contexts.first.snap_id << dendl;
 945
 946   Context *ctx = create_async_context_callback(
 947     *m_image_ctx, create_context_callback<
 948       ImageState<I>, &ImageState<I>::handle_set_snap>(this));
 949   image::SetSnapRequest<I> *req = image::SetSnapRequest<I>::create(
 950     *m_image_ctx, action_contexts.first.snap_id, ctx);
 951
 952   m_lock.unlock();
 953   req->send();
 954 }
 955
 956 template <typename I>
 957 void ImageState<I>::handle_set_snap(int r) {
 958   CephContext *cct = m_image_ctx->cct;
 959   ldout(cct, 10) << this << " " << __func__ << " r=" << r << dendl;
 960
 961   if (r < 0 && r != -ENOENT) {
 962     lderr(cct) << "failed to set snapshot: " << cpp_strerror(r) << dendl;
 963   }
 964
 965   m_lock.lock();
 966   complete_action_unlock(STATE_OPEN, r);
 967 }
 968
 969 template <typename I>
 970 void ImageState<I>::send_prepare_lock_unlock() {
 971   CephContext *cct = m_image_ctx->cct;
 972   ldout(cct, 10) << this << " " << __func__ << dendl;
 973
 974   ceph_assert(ceph_mutex_is_locked(m_lock));
 975   m_state = STATE_PREPARING_LOCK;
 976
 977   ceph_assert(!m_actions_contexts.empty());
 978   ActionContexts &action_contexts(m_actions_contexts.front());
 979   ceph_assert(action_contexts.first.action_type == ACTION_TYPE_LOCK);
 980
 981   Context *on_ready = action_contexts.first.on_ready;
 982   m_lock.unlock();
 983
 984   if (on_ready == nullptr) {
 985     complete_action_unlock(STATE_OPEN, 0);
 986     return;
 987   }
 988
 989   // wake up the lock handler now that its safe to proceed
 990   on_ready->complete(0);
 991 }
 992
 993 template <typename I>
 994 int ImageState<I>::register_quiesce_watcher(QuiesceWatchCtx *watcher,
 995                                                 uint64_t *handle) {
 996   CephContext *cct = m_image_ctx->cct;
 997   ldout(cct, 20) << __func__ << dendl;
 998
 999   m_quiesce_watchers->register_watcher(watcher, handle);
1000
1001   ldout(cct, 20) << __func__ << ": handle=" << *handle << dendl;
1002   return 0;
1003 }
1004
1005 template <typename I>
1006 int ImageState<I>::unregister_quiesce_watcher(uint64_t handle) {
1007   CephContext *cct = m_image_ctx->cct;
1008   ldout(cct, 20) << __func__ << ": handle=" << handle << dendl;
1009
1010   C_SaferCond ctx;
1011   m_quiesce_watchers->unregister_watcher(handle, &ctx);
1012   return ctx.wait();
1013 }
1014
1015 template <typename I>
1016 void ImageState<I>::notify_quiesce(Context *on_finish) {
1017   CephContext *cct = m_image_ctx->cct;
1018   ldout(cct, 20) << __func__ << dendl;
1019
1020   m_quiesce_watchers->notify_quiesce(on_finish);
1021 }
1022
1023 template <typename I>
1024 void ImageState<I>::notify_unquiesce(Context *on_finish) {
1025   CephContext *cct = m_image_ctx->cct;
1026   ldout(cct, 20) << __func__ << dendl;
1027
1028   m_quiesce_watchers->notify_unquiesce(on_finish);
1029 }
1030
1031 template <typename I>
1032 void ImageState<I>::quiesce_complete(uint64_t handle, int r) {
1033   CephContext *cct = m_image_ctx->cct;
1034   ldout(cct, 20) << __func__ << ": handle=" << handle << " r=" << r << dendl;
1035   m_quiesce_watchers->quiesce_complete(handle, r);
1036 }
1037
1038 } // namespace librbd
1039
1040 template class librbd::ImageState<librbd::ImageCtx>;