src/ceph/src/tools/rbd_mirror/Instances.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 "include/stringify.h"
   5 #include "common/Timer.h"
   6 #include "common/WorkQueue.h"
   7 #include "common/debug.h"
   8 #include "common/errno.h"
   9 #include "librbd/Utils.h"
  10 #include "InstanceWatcher.h"
  11 #include "Instances.h"
  12 #include "Threads.h"
  13
  14 #define dout_context g_ceph_context
  15 #define dout_subsys ceph_subsys_rbd_mirror
  16 #undef dout_prefix
  17 #define dout_prefix *_dout << "rbd::mirror::Instances: " \
  18                            << this << " " << __func__ << ": "
  19
  20 namespace rbd {
  21 namespace mirror {
  22
  23 using librbd::util::create_async_context_callback;
  24 using librbd::util::create_context_callback;
  25 using librbd::util::create_rados_callback;
  26
  27 template <typename I>
  28 Instances<I>::Instances(Threads<I> *threads, librados::IoCtx &ioctx) :
  29   m_threads(threads), m_ioctx(ioctx),
  30   m_cct(reinterpret_cast<CephContext *>(ioctx.cct())),
  31   m_lock("rbd::mirror::Instances " + ioctx.get_pool_name()) {
  32 }
  33
  34 template <typename I>
  35 Instances<I>::~Instances() {
  36 }
  37
  38 template <typename I>
  39 void Instances<I>::init(Context *on_finish) {
  40   dout(20) << dendl;
  41
  42   Mutex::Locker locker(m_lock);
  43   assert(m_on_finish == nullptr);
  44   m_on_finish = on_finish;
  45   get_instances();
  46 }
  47
  48 template <typename I>
  49 void Instances<I>::shut_down(Context *on_finish) {
  50   dout(20) << dendl;
  51
  52   Mutex::Locker locker(m_lock);
  53   assert(m_on_finish == nullptr);
  54   m_on_finish = on_finish;
  55
  56   Context *ctx = new FunctionContext(
  57     [this](int r) {
  58       Mutex::Locker timer_locker(m_threads->timer_lock);
  59       Mutex::Locker locker(m_lock);
  60
  61       for (auto it : m_instances) {
  62         cancel_remove_task(it.second);
  63       }
  64       wait_for_ops();
  65     });
  66
  67   m_threads->work_queue->queue(ctx, 0);
  68 }
  69
  70 template <typename I>
  71 void Instances<I>::notify(const std::string &instance_id) {
  72   dout(20) << instance_id << dendl;
  73
  74   Mutex::Locker locker(m_lock);
  75
  76   if (m_on_finish != nullptr) {
  77     dout(20) << "received on shut down, ignoring" << dendl;
  78     return;
  79   }
  80
  81   Context *ctx = new C_Notify(this, instance_id);
  82
  83   m_threads->work_queue->queue(ctx, 0);
  84 }
  85
  86 template <typename I>
  87 void Instances<I>::handle_notify(const std::string &instance_id) {
  88   dout(20) << instance_id << dendl;
  89
  90   Mutex::Locker timer_locker(m_threads->timer_lock);
  91   Mutex::Locker locker(m_lock);
  92
  93   if (m_on_finish != nullptr) {
  94     dout(20) << "handled on shut down, ignoring" << dendl;
  95     return;
  96   }
  97
  98   auto &instance = m_instances.insert(
  99     std::make_pair(instance_id, Instance(instance_id))).first->second;
 100
 101   schedule_remove_task(instance);
 102 }
 103
 104 template <typename I>
 105 void Instances<I>::list(std::vector<std::string> *instance_ids) {
 106   dout(20) << dendl;
 107
 108   Mutex::Locker locker(m_lock);
 109
 110   for (auto it : m_instances) {
 111     instance_ids->push_back(it.first);
 112   }
 113 }
 114
 115
 116 template <typename I>
 117 void Instances<I>::get_instances() {
 118   dout(20) << dendl;
 119
 120   assert(m_lock.is_locked());
 121
 122   Context *ctx = create_context_callback<
 123     Instances, &Instances<I>::handle_get_instances>(this);
 124
 125   InstanceWatcher<I>::get_instances(m_ioctx, &m_instance_ids, ctx);
 126 }
 127
 128 template <typename I>
 129 void Instances<I>::handle_get_instances(int r) {
 130   dout(20) << "r=" << r << dendl;
 131
 132   Context *on_finish = nullptr;
 133   {
 134     Mutex::Locker timer_locker(m_threads->timer_lock);
 135     Mutex::Locker locker(m_lock);
 136
 137     if (r < 0) {
 138       derr << "error retrieving instances: " << cpp_strerror(r) << dendl;
 139     } else {
 140       auto my_instance_id = stringify(m_ioctx.get_instance_id());
 141       for (auto &instance_id : m_instance_ids) {
 142         if (instance_id == my_instance_id) {
 143           continue;
 144         }
 145         auto &instance = m_instances.insert(
 146           std::make_pair(instance_id, Instance(instance_id))).first->second;
 147         schedule_remove_task(instance);
 148       }
 149     }
 150     std::swap(on_finish, m_on_finish);
 151   }
 152   on_finish->complete(r);
 153 }
 154
 155 template <typename I>
 156 void Instances<I>::wait_for_ops() {
 157   dout(20) << dendl;
 158
 159   assert(m_lock.is_locked());
 160
 161   Context *ctx = create_async_context_callback(
 162     m_threads->work_queue, create_context_callback<
 163     Instances, &Instances<I>::handle_wait_for_ops>(this));
 164
 165   m_async_op_tracker.wait_for_ops(ctx);
 166 }
 167
 168 template <typename I>
 169 void Instances<I>::handle_wait_for_ops(int r) {
 170   dout(20) << "r=" << r << dendl;
 171
 172   assert(r == 0);
 173
 174   Context *on_finish = nullptr;
 175   {
 176     Mutex::Locker locker(m_lock);
 177     std::swap(on_finish, m_on_finish);
 178   }
 179   on_finish->complete(r);
 180 }
 181
 182 template <typename I>
 183 void Instances<I>::remove_instance(Instance &instance) {
 184   assert(m_lock.is_locked());
 185
 186   dout(20) << instance.id << dendl;
 187
 188   Context *ctx = create_async_context_callback(
 189     m_threads->work_queue, create_context_callback<
 190     Instances, &Instances<I>::handle_remove_instance>(this));
 191
 192   m_async_op_tracker.start_op();
 193   InstanceWatcher<I>::remove_instance(m_ioctx, m_threads->work_queue,
 194                                       instance.id, ctx);
 195   m_instances.erase(instance.id);
 196 }
 197
 198 template <typename I>
 199 void Instances<I>::handle_remove_instance(int r) {
 200   Mutex::Locker locker(m_lock);
 201
 202   dout(20) << " r=" << r << dendl;
 203
 204   assert(r == 0);
 205
 206   m_async_op_tracker.finish_op();
 207 }
 208
 209 template <typename I>
 210 void Instances<I>::cancel_remove_task(Instance &instance) {
 211   assert(m_threads->timer_lock.is_locked());
 212   assert(m_lock.is_locked());
 213
 214   if (instance.timer_task == nullptr) {
 215     return;
 216   }
 217
 218   dout(20) << instance.timer_task << dendl;
 219
 220   bool canceled = m_threads->timer->cancel_event(instance.timer_task);
 221   assert(canceled);
 222   instance.timer_task = nullptr;
 223 }
 224
 225 template <typename I>
 226 void Instances<I>::schedule_remove_task(Instance &instance) {
 227   dout(20) << dendl;
 228
 229   cancel_remove_task(instance);
 230
 231   int after = m_cct->_conf->get_val<int64_t>("rbd_mirror_leader_heartbeat_interval") *
 232     (1 + m_cct->_conf->get_val<int64_t>("rbd_mirror_leader_max_missed_heartbeats") +
 233      m_cct->_conf->get_val<int64_t>("rbd_mirror_leader_max_acquire_attempts_before_break"));
 234
 235   instance.timer_task = new FunctionContext(
 236     [this, &instance](int r) {
 237       assert(m_threads->timer_lock.is_locked());
 238       Mutex::Locker locker(m_lock);
 239       instance.timer_task = nullptr;
 240       remove_instance(instance);
 241     });
 242
 243   dout(20) << "scheduling instance " << instance.id << " remove after " << after
 244            << " sec (task " << instance.timer_task << ")" << dendl;
 245
 246   m_threads->timer->add_event_after(after, instance.timer_task);
 247 }
 248
 249 } // namespace mirror
 250 } // namespace rbd
 251
 252 template class rbd::mirror::Instances<librbd::ImageCtx>;