diff options
Diffstat (limited to 'src/tools/rbd_mirror/ImageSyncThrottler.cc')
| -rw-r--r-- | src/tools/rbd_mirror/ImageSyncThrottler.cc | 227 |
1 files changed, 227 insertions, 0 deletions
diff --git a/src/tools/rbd_mirror/ImageSyncThrottler.cc b/src/tools/rbd_mirror/ImageSyncThrottler.cc new file mode 100644 index 00000000..b395a012 --- /dev/null +++ b/src/tools/rbd_mirror/ImageSyncThrottler.cc @@ -0,0 +1,227 @@ +// -*- 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(CephContext *cct) + : m_cct(cct), + m_lock(librbd::util::unique_lock_name("rbd::mirror::ImageSyncThrottler", + this)), + m_max_concurrent_syncs(cct->_conf.get_val<uint64_t>( + "rbd_mirror_concurrent_image_syncs")) { + dout(20) << "max_concurrent_syncs=" << m_max_concurrent_syncs << dendl; + m_cct->_conf.add_observer(this); +} + +template <typename I> +ImageSyncThrottler<I>::~ImageSyncThrottler() { + m_cct->_conf.remove_observer(this); + + Mutex::Locker locker(m_lock); + ceph_assert(m_inflight_ops.empty()); + ceph_assert(m_queue.empty()); +} + +template <typename I> +void ImageSyncThrottler<I>::start_op(const std::string &id, Context *on_start) { + dout(20) << "id=" << id << dendl; + + int r = 0; + { + Mutex::Locker locker(m_lock); + + if (m_inflight_ops.count(id) > 0) { + dout(20) << "duplicate for already started op " << id << dendl; + } else if (m_queued_ops.count(id) > 0) { + dout(20) << "duplicate for already queued op " << id << dendl; + std::swap(m_queued_ops[id], on_start); + r = -ENOENT; + } else if (m_max_concurrent_syncs == 0 || + m_inflight_ops.size() < m_max_concurrent_syncs) { + ceph_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(id); + std::swap(m_queued_ops[id], on_start); + dout(20) << "image sync for " << id << " has been queued" << dendl; + } + } + + if (on_start != nullptr) { + on_start->complete(r); + } +} + +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); + auto it = m_queued_ops.find(id); + if (it != m_queued_ops.end()) { + dout(20) << "canceled queued sync for " << id << dendl; + m_queue.remove(id); + on_start = it->second; + m_queued_ops.erase(it); + } + } + + 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 id = m_queue.front(); + auto it = m_queued_ops.find(id); + ceph_assert(it != m_queued_ops.end()); + m_inflight_ops.insert(id); + dout(20) << "ready to start sync for " << id << " [" + << m_inflight_ops.size() << "/" << m_max_concurrent_syncs << "]" + << dendl; + on_start = it->second; + m_queued_ops.erase(it); + 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::map<std::string, Context *> queued_ops; + { + Mutex::Locker locker(m_lock); + std::swap(m_queued_ops, queued_ops); + m_queue.clear(); + m_inflight_ops.clear(); + } + + for (auto &it : queued_ops) { + it.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 id = m_queue.front(); + m_inflight_ops.insert(id); + dout(20) << "ready to start sync for " << id << " [" + << m_inflight_ops.size() << "/" << m_max_concurrent_syncs << "]" + << dendl; + auto it = m_queued_ops.find(id); + ceph_assert(it != m_queued_ops.end()); + ops.push_back(it->second); + m_queued_ops.erase(it); + 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 ConfigProxy& 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>; |