diff options
Diffstat (limited to '')
| -rw-r--r-- | src/journal/JournalMetadata.cc | 1165 |
1 files changed, 1165 insertions, 0 deletions
diff --git a/src/journal/JournalMetadata.cc b/src/journal/JournalMetadata.cc new file mode 100644 index 000000000..bf04d4e1c --- /dev/null +++ b/src/journal/JournalMetadata.cc @@ -0,0 +1,1165 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "journal/JournalMetadata.h" +#include "journal/Utils.h" +#include "common/errno.h" +#include "common/Timer.h" +#include "cls/journal/cls_journal_client.h" +#include <functional> +#include <set> + +#define dout_subsys ceph_subsys_journaler +#undef dout_prefix +#define dout_prefix *_dout << "JournalMetadata: " << this << " " + +namespace journal { + +using namespace cls::journal; + +namespace { + +struct C_GetClient : public Context { + CephContext *cct; + librados::IoCtx &ioctx; + const std::string &oid; + AsyncOpTracker &async_op_tracker; + std::string client_id; + cls::journal::Client *client; + Context *on_finish; + + bufferlist out_bl; + + C_GetClient(CephContext *cct, librados::IoCtx &ioctx, const std::string &oid, + AsyncOpTracker &async_op_tracker, const std::string &client_id, + cls::journal::Client *client, Context *on_finish) + : cct(cct), ioctx(ioctx), oid(oid), async_op_tracker(async_op_tracker), + client_id(client_id), client(client), on_finish(on_finish) { + async_op_tracker.start_op(); + } + ~C_GetClient() override { + async_op_tracker.finish_op(); + } + + virtual void send() { + send_get_client(); + } + + void send_get_client() { + ldout(cct, 20) << "C_GetClient: " << __func__ << dendl; + + librados::ObjectReadOperation op; + client::get_client_start(&op, client_id); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_GetClient, &C_GetClient::handle_get_client>); + + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_get_client(int r) { + ldout(cct, 20) << "C_GetClient: " << __func__ << ": r=" << r << dendl; + + if (r == 0) { + auto it = out_bl.cbegin(); + r = client::get_client_finish(&it, client); + } + complete(r); + } + + void finish(int r) override { + on_finish->complete(r); + } +}; + +struct C_AllocateTag : public Context { + CephContext *cct; + librados::IoCtx &ioctx; + const std::string &oid; + AsyncOpTracker &async_op_tracker; + uint64_t tag_class; + Tag *tag; + Context *on_finish; + + bufferlist out_bl; + + C_AllocateTag(CephContext *cct, librados::IoCtx &ioctx, + const std::string &oid, AsyncOpTracker &async_op_tracker, + uint64_t tag_class, const bufferlist &data, Tag *tag, + Context *on_finish) + : cct(cct), ioctx(ioctx), oid(oid), async_op_tracker(async_op_tracker), + tag_class(tag_class), tag(tag), on_finish(on_finish) { + async_op_tracker.start_op(); + tag->data = data; + } + ~C_AllocateTag() override { + async_op_tracker.finish_op(); + } + + void send() { + send_get_next_tag_tid(); + } + + void send_get_next_tag_tid() { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << dendl; + + librados::ObjectReadOperation op; + client::get_next_tag_tid_start(&op); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_AllocateTag, &C_AllocateTag::handle_get_next_tag_tid>); + + out_bl.clear(); + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_get_next_tag_tid(int r) { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << ": r=" << r << dendl; + + if (r == 0) { + auto iter = out_bl.cbegin(); + r = client::get_next_tag_tid_finish(&iter, &tag->tid); + } + if (r < 0) { + complete(r); + return; + } + send_tag_create(); + } + + void send_tag_create() { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << dendl; + + librados::ObjectWriteOperation op; + client::tag_create(&op, tag->tid, tag_class, tag->data); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_AllocateTag, &C_AllocateTag::handle_tag_create>); + + int r = ioctx.aio_operate(oid, comp, &op); + ceph_assert(r == 0); + comp->release(); + } + + void handle_tag_create(int r) { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << ": r=" << r << dendl; + + if (r == -ESTALE) { + send_get_next_tag_tid(); + return; + } else if (r < 0) { + complete(r); + return; + } + + send_get_tag(); + } + + void send_get_tag() { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << dendl; + + librados::ObjectReadOperation op; + client::get_tag_start(&op, tag->tid); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_AllocateTag, &C_AllocateTag::handle_get_tag>); + + out_bl.clear(); + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_get_tag(int r) { + ldout(cct, 20) << "C_AllocateTag: " << __func__ << ": r=" << r << dendl; + + if (r == 0) { + auto iter = out_bl.cbegin(); + + cls::journal::Tag journal_tag; + r = client::get_tag_finish(&iter, &journal_tag); + if (r == 0) { + *tag = journal_tag; + } + } + complete(r); + } + + void finish(int r) override { + on_finish->complete(r); + } +}; + +struct C_GetTag : public Context { + CephContext *cct; + librados::IoCtx &ioctx; + const std::string &oid; + AsyncOpTracker &async_op_tracker; + uint64_t tag_tid; + JournalMetadata::Tag *tag; + Context *on_finish; + + bufferlist out_bl; + + C_GetTag(CephContext *cct, librados::IoCtx &ioctx, const std::string &oid, + AsyncOpTracker &async_op_tracker, uint64_t tag_tid, + JournalMetadata::Tag *tag, Context *on_finish) + : cct(cct), ioctx(ioctx), oid(oid), async_op_tracker(async_op_tracker), + tag_tid(tag_tid), tag(tag), on_finish(on_finish) { + async_op_tracker.start_op(); + } + ~C_GetTag() override { + async_op_tracker.finish_op(); + } + + void send() { + send_get_tag(); + } + + void send_get_tag() { + librados::ObjectReadOperation op; + client::get_tag_start(&op, tag_tid); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_GetTag, &C_GetTag::handle_get_tag>); + + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_get_tag(int r) { + if (r == 0) { + auto iter = out_bl.cbegin(); + r = client::get_tag_finish(&iter, tag); + } + complete(r); + } + + void finish(int r) override { + on_finish->complete(r); + } +}; + +struct C_GetTags : public Context { + CephContext *cct; + librados::IoCtx &ioctx; + const std::string &oid; + const std::string &client_id; + AsyncOpTracker &async_op_tracker; + uint64_t start_after_tag_tid; + boost::optional<uint64_t> tag_class; + JournalMetadata::Tags *tags; + Context *on_finish; + + const uint64_t MAX_RETURN = 64; + bufferlist out_bl; + + C_GetTags(CephContext *cct, librados::IoCtx &ioctx, const std::string &oid, + const std::string &client_id, AsyncOpTracker &async_op_tracker, + uint64_t start_after_tag_tid, + const boost::optional<uint64_t> &tag_class, + JournalMetadata::Tags *tags, Context *on_finish) + : cct(cct), ioctx(ioctx), oid(oid), client_id(client_id), + async_op_tracker(async_op_tracker), + start_after_tag_tid(start_after_tag_tid), tag_class(tag_class), + tags(tags), on_finish(on_finish) { + async_op_tracker.start_op(); + } + ~C_GetTags() override { + async_op_tracker.finish_op(); + } + + void send() { + send_tag_list(); + } + + void send_tag_list() { + librados::ObjectReadOperation op; + client::tag_list_start(&op, start_after_tag_tid, MAX_RETURN, client_id, + tag_class); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_GetTags, &C_GetTags::handle_tag_list>); + + out_bl.clear(); + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_tag_list(int r) { + if (r == 0) { + std::set<cls::journal::Tag> journal_tags; + auto iter = out_bl.cbegin(); + r = client::tag_list_finish(&iter, &journal_tags); + if (r == 0) { + for (auto &journal_tag : journal_tags) { + tags->push_back(journal_tag); + start_after_tag_tid = journal_tag.tid; + } + + if (journal_tags.size() == MAX_RETURN) { + send_tag_list(); + return; + } + } + } + complete(r); + } + + void finish(int r) override { + on_finish->complete(r); + } +}; + +struct C_FlushCommitPosition : public Context { + Context *commit_position_ctx; + Context *on_finish; + + C_FlushCommitPosition(Context *commit_position_ctx, Context *on_finish) + : commit_position_ctx(commit_position_ctx), on_finish(on_finish) { + } + void finish(int r) override { + if (commit_position_ctx != nullptr) { + commit_position_ctx->complete(r); + } + on_finish->complete(r); + } +}; + +struct C_AssertActiveTag : public Context { + CephContext *cct; + librados::IoCtx &ioctx; + const std::string &oid; + AsyncOpTracker &async_op_tracker; + std::string client_id; + uint64_t tag_tid; + Context *on_finish; + + bufferlist out_bl; + + C_AssertActiveTag(CephContext *cct, librados::IoCtx &ioctx, + const std::string &oid, AsyncOpTracker &async_op_tracker, + const std::string &client_id, uint64_t tag_tid, + Context *on_finish) + : cct(cct), ioctx(ioctx), oid(oid), async_op_tracker(async_op_tracker), + client_id(client_id), tag_tid(tag_tid), on_finish(on_finish) { + async_op_tracker.start_op(); + } + ~C_AssertActiveTag() override { + async_op_tracker.finish_op(); + } + + void send() { + ldout(cct, 20) << "C_AssertActiveTag: " << __func__ << dendl; + + librados::ObjectReadOperation op; + client::tag_list_start(&op, tag_tid, 2, client_id, boost::none); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + this, &utils::rados_state_callback< + C_AssertActiveTag, &C_AssertActiveTag::handle_send>); + + int r = ioctx.aio_operate(oid, comp, &op, &out_bl); + ceph_assert(r == 0); + comp->release(); + } + + void handle_send(int r) { + ldout(cct, 20) << "C_AssertActiveTag: " << __func__ << ": r=" << r << dendl; + + std::set<cls::journal::Tag> tags; + if (r == 0) { + auto it = out_bl.cbegin(); + r = client::tag_list_finish(&it, &tags); + } + + // NOTE: since 0 is treated as an uninitialized list filter, we need to + // load to entries and look at the last tid + if (r == 0 && !tags.empty() && tags.rbegin()->tid > tag_tid) { + r = -ESTALE; + } + complete(r); + } + + void finish(int r) override { + on_finish->complete(r); + } +}; + +} // anonymous namespace + +JournalMetadata::JournalMetadata(ContextWQ *work_queue, SafeTimer *timer, + ceph::mutex *timer_lock, librados::IoCtx &ioctx, + const std::string &oid, + const std::string &client_id, + const Settings &settings) + : m_oid(oid), + m_client_id(client_id), m_settings(settings), + m_work_queue(work_queue), m_timer(timer), m_timer_lock(timer_lock), + m_watch_ctx(this) +{ + m_ioctx.dup(ioctx); + m_cct = reinterpret_cast<CephContext*>(m_ioctx.cct()); +} + +JournalMetadata::~JournalMetadata() { + std::lock_guard locker{m_lock}; + ceph_assert(!m_initialized); +} + +void JournalMetadata::init(Context *on_finish) { + { + std::lock_guard locker{m_lock}; + ceph_assert(!m_initialized); + m_initialized = true; + } + + // chain the init sequence (reverse order) + on_finish = utils::create_async_context_callback( + this, on_finish); + on_finish = new C_ImmutableMetadata(this, on_finish); + on_finish = new LambdaContext([this, on_finish](int r) { + if (r < 0) { + lderr(m_cct) << __func__ << ": failed to watch journal" + << cpp_strerror(r) << dendl; + std::lock_guard locker{m_lock}; + m_watch_handle = 0; + on_finish->complete(r); + return; + } + + get_immutable_metadata(&m_order, &m_splay_width, &m_pool_id, on_finish); + }); + + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + on_finish, utils::rados_ctx_callback); + int r = m_ioctx.aio_watch(m_oid, comp, &m_watch_handle, &m_watch_ctx); + ceph_assert(r == 0); + comp->release(); +} + +void JournalMetadata::shut_down(Context *on_finish) { + + ldout(m_cct, 20) << __func__ << dendl; + + uint64_t watch_handle = 0; + { + std::lock_guard locker{m_lock}; + m_initialized = false; + std::swap(watch_handle, m_watch_handle); + } + + // chain the shut down sequence (reverse order) + on_finish = utils::create_async_context_callback( + this, on_finish); + on_finish = new LambdaContext([this, on_finish](int r) { + ldout(m_cct, 20) << "shut_down: waiting for ops" << dendl; + m_async_op_tracker.wait_for_ops(on_finish); + }); + on_finish = new LambdaContext([this, on_finish](int r) { + ldout(m_cct, 20) << "shut_down: flushing watch" << dendl; + librados::Rados rados(m_ioctx); + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + on_finish, utils::rados_ctx_callback); + r = rados.aio_watch_flush(comp); + ceph_assert(r == 0); + comp->release(); + }); + on_finish = new LambdaContext([this, on_finish](int r) { + flush_commit_position(on_finish); + }); + if (watch_handle != 0) { + librados::AioCompletion *comp = librados::Rados::aio_create_completion( + on_finish, utils::rados_ctx_callback); + int r = m_ioctx.aio_unwatch(watch_handle, comp); + ceph_assert(r == 0); + comp->release(); + } else { + on_finish->complete(0); + } +} + +void JournalMetadata::get_immutable_metadata(uint8_t *order, + uint8_t *splay_width, + int64_t *pool_id, + Context *on_finish) { + client::get_immutable_metadata(m_ioctx, m_oid, order, splay_width, pool_id, + on_finish); +} + +void JournalMetadata::get_mutable_metadata(uint64_t *minimum_set, + uint64_t *active_set, + RegisteredClients *clients, + Context *on_finish) { + client::get_mutable_metadata(m_ioctx, m_oid, minimum_set, active_set, clients, + on_finish); +} + +void JournalMetadata::register_client(const bufferlist &data, + Context *on_finish) { + ldout(m_cct, 10) << __func__ << ": " << m_client_id << dendl; + librados::ObjectWriteOperation op; + client::client_register(&op, m_client_id, data); + + C_NotifyUpdate *ctx = new C_NotifyUpdate(this, on_finish); + + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, + utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); +} + +void JournalMetadata::update_client(const bufferlist &data, + Context *on_finish) { + ldout(m_cct, 10) << __func__ << ": " << m_client_id << dendl; + librados::ObjectWriteOperation op; + client::client_update_data(&op, m_client_id, data); + + C_NotifyUpdate *ctx = new C_NotifyUpdate(this, on_finish); + + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); +} + +void JournalMetadata::unregister_client(Context *on_finish) { + ceph_assert(!m_client_id.empty()); + + ldout(m_cct, 10) << __func__ << ": " << m_client_id << dendl; + librados::ObjectWriteOperation op; + client::client_unregister(&op, m_client_id); + + C_NotifyUpdate *ctx = new C_NotifyUpdate(this, on_finish); + + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); +} + +void JournalMetadata::allocate_tag(uint64_t tag_class, const bufferlist &data, + Tag *tag, Context *on_finish) { + on_finish = new C_NotifyUpdate(this, on_finish); + C_AllocateTag *ctx = new C_AllocateTag(m_cct, m_ioctx, m_oid, + m_async_op_tracker, tag_class, + data, tag, on_finish); + ctx->send(); +} + +void JournalMetadata::get_client(const std::string &client_id, + cls::journal::Client *client, + Context *on_finish) { + C_GetClient *ctx = new C_GetClient(m_cct, m_ioctx, m_oid, m_async_op_tracker, + client_id, client, on_finish); + ctx->send(); +} + +void JournalMetadata::get_tag(uint64_t tag_tid, Tag *tag, Context *on_finish) { + C_GetTag *ctx = new C_GetTag(m_cct, m_ioctx, m_oid, m_async_op_tracker, + tag_tid, tag, on_finish); + ctx->send(); +} + +void JournalMetadata::get_tags(uint64_t start_after_tag_tid, + const boost::optional<uint64_t> &tag_class, + Tags *tags, Context *on_finish) { + C_GetTags *ctx = new C_GetTags(m_cct, m_ioctx, m_oid, m_client_id, + m_async_op_tracker, start_after_tag_tid, + tag_class, tags, on_finish); + ctx->send(); +} + +void JournalMetadata::add_listener(JournalMetadataListener *listener) { + std::unique_lock locker{m_lock}; + m_update_cond.wait(locker, [this] { + return m_update_notifications <= 0; + }); + m_listeners.push_back(listener); +} + +void JournalMetadata::remove_listener(JournalMetadataListener *listener) { + std::unique_lock locker{m_lock}; + m_update_cond.wait(locker, [this] { + return m_update_notifications <= 0; + }); + m_listeners.remove(listener); +} + +void JournalMetadata::set_minimum_set(uint64_t object_set) { + std::lock_guard locker{m_lock}; + + ldout(m_cct, 20) << __func__ << ": current=" << m_minimum_set + << ", new=" << object_set << dendl; + if (m_minimum_set >= object_set) { + return; + } + + librados::ObjectWriteOperation op; + client::set_minimum_set(&op, object_set); + + C_NotifyUpdate *ctx = new C_NotifyUpdate(this); + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); + + m_minimum_set = object_set; +} + +int JournalMetadata::set_active_set(uint64_t object_set) { + C_SaferCond ctx; + set_active_set(object_set, &ctx); + return ctx.wait(); +} + +void JournalMetadata::set_active_set(uint64_t object_set, Context *on_finish) { + std::lock_guard locker{m_lock}; + + ldout(m_cct, 20) << __func__ << ": current=" << m_active_set + << ", new=" << object_set << dendl; + if (m_active_set >= object_set) { + m_work_queue->queue(on_finish, 0); + return; + } + + librados::ObjectWriteOperation op; + client::set_active_set(&op, object_set); + + C_NotifyUpdate *ctx = new C_NotifyUpdate(this, on_finish); + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); + + m_active_set = object_set; +} + +void JournalMetadata::assert_active_tag(uint64_t tag_tid, Context *on_finish) { + std::lock_guard locker{m_lock}; + + C_AssertActiveTag *ctx = new C_AssertActiveTag(m_cct, m_ioctx, m_oid, + m_async_op_tracker, + m_client_id, tag_tid, + on_finish); + ctx->send(); +} + +void JournalMetadata::flush_commit_position() { + ldout(m_cct, 20) << __func__ << dendl; + + C_SaferCond ctx; + flush_commit_position(&ctx); + ctx.wait(); +} + +void JournalMetadata::flush_commit_position(Context *on_safe) { + ldout(m_cct, 20) << __func__ << dendl; + + std::scoped_lock locker{*m_timer_lock, m_lock}; + if (m_commit_position_ctx == nullptr && m_flush_commits_in_progress == 0) { + // nothing to flush + if (on_safe != nullptr) { + m_work_queue->queue(on_safe, 0); + } + return; + } + + if (on_safe != nullptr) { + m_flush_commit_position_ctxs.push_back(on_safe); + } + if (m_commit_position_ctx == nullptr) { + return; + } + + cancel_commit_task(); + handle_commit_position_task(); +} + +void JournalMetadata::reserve_entry_tid(uint64_t tag_tid, uint64_t entry_tid) { + std::lock_guard locker{m_lock}; + uint64_t &allocated_entry_tid = m_allocated_entry_tids[tag_tid]; + if (allocated_entry_tid <= entry_tid) { + allocated_entry_tid = entry_tid + 1; + } +} + +bool JournalMetadata::get_last_allocated_entry_tid(uint64_t tag_tid, + uint64_t *entry_tid) const { + std::lock_guard locker{m_lock}; + + AllocatedEntryTids::const_iterator it = m_allocated_entry_tids.find(tag_tid); + if (it == m_allocated_entry_tids.end()) { + return false; + } + + ceph_assert(it->second > 0); + *entry_tid = it->second - 1; + return true; +} + +void JournalMetadata::handle_immutable_metadata(int r, Context *on_init) { + if (r < 0) { + lderr(m_cct) << "failed to initialize immutable metadata: " + << cpp_strerror(r) << dendl; + on_init->complete(r); + return; + } + + ldout(m_cct, 10) << "initialized immutable metadata" << dendl; + refresh(on_init); +} + +void JournalMetadata::refresh(Context *on_complete) { + ldout(m_cct, 10) << "refreshing mutable metadata" << dendl; + + { + std::lock_guard locker{m_lock}; + if (on_complete != nullptr) { + m_refresh_ctxs.push_back(on_complete); + } + ++m_refreshes_in_progress; + } + + auto refresh = new C_Refresh(this); + get_mutable_metadata(&refresh->minimum_set, &refresh->active_set, + &refresh->registered_clients, refresh); +} + +void JournalMetadata::handle_refresh_complete(C_Refresh *refresh, int r) { + ldout(m_cct, 10) << "refreshed mutable metadata: r=" << r << dendl; + + m_lock.lock(); + if (r == 0) { + Client client(m_client_id, bufferlist()); + RegisteredClients::iterator it = refresh->registered_clients.find(client); + if (it != refresh->registered_clients.end()) { + if (it->state == cls::journal::CLIENT_STATE_DISCONNECTED) { + ldout(m_cct, 0) << "client flagged disconnected: " << m_client_id + << dendl; + } + m_minimum_set = std::max(m_minimum_set, refresh->minimum_set); + m_active_set = std::max(m_active_set, refresh->active_set); + m_registered_clients = refresh->registered_clients; + m_client = *it; + + ++m_update_notifications; + m_lock.unlock(); + for (Listeners::iterator it = m_listeners.begin(); + it != m_listeners.end(); ++it) { + (*it)->handle_update(this); + } + m_lock.lock(); + if (--m_update_notifications == 0) { + m_update_cond.notify_all(); + } + } else { + lderr(m_cct) << "failed to locate client: " << m_client_id << dendl; + r = -ENOENT; + } + } + + Contexts refresh_ctxs; + ceph_assert(m_refreshes_in_progress > 0); + --m_refreshes_in_progress; + if (m_refreshes_in_progress == 0) { + std::swap(refresh_ctxs, m_refresh_ctxs); + } + m_lock.unlock(); + + for (auto ctx : refresh_ctxs) { + ctx->complete(r); + } +} + +void JournalMetadata::cancel_commit_task() { + ldout(m_cct, 20) << __func__ << dendl; + + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + ceph_assert(ceph_mutex_is_locked(m_lock)); + ceph_assert(m_commit_position_ctx != nullptr); + ceph_assert(m_commit_position_task_ctx != nullptr); + m_timer->cancel_event(m_commit_position_task_ctx); + m_commit_position_task_ctx = NULL; +} + +void JournalMetadata::schedule_commit_task() { + ldout(m_cct, 20) << __func__ << dendl; + + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + ceph_assert(ceph_mutex_is_locked(m_lock)); + ceph_assert(m_commit_position_ctx != nullptr); + if (m_commit_position_task_ctx == nullptr) { + m_commit_position_task_ctx = + m_timer->add_event_after(m_settings.commit_interval, + new C_CommitPositionTask(this)); + } +} + +void JournalMetadata::handle_commit_position_task() { + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + ceph_assert(ceph_mutex_is_locked(m_lock)); + ldout(m_cct, 20) << __func__ << ": " + << "client_id=" << m_client_id << ", " + << "commit_position=" << m_commit_position << dendl; + + m_commit_position_task_ctx = nullptr; + Context* commit_position_ctx = nullptr; + std::swap(commit_position_ctx, m_commit_position_ctx); + + m_async_op_tracker.start_op(); + ++m_flush_commits_in_progress; + + Context* ctx = new LambdaContext([this, commit_position_ctx](int r) { + Contexts flush_commit_position_ctxs; + m_lock.lock(); + ceph_assert(m_flush_commits_in_progress > 0); + --m_flush_commits_in_progress; + if (m_flush_commits_in_progress == 0) { + std::swap(flush_commit_position_ctxs, m_flush_commit_position_ctxs); + } + m_lock.unlock(); + + commit_position_ctx->complete(0); + for (auto ctx : flush_commit_position_ctxs) { + ctx->complete(0); + } + m_async_op_tracker.finish_op(); + }); + ctx = new C_NotifyUpdate(this, ctx); + ctx = new LambdaContext([this, ctx](int r) { + // manually kick of a refresh in case the notification is missed + // and ignore the next notification that we are about to send + m_lock.lock(); + ++m_ignore_watch_notifies; + m_lock.unlock(); + + refresh(ctx); + }); + ctx = new LambdaContext([this, ctx](int r) { + schedule_laggy_clients_disconnect(ctx); + }); + + librados::ObjectWriteOperation op; + client::client_commit(&op, m_client_id, m_commit_position); + + auto comp = librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); +} + +void JournalMetadata::schedule_watch_reset() { + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + m_timer->add_event_after(1, new C_WatchReset(this)); +} + +void JournalMetadata::handle_watch_reset() { + ceph_assert(ceph_mutex_is_locked(*m_timer_lock)); + if (!m_initialized) { + return; + } + + int r = m_ioctx.watch2(m_oid, &m_watch_handle, &m_watch_ctx); + if (r < 0) { + if (r == -ENOENT) { + ldout(m_cct, 5) << __func__ << ": journal header not found" << dendl; + } else if (r == -EBLOCKLISTED) { + ldout(m_cct, 5) << __func__ << ": client blocklisted" << dendl; + } else { + lderr(m_cct) << __func__ << ": failed to watch journal: " + << cpp_strerror(r) << dendl; + } + schedule_watch_reset(); + } else { + ldout(m_cct, 10) << __func__ << ": reset journal watch" << dendl; + refresh(NULL); + } +} + +void JournalMetadata::handle_watch_notify(uint64_t notify_id, uint64_t cookie) { + ldout(m_cct, 10) << "journal header updated" << dendl; + + bufferlist bl; + m_ioctx.notify_ack(m_oid, notify_id, cookie, bl); + + { + std::lock_guard locker{m_lock}; + if (m_ignore_watch_notifies > 0) { + --m_ignore_watch_notifies; + return; + } + } + + refresh(NULL); +} + +void JournalMetadata::handle_watch_error(int err) { + if (err == -ENOTCONN) { + ldout(m_cct, 5) << "journal watch error: header removed" << dendl; + } else if (err == -EBLOCKLISTED) { + lderr(m_cct) << "journal watch error: client blocklisted" << dendl; + } else { + lderr(m_cct) << "journal watch error: " << cpp_strerror(err) << dendl; + } + + std::scoped_lock locker{*m_timer_lock, m_lock}; + + // release old watch on error + if (m_watch_handle != 0) { + m_ioctx.unwatch2(m_watch_handle); + m_watch_handle = 0; + } + + if (m_initialized && err != -ENOENT) { + schedule_watch_reset(); + } +} + +uint64_t JournalMetadata::allocate_commit_tid(uint64_t object_num, + uint64_t tag_tid, + uint64_t entry_tid) { + std::lock_guard locker{m_lock}; + uint64_t commit_tid = ++m_commit_tid; + m_pending_commit_tids[commit_tid] = CommitEntry(object_num, tag_tid, + entry_tid); + + ldout(m_cct, 20) << "allocated commit tid: commit_tid=" << commit_tid << " [" + << "object_num=" << object_num << ", " + << "tag_tid=" << tag_tid << ", " + << "entry_tid=" << entry_tid << "]" + << dendl; + return commit_tid; +} + +void JournalMetadata::overflow_commit_tid(uint64_t commit_tid, + uint64_t object_num) { + std::lock_guard locker{m_lock}; + + auto it = m_pending_commit_tids.find(commit_tid); + ceph_assert(it != m_pending_commit_tids.end()); + ceph_assert(it->second.object_num < object_num); + + ldout(m_cct, 20) << __func__ << ": " + << "commit_tid=" << commit_tid << ", " + << "old_object_num=" << it->second.object_num << ", " + << "new_object_num=" << object_num << dendl; + it->second.object_num = object_num; +} + +void JournalMetadata::get_commit_entry(uint64_t commit_tid, + uint64_t *object_num, + uint64_t *tag_tid, uint64_t *entry_tid) { + std::lock_guard locker{m_lock}; + + auto it = m_pending_commit_tids.find(commit_tid); + ceph_assert(it != m_pending_commit_tids.end()); + + *object_num = it->second.object_num; + *tag_tid = it->second.tag_tid; + *entry_tid = it->second.entry_tid; +} + +void JournalMetadata::committed(uint64_t commit_tid, + const CreateContext &create_context) { + ldout(m_cct, 20) << "committed tid=" << commit_tid << dendl; + + ObjectSetPosition commit_position; + Context *stale_ctx = nullptr; + { + std::scoped_lock locker{*m_timer_lock, m_lock}; + ceph_assert(commit_tid > m_commit_position_tid); + + if (!m_commit_position.object_positions.empty()) { + // in-flight commit position update + commit_position = m_commit_position; + } else { + // safe commit position + commit_position = m_client.commit_position; + } + + CommitTids::iterator it = m_pending_commit_tids.find(commit_tid); + ceph_assert(it != m_pending_commit_tids.end()); + + CommitEntry &commit_entry = it->second; + commit_entry.committed = true; + + bool update_commit_position = false; + while (!m_pending_commit_tids.empty()) { + CommitTids::iterator it = m_pending_commit_tids.begin(); + CommitEntry &commit_entry = it->second; + if (!commit_entry.committed) { + break; + } + + commit_position.object_positions.emplace_front( + commit_entry.object_num, commit_entry.tag_tid, + commit_entry.entry_tid); + m_pending_commit_tids.erase(it); + update_commit_position = true; + } + + if (!update_commit_position) { + return; + } + + // prune the position to have one position per splay offset + std::set<uint8_t> in_use_splay_offsets; + ObjectPositions::iterator ob_it = commit_position.object_positions.begin(); + while (ob_it != commit_position.object_positions.end()) { + uint8_t splay_offset = ob_it->object_number % m_splay_width; + if (!in_use_splay_offsets.insert(splay_offset).second) { + ob_it = commit_position.object_positions.erase(ob_it); + } else { + ++ob_it; + } + } + + stale_ctx = m_commit_position_ctx; + m_commit_position_ctx = create_context(); + m_commit_position = commit_position; + m_commit_position_tid = commit_tid; + + ldout(m_cct, 20) << "updated commit position: " << commit_position << ", " + << "on_safe=" << m_commit_position_ctx << dendl; + schedule_commit_task(); + } + + + if (stale_ctx != nullptr) { + ldout(m_cct, 20) << "canceling stale commit: on_safe=" << stale_ctx + << dendl; + stale_ctx->complete(-ESTALE); + } +} + +void JournalMetadata::notify_update() { + ldout(m_cct, 10) << "notifying journal header update" << dendl; + + bufferlist bl; + m_ioctx.notify2(m_oid, bl, 5000, NULL); +} + +void JournalMetadata::async_notify_update(Context *on_safe) { + ldout(m_cct, 10) << "async notifying journal header update" << dendl; + + C_AioNotify *ctx = new C_AioNotify(this, on_safe); + librados::AioCompletion *comp = + librados::Rados::aio_create_completion(ctx, utils::rados_ctx_callback); + + bufferlist bl; + int r = m_ioctx.aio_notify(m_oid, comp, bl, 5000, NULL); + ceph_assert(r == 0); + + comp->release(); +} + +void JournalMetadata::wait_for_ops() { + C_SaferCond ctx; + m_async_op_tracker.wait_for_ops(&ctx); + ctx.wait(); +} + +void JournalMetadata::handle_notified(int r) { + ldout(m_cct, 10) << "notified journal header update: r=" << r << dendl; +} + +void JournalMetadata::schedule_laggy_clients_disconnect(Context *on_finish) { + ldout(m_cct, 20) << __func__ << dendl; + if (m_settings.max_concurrent_object_sets <= 0) { + on_finish->complete(0); + return; + } + + Context *ctx = on_finish; + { + std::lock_guard locker{m_lock}; + for (auto &c : m_registered_clients) { + if (c.state == cls::journal::CLIENT_STATE_DISCONNECTED || + c.id == m_client_id || + m_settings.ignored_laggy_clients.count(c.id) > 0) { + continue; + } + const std::string &client_id = c.id; + uint64_t object_set = 0; + if (!c.commit_position.object_positions.empty()) { + auto &position = *(c.commit_position.object_positions.begin()); + object_set = position.object_number / m_splay_width; + } + + if (m_active_set > object_set + m_settings.max_concurrent_object_sets) { + ldout(m_cct, 1) << __func__ << ": " << client_id + << ": scheduling disconnect" << dendl; + + ctx = new LambdaContext([this, client_id, ctx](int r1) { + ldout(m_cct, 10) << __func__ << ": " << client_id + << ": flagging disconnected" << dendl; + + librados::ObjectWriteOperation op; + client::client_update_state( + &op, client_id, cls::journal::CLIENT_STATE_DISCONNECTED); + + auto comp = librados::Rados::aio_create_completion( + ctx, utils::rados_ctx_callback); + int r = m_ioctx.aio_operate(m_oid, comp, &op); + ceph_assert(r == 0); + comp->release(); + }); + } + } + } + + if (ctx == on_finish) { + ldout(m_cct, 20) << __func__ << ": no laggy clients to disconnect" << dendl; + } + ctx->complete(0); +} + +std::ostream &operator<<(std::ostream &os, + const JournalMetadata::RegisteredClients &clients) { + os << "["; + for (JournalMetadata::RegisteredClients::const_iterator c = clients.begin(); + c != clients.end(); ++c) { + os << (c == clients.begin() ? "" : ", " ) << *c; + } + os << "]"; + return os; +} + +std::ostream &operator<<(std::ostream &os, + const JournalMetadata &jm) { + std::lock_guard locker{jm.m_lock}; + os << "[oid=" << jm.m_oid << ", " + << "initialized=" << jm.m_initialized << ", " + << "order=" << (int)jm.m_order << ", " + << "splay_width=" << (int)jm.m_splay_width << ", " + << "pool_id=" << jm.m_pool_id << ", " + << "minimum_set=" << jm.m_minimum_set << ", " + << "active_set=" << jm.m_active_set << ", " + << "client_id=" << jm.m_client_id << ", " + << "commit_tid=" << jm.m_commit_tid << ", " + << "commit_interval=" << jm.m_settings.commit_interval << ", " + << "commit_position=" << jm.m_commit_position << ", " + << "registered_clients=" << jm.m_registered_clients << "]"; + return os; +} + +} // namespace journal |