ceph/src/librbd/mirror/DemoteRequest.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/mirror/DemoteRequest.h"
   5 #include "common/dout.h"
   6 #include "common/errno.h"
   7 #include "cls/rbd/cls_rbd_client.h"
   8 #include "librbd/ExclusiveLock.h"
   9 #include "librbd/ImageCtx.h"
  10 #include "librbd/ImageState.h"
  11 #include "librbd/Journal.h"
  12 #include "librbd/Utils.h"
  13 #include "librbd/mirror/GetInfoRequest.h"
  14
  15 #define dout_subsys ceph_subsys_rbd
  16 #undef dout_prefix
  17 #define dout_prefix *_dout << "librbd::mirror::DemoteRequest: " << this \
  18                            << " " << __func__ << ": "
  19
  20 namespace librbd {
  21 namespace mirror {
  22
  23 using librbd::util::create_context_callback;
  24
  25 template <typename I>
  26 void DemoteRequest<I>::send() {
  27   get_info();
  28 }
  29
  30 template <typename I>
  31 void DemoteRequest<I>::get_info() {
  32   CephContext *cct = m_image_ctx.cct;
  33   ldout(cct, 20) << dendl;
  34
  35   auto ctx = create_context_callback<
  36     DemoteRequest<I>, &DemoteRequest<I>::handle_get_info>(this);
  37   auto req = GetInfoRequest<I>::create(m_image_ctx, &m_mirror_image,
  38                                        &m_promotion_state, ctx);
  39   req->send();
  40 }
  41
  42 template <typename I>
  43 void DemoteRequest<I>::handle_get_info(int r) {
  44   CephContext *cct = m_image_ctx.cct;
  45   ldout(cct, 20) << "r=" << r << dendl;
  46
  47   if (r < 0) {
  48     lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
  49                << dendl;
  50     finish(r);
  51     return;
  52   } else if (m_mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
  53     lderr(cct) << "mirroring is not currently enabled" << dendl;
  54     finish(-EINVAL);
  55     return;
  56   } else if (m_promotion_state != PROMOTION_STATE_PRIMARY) {
  57     lderr(cct) << "image is not primary" << dendl;
  58     finish(-EINVAL);
  59     return;
  60   }
  61
  62   acquire_lock();
  63 }
  64
  65 template <typename I>
  66 void DemoteRequest<I>::acquire_lock() {
  67   CephContext *cct = m_image_ctx.cct;
  68
  69   m_image_ctx.owner_lock.get_read();
  70   if (m_image_ctx.exclusive_lock == nullptr) {
  71     m_image_ctx.owner_lock.put_read();
  72     lderr(cct) << "exclusive lock is not active" << dendl;
  73     finish(-EINVAL);
  74     return;
  75   }
  76
  77   // avoid accepting new requests from peers while we demote
  78   // the image
  79   m_image_ctx.exclusive_lock->block_requests(0);
  80   m_blocked_requests = true;
  81
  82   if (m_image_ctx.exclusive_lock->is_lock_owner()) {
  83     m_image_ctx.owner_lock.put_read();
  84     demote();
  85     return;
  86   }
  87
  88   ldout(cct, 20) << dendl;
  89
  90   auto ctx = create_context_callback<
  91     DemoteRequest<I>, &DemoteRequest<I>::handle_acquire_lock>(this);
  92   m_image_ctx.exclusive_lock->acquire_lock(ctx);
  93   m_image_ctx.owner_lock.put_read();
  94 }
  95
  96 template <typename I>
  97 void DemoteRequest<I>::handle_acquire_lock(int r) {
  98   CephContext *cct = m_image_ctx.cct;
  99   ldout(cct, 20) << "r=" << r << dendl;
 100
 101   if (r < 0) {
 102     lderr(cct) << "failed to lock image: " << cpp_strerror(r) << dendl;
 103     finish(r);
 104     return;
 105   }
 106
 107   m_image_ctx.owner_lock.get_read();
 108   if (m_image_ctx.exclusive_lock != nullptr &&
 109       !m_image_ctx.exclusive_lock->is_lock_owner()) {
 110     r = m_image_ctx.exclusive_lock->get_unlocked_op_error();
 111     m_image_ctx.owner_lock.put_read();
 112     lderr(cct) << "failed to acquire exclusive lock" << dendl;
 113     finish(r);
 114     return;
 115   }
 116   m_image_ctx.owner_lock.put_read();
 117
 118   demote();
 119 }
 120
 121 template <typename I>
 122 void DemoteRequest<I>::demote() {
 123   CephContext *cct = m_image_ctx.cct;
 124   ldout(cct, 20) << dendl;
 125
 126   auto ctx = create_context_callback<
 127     DemoteRequest<I>, &DemoteRequest<I>::handle_demote>(this);
 128   Journal<I>::demote(&m_image_ctx, ctx);
 129 }
 130
 131 template <typename I>
 132 void DemoteRequest<I>::handle_demote(int r) {
 133   CephContext *cct = m_image_ctx.cct;
 134   ldout(cct, 20) << "r=" << r << dendl;
 135
 136   if (r < 0) {
 137     m_ret_val = r;
 138     lderr(cct) << "failed to demote image: " << cpp_strerror(r) << dendl;
 139   }
 140
 141   release_lock();
 142 }
 143
 144 template <typename I>
 145 void DemoteRequest<I>::release_lock() {
 146   CephContext *cct = m_image_ctx.cct;
 147   ldout(cct, 20) << dendl;
 148
 149   m_image_ctx.owner_lock.get_read();
 150   if (m_image_ctx.exclusive_lock == nullptr) {
 151     m_image_ctx.owner_lock.put_read();
 152     finish(0);
 153     return;
 154   }
 155
 156   auto ctx = create_context_callback<
 157     DemoteRequest<I>, &DemoteRequest<I>::handle_release_lock>(this);
 158   m_image_ctx.exclusive_lock->release_lock(ctx);
 159   m_image_ctx.owner_lock.put_read();
 160 }
 161
 162 template <typename I>
 163 void DemoteRequest<I>::handle_release_lock(int r) {
 164   CephContext *cct = m_image_ctx.cct;
 165   ldout(cct, 20) << "r=" << r << dendl;
 166
 167   if (r < 0) {
 168     lderr(cct) << "failed to release exclusive lock: " << cpp_strerror(r)
 169                << dendl;
 170   }
 171
 172   finish(r);
 173 }
 174
 175 template <typename I>
 176 void DemoteRequest<I>::finish(int r) {
 177   if (m_ret_val < 0) {
 178     r = m_ret_val;
 179   }
 180
 181   {
 182     RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
 183     if (m_blocked_requests && m_image_ctx.exclusive_lock != nullptr) {
 184       m_image_ctx.exclusive_lock->unblock_requests();
 185     }
 186   }
 187
 188   CephContext *cct = m_image_ctx.cct;
 189   ldout(cct, 20) << "r=" << r << dendl;
 190
 191   m_on_finish->complete(r);
 192   delete this;
 193 }
 194
 195 } // namespace mirror
 196 } // namespace librbd
 197
 198 template class librbd::mirror::DemoteRequest<librbd::ImageCtx>;