--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2016 SUSE LINUX GmbH
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include "ImageSyncThrottler.h"
+#include "common/Formatter.h"
+#include "common/debug.h"
+#include "common/errno.h"
+#include "librbd/Utils.h"
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rbd_mirror
+#undef dout_prefix
+#define dout_prefix *_dout << "rbd::mirror::ImageSyncThrottler:: " << this \
+ << " " << __func__ << ": "
+
+namespace rbd {
+namespace mirror {
+
+template <typename I>
+ImageSyncThrottler<I>::ImageSyncThrottler()
+ : m_lock(librbd::util::unique_lock_name("rbd::mirror::ImageSyncThrottler",
+ this)),
+ m_max_concurrent_syncs(g_ceph_context->_conf->get_val<uint64_t>(
+ "rbd_mirror_concurrent_image_syncs")) {
+ dout(20) << "max_concurrent_syncs=" << m_max_concurrent_syncs << dendl;
+ g_ceph_context->_conf->add_observer(this);
+}
+
+template <typename I>
+ImageSyncThrottler<I>::~ImageSyncThrottler() {
+ g_ceph_context->_conf->remove_observer(this);
+
+ Mutex::Locker locker(m_lock);
+ assert(m_inflight_ops.empty());
+ assert(m_queue.empty());
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::start_op(const std::string &id, Context *on_start) {
+ dout(20) << "id=" << id << dendl;
+
+ {
+ Mutex::Locker locker(m_lock);
+
+ if (m_inflight_ops.count(id) > 0) {
+ dout(20) << "duplicate for already started op " << id << dendl;
+ } else if (m_max_concurrent_syncs == 0 ||
+ m_inflight_ops.size() < m_max_concurrent_syncs) {
+ assert(m_queue.empty());
+ m_inflight_ops.insert(id);
+ dout(20) << "ready to start sync for " << id << " ["
+ << m_inflight_ops.size() << "/" << m_max_concurrent_syncs << "]"
+ << dendl;
+ } else {
+ m_queue.push_back(std::make_pair(id, on_start));
+ on_start = nullptr;
+ dout(20) << "image sync for " << id << " has been queued" << dendl;
+ }
+ }
+
+ if (on_start != nullptr) {
+ on_start->complete(0);
+ }
+}
+
+template <typename I>
+bool ImageSyncThrottler<I>::cancel_op(const std::string &id) {
+ dout(20) << "id=" << id << dendl;
+
+ Context *on_start = nullptr;
+ {
+ Mutex::Locker locker(m_lock);
+ for (auto it = m_queue.begin(); it != m_queue.end(); ++it) {
+ if (it->first == id) {
+ on_start = it->second;
+ dout(20) << "canceled queued sync for " << id << dendl;
+ m_queue.erase(it);
+ break;
+ }
+ }
+ }
+
+ if (on_start == nullptr) {
+ return false;
+ }
+
+ on_start->complete(-ECANCELED);
+ return true;
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::finish_op(const std::string &id) {
+ dout(20) << "id=" << id << dendl;
+
+ if (cancel_op(id)) {
+ return;
+ }
+
+ Context *on_start = nullptr;
+ {
+ Mutex::Locker locker(m_lock);
+
+ m_inflight_ops.erase(id);
+
+ if (m_inflight_ops.size() < m_max_concurrent_syncs && !m_queue.empty()) {
+ auto pair = m_queue.front();
+ m_inflight_ops.insert(pair.first);
+ dout(20) << "ready to start sync for " << pair.first << " ["
+ << m_inflight_ops.size() << "/" << m_max_concurrent_syncs << "]"
+ << dendl;
+ on_start= pair.second;
+ m_queue.pop_front();
+ }
+ }
+
+ if (on_start != nullptr) {
+ on_start->complete(0);
+ }
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::drain(int r) {
+ dout(20) << dendl;
+
+ std::list<std::pair<std::string, Context *>> queue;
+ {
+ Mutex::Locker locker(m_lock);
+ std::swap(m_queue, queue);
+ m_inflight_ops.clear();
+ }
+
+ for (auto &pair : queue) {
+ pair.second->complete(r);
+ }
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::set_max_concurrent_syncs(uint32_t max) {
+ dout(20) << "max=" << max << dendl;
+
+ std::list<Context *> ops;
+ {
+ Mutex::Locker locker(m_lock);
+ m_max_concurrent_syncs = max;
+
+ // Start waiting ops in the case of available free slots
+ while ((m_max_concurrent_syncs == 0 ||
+ m_inflight_ops.size() < m_max_concurrent_syncs) &&
+ !m_queue.empty()) {
+ auto pair = m_queue.front();
+ m_inflight_ops.insert(pair.first);
+ dout(20) << "ready to start sync for " << pair.first << " ["
+ << m_inflight_ops.size() << "/" << m_max_concurrent_syncs << "]"
+ << dendl;
+ ops.push_back(pair.second);
+ m_queue.pop_front();
+ }
+ }
+
+ for (const auto& ctx : ops) {
+ ctx->complete(0);
+ }
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::print_status(Formatter *f, std::stringstream *ss) {
+ dout(20) << dendl;
+
+ Mutex::Locker locker(m_lock);
+
+ if (f) {
+ f->dump_int("max_parallel_syncs", m_max_concurrent_syncs);
+ f->dump_int("running_syncs", m_inflight_ops.size());
+ f->dump_int("waiting_syncs", m_queue.size());
+ f->flush(*ss);
+ } else {
+ *ss << "[ ";
+ *ss << "max_parallel_syncs=" << m_max_concurrent_syncs << ", ";
+ *ss << "running_syncs=" << m_inflight_ops.size() << ", ";
+ *ss << "waiting_syncs=" << m_queue.size() << " ]";
+ }
+}
+
+template <typename I>
+const char** ImageSyncThrottler<I>::get_tracked_conf_keys() const {
+ static const char* KEYS[] = {
+ "rbd_mirror_concurrent_image_syncs",
+ NULL
+ };
+ return KEYS;
+}
+
+template <typename I>
+void ImageSyncThrottler<I>::handle_conf_change(const struct md_config_t *conf,
+ const set<string> &changed) {
+ if (changed.count("rbd_mirror_concurrent_image_syncs")) {
+ set_max_concurrent_syncs(conf->get_val<uint64_t>("rbd_mirror_concurrent_image_syncs"));
+ }
+}
+
+} // namespace mirror
+} // namespace rbd
+
+template class rbd::mirror::ImageSyncThrottler<librbd::ImageCtx>;
Code Review / stor4nfv.git / blobdiff