Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 397 insertions, 0 deletions

diff --git a/src/tools/rbd_mirror/MirrorStatusUpdater.cc b/src/tools/rbd_mirror/MirrorStatusUpdater.cc
new file mode 100644
index 000000000..257cb1df2
--- /dev/null
+++ b/src/tools/rbd_mirror/MirrorStatusUpdater.cc
@@ -0,0 +1,397 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "tools/rbd_mirror/MirrorStatusUpdater.h"
+#include "include/Context.h"
+#include "include/stringify.h"
+#include "common/debug.h"
+#include "common/errno.h"
+#include "common/Timer.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Utils.h"
+#include "librbd/asio/ContextWQ.h"
+#include "tools/rbd_mirror/MirrorStatusWatcher.h"
+#include "tools/rbd_mirror/Threads.h"
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rbd_mirror
+#undef dout_prefix
+#define dout_prefix *_dout << "rbd::mirror::MirrorStatusUpdater " << this \
+                           << " " << __func__ << ": "
+
+namespace rbd {
+namespace mirror {
+
+static const double UPDATE_INTERVAL_SECONDS = 30;
+static const uint32_t MAX_UPDATES_PER_OP = 100;
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+MirrorStatusUpdater<I>::MirrorStatusUpdater(
+    librados::IoCtx& io_ctx, Threads<I> *threads,
+    const std::string& local_mirror_uuid)
+  : m_io_ctx(io_ctx), m_threads(threads),
+    m_local_mirror_uuid(local_mirror_uuid),
+    m_lock(ceph::make_mutex("rbd::mirror::MirrorStatusUpdater " +
+                              stringify(m_io_ctx.get_id()))) {
+  dout(10) << "local_mirror_uuid=" << local_mirror_uuid << ", "
+           << "pool_id=" << m_io_ctx.get_id() << dendl;
+}
+
+template <typename I>
+MirrorStatusUpdater<I>::~MirrorStatusUpdater() {
+  ceph_assert(!m_initialized);
+  delete m_mirror_status_watcher;
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::init(Context* on_finish) {
+  dout(10) << dendl;
+
+  ceph_assert(!m_initialized);
+  m_initialized = true;
+
+  {
+    std::lock_guard timer_locker{m_threads->timer_lock};
+    schedule_timer_task();
+  }
+
+  init_mirror_status_watcher(on_finish);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::init_mirror_status_watcher(Context* on_finish) {
+  dout(10) << dendl;
+
+  auto ctx = new LambdaContext([this, on_finish](int r) {
+      handle_init_mirror_status_watcher(r, on_finish);
+    });
+  m_mirror_status_watcher = MirrorStatusWatcher<I>::create(
+    m_io_ctx, m_threads->work_queue);
+  m_mirror_status_watcher->init(ctx);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::handle_init_mirror_status_watcher(
+    int r, Context* on_finish) {
+  dout(10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    derr << "failed to init mirror status watcher: " << cpp_strerror(r)
+         << dendl;
+
+    delete m_mirror_status_watcher;
+    m_mirror_status_watcher = nullptr;
+
+    on_finish = new LambdaContext([r, on_finish](int) {
+        on_finish->complete(r);
+      });
+    shut_down(on_finish);
+    return;
+  }
+
+  m_threads->work_queue->queue(on_finish, 0);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::shut_down(Context* on_finish) {
+  dout(10) << dendl;
+
+  {
+    std::lock_guard timer_locker{m_threads->timer_lock};
+    ceph_assert(m_timer_task != nullptr);
+    m_threads->timer->cancel_event(m_timer_task);
+  }
+
+  {
+    std::unique_lock locker(m_lock);
+    ceph_assert(m_initialized);
+    m_initialized = false;
+  }
+
+  shut_down_mirror_status_watcher(on_finish);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::shut_down_mirror_status_watcher(
+    Context* on_finish) {
+  if (m_mirror_status_watcher == nullptr) {
+    finalize_shutdown(0, on_finish);
+    return;
+  }
+
+  dout(10) << dendl;
+
+  auto ctx = new LambdaContext([this, on_finish](int r) {
+      handle_shut_down_mirror_status_watcher(r, on_finish);
+    });
+  m_mirror_status_watcher->shut_down(ctx);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::handle_shut_down_mirror_status_watcher(
+    int r, Context* on_finish) {
+  dout(10) << "r=" << r << dendl;
+
+  if (r < 0) {
+    derr << "failed to shut down mirror status watcher: " << cpp_strerror(r)
+         << dendl;
+  }
+
+  finalize_shutdown(r, on_finish);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::finalize_shutdown(int r, Context* on_finish) {
+  dout(10) << dendl;
+
+  {
+    std::unique_lock locker(m_lock);
+    if (m_update_in_progress) {
+      if (r < 0) {
+        on_finish = new LambdaContext([r, on_finish](int) {
+            on_finish->complete(r);
+          });
+      }
+
+      m_update_on_finish_ctxs.push_back(on_finish);
+      return;
+    }
+  }
+
+  m_threads->work_queue->queue(on_finish, r);
+}
+
+template <typename I>
+bool MirrorStatusUpdater<I>::exists(const std::string& global_image_id) {
+  dout(15) << "global_image_id=" << global_image_id << dendl;
+
+  std::unique_lock locker(m_lock);
+  return (m_global_image_status.count(global_image_id) > 0);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::set_mirror_image_status(
+    const std::string& global_image_id,
+    const cls::rbd::MirrorImageSiteStatus& mirror_image_site_status,
+    bool immediate_update) {
+  dout(15) << "global_image_id=" << global_image_id << ", "
+           << "mirror_image_site_status=" << mirror_image_site_status << dendl;
+
+  std::unique_lock locker(m_lock);
+
+  m_global_image_status[global_image_id] = mirror_image_site_status;
+  if (immediate_update) {
+    m_update_global_image_ids.insert(global_image_id);
+    queue_update_task(std::move(locker));
+  }
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::remove_refresh_mirror_image_status(
+    const std::string& global_image_id,
+    Context* on_finish) {
+  if (try_remove_mirror_image_status(global_image_id, false, false,
+                                     on_finish)) {
+    m_threads->work_queue->queue(on_finish, 0);
+  }
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::remove_mirror_image_status(
+    const std::string& global_image_id, bool immediate_update,
+    Context* on_finish) {
+  if (try_remove_mirror_image_status(global_image_id, true, immediate_update,
+                                     on_finish)) {
+    m_threads->work_queue->queue(on_finish, 0);
+  }
+}
+
+template <typename I>
+bool MirrorStatusUpdater<I>::try_remove_mirror_image_status(
+    const std::string& global_image_id, bool queue_update,
+    bool immediate_update, Context* on_finish) {
+  dout(15) << "global_image_id=" << global_image_id << ", "
+           << "queue_update=" << queue_update << ", "
+           << "immediate_update=" << immediate_update << dendl;
+
+  std::unique_lock locker(m_lock);
+  if ((m_update_in_flight &&
+       m_updating_global_image_ids.count(global_image_id) > 0) ||
+      ((m_update_in_progress || m_update_requested) &&
+       m_update_global_image_ids.count(global_image_id) > 0)) {
+    // if update is scheduled/in-progress, wait for it to complete
+    on_finish = new LambdaContext(
+      [this, global_image_id, queue_update, immediate_update,
+             on_finish](int r) {
+        if (try_remove_mirror_image_status(global_image_id, queue_update,
+                                           immediate_update, on_finish)) {
+          on_finish->complete(0);
+        }
+      });
+    m_update_on_finish_ctxs.push_back(on_finish);
+    return false;
+  }
+
+  m_global_image_status.erase(global_image_id);
+  if (queue_update) {
+    m_update_global_image_ids.insert(global_image_id);
+    if (immediate_update) {
+      queue_update_task(std::move(locker));
+    }
+  }
+
+  return true;
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::schedule_timer_task() {
+  dout(10) << dendl;
+
+  ceph_assert(ceph_mutex_is_locked(m_threads->timer_lock));
+  ceph_assert(m_timer_task == nullptr);
+  m_timer_task = create_context_callback<
+    MirrorStatusUpdater<I>,
+    &MirrorStatusUpdater<I>::handle_timer_task>(this);
+  m_threads->timer->add_event_after(UPDATE_INTERVAL_SECONDS, m_timer_task);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::handle_timer_task(int r) {
+  dout(10) << dendl;
+
+  ceph_assert(ceph_mutex_is_locked(m_threads->timer_lock));
+  ceph_assert(m_timer_task != nullptr);
+  m_timer_task = nullptr;
+  schedule_timer_task();
+
+  std::unique_lock locker(m_lock);
+  for (auto& pair : m_global_image_status) {
+    m_update_global_image_ids.insert(pair.first);
+  }
+
+  queue_update_task(std::move(locker));
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::queue_update_task(
+  std::unique_lock<ceph::mutex>&& locker) {
+  if (!m_initialized) {
+    return;
+  }
+
+  if (m_update_in_progress) {
+    if (m_update_in_flight) {
+      dout(10) << "deferring update due to in-flight ops" << dendl;
+      m_update_requested = true;
+    }
+    return;
+  }
+
+  m_update_in_progress = true;
+  ceph_assert(!m_update_in_flight);
+  ceph_assert(!m_update_requested);
+  locker.unlock();
+
+  dout(10) << dendl;
+  auto ctx = create_context_callback<
+    MirrorStatusUpdater<I>,
+    &MirrorStatusUpdater<I>::update_task>(this);
+  m_threads->work_queue->queue(ctx);
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::update_task(int r) {
+  dout(10) << dendl;
+
+  std::unique_lock locker(m_lock);
+  ceph_assert(m_update_in_progress);
+  ceph_assert(!m_update_in_flight);
+  m_update_in_flight = true;
+
+  std::swap(m_updating_global_image_ids, m_update_global_image_ids);
+  auto updating_global_image_ids = m_updating_global_image_ids;
+  auto global_image_status = m_global_image_status;
+  locker.unlock();
+
+  Context* ctx = create_context_callback<
+    MirrorStatusUpdater<I>,
+    &MirrorStatusUpdater<I>::handle_update_task>(this);
+  if (updating_global_image_ids.empty()) {
+    ctx->complete(0);
+    return;
+  }
+
+  auto gather = new C_Gather(g_ceph_context, ctx);
+
+  auto it = updating_global_image_ids.begin();
+  while (it != updating_global_image_ids.end()) {
+    librados::ObjectWriteOperation op;
+    uint32_t op_count = 0;
+
+    while (it != updating_global_image_ids.end() &&
+           op_count < MAX_UPDATES_PER_OP) {
+      auto& global_image_id = *it;
+      ++it;
+
+      auto status_it = global_image_status.find(global_image_id);
+      if (status_it == global_image_status.end()) {
+        librbd::cls_client::mirror_image_status_remove(&op, global_image_id);
+        ++op_count;
+        continue;
+      }
+
+      status_it->second.mirror_uuid = m_local_mirror_uuid;
+      librbd::cls_client::mirror_image_status_set(&op, global_image_id,
+                                                  status_it->second);
+      ++op_count;
+    }
+
+    auto aio_comp = create_rados_callback(gather->new_sub());
+    int r = m_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op);
+    ceph_assert(r == 0);
+    aio_comp->release();
+  }
+
+  gather->activate();
+}
+
+template <typename I>
+void MirrorStatusUpdater<I>::handle_update_task(int r) {
+  dout(10) << dendl;
+  if (r < 0) {
+    derr << "failed to update mirror image statuses: " << cpp_strerror(r)
+         << dendl;
+  }
+
+  std::unique_lock locker(m_lock);
+
+  Contexts on_finish_ctxs;
+  std::swap(on_finish_ctxs, m_update_on_finish_ctxs);
+
+  ceph_assert(m_update_in_progress);
+  m_update_in_progress = false;
+
+  ceph_assert(m_update_in_flight);
+  m_update_in_flight = false;
+
+  m_updating_global_image_ids.clear();
+
+  if (m_update_requested) {
+    m_update_requested = false;
+    queue_update_task(std::move(locker));
+  } else {
+    locker.unlock();
+  }
+
+  for (auto on_finish : on_finish_ctxs) {
+    on_finish->complete(0);
+  }
+}
+
+} // namespace mirror
+} // namespace rbd
+
+template class rbd::mirror::MirrorStatusUpdater<librbd::ImageCtx>;