--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "journal/ObjectPlayer.h"
+#include "journal/Utils.h"
+#include "common/Timer.h"
+#include <limits>
+
+#define dout_subsys ceph_subsys_journaler
+#undef dout_prefix
+#define dout_prefix *_dout << "ObjectPlayer: " << this << " "
+
+namespace journal {
+
+ObjectPlayer::ObjectPlayer(librados::IoCtx &ioctx,
+ const std::string &object_oid_prefix,
+ uint64_t object_num, SafeTimer &timer,
+ Mutex &timer_lock, uint8_t order,
+ uint64_t max_fetch_bytes)
+ : RefCountedObject(NULL, 0), m_object_num(object_num),
+ m_oid(utils::get_object_name(object_oid_prefix, m_object_num)),
+ m_cct(NULL), m_timer(timer), m_timer_lock(timer_lock), m_order(order),
+ m_max_fetch_bytes(max_fetch_bytes > 0 ? max_fetch_bytes : 2 << order),
+ m_watch_interval(0), m_watch_task(NULL),
+ m_lock(utils::unique_lock_name("ObjectPlayer::m_lock", this)),
+ m_fetch_in_progress(false) {
+ m_ioctx.dup(ioctx);
+ m_cct = reinterpret_cast<CephContext*>(m_ioctx.cct());
+}
+
+ObjectPlayer::~ObjectPlayer() {
+ {
+ Mutex::Locker timer_locker(m_timer_lock);
+ Mutex::Locker locker(m_lock);
+ assert(!m_fetch_in_progress);
+ assert(m_watch_ctx == nullptr);
+ }
+}
+
+void ObjectPlayer::fetch(Context *on_finish) {
+ ldout(m_cct, 10) << __func__ << ": " << m_oid << dendl;
+
+ Mutex::Locker locker(m_lock);
+ assert(!m_fetch_in_progress);
+ m_fetch_in_progress = true;
+
+ C_Fetch *context = new C_Fetch(this, on_finish);
+ librados::ObjectReadOperation op;
+ op.read(m_read_off, m_max_fetch_bytes, &context->read_bl, NULL);
+ op.set_op_flags2(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
+
+ librados::AioCompletion *rados_completion =
+ librados::Rados::aio_create_completion(context, utils::rados_ctx_callback,
+ NULL);
+ int r = m_ioctx.aio_operate(m_oid, rados_completion, &op, 0, NULL);
+ assert(r == 0);
+ rados_completion->release();
+}
+
+void ObjectPlayer::watch(Context *on_fetch, double interval) {
+ ldout(m_cct, 20) << __func__ << ": " << m_oid << " watch" << dendl;
+
+ Mutex::Locker timer_locker(m_timer_lock);
+ m_watch_interval = interval;
+
+ assert(m_watch_ctx == nullptr);
+ m_watch_ctx = on_fetch;
+
+ schedule_watch();
+}
+
+void ObjectPlayer::unwatch() {
+ ldout(m_cct, 20) << __func__ << ": " << m_oid << " unwatch" << dendl;
+ Context *watch_ctx = nullptr;
+ {
+ Mutex::Locker timer_locker(m_timer_lock);
+ assert(!m_unwatched);
+ m_unwatched = true;
+
+ if (!cancel_watch()) {
+ return;
+ }
+
+ std::swap(watch_ctx, m_watch_ctx);
+ }
+
+ if (watch_ctx != nullptr) {
+ watch_ctx->complete(-ECANCELED);
+ }
+}
+
+void ObjectPlayer::front(Entry *entry) const {
+ Mutex::Locker locker(m_lock);
+ assert(!m_entries.empty());
+ *entry = m_entries.front();
+}
+
+void ObjectPlayer::pop_front() {
+ Mutex::Locker locker(m_lock);
+ assert(!m_entries.empty());
+
+ auto &entry = m_entries.front();
+ m_entry_keys.erase({entry.get_tag_tid(), entry.get_entry_tid()});
+ m_entries.pop_front();
+}
+
+int ObjectPlayer::handle_fetch_complete(int r, const bufferlist &bl,
+ bool *refetch) {
+ ldout(m_cct, 10) << __func__ << ": " << m_oid << ", r=" << r << ", len="
+ << bl.length() << dendl;
+
+ *refetch = false;
+ if (r == -ENOENT) {
+ return 0;
+ } else if (r < 0) {
+ return r;
+ } else if (bl.length() == 0) {
+ return 0;
+ }
+
+ Mutex::Locker locker(m_lock);
+ assert(m_fetch_in_progress);
+ m_read_off += bl.length();
+ m_read_bl.append(bl);
+ m_refetch_state = REFETCH_STATE_REQUIRED;
+
+ bool full_fetch = (m_max_fetch_bytes == 2U << m_order);
+ bool partial_entry = false;
+ bool invalid = false;
+ uint32_t invalid_start_off = 0;
+
+ clear_invalid_range(m_read_bl_off, m_read_bl.length());
+ bufferlist::iterator iter(&m_read_bl, 0);
+ while (!iter.end()) {
+ uint32_t bytes_needed;
+ uint32_t bl_off = iter.get_off();
+ if (!Entry::is_readable(iter, &bytes_needed)) {
+ if (bytes_needed != 0) {
+ invalid_start_off = m_read_bl_off + bl_off;
+ invalid = true;
+ partial_entry = true;
+ if (full_fetch) {
+ lderr(m_cct) << ": partial record at offset " << invalid_start_off
+ << dendl;
+ } else {
+ ldout(m_cct, 20) << ": partial record detected, will re-fetch"
+ << dendl;
+ }
+ break;
+ }
+
+ if (!invalid) {
+ invalid_start_off = m_read_bl_off + bl_off;
+ invalid = true;
+ lderr(m_cct) << ": detected corrupt journal entry at offset "
+ << invalid_start_off << dendl;
+ }
+ ++iter;
+ continue;
+ }
+
+ Entry entry;
+ ::decode(entry, iter);
+ ldout(m_cct, 20) << ": " << entry << " decoded" << dendl;
+
+ uint32_t entry_len = iter.get_off() - bl_off;
+ if (invalid) {
+ // new corrupt region detected
+ uint32_t invalid_end_off = m_read_bl_off + bl_off;
+ lderr(m_cct) << ": corruption range [" << invalid_start_off
+ << ", " << invalid_end_off << ")" << dendl;
+ m_invalid_ranges.insert(invalid_start_off,
+ invalid_end_off - invalid_start_off);
+ invalid = false;
+ }
+
+ EntryKey entry_key(std::make_pair(entry.get_tag_tid(),
+ entry.get_entry_tid()));
+ if (m_entry_keys.find(entry_key) == m_entry_keys.end()) {
+ m_entry_keys[entry_key] = m_entries.insert(m_entries.end(), entry);
+ } else {
+ ldout(m_cct, 10) << ": " << entry << " is duplicate, replacing" << dendl;
+ *m_entry_keys[entry_key] = entry;
+ }
+
+ // prune decoded / corrupted journal entries from front of bl
+ bufferlist sub_bl;
+ sub_bl.substr_of(m_read_bl, iter.get_off(),
+ m_read_bl.length() - iter.get_off());
+ sub_bl.swap(m_read_bl);
+ iter = bufferlist::iterator(&m_read_bl, 0);
+
+ // advance the decoded entry offset
+ m_read_bl_off += entry_len;
+ }
+
+ if (invalid) {
+ uint32_t invalid_end_off = m_read_bl_off + m_read_bl.length();
+ if (!partial_entry) {
+ lderr(m_cct) << ": corruption range [" << invalid_start_off
+ << ", " << invalid_end_off << ")" << dendl;
+ }
+ m_invalid_ranges.insert(invalid_start_off,
+ invalid_end_off - invalid_start_off);
+ }
+
+ if (!m_invalid_ranges.empty() && !partial_entry) {
+ return -EBADMSG;
+ } else if (partial_entry && (full_fetch || m_entries.empty())) {
+ *refetch = true;
+ return -EAGAIN;
+ }
+
+ return 0;
+}
+
+void ObjectPlayer::clear_invalid_range(uint32_t off, uint32_t len) {
+ // possibly remove previously partial record region
+ InvalidRanges decode_range;
+ decode_range.insert(off, len);
+ InvalidRanges intersect_range;
+ intersect_range.intersection_of(m_invalid_ranges, decode_range);
+ if (!intersect_range.empty()) {
+ ldout(m_cct, 20) << ": clearing invalid range: " << intersect_range
+ << dendl;
+ m_invalid_ranges.subtract(intersect_range);
+ }
+}
+
+void ObjectPlayer::schedule_watch() {
+ assert(m_timer_lock.is_locked());
+ if (m_watch_ctx == NULL) {
+ return;
+ }
+
+ ldout(m_cct, 20) << __func__ << ": " << m_oid << " scheduling watch" << dendl;
+ assert(m_watch_task == nullptr);
+ m_watch_task = m_timer.add_event_after(
+ m_watch_interval,
+ new FunctionContext([this](int) {
+ handle_watch_task();
+ }));
+}
+
+bool ObjectPlayer::cancel_watch() {
+ assert(m_timer_lock.is_locked());
+ ldout(m_cct, 20) << __func__ << ": " << m_oid << " cancelling watch" << dendl;
+ if (m_watch_task != nullptr) {
+ bool canceled = m_timer.cancel_event(m_watch_task);
+ assert(canceled);
+
+ m_watch_task = nullptr;
+ return true;
+ }
+ return false;
+}
+
+void ObjectPlayer::handle_watch_task() {
+ assert(m_timer_lock.is_locked());
+
+ ldout(m_cct, 10) << __func__ << ": " << m_oid << " polling" << dendl;
+ assert(m_watch_ctx != nullptr);
+ assert(m_watch_task != nullptr);
+
+ m_watch_task = nullptr;
+ fetch(new C_WatchFetch(this));
+}
+
+void ObjectPlayer::handle_watch_fetched(int r) {
+ ldout(m_cct, 10) << __func__ << ": " << m_oid << " poll complete, r=" << r
+ << dendl;
+
+ Context *watch_ctx = nullptr;
+ {
+ Mutex::Locker timer_locker(m_timer_lock);
+ std::swap(watch_ctx, m_watch_ctx);
+
+ if (m_unwatched) {
+ m_unwatched = false;
+ r = -ECANCELED;
+ }
+ }
+
+ if (watch_ctx != nullptr) {
+ watch_ctx->complete(r);
+ }
+}
+
+void ObjectPlayer::C_Fetch::finish(int r) {
+ bool refetch = false;
+ r = object_player->handle_fetch_complete(r, read_bl, &refetch);
+
+ {
+ Mutex::Locker locker(object_player->m_lock);
+ object_player->m_fetch_in_progress = false;
+ }
+
+ if (refetch) {
+ object_player->fetch(on_finish);
+ return;
+ }
+
+ object_player.reset();
+ on_finish->complete(r);
+}
+
+void ObjectPlayer::C_WatchFetch::finish(int r) {
+ object_player->handle_watch_fetched(r);
+}
+
+} // namespace journal
Code Review / stor4nfv.git / blobdiff