+++ /dev/null
-// -*- 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, nullptr, &utils::rados_state_callback<
- C_GetClient, &C_GetClient::handle_get_client>);
-
- int r = ioctx.aio_operate(oid, comp, &op, &out_bl);
- assert(r == 0);
- comp->release();
- }
-
- void handle_get_client(int r) {
- ldout(cct, 20) << "C_GetClient: " << __func__ << ": r=" << r << dendl;
-
- if (r == 0) {
- bufferlist::iterator it = out_bl.begin();
- 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, nullptr, &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);
- 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) {
- bufferlist::iterator iter = out_bl.begin();
- 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, nullptr, &utils::rados_state_callback<
- C_AllocateTag, &C_AllocateTag::handle_tag_create>);
-
- int r = ioctx.aio_operate(oid, comp, &op);
- 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, nullptr, &utils::rados_state_callback<
- C_AllocateTag, &C_AllocateTag::handle_get_tag>);
-
- out_bl.clear();
- int r = ioctx.aio_operate(oid, comp, &op, &out_bl);
- assert(r == 0);
- comp->release();
- }
-
- void handle_get_tag(int r) {
- ldout(cct, 20) << "C_AllocateTag: " << __func__ << ": r=" << r << dendl;
-
- if (r == 0) {
- bufferlist::iterator iter = out_bl.begin();
-
- 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, nullptr, &utils::rados_state_callback<
- C_GetTag, &C_GetTag::handle_get_tag>);
-
- int r = ioctx.aio_operate(oid, comp, &op, &out_bl);
- assert(r == 0);
- comp->release();
- }
-
- void handle_get_tag(int r) {
- if (r == 0) {
- bufferlist::iterator iter = out_bl.begin();
- 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, nullptr, &utils::rados_state_callback<
- C_GetTags, &C_GetTags::handle_tag_list>);
-
- out_bl.clear();
- int r = ioctx.aio_operate(oid, comp, &op, &out_bl);
- assert(r == 0);
- comp->release();
- }
-
- void handle_tag_list(int r) {
- if (r == 0) {
- std::set<cls::journal::Tag> journal_tags;
- bufferlist::iterator iter = out_bl.begin();
- 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, nullptr, &utils::rados_state_callback<
- C_AssertActiveTag, &C_AssertActiveTag::handle_send>);
-
- int r = ioctx.aio_operate(oid, comp, &op, &out_bl);
- 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) {
- bufferlist::iterator it = out_bl.begin();
- 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,
- Mutex *timer_lock, librados::IoCtx &ioctx,
- const std::string &oid,
- const std::string &client_id,
- const Settings &settings)
- : RefCountedObject(NULL, 0), m_cct(NULL), m_oid(oid),
- m_client_id(client_id), m_settings(settings), m_order(0),
- m_splay_width(0), m_pool_id(-1), m_initialized(false),
- m_work_queue(work_queue), m_timer(timer), m_timer_lock(timer_lock),
- m_lock("JournalMetadata::m_lock"), m_commit_tid(0), m_watch_ctx(this),
- m_watch_handle(0), m_minimum_set(0), m_active_set(0),
- m_update_notifications(0), m_commit_position_ctx(NULL),
- m_commit_position_task_ctx(NULL) {
- m_ioctx.dup(ioctx);
- m_cct = reinterpret_cast<CephContext*>(m_ioctx.cct());
-}
-
-JournalMetadata::~JournalMetadata() {
- Mutex::Locker locker(m_lock);
- assert(!m_initialized);
-}
-
-void JournalMetadata::init(Context *on_finish) {
- {
- Mutex::Locker locker(m_lock);
- 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 FunctionContext([this, on_finish](int r) {
- if (r < 0) {
- lderr(m_cct) << __func__ << ": failed to watch journal"
- << cpp_strerror(r) << dendl;
- Mutex::Locker 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, nullptr, utils::rados_ctx_callback);
- int r = m_ioctx.aio_watch(m_oid, comp, &m_watch_handle, &m_watch_ctx);
- assert(r == 0);
- comp->release();
-}
-
-void JournalMetadata::shut_down(Context *on_finish) {
-
- ldout(m_cct, 20) << __func__ << dendl;
-
- uint64_t watch_handle = 0;
- {
- Mutex::Locker 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 FunctionContext([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 FunctionContext([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, nullptr, utils::rados_ctx_callback);
- r = rados.aio_watch_flush(comp);
- assert(r == 0);
- comp->release();
- });
- on_finish = new FunctionContext([this, on_finish](int r) {
- flush_commit_position(on_finish);
- });
- if (watch_handle != 0) {
- librados::AioCompletion *comp = librados::Rados::aio_create_completion(
- on_finish, nullptr, utils::rados_ctx_callback);
- int r = m_ioctx.aio_unwatch(watch_handle, comp);
- 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, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- 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, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- assert(r == 0);
- comp->release();
-}
-
-void JournalMetadata::unregister_client(Context *on_finish) {
- 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, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- 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) {
- Mutex::Locker locker(m_lock);
- while (m_update_notifications > 0) {
- m_update_cond.Wait(m_lock);
- }
- m_listeners.push_back(listener);
-}
-
-void JournalMetadata::remove_listener(JournalMetadataListener *listener) {
- Mutex::Locker locker(m_lock);
- while (m_update_notifications > 0) {
- m_update_cond.Wait(m_lock);
- }
- m_listeners.remove(listener);
-}
-
-void JournalMetadata::set_minimum_set(uint64_t object_set) {
- Mutex::Locker 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, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- 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) {
- Mutex::Locker 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, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- assert(r == 0);
- comp->release();
-
- m_active_set = object_set;
-}
-
-void JournalMetadata::assert_active_tag(uint64_t tag_tid, Context *on_finish) {
- Mutex::Locker 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;
-
- Mutex::Locker timer_locker(*m_timer_lock);
- Mutex::Locker locker(m_lock);
- if (m_commit_position_ctx == nullptr) {
- return;
- }
-
- cancel_commit_task();
- handle_commit_position_task();
-}
-
-void JournalMetadata::flush_commit_position(Context *on_safe) {
- ldout(m_cct, 20) << __func__ << dendl;
-
- Mutex::Locker timer_locker(*m_timer_lock);
- Mutex::Locker locker(m_lock);
- if (m_commit_position_ctx == nullptr) {
- // nothing to flush
- if (on_safe != nullptr) {
- m_work_queue->queue(on_safe, 0);
- }
- return;
- }
-
- if (on_safe != nullptr) {
- m_commit_position_ctx = new C_FlushCommitPosition(
- m_commit_position_ctx, on_safe);
- }
- cancel_commit_task();
- handle_commit_position_task();
-}
-
-void JournalMetadata::reserve_entry_tid(uint64_t tag_tid, uint64_t entry_tid) {
- Mutex::Locker 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 {
- Mutex::Locker locker(m_lock);
-
- AllocatedEntryTids::const_iterator it = m_allocated_entry_tids.find(tag_tid);
- if (it == m_allocated_entry_tids.end()) {
- return false;
- }
-
- 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;
- C_Refresh *refresh = new C_Refresh(this, on_complete);
- 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;
- if (r == 0) {
- Mutex::Locker locker(m_lock);
-
- 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 = MAX(m_minimum_set, refresh->minimum_set);
- m_active_set = 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.Signal();
- }
- } else {
- lderr(m_cct) << "failed to locate client: " << m_client_id << dendl;
- r = -ENOENT;
- }
- }
-
- if (refresh->on_finish != NULL) {
- refresh->on_finish->complete(r);
- }
-}
-
-void JournalMetadata::cancel_commit_task() {
- ldout(m_cct, 20) << __func__ << dendl;
-
- assert(m_timer_lock->is_locked());
- assert(m_lock.is_locked());
- assert(m_commit_position_ctx != nullptr);
- 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;
-
- assert(m_timer_lock->is_locked());
- assert(m_lock.is_locked());
- assert(m_commit_position_ctx != nullptr);
- if (m_commit_position_task_ctx == NULL) {
- m_commit_position_task_ctx =
- m_timer->add_event_after(m_settings.commit_interval,
- new C_CommitPositionTask(this));
- }
-}
-
-void JournalMetadata::handle_commit_position_task() {
- assert(m_timer_lock->is_locked());
- assert(m_lock.is_locked());
- ldout(m_cct, 20) << __func__ << ": "
- << "client_id=" << m_client_id << ", "
- << "commit_position=" << m_commit_position << dendl;
-
- librados::ObjectWriteOperation op;
- client::client_commit(&op, m_client_id, m_commit_position);
-
- Context *ctx = new C_NotifyUpdate(this, m_commit_position_ctx);
- m_commit_position_ctx = NULL;
-
- ctx = schedule_laggy_clients_disconnect(ctx);
-
- librados::AioCompletion *comp =
- librados::Rados::aio_create_completion(ctx, NULL,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- assert(r == 0);
- comp->release();
-
- m_commit_position_task_ctx = NULL;
-}
-
-void JournalMetadata::schedule_watch_reset() {
- assert(m_timer_lock->is_locked());
- m_timer->add_event_after(1, new C_WatchReset(this));
-}
-
-void JournalMetadata::handle_watch_reset() {
- assert(m_timer_lock->is_locked());
- 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 == -EBLACKLISTED) {
- ldout(m_cct, 5) << __func__ << ": client blacklisted" << 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);
-
- 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 == -EBLACKLISTED) {
- lderr(m_cct) << "journal watch error: client blacklisted" << dendl;
- } else {
- lderr(m_cct) << "journal watch error: " << cpp_strerror(err) << dendl;
- }
-
- Mutex::Locker timer_locker(*m_timer_lock);
- Mutex::Locker locker(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) {
- Mutex::Locker 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) {
- Mutex::Locker locker(m_lock);
-
- auto it = m_pending_commit_tids.find(commit_tid);
- assert(it != m_pending_commit_tids.end());
- 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) {
- Mutex::Locker locker(m_lock);
-
- auto it = m_pending_commit_tids.find(commit_tid);
- 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;
- {
- Mutex::Locker timer_locker(*m_timer_lock);
- Mutex::Locker locker(m_lock);
- 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);
- 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, NULL,
- utils::rados_ctx_callback);
-
- bufferlist bl;
- int r = m_ioctx.aio_notify(m_oid, comp, bl, 5000, NULL);
- 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;
-}
-
-Context *JournalMetadata::schedule_laggy_clients_disconnect(Context *on_finish) {
- assert(m_lock.is_locked());
-
- ldout(m_cct, 20) << __func__ << dendl;
-
- if (m_settings.max_concurrent_object_sets <= 0) {
- return on_finish;
- }
-
- Context *ctx = on_finish;
-
- for (auto &c : m_registered_clients) {
- if (c.state == cls::journal::CLIENT_STATE_DISCONNECTED ||
- c.id == m_client_id ||
- m_settings.whitelisted_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 FunctionContext([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);
-
- librados::AioCompletion *comp =
- librados::Rados::aio_create_completion(ctx, nullptr,
- utils::rados_ctx_callback);
- int r = m_ioctx.aio_operate(m_oid, comp, &op);
- assert(r == 0);
- comp->release();
- });
- }
- }
-
- if (ctx == on_finish) {
- ldout(m_cct, 20) << __func__ << ": no laggy clients to disconnect" << dendl;
- }
-
- return ctx;
-}
-
-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) {
- Mutex::Locker 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