+++ /dev/null
-// -*- 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) 2004-2006 Sage Weil <sage@newdream.net>
- *
- * 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 "MDSRank.h"
-#include "MDLog.h"
-#include "MDCache.h"
-#include "LogEvent.h"
-#include "MDSContext.h"
-
-#include "osdc/Journaler.h"
-#include "mds/JournalPointer.h"
-
-#include "common/entity_name.h"
-#include "common/perf_counters.h"
-#include "common/Cond.h"
-
-#include "events/ESubtreeMap.h"
-
-#include "common/config.h"
-#include "common/errno.h"
-#include "include/assert.h"
-
-#define dout_context g_ceph_context
-#define dout_subsys ceph_subsys_mds
-#undef dout_prefix
-#define dout_prefix *_dout << "mds." << mds->get_nodeid() << ".log "
-
-// cons/des
-MDLog::~MDLog()
-{
- if (journaler) { delete journaler; journaler = 0; }
- if (logger) {
- g_ceph_context->get_perfcounters_collection()->remove(logger);
- delete logger;
- logger = 0;
- }
-}
-
-
-void MDLog::create_logger()
-{
- PerfCountersBuilder plb(g_ceph_context, "mds_log", l_mdl_first, l_mdl_last);
-
- plb.add_u64_counter(l_mdl_evadd, "evadd",
- "Events submitted", "subm", PerfCountersBuilder::PRIO_INTERESTING);
- plb.add_u64_counter(l_mdl_evex, "evex", "Total expired events");
- plb.add_u64_counter(l_mdl_evtrm, "evtrm", "Trimmed events");
- plb.add_u64(l_mdl_ev, "ev",
- "Events", "evts", PerfCountersBuilder::PRIO_INTERESTING);
- plb.add_u64(l_mdl_evexg, "evexg", "Expiring events");
- plb.add_u64(l_mdl_evexd, "evexd", "Current expired events");
-
- plb.add_u64_counter(l_mdl_segadd, "segadd", "Segments added");
- plb.add_u64_counter(l_mdl_segex, "segex", "Total expired segments");
- plb.add_u64_counter(l_mdl_segtrm, "segtrm", "Trimmed segments");
- plb.add_u64(l_mdl_seg, "seg",
- "Segments", "segs", PerfCountersBuilder::PRIO_INTERESTING);
- plb.add_u64(l_mdl_segexg, "segexg", "Expiring segments");
- plb.add_u64(l_mdl_segexd, "segexd", "Current expired segments");
-
- plb.add_u64(l_mdl_expos, "expos", "Journaler xpire position");
- plb.add_u64(l_mdl_wrpos, "wrpos", "Journaler write position");
- plb.add_u64(l_mdl_rdpos, "rdpos", "Journaler read position");
- plb.add_time_avg(l_mdl_jlat, "jlat", "Journaler flush latency");
-
- plb.add_u64_counter(l_mdl_replayed, "replayed", "Events replayed");
-
- // logger
- logger = plb.create_perf_counters();
- g_ceph_context->get_perfcounters_collection()->add(logger);
-}
-
-void MDLog::set_write_iohint(unsigned iohint_flags)
-{
- journaler->set_write_iohint(iohint_flags);
-}
-
-class C_MDL_WriteError : public MDSIOContextBase {
- protected:
- MDLog *mdlog;
- MDSRank *get_mds() override {return mdlog->mds;}
-
- void finish(int r) override {
- MDSRank *mds = get_mds();
- // assume journal is reliable, so don't choose action based on
- // g_conf->mds_action_on_write_error.
- if (r == -EBLACKLISTED) {
- derr << "we have been blacklisted (fenced), respawning..." << dendl;
- mds->respawn();
- } else {
- derr << "unhandled error " << cpp_strerror(r) << ", shutting down..." << dendl;
- // Although it's possible that this could be something transient,
- // it's severe and scary, so disable this rank until an administrator
- // intervenes.
- mds->clog->error() << "Unhandled journal write error on MDS rank " <<
- mds->get_nodeid() << ": " << cpp_strerror(r) << ", shutting down.";
- mds->damaged();
- ceph_abort(); // damaged should never return
- }
- }
-
- public:
- explicit C_MDL_WriteError(MDLog *m) : mdlog(m) {}
-};
-
-
-void MDLog::write_head(MDSInternalContextBase *c)
-{
- Context *fin = NULL;
- if (c != NULL) {
- fin = new C_IO_Wrapper(mds, c);
- }
- journaler->write_head(fin);
-}
-
-uint64_t MDLog::get_read_pos() const
-{
- return journaler->get_read_pos();
-}
-
-uint64_t MDLog::get_write_pos() const
-{
- return journaler->get_write_pos();
-}
-
-uint64_t MDLog::get_safe_pos() const
-{
- return journaler->get_write_safe_pos();
-}
-
-
-
-void MDLog::create(MDSInternalContextBase *c)
-{
- dout(5) << "create empty log" << dendl;
-
- C_GatherBuilder gather(g_ceph_context);
- // This requires an OnFinisher wrapper because Journaler will call back the completion for write_head inside its own lock
- // XXX but should maybe that be handled inside Journaler?
- gather.set_finisher(new C_IO_Wrapper(mds, c));
-
- // The inode of the default Journaler we will create
- ino = MDS_INO_LOG_OFFSET + mds->get_nodeid();
-
- // Instantiate Journaler and start async write to RADOS
- assert(journaler == NULL);
- journaler = new Journaler("mdlog", ino, mds->mdsmap->get_metadata_pool(),
- CEPH_FS_ONDISK_MAGIC, mds->objecter, logger,
- l_mdl_jlat, mds->finisher);
- assert(journaler->is_readonly());
- journaler->set_write_error_handler(new C_MDL_WriteError(this));
- journaler->set_writeable();
- journaler->create(&mds->mdcache->default_log_layout, g_conf->mds_journal_format);
- journaler->write_head(gather.new_sub());
-
- // Async write JournalPointer to RADOS
- JournalPointer jp(mds->get_nodeid(), mds->mdsmap->get_metadata_pool());
- jp.front = ino;
- jp.back = 0;
- jp.save(mds->objecter, gather.new_sub());
-
- gather.activate();
-
- logger->set(l_mdl_expos, journaler->get_expire_pos());
- logger->set(l_mdl_wrpos, journaler->get_write_pos());
-
- submit_thread.create("md_submit");
-}
-
-void MDLog::open(MDSInternalContextBase *c)
-{
- dout(5) << "open discovering log bounds" << dendl;
-
- assert(!recovery_thread.is_started());
- recovery_thread.set_completion(c);
- recovery_thread.create("md_recov_open");
-
- submit_thread.create("md_submit");
- // either append() or replay() will follow.
-}
-
-/**
- * Final part of reopen() procedure, after recovery_thread
- * has done its thing we call append()
- */
-class C_ReopenComplete : public MDSInternalContext {
- MDLog *mdlog;
- MDSInternalContextBase *on_complete;
-public:
- C_ReopenComplete(MDLog *mdlog_, MDSInternalContextBase *on_complete_) : MDSInternalContext(mdlog_->mds), mdlog(mdlog_), on_complete(on_complete_) {}
- void finish(int r) override {
- mdlog->append();
- on_complete->complete(r);
- }
-};
-
-/**
- * Given that open() has been called in the past, go through the journal
- * recovery procedure again, potentially reformatting the journal if it
- * was in an old format.
- */
-void MDLog::reopen(MDSInternalContextBase *c)
-{
- dout(5) << "reopen" << dendl;
-
- // Because we will call append() at the completion of this, check that we have already
- // read the whole journal.
- assert(journaler != NULL);
- assert(journaler->get_read_pos() == journaler->get_write_pos());
-
- delete journaler;
- journaler = NULL;
-
- // recovery_thread was started at some point in the past. Although
- // it has called it's completion if we made it back here, it might
- // still not have been cleaned up: join it.
- recovery_thread.join();
-
- recovery_thread.set_completion(new C_ReopenComplete(this, c));
- recovery_thread.create("md_recov_reopen");
-}
-
-void MDLog::append()
-{
- dout(5) << "append positioning at end and marking writeable" << dendl;
- journaler->set_read_pos(journaler->get_write_pos());
- journaler->set_expire_pos(journaler->get_write_pos());
-
- journaler->set_writeable();
-
- logger->set(l_mdl_expos, journaler->get_write_pos());
-}
-
-
-
-// -------------------------------------------------
-
-void MDLog::_start_entry(LogEvent *e)
-{
- assert(submit_mutex.is_locked_by_me());
-
- assert(cur_event == NULL);
- cur_event = e;
-
- event_seq++;
-
- EMetaBlob *metablob = e->get_metablob();
- if (metablob) {
- metablob->event_seq = event_seq;
- metablob->last_subtree_map = get_last_segment_seq();
- }
-}
-
-void MDLog::cancel_entry(LogEvent *le)
-{
- assert(le == cur_event);
- cur_event = NULL;
- delete le;
-}
-
-void MDLog::_submit_entry(LogEvent *le, MDSLogContextBase *c)
-{
- assert(submit_mutex.is_locked_by_me());
- assert(!mds->is_any_replay());
- assert(!capped);
-
- assert(le == cur_event);
- cur_event = NULL;
-
- // let the event register itself in the segment
- assert(!segments.empty());
- LogSegment *ls = segments.rbegin()->second;
- ls->num_events++;
-
- le->_segment = ls;
- le->update_segment();
- le->set_stamp(ceph_clock_now());
-
- mdsmap_up_features = mds->mdsmap->get_up_features();
- pending_events[ls->seq].push_back(PendingEvent(le, c));
- num_events++;
-
- if (logger) {
- logger->inc(l_mdl_evadd);
- logger->set(l_mdl_ev, num_events);
- }
-
- unflushed++;
-
- uint64_t period = journaler->get_layout_period();
- // start a new segment?
- if (le->get_type() == EVENT_SUBTREEMAP ||
- (le->get_type() == EVENT_IMPORTFINISH && mds->is_resolve())) {
- // avoid infinite loop when ESubtreeMap is very large.
- // do not insert ESubtreeMap among EImportFinish events that finish
- // disambiguate imports. Because the ESubtreeMap reflects the subtree
- // state when all EImportFinish events are replayed.
- } else if (ls->end/period != ls->offset/period ||
- ls->num_events >= g_conf->mds_log_events_per_segment) {
- dout(10) << "submit_entry also starting new segment: last = "
- << ls->seq << "/" << ls->offset << ", event seq = " << event_seq << dendl;
- _start_new_segment();
- } else if (g_conf->mds_debug_subtrees &&
- le->get_type() != EVENT_SUBTREEMAP_TEST) {
- // debug: journal this every time to catch subtree replay bugs.
- // use a different event id so it doesn't get interpreted as a
- // LogSegment boundary on replay.
- LogEvent *sle = mds->mdcache->create_subtree_map();
- sle->set_type(EVENT_SUBTREEMAP_TEST);
- _submit_entry(sle, NULL);
- }
-}
-
-/**
- * Invoked on the flush after each entry submitted
- */
-class C_MDL_Flushed : public MDSLogContextBase {
-protected:
- MDLog *mdlog;
- MDSRank *get_mds() override {return mdlog->mds;}
- MDSInternalContextBase *wrapped;
-
- void finish(int r) override {
- if (wrapped)
- wrapped->complete(r);
- }
-
-public:
- C_MDL_Flushed(MDLog *m, MDSInternalContextBase *w)
- : mdlog(m), wrapped(w) {}
- C_MDL_Flushed(MDLog *m, uint64_t wp) : mdlog(m), wrapped(NULL) {
- set_write_pos(wp);
- }
-};
-
-void MDLog::_submit_thread()
-{
- dout(10) << "_submit_thread start" << dendl;
-
- submit_mutex.Lock();
-
- while (!mds->is_daemon_stopping()) {
- if (g_conf->mds_log_pause) {
- submit_cond.Wait(submit_mutex);
- continue;
- }
-
- map<uint64_t,list<PendingEvent> >::iterator it = pending_events.begin();
- if (it == pending_events.end()) {
- submit_cond.Wait(submit_mutex);
- continue;
- }
-
- if (it->second.empty()) {
- pending_events.erase(it);
- continue;
- }
-
- int64_t features = mdsmap_up_features;
- PendingEvent data = it->second.front();
- it->second.pop_front();
-
- submit_mutex.Unlock();
-
- if (data.le) {
- LogEvent *le = data.le;
- LogSegment *ls = le->_segment;
- // encode it, with event type
- bufferlist bl;
- le->encode_with_header(bl, features);
-
- uint64_t write_pos = journaler->get_write_pos();
-
- le->set_start_off(write_pos);
- if (le->get_type() == EVENT_SUBTREEMAP)
- ls->offset = write_pos;
-
- dout(5) << "_submit_thread " << write_pos << "~" << bl.length()
- << " : " << *le << dendl;
-
- // journal it.
- const uint64_t new_write_pos = journaler->append_entry(bl); // bl is destroyed.
- ls->end = new_write_pos;
-
- MDSLogContextBase *fin;
- if (data.fin) {
- fin = dynamic_cast<MDSLogContextBase*>(data.fin);
- assert(fin);
- fin->set_write_pos(new_write_pos);
- } else {
- fin = new C_MDL_Flushed(this, new_write_pos);
- }
-
- journaler->wait_for_flush(fin);
-
- if (data.flush)
- journaler->flush();
-
- if (logger)
- logger->set(l_mdl_wrpos, ls->end);
-
- delete le;
- } else {
- if (data.fin) {
- MDSInternalContextBase* fin =
- dynamic_cast<MDSInternalContextBase*>(data.fin);
- assert(fin);
- C_MDL_Flushed *fin2 = new C_MDL_Flushed(this, fin);
- fin2->set_write_pos(journaler->get_write_pos());
- journaler->wait_for_flush(fin2);
- }
- if (data.flush)
- journaler->flush();
- }
-
- submit_mutex.Lock();
- if (data.flush)
- unflushed = 0;
- else if (data.le)
- unflushed++;
- }
-
- submit_mutex.Unlock();
-}
-
-void MDLog::wait_for_safe(MDSInternalContextBase *c)
-{
- submit_mutex.Lock();
-
- bool no_pending = true;
- if (!pending_events.empty()) {
- pending_events.rbegin()->second.push_back(PendingEvent(NULL, c));
- no_pending = false;
- submit_cond.Signal();
- }
-
- submit_mutex.Unlock();
-
- if (no_pending && c)
- journaler->wait_for_flush(new C_IO_Wrapper(mds, c));
-}
-
-void MDLog::flush()
-{
- submit_mutex.Lock();
-
- bool do_flush = unflushed > 0;
- unflushed = 0;
- if (!pending_events.empty()) {
- pending_events.rbegin()->second.push_back(PendingEvent(NULL, NULL, true));
- do_flush = false;
- submit_cond.Signal();
- }
-
- submit_mutex.Unlock();
-
- if (do_flush)
- journaler->flush();
-}
-
-void MDLog::kick_submitter()
-{
- Mutex::Locker l(submit_mutex);
- submit_cond.Signal();
-}
-
-void MDLog::cap()
-{
- dout(5) << "cap" << dendl;
- capped = true;
-}
-
-void MDLog::shutdown()
-{
- assert(mds->mds_lock.is_locked_by_me());
-
- dout(5) << "shutdown" << dendl;
- if (submit_thread.is_started()) {
- assert(mds->is_daemon_stopping());
-
- if (submit_thread.am_self()) {
- // Called suicide from the thread: trust it to do no work after
- // returning from suicide, and subsequently respect mds->is_daemon_stopping()
- // and fall out of its loop.
- } else {
- mds->mds_lock.Unlock();
- // Because MDS::stopping is true, it's safe to drop mds_lock: nobody else
- // picking it up will do anything with it.
-
- submit_mutex.Lock();
- submit_cond.Signal();
- submit_mutex.Unlock();
-
- mds->mds_lock.Lock();
-
- submit_thread.join();
- }
- }
-
- // Replay thread can be stuck inside e.g. Journaler::wait_for_readable,
- // so we need to shutdown the journaler first.
- if (journaler) {
- journaler->shutdown();
- }
-
- if (replay_thread.is_started() && !replay_thread.am_self()) {
- mds->mds_lock.Unlock();
- replay_thread.join();
- mds->mds_lock.Lock();
- }
-
- if (recovery_thread.is_started() && !recovery_thread.am_self()) {
- mds->mds_lock.Unlock();
- recovery_thread.join();
- mds->mds_lock.Lock();
- }
-}
-
-
-// -----------------------------
-// segments
-
-void MDLog::_start_new_segment()
-{
- _prepare_new_segment();
- _journal_segment_subtree_map(NULL);
-}
-
-void MDLog::_prepare_new_segment()
-{
- assert(submit_mutex.is_locked_by_me());
-
- uint64_t seq = event_seq + 1;
- dout(7) << __func__ << " seq " << seq << dendl;
-
- segments[seq] = new LogSegment(seq);
-
- logger->inc(l_mdl_segadd);
- logger->set(l_mdl_seg, segments.size());
-
- // Adjust to next stray dir
- dout(10) << "Advancing to next stray directory on mds " << mds->get_nodeid()
- << dendl;
- mds->mdcache->advance_stray();
-}
-
-void MDLog::_journal_segment_subtree_map(MDSInternalContextBase *onsync)
-{
- assert(submit_mutex.is_locked_by_me());
-
- dout(7) << __func__ << dendl;
- ESubtreeMap *sle = mds->mdcache->create_subtree_map();
- sle->event_seq = get_last_segment_seq();
-
- _submit_entry(sle, new C_MDL_Flushed(this, onsync));
-}
-
-void MDLog::trim(int m)
-{
- unsigned max_segments = g_conf->mds_log_max_segments;
- int max_events = g_conf->mds_log_max_events;
- if (m >= 0)
- max_events = m;
-
- if (mds->mdcache->is_readonly()) {
- dout(10) << "trim, ignoring read-only FS" << dendl;
- return;
- }
-
- // Clamp max_events to not be smaller than events per segment
- if (max_events > 0 && max_events <= g_conf->mds_log_events_per_segment) {
- max_events = g_conf->mds_log_events_per_segment + 1;
- }
-
- submit_mutex.Lock();
-
- // trim!
- dout(10) << "trim "
- << segments.size() << " / " << max_segments << " segments, "
- << num_events << " / " << max_events << " events"
- << ", " << expiring_segments.size() << " (" << expiring_events << ") expiring"
- << ", " << expired_segments.size() << " (" << expired_events << ") expired"
- << dendl;
-
- if (segments.empty()) {
- submit_mutex.Unlock();
- return;
- }
-
- // hack: only trim for a few seconds at a time
- utime_t stop = ceph_clock_now();
- stop += 2.0;
-
- map<uint64_t,LogSegment*>::iterator p = segments.begin();
- while (p != segments.end() &&
- ((max_events >= 0 &&
- num_events - expiring_events - expired_events > max_events) ||
- (segments.size() - expiring_segments.size() - expired_segments.size() > max_segments))) {
-
- if (stop < ceph_clock_now())
- break;
-
- int num_expiring_segments = (int)expiring_segments.size();
- if (num_expiring_segments >= g_conf->mds_log_max_expiring)
- break;
-
- int op_prio = CEPH_MSG_PRIO_LOW +
- (CEPH_MSG_PRIO_HIGH - CEPH_MSG_PRIO_LOW) *
- num_expiring_segments / g_conf->mds_log_max_expiring;
-
- // look at first segment
- LogSegment *ls = p->second;
- assert(ls);
- ++p;
-
- if (pending_events.count(ls->seq) ||
- ls->end > safe_pos) {
- dout(5) << "trim segment " << ls->seq << "/" << ls->offset << ", not fully flushed yet, safe "
- << journaler->get_write_safe_pos() << " < end " << ls->end << dendl;
- break;
- }
- if (expiring_segments.count(ls)) {
- dout(5) << "trim already expiring segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
- } else if (expired_segments.count(ls)) {
- dout(5) << "trim already expired segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
- } else {
- assert(expiring_segments.count(ls) == 0);
- expiring_segments.insert(ls);
- expiring_events += ls->num_events;
- submit_mutex.Unlock();
-
- uint64_t last_seq = ls->seq;
- try_expire(ls, op_prio);
-
- submit_mutex.Lock();
- p = segments.lower_bound(last_seq + 1);
- }
- }
-
- // discard expired segments and unlock submit_mutex
- _trim_expired_segments();
-}
-
-class C_MaybeExpiredSegment : public MDSInternalContext {
- MDLog *mdlog;
- LogSegment *ls;
- int op_prio;
- public:
- C_MaybeExpiredSegment(MDLog *mdl, LogSegment *s, int p) :
- MDSInternalContext(mdl->mds), mdlog(mdl), ls(s), op_prio(p) {}
- void finish(int res) override {
- if (res < 0)
- mdlog->mds->handle_write_error(res);
- mdlog->_maybe_expired(ls, op_prio);
- }
-};
-
-/**
- * Like MDLog::trim, but instead of trimming to max_segments, trim all but the latest
- * segment.
- */
-int MDLog::trim_all()
-{
- submit_mutex.Lock();
-
- dout(10) << __func__ << ": "
- << segments.size()
- << "/" << expiring_segments.size()
- << "/" << expired_segments.size() << dendl;
-
- uint64_t last_seq = 0;
- if (!segments.empty())
- last_seq = get_last_segment_seq();
-
- map<uint64_t,LogSegment*>::iterator p = segments.begin();
- while (p != segments.end() &&
- p->first < last_seq && p->second->end <= safe_pos) {
- LogSegment *ls = p->second;
- ++p;
-
- // Caller should have flushed journaler before calling this
- if (pending_events.count(ls->seq)) {
- dout(5) << __func__ << ": segment " << ls->seq << " has pending events" << dendl;
- submit_mutex.Unlock();
- return -EAGAIN;
- }
-
- if (expiring_segments.count(ls)) {
- dout(5) << "trim already expiring segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
- } else if (expired_segments.count(ls)) {
- dout(5) << "trim already expired segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
- } else {
- assert(expiring_segments.count(ls) == 0);
- expiring_segments.insert(ls);
- expiring_events += ls->num_events;
- submit_mutex.Unlock();
-
- uint64_t next_seq = ls->seq + 1;
- try_expire(ls, CEPH_MSG_PRIO_DEFAULT);
-
- submit_mutex.Lock();
- p = segments.lower_bound(next_seq);
- }
- }
-
- _trim_expired_segments();
-
- return 0;
-}
-
-
-void MDLog::try_expire(LogSegment *ls, int op_prio)
-{
- MDSGatherBuilder gather_bld(g_ceph_context);
- ls->try_to_expire(mds, gather_bld, op_prio);
-
- if (gather_bld.has_subs()) {
- dout(5) << "try_expire expiring segment " << ls->seq << "/" << ls->offset << dendl;
- gather_bld.set_finisher(new C_MaybeExpiredSegment(this, ls, op_prio));
- gather_bld.activate();
- } else {
- dout(10) << "try_expire expired segment " << ls->seq << "/" << ls->offset << dendl;
- submit_mutex.Lock();
- assert(expiring_segments.count(ls));
- expiring_segments.erase(ls);
- expiring_events -= ls->num_events;
- _expired(ls);
- submit_mutex.Unlock();
- }
-
- logger->set(l_mdl_segexg, expiring_segments.size());
- logger->set(l_mdl_evexg, expiring_events);
-}
-
-void MDLog::_maybe_expired(LogSegment *ls, int op_prio)
-{
- if (mds->mdcache->is_readonly()) {
- dout(10) << "_maybe_expired, ignoring read-only FS" << dendl;
- return;
- }
-
- dout(10) << "_maybe_expired segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
- try_expire(ls, op_prio);
-}
-
-void MDLog::_trim_expired_segments()
-{
- assert(submit_mutex.is_locked_by_me());
-
- // trim expired segments?
- bool trimmed = false;
- while (!segments.empty()) {
- LogSegment *ls = segments.begin()->second;
- if (!expired_segments.count(ls)) {
- dout(10) << "_trim_expired_segments waiting for " << ls->seq << "/" << ls->offset
- << " to expire" << dendl;
- break;
- }
-
- dout(10) << "_trim_expired_segments trimming expired "
- << ls->seq << "/0x" << std::hex << ls->offset << std::dec << dendl;
- expired_events -= ls->num_events;
- expired_segments.erase(ls);
- num_events -= ls->num_events;
-
- // this was the oldest segment, adjust expire pos
- if (journaler->get_expire_pos() < ls->end) {
- journaler->set_expire_pos(ls->end);
- logger->set(l_mdl_expos, ls->end);
- } else {
- logger->set(l_mdl_expos, ls->offset);
- }
-
- logger->inc(l_mdl_segtrm);
- logger->inc(l_mdl_evtrm, ls->num_events);
-
- segments.erase(ls->seq);
- delete ls;
- trimmed = true;
- }
-
- submit_mutex.Unlock();
-
- if (trimmed)
- journaler->write_head(0);
-}
-
-void MDLog::trim_expired_segments()
-{
- submit_mutex.Lock();
- _trim_expired_segments();
-}
-
-void MDLog::_expired(LogSegment *ls)
-{
- assert(submit_mutex.is_locked_by_me());
-
- dout(5) << "_expired segment " << ls->seq << "/" << ls->offset
- << ", " << ls->num_events << " events" << dendl;
-
- if (!capped && ls == peek_current_segment()) {
- dout(5) << "_expired not expiring " << ls->seq << "/" << ls->offset
- << ", last one and !capped" << dendl;
- } else {
- // expired.
- expired_segments.insert(ls);
- expired_events += ls->num_events;
-
- // Trigger all waiters
- for (std::list<MDSInternalContextBase*>::iterator i = ls->expiry_waiters.begin();
- i != ls->expiry_waiters.end(); ++i) {
- (*i)->complete(0);
- }
- ls->expiry_waiters.clear();
-
- logger->inc(l_mdl_evex, ls->num_events);
- logger->inc(l_mdl_segex);
- }
-
- logger->set(l_mdl_ev, num_events);
- logger->set(l_mdl_evexd, expired_events);
- logger->set(l_mdl_seg, segments.size());
- logger->set(l_mdl_segexd, expired_segments.size());
-}
-
-
-
-void MDLog::replay(MDSInternalContextBase *c)
-{
- assert(journaler->is_active());
- assert(journaler->is_readonly());
-
- // empty?
- if (journaler->get_read_pos() == journaler->get_write_pos()) {
- dout(10) << "replay - journal empty, done." << dendl;
- mds->mdcache->trim();
- if (c) {
- c->complete(0);
- }
- return;
- }
-
- // add waiter
- if (c)
- waitfor_replay.push_back(c);
-
- // go!
- dout(10) << "replay start, from " << journaler->get_read_pos()
- << " to " << journaler->get_write_pos() << dendl;
-
- assert(num_events == 0 || already_replayed);
- if (already_replayed) {
- // Ensure previous instance of ReplayThread is joined before
- // we create another one
- replay_thread.join();
- }
- already_replayed = true;
-
- replay_thread.create("md_log_replay");
-}
-
-
-/**
- * Resolve the JournalPointer object to a journal file, and
- * instantiate a Journaler object. This may re-write the journal
- * if the journal in RADOS appears to be in an old format.
- *
- * This is a separate thread because of the way it is initialized from inside
- * the mds lock, which is also the global objecter lock -- rather than split
- * it up into hard-to-read async operations linked up by contexts,
- *
- * When this function completes, the `journaler` attribute will be set to
- * a Journaler instance using the latest available serialization format.
- */
-void MDLog::_recovery_thread(MDSInternalContextBase *completion)
-{
- assert(journaler == NULL);
- if (g_conf->mds_journal_format > JOURNAL_FORMAT_MAX) {
- dout(0) << "Configuration value for mds_journal_format is out of bounds, max is "
- << JOURNAL_FORMAT_MAX << dendl;
-
- // Oh dear, something unreadable in the store for this rank: require
- // operator intervention.
- mds->damaged();
- ceph_abort(); // damaged should not return
- }
-
- // First, read the pointer object.
- // If the pointer object is not present, then create it with
- // front = default ino and back = null
- JournalPointer jp(mds->get_nodeid(), mds->mdsmap->get_metadata_pool());
- const int read_result = jp.load(mds->objecter);
- if (read_result == -ENOENT) {
- inodeno_t const default_log_ino = MDS_INO_LOG_OFFSET + mds->get_nodeid();
- jp.front = default_log_ino;
- int write_result = jp.save(mds->objecter);
- // Nothing graceful we can do for this
- assert(write_result >= 0);
- } else if (read_result == -EBLACKLISTED) {
- derr << "Blacklisted during JournalPointer read! Respawning..." << dendl;
- mds->respawn();
- ceph_abort(); // Should be unreachable because respawn calls execv
- } else if (read_result != 0) {
- mds->clog->error() << "failed to read JournalPointer: " << read_result
- << " (" << cpp_strerror(read_result) << ")";
- mds->damaged_unlocked();
- ceph_abort(); // Should be unreachable because damaged() calls respawn()
- }
-
- // If the back pointer is non-null, that means that a journal
- // rewrite failed part way through. Erase the back journal
- // to clean up.
- if (jp.back) {
- if (mds->is_standby_replay()) {
- dout(1) << "Journal " << jp.front << " is being rewritten, "
- << "cannot replay in standby until an active MDS completes rewrite" << dendl;
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- return;
- }
- completion->complete(-EAGAIN);
- return;
- }
- dout(1) << "Erasing journal " << jp.back << dendl;
- C_SaferCond erase_waiter;
- Journaler back("mdlog", jp.back, mds->mdsmap->get_metadata_pool(),
- CEPH_FS_ONDISK_MAGIC, mds->objecter, logger, l_mdl_jlat,
- mds->finisher);
-
- // Read all about this journal (header + extents)
- C_SaferCond recover_wait;
- back.recover(&recover_wait);
- int recovery_result = recover_wait.wait();
- if (recovery_result == -EBLACKLISTED) {
- derr << "Blacklisted during journal recovery! Respawning..." << dendl;
- mds->respawn();
- ceph_abort(); // Should be unreachable because respawn calls execv
- } else if (recovery_result != 0) {
- // Journaler.recover succeeds if no journal objects are present: an error
- // means something worse like a corrupt header, which we can't handle here.
- mds->clog->error() << "Error recovering journal " << jp.front << ": "
- << cpp_strerror(recovery_result);
- mds->damaged_unlocked();
- assert(recovery_result == 0); // Unreachable because damaged() calls respawn()
- }
-
- // We could read journal, so we can erase it.
- back.erase(&erase_waiter);
- int erase_result = erase_waiter.wait();
-
- // If we are successful, or find no data, we can update the JournalPointer to
- // reflect that the back journal is gone.
- if (erase_result != 0 && erase_result != -ENOENT) {
- derr << "Failed to erase journal " << jp.back << ": " << cpp_strerror(erase_result) << dendl;
- } else {
- dout(1) << "Successfully erased journal, updating journal pointer" << dendl;
- jp.back = 0;
- int write_result = jp.save(mds->objecter);
- // Nothing graceful we can do for this
- assert(write_result >= 0);
- }
- }
-
- /* Read the header from the front journal */
- Journaler *front_journal = new Journaler("mdlog", jp.front,
- mds->mdsmap->get_metadata_pool(), CEPH_FS_ONDISK_MAGIC, mds->objecter,
- logger, l_mdl_jlat, mds->finisher);
-
- // Assign to ::journaler so that we can be aborted by ::shutdown while
- // waiting for journaler recovery
- {
- Mutex::Locker l(mds->mds_lock);
- journaler = front_journal;
- }
-
- C_SaferCond recover_wait;
- front_journal->recover(&recover_wait);
- dout(4) << "Waiting for journal " << jp.front << " to recover..." << dendl;
- int recovery_result = recover_wait.wait();
- dout(4) << "Journal " << jp.front << " recovered." << dendl;
-
- if (recovery_result == -EBLACKLISTED) {
- derr << "Blacklisted during journal recovery! Respawning..." << dendl;
- mds->respawn();
- ceph_abort(); // Should be unreachable because respawn calls execv
- } else if (recovery_result != 0) {
- mds->clog->error() << "Error recovering journal " << jp.front << ": "
- << cpp_strerror(recovery_result);
- mds->damaged_unlocked();
- assert(recovery_result == 0); // Unreachable because damaged() calls respawn()
- }
-
- /* Check whether the front journal format is acceptable or needs re-write */
- if (front_journal->get_stream_format() > JOURNAL_FORMAT_MAX) {
- dout(0) << "Journal " << jp.front << " is in unknown format " << front_journal->get_stream_format()
- << ", does this MDS daemon require upgrade?" << dendl;
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- journaler = NULL;
- delete front_journal;
- return;
- }
- completion->complete(-EINVAL);
- }
- } else if (mds->is_standby_replay() || front_journal->get_stream_format() >= g_conf->mds_journal_format) {
- /* The journal is of configured format, or we are in standbyreplay and will
- * tolerate replaying old journals until we have to go active. Use front_journal as
- * our journaler attribute and complete */
- dout(4) << "Recovered journal " << jp.front << " in format " << front_journal->get_stream_format() << dendl;
- journaler->set_write_error_handler(new C_MDL_WriteError(this));
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- return;
- }
- completion->complete(0);
- }
- } else {
- /* Hand off to reformat routine, which will ultimately set the
- * completion when it has done its thing */
- dout(1) << "Journal " << jp.front << " has old format "
- << front_journal->get_stream_format() << ", it will now be updated" << dendl;
- _reformat_journal(jp, front_journal, completion);
- }
-}
-
-/**
- * Blocking rewrite of the journal to a new file, followed by
- * swap of journal pointer to point to the new one.
- *
- * We write the new journal to the 'back' journal from the JournalPointer,
- * swapping pointers to make that one the front journal only when we have
- * safely completed.
- */
-void MDLog::_reformat_journal(JournalPointer const &jp_in, Journaler *old_journal, MDSInternalContextBase *completion)
-{
- assert(!jp_in.is_null());
- assert(completion != NULL);
- assert(old_journal != NULL);
-
- JournalPointer jp = jp_in;
-
- /* Set JournalPointer.back to the location we will write the new journal */
- inodeno_t primary_ino = MDS_INO_LOG_OFFSET + mds->get_nodeid();
- inodeno_t secondary_ino = MDS_INO_LOG_BACKUP_OFFSET + mds->get_nodeid();
- jp.back = (jp.front == primary_ino ? secondary_ino : primary_ino);
- int write_result = jp.save(mds->objecter);
- assert(write_result == 0);
-
- /* Create the new Journaler file */
- Journaler *new_journal = new Journaler("mdlog", jp.back,
- mds->mdsmap->get_metadata_pool(), CEPH_FS_ONDISK_MAGIC, mds->objecter, logger, l_mdl_jlat, mds->finisher);
- dout(4) << "Writing new journal header " << jp.back << dendl;
- file_layout_t new_layout = old_journal->get_layout();
- new_journal->set_writeable();
- new_journal->create(&new_layout, g_conf->mds_journal_format);
-
- /* Write the new journal header to RADOS */
- C_SaferCond write_head_wait;
- new_journal->write_head(&write_head_wait);
- write_head_wait.wait();
-
- // Read in the old journal, and whenever we have readable events,
- // write them to the new journal.
- int r = 0;
-
- // In old format journals before event_seq was introduced, the serialized
- // offset of a SubtreeMap message in the log is used as the unique ID for
- // a log segment. Because we change serialization, this will end up changing
- // for us, so we have to explicitly update the fields that point back to that
- // log segment.
- std::map<log_segment_seq_t, log_segment_seq_t> segment_pos_rewrite;
-
- // The logic in here borrowed from replay_thread expects mds_lock to be held,
- // e.g. between checking readable and doing wait_for_readable so that journaler
- // state doesn't change in between.
- uint32_t events_transcribed = 0;
- while (1) {
- while (!old_journal->is_readable() &&
- old_journal->get_read_pos() < old_journal->get_write_pos() &&
- !old_journal->get_error()) {
-
- // Issue a journal prefetch
- C_SaferCond readable_waiter;
- old_journal->wait_for_readable(&readable_waiter);
-
- // Wait for a journal prefetch to complete
- readable_waiter.wait();
- }
- if (old_journal->get_error()) {
- r = old_journal->get_error();
- dout(0) << "_replay journaler got error " << r << ", aborting" << dendl;
- break;
- }
-
- if (!old_journal->is_readable() &&
- old_journal->get_read_pos() == old_journal->get_write_pos())
- break;
-
- // Read one serialized LogEvent
- assert(old_journal->is_readable());
- bufferlist bl;
- uint64_t le_pos = old_journal->get_read_pos();
- bool r = old_journal->try_read_entry(bl);
- if (!r && old_journal->get_error())
- continue;
- assert(r);
-
- // Update segment_pos_rewrite
- LogEvent *le = LogEvent::decode(bl);
- if (le) {
- bool modified = false;
-
- if (le->get_type() == EVENT_SUBTREEMAP ||
- le->get_type() == EVENT_RESETJOURNAL) {
- ESubtreeMap *sle = dynamic_cast<ESubtreeMap*>(le);
- if (sle == NULL || sle->event_seq == 0) {
- // A non-explicit event seq: the effective sequence number
- // of this segment is it's position in the old journal and
- // the new effective sequence number will be its position
- // in the new journal.
- segment_pos_rewrite[le_pos] = new_journal->get_write_pos();
- dout(20) << __func__ << " discovered segment seq mapping "
- << le_pos << " -> " << new_journal->get_write_pos() << dendl;
- }
- } else {
- event_seq++;
- }
-
- // Rewrite segment references if necessary
- EMetaBlob *blob = le->get_metablob();
- if (blob) {
- modified = blob->rewrite_truncate_finish(mds, segment_pos_rewrite);
- }
-
- // Zero-out expire_pos in subtreemap because offsets have changed
- // (expire_pos is just an optimization so it's safe to eliminate it)
- if (le->get_type() == EVENT_SUBTREEMAP
- || le->get_type() == EVENT_SUBTREEMAP_TEST) {
- ESubtreeMap *sle = dynamic_cast<ESubtreeMap*>(le);
- assert(sle != NULL);
- dout(20) << __func__ << " zeroing expire_pos in subtreemap event at "
- << le_pos << " seq=" << sle->event_seq << dendl;
- sle->expire_pos = 0;
- modified = true;
- }
-
- if (modified) {
- bl.clear();
- le->encode_with_header(bl, mds->mdsmap->get_up_features());
- }
-
- delete le;
- } else {
- // Failure from LogEvent::decode, our job is to change the journal wrapper,
- // not validate the contents, so pass it through.
- dout(1) << __func__ << " transcribing un-decodable LogEvent at old position "
- << old_journal->get_read_pos() << ", new position " << new_journal->get_write_pos()
- << dendl;
- }
-
- // Write (buffered, synchronous) one serialized LogEvent
- events_transcribed += 1;
- new_journal->append_entry(bl);
- }
-
- dout(1) << "Transcribed " << events_transcribed << " events, flushing new journal" << dendl;
- C_SaferCond flush_waiter;
- new_journal->flush(&flush_waiter);
- flush_waiter.wait();
-
- // If failed to rewrite journal, leave the part written journal
- // as garbage to be cleaned up next startup.
- assert(r == 0);
-
- /* Now that the new journal is safe, we can flip the pointers */
- inodeno_t const tmp = jp.front;
- jp.front = jp.back;
- jp.back = tmp;
- write_result = jp.save(mds->objecter);
- assert(write_result == 0);
-
- /* Delete the old journal to free space */
- dout(1) << "New journal flushed, erasing old journal" << dendl;
- C_SaferCond erase_waiter;
- old_journal->erase(&erase_waiter);
- int erase_result = erase_waiter.wait();
- assert(erase_result == 0);
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- delete new_journal;
- return;
- }
- assert(journaler == old_journal);
- journaler = NULL;
- delete old_journal;
- }
-
- /* Update the pointer to reflect we're back in clean single journal state. */
- jp.back = 0;
- write_result = jp.save(mds->objecter);
- assert(write_result == 0);
-
- /* Reset the Journaler object to its default state */
- dout(1) << "Journal rewrite complete, continuing with normal startup" << dendl;
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- delete new_journal;
- return;
- }
- journaler = new_journal;
- journaler->set_readonly();
- journaler->set_write_error_handler(new C_MDL_WriteError(this));
- }
-
- /* Trigger completion */
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- return;
- }
- completion->complete(0);
- }
-}
-
-
-// i am a separate thread
-void MDLog::_replay_thread()
-{
- dout(10) << "_replay_thread start" << dendl;
-
- // loop
- int r = 0;
- while (1) {
- // wait for read?
- while (!journaler->is_readable() &&
- journaler->get_read_pos() < journaler->get_write_pos() &&
- !journaler->get_error()) {
- C_SaferCond readable_waiter;
- journaler->wait_for_readable(&readable_waiter);
- r = readable_waiter.wait();
- }
- if (journaler->get_error()) {
- r = journaler->get_error();
- dout(0) << "_replay journaler got error " << r << ", aborting" << dendl;
- if (r == -ENOENT) {
- if (mds->is_standby_replay()) {
- // journal has been trimmed by somebody else
- r = -EAGAIN;
- } else {
- mds->clog->error() << "missing journal object";
- mds->damaged_unlocked();
- ceph_abort(); // Should be unreachable because damaged() calls respawn()
- }
- } else if (r == -EINVAL) {
- if (journaler->get_read_pos() < journaler->get_expire_pos()) {
- // this should only happen if you're following somebody else
- if(journaler->is_readonly()) {
- dout(0) << "expire_pos is higher than read_pos, returning EAGAIN" << dendl;
- r = -EAGAIN;
- } else {
- mds->clog->error() << "invalid journaler offsets";
- mds->damaged_unlocked();
- ceph_abort(); // Should be unreachable because damaged() calls respawn()
- }
- } else {
- /* re-read head and check it
- * Given that replay happens in a separate thread and
- * the MDS is going to either shut down or restart when
- * we return this error, doing it synchronously is fine
- * -- as long as we drop the main mds lock--. */
- C_SaferCond reread_fin;
- journaler->reread_head(&reread_fin);
- int err = reread_fin.wait();
- if (err) {
- if (err == -ENOENT && mds->is_standby_replay()) {
- r = -EAGAIN;
- dout(1) << "Journal header went away while in standby replay, journal rewritten?"
- << dendl;
- break;
- } else {
- dout(0) << "got error while reading head: " << cpp_strerror(err)
- << dendl;
-
- mds->clog->error() << "error reading journal header";
- mds->damaged_unlocked();
- ceph_abort(); // Should be unreachable because damaged() calls
- // respawn()
- }
- }
- standby_trim_segments();
- if (journaler->get_read_pos() < journaler->get_expire_pos()) {
- dout(0) << "expire_pos is higher than read_pos, returning EAGAIN" << dendl;
- r = -EAGAIN;
- }
- }
- }
- break;
- }
-
- if (!journaler->is_readable() &&
- journaler->get_read_pos() == journaler->get_write_pos())
- break;
-
- assert(journaler->is_readable() || mds->is_daemon_stopping());
-
- // read it
- uint64_t pos = journaler->get_read_pos();
- bufferlist bl;
- bool r = journaler->try_read_entry(bl);
- if (!r && journaler->get_error())
- continue;
- assert(r);
-
- // unpack event
- LogEvent *le = LogEvent::decode(bl);
- if (!le) {
- dout(0) << "_replay " << pos << "~" << bl.length() << " / " << journaler->get_write_pos()
- << " -- unable to decode event" << dendl;
- dout(0) << "dump of unknown or corrupt event:\n";
- bl.hexdump(*_dout);
- *_dout << dendl;
-
- mds->clog->error() << "corrupt journal event at " << pos << "~"
- << bl.length() << " / "
- << journaler->get_write_pos();
- if (g_conf->mds_log_skip_corrupt_events) {
- continue;
- } else {
- mds->damaged_unlocked();
- ceph_abort(); // Should be unreachable because damaged() calls
- // respawn()
- }
-
- }
- le->set_start_off(pos);
-
- // new segment?
- if (le->get_type() == EVENT_SUBTREEMAP ||
- le->get_type() == EVENT_RESETJOURNAL) {
- ESubtreeMap *sle = dynamic_cast<ESubtreeMap*>(le);
- if (sle && sle->event_seq > 0)
- event_seq = sle->event_seq;
- else
- event_seq = pos;
- segments[event_seq] = new LogSegment(event_seq, pos);
- logger->set(l_mdl_seg, segments.size());
- } else {
- event_seq++;
- }
-
- // have we seen an import map yet?
- if (segments.empty()) {
- dout(10) << "_replay " << pos << "~" << bl.length() << " / " << journaler->get_write_pos()
- << " " << le->get_stamp() << " -- waiting for subtree_map. (skipping " << *le << ")" << dendl;
- } else {
- dout(10) << "_replay " << pos << "~" << bl.length() << " / " << journaler->get_write_pos()
- << " " << le->get_stamp() << ": " << *le << dendl;
- le->_segment = get_current_segment(); // replay may need this
- le->_segment->num_events++;
- le->_segment->end = journaler->get_read_pos();
- num_events++;
-
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- return;
- }
- logger->inc(l_mdl_replayed);
- le->replay(mds);
- }
- }
- delete le;
-
- logger->set(l_mdl_rdpos, pos);
- }
-
- // done!
- if (r == 0) {
- assert(journaler->get_read_pos() == journaler->get_write_pos());
- dout(10) << "_replay - complete, " << num_events
- << " events" << dendl;
-
- logger->set(l_mdl_expos, journaler->get_expire_pos());
- }
-
- safe_pos = journaler->get_write_safe_pos();
-
- dout(10) << "_replay_thread kicking waiters" << dendl;
- {
- Mutex::Locker l(mds->mds_lock);
- if (mds->is_daemon_stopping()) {
- return;
- }
- finish_contexts(g_ceph_context, waitfor_replay, r);
- }
-
- dout(10) << "_replay_thread finish" << dendl;
-}
-
-void MDLog::standby_trim_segments()
-{
- dout(10) << "standby_trim_segments" << dendl;
- uint64_t expire_pos = journaler->get_expire_pos();
- dout(10) << " expire_pos=" << expire_pos << dendl;
- bool removed_segment = false;
- while (have_any_segments()) {
- LogSegment *seg = get_oldest_segment();
- dout(10) << " segment seq=" << seg->seq << " " << seg->offset <<
- "~" << seg->end - seg->offset << dendl;
-
- if (seg->end > expire_pos) {
- dout(10) << " won't remove, not expired!" << dendl;
- break;
- }
-
- if (segments.size() == 1) {
- dout(10) << " won't remove, last segment!" << dendl;
- break;
- }
-
- dout(10) << " removing segment" << dendl;
- mds->mdcache->standby_trim_segment(seg);
- remove_oldest_segment();
- removed_segment = true;
- }
-
- if (removed_segment) {
- dout(20) << " calling mdcache->trim!" << dendl;
- mds->mdcache->trim();
- } else {
- dout(20) << " removed no segments!" << dendl;
- }
-}
-
-void MDLog::dump_replay_status(Formatter *f) const
-{
- f->open_object_section("replay_status");
- f->dump_unsigned("journal_read_pos", journaler ? journaler->get_read_pos() : 0);
- f->dump_unsigned("journal_write_pos", journaler ? journaler->get_write_pos() : 0);
- f->dump_unsigned("journal_expire_pos", journaler ? journaler->get_expire_pos() : 0);
- f->dump_unsigned("num_events", get_num_events());
- f->dump_unsigned("num_segments", get_num_segments());
- f->close_section();
-}
Code Review / stor4nfv.git / blobdiff