From 483eb2f56657e8e7f419ab1a4fab8dce9ade8609 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sat, 27 Apr 2024 20:24:20 +0200 Subject: Adding upstream version 14.2.21. Signed-off-by: Daniel Baumann --- src/librbd/watcher/Notifier.cc | 98 +++++++++++++++++++++++++++++++ src/librbd/watcher/Notifier.h | 63 ++++++++++++++++++++ src/librbd/watcher/RewatchRequest.cc | 108 +++++++++++++++++++++++++++++++++++ src/librbd/watcher/RewatchRequest.h | 75 ++++++++++++++++++++++++ src/librbd/watcher/Types.cc | 45 +++++++++++++++ src/librbd/watcher/Types.h | 71 +++++++++++++++++++++++ src/librbd/watcher/Utils.h | 74 ++++++++++++++++++++++++ 7 files changed, 534 insertions(+) create mode 100644 src/librbd/watcher/Notifier.cc create mode 100644 src/librbd/watcher/Notifier.h create mode 100644 src/librbd/watcher/RewatchRequest.cc create mode 100644 src/librbd/watcher/RewatchRequest.h create mode 100644 src/librbd/watcher/Types.cc create mode 100644 src/librbd/watcher/Types.h create mode 100644 src/librbd/watcher/Utils.h (limited to 'src/librbd/watcher') diff --git a/src/librbd/watcher/Notifier.cc b/src/librbd/watcher/Notifier.cc new file mode 100644 index 00000000..dfb95aec --- /dev/null +++ b/src/librbd/watcher/Notifier.cc @@ -0,0 +1,98 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/watcher/Notifier.h" +#include "common/WorkQueue.h" +#include "librbd/ImageCtx.h" +#include "librbd/Utils.h" +#include "librbd/watcher/Types.h" + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::watcher::Notifier: " \ + << this << " " << __func__ << ": " + +namespace librbd { +namespace watcher { + +const uint64_t Notifier::NOTIFY_TIMEOUT = 5000; + +Notifier::C_AioNotify::C_AioNotify(Notifier *notifier, NotifyResponse *response, + Context *on_finish) + : notifier(notifier), response(response), on_finish(on_finish) { +} + +void Notifier::C_AioNotify::finish(int r) { + if (response != nullptr) { + if (r == 0 || r == -ETIMEDOUT) { + try { + auto it = out_bl.cbegin(); + decode(*response, it); + } catch (const buffer::error &err) { + r = -EBADMSG; + } + } + } + notifier->handle_notify(r, on_finish); +} + +Notifier::Notifier(ContextWQ *work_queue, IoCtx &ioctx, const std::string &oid) + : m_work_queue(work_queue), m_ioctx(ioctx), m_oid(oid), + m_aio_notify_lock(util::unique_lock_name( + "librbd::object_watcher::Notifier::m_aio_notify_lock", this)) { + m_cct = reinterpret_cast(m_ioctx.cct()); +} + +Notifier::~Notifier() { + Mutex::Locker aio_notify_locker(m_aio_notify_lock); + ceph_assert(m_pending_aio_notifies == 0); +} + +void Notifier::flush(Context *on_finish) { + Mutex::Locker aio_notify_locker(m_aio_notify_lock); + if (m_pending_aio_notifies == 0) { + m_work_queue->queue(on_finish, 0); + return; + } + + m_aio_notify_flush_ctxs.push_back(on_finish); +} + +void Notifier::notify(bufferlist &bl, NotifyResponse *response, + Context *on_finish) { + { + Mutex::Locker aio_notify_locker(m_aio_notify_lock); + ++m_pending_aio_notifies; + + ldout(m_cct, 20) << "pending=" << m_pending_aio_notifies << dendl; + } + + C_AioNotify *ctx = new C_AioNotify(this, response, on_finish); + librados::AioCompletion *comp = util::create_rados_callback(ctx); + int r = m_ioctx.aio_notify(m_oid, comp, bl, NOTIFY_TIMEOUT, &ctx->out_bl); + ceph_assert(r == 0); + comp->release(); +} + +void Notifier::handle_notify(int r, Context *on_finish) { + ldout(m_cct, 20) << "r=" << r << dendl; + + Mutex::Locker aio_notify_locker(m_aio_notify_lock); + ceph_assert(m_pending_aio_notifies > 0); + --m_pending_aio_notifies; + + ldout(m_cct, 20) << "pending=" << m_pending_aio_notifies << dendl; + if (m_pending_aio_notifies == 0) { + for (auto ctx : m_aio_notify_flush_ctxs) { + m_work_queue->queue(ctx, 0); + } + m_aio_notify_flush_ctxs.clear(); + } + + if (on_finish != nullptr) { + m_work_queue->queue(on_finish, r); + } +} + +} // namespace watcher +} // namespace librbd diff --git a/src/librbd/watcher/Notifier.h b/src/librbd/watcher/Notifier.h new file mode 100644 index 00000000..8b0ad37b --- /dev/null +++ b/src/librbd/watcher/Notifier.h @@ -0,0 +1,63 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_WATCHER_NOTIFIER_H +#define CEPH_LIBRBD_WATCHER_NOTIFIER_H + +#include "include/int_types.h" +#include "include/buffer_fwd.h" +#include "include/Context.h" +#include "include/rados/librados.hpp" +#include "common/Mutex.h" +#include "common/WorkQueue.h" +#include + +namespace librbd { + +namespace watcher { + +struct NotifyResponse; + +class Notifier { +public: + static const uint64_t NOTIFY_TIMEOUT; + + Notifier(ContextWQ *work_queue, librados::IoCtx &ioctx, + const std::string &oid); + ~Notifier(); + + void flush(Context *on_finish); + void notify(bufferlist &bl, NotifyResponse *response, Context *on_finish); + +private: + typedef std::list Contexts; + + struct C_AioNotify : public Context { + Notifier *notifier; + NotifyResponse *response; + Context *on_finish; + bufferlist out_bl; + + C_AioNotify(Notifier *notifier, NotifyResponse *response, + Context *on_finish); + + void finish(int r) override; + }; + + ContextWQ *m_work_queue; + librados::IoCtx &m_ioctx; + CephContext *m_cct; + std::string m_oid; + + Mutex m_aio_notify_lock; + size_t m_pending_aio_notifies = 0; + Contexts m_aio_notify_flush_ctxs; + + void handle_notify(int r, Context *on_finish); + +}; + +} // namespace watcher +} // namespace librbd + +#endif // CEPH_LIBRBD_WATCHER_NOTIFIER_H diff --git a/src/librbd/watcher/RewatchRequest.cc b/src/librbd/watcher/RewatchRequest.cc new file mode 100644 index 00000000..40c3dfe7 --- /dev/null +++ b/src/librbd/watcher/RewatchRequest.cc @@ -0,0 +1,108 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/watcher/RewatchRequest.h" +#include "common/RWLock.h" +#include "common/errno.h" +#include "librbd/Utils.h" + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::watcher::RewatchRequest: " \ + << this << " " << __func__ << " " + +namespace librbd { + +using util::create_context_callback; +using util::create_rados_callback; + +namespace watcher { + +using std::string; + +RewatchRequest::RewatchRequest(librados::IoCtx& ioctx, const string& oid, + RWLock &watch_lock, + librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish) + : m_ioctx(ioctx), m_oid(oid), m_watch_lock(watch_lock), + m_watch_ctx(watch_ctx), m_watch_handle(watch_handle), + m_on_finish(on_finish) { +} + +void RewatchRequest::send() { + unwatch(); +} + +void RewatchRequest::unwatch() { + ceph_assert(m_watch_lock.is_wlocked()); + if (*m_watch_handle == 0) { + rewatch(); + return; + } + + CephContext *cct = reinterpret_cast(m_ioctx.cct()); + ldout(cct, 10) << dendl; + + uint64_t watch_handle = 0; + std::swap(*m_watch_handle, watch_handle); + + librados::AioCompletion *aio_comp = create_rados_callback< + RewatchRequest, &RewatchRequest::handle_unwatch>(this); + int r = m_ioctx.aio_unwatch(watch_handle, aio_comp); + ceph_assert(r == 0); + aio_comp->release(); +} + +void RewatchRequest::handle_unwatch(int r) { + CephContext *cct = reinterpret_cast(m_ioctx.cct()); + ldout(cct, 10) << "r=" << r << dendl; + + if (r == -EBLACKLISTED) { + lderr(cct) << "client blacklisted" << dendl; + finish(r); + return; + } else if (r < 0) { + lderr(cct) << "failed to unwatch: " << cpp_strerror(r) << dendl; + } + rewatch(); +} + +void RewatchRequest::rewatch() { + CephContext *cct = reinterpret_cast(m_ioctx.cct()); + ldout(cct, 10) << dendl; + + librados::AioCompletion *aio_comp = create_rados_callback< + RewatchRequest, &RewatchRequest::handle_rewatch>(this); + int r = m_ioctx.aio_watch(m_oid, aio_comp, &m_rewatch_handle, m_watch_ctx); + ceph_assert(r == 0); + aio_comp->release(); +} + +void RewatchRequest::handle_rewatch(int r) { + CephContext *cct = reinterpret_cast(m_ioctx.cct()); + ldout(cct, 10) << "r=" << r << dendl; + if (r < 0) { + lderr(cct) << "failed to watch object: " << cpp_strerror(r) + << dendl; + m_rewatch_handle = 0; + } + + { + RWLock::WLocker watch_locker(m_watch_lock); + *m_watch_handle = m_rewatch_handle; + } + + finish(r); +} + +void RewatchRequest::finish(int r) { + CephContext *cct = reinterpret_cast(m_ioctx.cct()); + ldout(cct, 10) << "r=" << r << dendl; + + m_on_finish->complete(r); + delete this; +} + +} // namespace watcher +} // namespace librbd + diff --git a/src/librbd/watcher/RewatchRequest.h b/src/librbd/watcher/RewatchRequest.h new file mode 100644 index 00000000..d4fc250a --- /dev/null +++ b/src/librbd/watcher/RewatchRequest.h @@ -0,0 +1,75 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_WATCHER_REWATCH_REQUEST_H +#define CEPH_LIBRBD_WATCHER_REWATCH_REQUEST_H + +#include "include/int_types.h" +#include "include/rados/librados.hpp" + +struct Context; +struct RWLock; + +namespace librbd { + +namespace watcher { + +class RewatchRequest { +public: + + static RewatchRequest *create(librados::IoCtx& ioctx, const std::string& oid, + RWLock &watch_lock, + librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish) { + return new RewatchRequest(ioctx, oid, watch_lock, watch_ctx, watch_handle, + on_finish); + } + + RewatchRequest(librados::IoCtx& ioctx, const std::string& oid, + RWLock &watch_lock, librados::WatchCtx2 *watch_ctx, + uint64_t *watch_handle, Context *on_finish); + + void send(); + +private: + /** + * @verbatim + * + * + * | + * v + * UNWATCH + * | + * | . . . . + * | . . (recoverable error) + * v v . + * REWATCH . . . + * | + * v + * + * + * @endverbatim + */ + + librados::IoCtx& m_ioctx; + std::string m_oid; + RWLock &m_watch_lock; + librados::WatchCtx2 *m_watch_ctx; + uint64_t *m_watch_handle; + Context *m_on_finish; + + uint64_t m_rewatch_handle = 0; + + void unwatch(); + void handle_unwatch(int r); + + void rewatch(); + void handle_rewatch(int r); + + void finish(int r); +}; + +} // namespace watcher +} // namespace librbd + +#endif // CEPH_LIBRBD_WATCHER_REWATCH_REQUEST_H diff --git a/src/librbd/watcher/Types.cc b/src/librbd/watcher/Types.cc new file mode 100644 index 00000000..8f1991d7 --- /dev/null +++ b/src/librbd/watcher/Types.cc @@ -0,0 +1,45 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/watcher/Types.h" +#include "common/Formatter.h" + +namespace librbd { +namespace watcher { + +void ClientId::encode(bufferlist &bl) const { + using ceph::encode; + encode(gid, bl); + encode(handle, bl); +} + +void ClientId::decode(bufferlist::const_iterator &iter) { + using ceph::decode; + decode(gid, iter); + decode(handle, iter); +} + +void ClientId::dump(Formatter *f) const { + f->dump_unsigned("gid", gid); + f->dump_unsigned("handle", handle); +} + +void NotifyResponse::encode(bufferlist& bl) const { + using ceph::encode; + encode(acks, bl); + encode(timeouts, bl); +} + +void NotifyResponse::decode(bufferlist::const_iterator& iter) { + using ceph::decode; + decode(acks, iter); + decode(timeouts, iter); +} +std::ostream &operator<<(std::ostream &out, + const ClientId &client_id) { + out << "[" << client_id.gid << "," << client_id.handle << "]"; + return out; +} + +} // namespace watcher +} // namespace librbd diff --git a/src/librbd/watcher/Types.h b/src/librbd/watcher/Types.h new file mode 100644 index 00000000..d1517fb0 --- /dev/null +++ b/src/librbd/watcher/Types.h @@ -0,0 +1,71 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_WATCHER_TYPES_H +#define CEPH_LIBRBD_WATCHER_TYPES_H + +#include "include/int_types.h" +#include "include/buffer_fwd.h" +#include "include/encoding.h" + +namespace ceph { class Formatter; } + +namespace librbd { + +class Watcher; + +namespace watcher { + +struct ClientId { + uint64_t gid; + uint64_t handle; + + ClientId() : gid(0), handle(0) {} + ClientId(uint64_t gid, uint64_t handle) : gid(gid), handle(handle) {} + + void encode(bufferlist& bl) const; + void decode(bufferlist::const_iterator& it); + void dump(Formatter *f) const; + + inline bool is_valid() const { + return (*this != ClientId()); + } + + inline bool operator==(const ClientId &rhs) const { + return (gid == rhs.gid && handle == rhs.handle); + } + inline bool operator!=(const ClientId &rhs) const { + return !(*this == rhs); + } + inline bool operator<(const ClientId &rhs) const { + if (gid != rhs.gid) { + return gid < rhs.gid; + } else { + return handle < rhs.handle; + } + } +}; + +struct NotifyResponse { + std::map acks; + std::vector timeouts; + + void encode(bufferlist& bl) const; + void decode(bufferlist::const_iterator& it); +}; + +template +struct Traits { + typedef librbd::Watcher Watcher; +}; + +std::ostream &operator<<(std::ostream &out, + const ClientId &client); + +WRITE_CLASS_ENCODER(ClientId); +WRITE_CLASS_ENCODER(NotifyResponse); + +} // namespace watcher +} // namespace librbd + +#endif // CEPH_LIBRBD_WATCHER_TYPES_H diff --git a/src/librbd/watcher/Utils.h b/src/librbd/watcher/Utils.h new file mode 100644 index 00000000..d2510aaf --- /dev/null +++ b/src/librbd/watcher/Utils.h @@ -0,0 +1,74 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#ifndef CEPH_LIBRBD_WATCHER_UTILS_H +#define CEPH_LIBRBD_WATCHER_UTILS_H + +#include "include/buffer_fwd.h" +#include "include/encoding.h" +#include "include/Context.h" +#include "librbd/Watcher.h" + +namespace ceph { class Formatter; } + +namespace librbd { +namespace watcher { +namespace util { + +template +struct HandlePayloadVisitor : public boost::static_visitor { + Watcher *watcher; + uint64_t notify_id; + uint64_t handle; + + HandlePayloadVisitor(Watcher *watcher_, uint64_t notify_id_, + uint64_t handle_) + : watcher(watcher_), notify_id(notify_id_), handle(handle_) + { + } + + template + inline void operator()(const P &payload) const { + typename Watcher::C_NotifyAck *ctx = + new typename Watcher::C_NotifyAck(watcher, notify_id, handle); + if (watcher->handle_payload(payload, ctx)) { + ctx->complete(0); + } + } +}; + +class EncodePayloadVisitor : public boost::static_visitor { +public: + explicit EncodePayloadVisitor(bufferlist &bl) : m_bl(bl) {} + + template + inline void operator()(const P &payload) const { + using ceph::encode; + encode(static_cast(P::NOTIFY_OP), m_bl); + payload.encode(m_bl); + } + +private: + bufferlist &m_bl; +}; + +class DecodePayloadVisitor : public boost::static_visitor { +public: + DecodePayloadVisitor(__u8 version, bufferlist::const_iterator &iter) + : m_version(version), m_iter(iter) {} + + template + inline void operator()(P &payload) const { + payload.decode(m_version, m_iter); + } + +private: + __u8 m_version; + bufferlist::const_iterator &m_iter; +}; + +} // namespace util +} // namespace watcher +} // namespace librbd + +#endif // CEPH_LIBRBD_WATCHER_UTILS_H -- cgit v1.2.3