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     m_image_ctx.owner_lock.put_read();
 111     lderr(cct) << "failed to acquire exclusive lock" << dendl;
 112     finish(-EROFS);
 113     return;
 114   }
 115   m_image_ctx.owner_lock.put_read();
 116
 117   demote();
 118 }
 119
 120 template <typename I>
 121 void DemoteRequest<I>::demote() {
 122   CephContext *cct = m_image_ctx.cct;
 123   ldout(cct, 20) << dendl;
 124
 125   auto ctx = create_context_callback<
 126     DemoteRequest<I>, &DemoteRequest<I>::handle_demote>(this);
 127   Journal<I>::demote(&m_image_ctx, ctx);
 128 }
 129
 130 template <typename I>
 131 void DemoteRequest<I>::handle_demote(int r) {
 132   CephContext *cct = m_image_ctx.cct;
 133   ldout(cct, 20) << "r=" << r << dendl;
 134
 135   if (r < 0) {
 136     m_ret_val = r;
 137     lderr(cct) << "failed to demote image: " << cpp_strerror(r) << dendl;
 138   }
 139
 140   release_lock();
 141 }
 142
 143 template <typename I>
 144 void DemoteRequest<I>::release_lock() {
 145   CephContext *cct = m_image_ctx.cct;
 146   ldout(cct, 20) << dendl;
 147
 148   m_image_ctx.owner_lock.get_read();
 149   if (m_image_ctx.exclusive_lock == nullptr) {
 150     m_image_ctx.owner_lock.put_read();
 151     finish(0);
 152     return;
 153   }
 154
 155   auto ctx = create_context_callback<
 156     DemoteRequest<I>, &DemoteRequest<I>::handle_release_lock>(this);
 157   m_image_ctx.exclusive_lock->release_lock(ctx);
 158   m_image_ctx.owner_lock.put_read();
 159 }
 160
 161 template <typename I>
 162 void DemoteRequest<I>::handle_release_lock(int r) {
 163   CephContext *cct = m_image_ctx.cct;
 164   ldout(cct, 20) << "r=" << r << dendl;
 165
 166   if (r < 0) {
 167     lderr(cct) << "failed to release exclusive lock: " << cpp_strerror(r)
 168                << dendl;
 169   }
 170
 171   finish(r);
 172 }
 173
 174 template <typename I>
 175 void DemoteRequest<I>::finish(int r) {
 176   if (m_ret_val < 0) {
 177     r = m_ret_val;
 178   }
 179
 180   {
 181     RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
 182     if (m_blocked_requests && m_image_ctx.exclusive_lock != nullptr) {
 183       m_image_ctx.exclusive_lock->unblock_requests();
 184     }
 185   }
 186
 187   CephContext *cct = m_image_ctx.cct;
 188   ldout(cct, 20) << "r=" << r << dendl;
 189
 190   m_on_finish->complete(r);
 191   delete this;
 192 }
 193
 194 } // namespace mirror
 195 } // namespace librbd
 196
 197 template class librbd::mirror::DemoteRequest<librbd::ImageCtx>;