PGLog: add rollback_info_trimmed_to_riter to IndexedLog and use in PG

Signed-off-by: Samuel Just <sam.just@inktank.com>

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 2c86f3ba2d2511875be56256b2f2dedbe98f2da2..d729de3759e5e3e8a54072e4219069a55f7bf42e 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -6577,6 +6577,7 @@ boost::statechart::result PG::RecoveryState::Stray::react(const MLogRec& logevt)
   MOSDPGLog *msg = logevt.msg.get();
   dout(10) << "got info+log from osd." << logevt.from << " " << msg->info << " " << msg->log << dendl;
 
+  ObjectStore::Transaction* t = context<RecoveryMachine>().get_cur_transaction();
   if (msg->info.last_backfill == hobject_t()) {
     // restart backfill
     pg->unreg_next_scrub();
@@ -6584,10 +6585,13 @@ boost::statechart::result PG::RecoveryState::Stray::react(const MLogRec& logevt)
     pg->reg_next_scrub();
     pg->dirty_info = true;
     pg->dirty_big_info = true;  // maybe.
-    pg->pg_log.claim_log(msg->log);
+
+    PGLogEntryHandler rollbacker;
+    pg->pg_log.claim_log_and_clear_rollback_info(msg->log, &rollbacker);
+    rollbacker.apply(pg, t);
+
     pg->pg_log.reset_backfill();
   } else {
-    ObjectStore::Transaction* t = context<RecoveryMachine>().get_cur_transaction();
     pg->merge_log(*t, msg->info, msg->log, logevt.from);
   }
 

diff --git a/src/osd/PGLog.cc b/src/osd/PGLog.cc
index ecf54a8e6ba6ebf808d0638c25d40bd61ef5e47a..36b26e73adf5b8ed3fd64413a127ce427c88200f 100644 (file)
--- a/src/osd/PGLog.cc
+++ b/src/osd/PGLog.cc
@@ -24,6 +24,25 @@
 
 //////////////////// PGLog::IndexedLog ////////////////////
 
+void PGLog::IndexedLog::advance_rollback_info_trimmed_to(
+  eversion_t to,
+  LogEntryHandler *h)
+{
+  assert(to <= can_rollback_to);
+
+  if (to > rollback_info_trimmed_to)
+    rollback_info_trimmed_to = to;
+
+  while (rollback_info_trimmed_to_riter != log.rbegin()) {
+    --rollback_info_trimmed_to_riter;
+    if (rollback_info_trimmed_to_riter->version > rollback_info_trimmed_to) {
+      ++rollback_info_trimmed_to_riter;
+      break;
+    }
+    h->trim(*rollback_info_trimmed_to_riter);
+  }
+}
+
 void PGLog::IndexedLog::split_into(
   pg_t child_pgid,
   unsigned split_bits,
@@ -47,9 +66,11 @@ void PGLog::IndexedLog::split_into(
     oldlog.erase(i++);
   }
 
+
+  olog->can_rollback_to = can_rollback_to;
+
   olog->index();
   index();
-  olog->can_rollback_to = can_rollback_to;
 }
 
 void PGLog::IndexedLog::trim(
@@ -64,6 +85,10 @@ void PGLog::IndexedLog::trim(
                    << " on " << *this << dendl;
   }
 
+  if (s > can_rollback_to)
+    can_rollback_to = s;
+  advance_rollback_info_trimmed_to(s, handler);
+
   while (!log.empty()) {
     pg_log_entry_t &e = *log.begin();
     if (e.version > s)
@@ -71,9 +96,15 @@ void PGLog::IndexedLog::trim(
     generic_dout(20) << "trim " << e << dendl;
     if (trimmed)
       trimmed->insert(e.version);
-    handler->trim(e);
+
     unindex(e);         // remove from index,
-    log.pop_front();    // from log
+
+    if (e.version == rollback_info_trimmed_to_riter->version) {
+      log.pop_front();
+      rollback_info_trimmed_to_riter = log.rend();
+    } else {
+      log.pop_front();
+    }
   }
 
   // raise tail?

diff --git a/src/osd/PGLog.h b/src/osd/PGLog.h
index 1ae6af40eaa62d2183d2c66e738dc53be1fef4cc..c46f63bfa39f653479f3c5c939667878a28723a0 100644 (file)
--- a/src/osd/PGLog.h
+++ b/src/osd/PGLog.h
@@ -62,13 +62,33 @@ struct PGLog {
     list<pg_log_entry_t>::iterator complete_to;  // not inclusive of referenced item
     version_t last_requested;           // last object requested by primary
 
+    //
+  private:
+    /**
+     * rollback_info_trimmed_to_riter points to the first log entry <=
+     * rollback_info_trimmed_to
+     *
+     * It's a reverse_iterator because rend() is a natural representation for
+     * tail, and rbegin() works nicely for head.
+     */
+    list<pg_log_entry_t>::reverse_iterator rollback_info_trimmed_to_riter;
+  public:
+    void advance_rollback_info_trimmed_to(eversion_t to, LogEntryHandler *h);
+
     /****/
     IndexedLog() :
       complete_to(log.end()),
-      last_requested(0) {}
+      last_requested(0),
+      rollback_info_trimmed_to_riter(log.rbegin())
+      {}
+
+    void claim_log_and_clear_rollback_info(const pg_log_t& o) {
+      // we must have already trimmed the old entries
+      assert(rollback_info_trimmed_to == head);
+      assert(rollback_info_trimmed_to_riter == log.rbegin());
 
-    void claim_log(const pg_log_t& o) {
       log = o.log;
+      rollback_info_trimmed_to = head;
       head = o.head;
       tail = o.tail;
       index();
@@ -80,8 +100,13 @@ struct PGLog {
       IndexedLog *olog);
 
     void zero() {
+      // we must have already trimmed the old entries
+      assert(rollback_info_trimmed_to == head);
+      assert(rollback_info_trimmed_to_riter == log.rbegin());
+
       unindex();
       pg_log_t::clear();
+      rollback_info_trimmed_to_riter = log.rbegin();
       reset_recovery_pointers();
     }
     void reset_recovery_pointers() {
@@ -114,6 +139,11 @@ struct PGLog {
          caller_ops[i->reqid] = &(*i);
        }
       }
+
+      rollback_info_trimmed_to_riter = log.rbegin();
+      while (rollback_info_trimmed_to_riter != log.rend() &&
+            rollback_info_trimmed_to_riter->version > rollback_info_trimmed_to)
+       rollback_info_trimmed_to_riter++;
     }
 
     void index(pg_log_entry_t& e) {
@@ -143,6 +173,11 @@ struct PGLog {
     void add(pg_log_entry_t& e) {
       // add to log
       log.push_back(e);
+
+      // riter previously pointed to the previous entry
+      if (rollback_info_trimmed_to_riter == log.rbegin())
+       ++rollback_info_trimmed_to_riter;
+
       assert(e.version > head);
       assert(head.version == 0 || e.version.version > head.version);
       head = e.version;
@@ -333,8 +368,10 @@ public:
 
   //////////////////// get or set log & missing ////////////////////
 
-  void claim_log(const pg_log_t &o) {
-    log.claim_log(o);
+  void claim_log_and_clear_rollback_info(const pg_log_t &o, LogEntryHandler *h) {
+    log.can_rollback_to = log.head;
+    log.advance_rollback_info_trimmed_to(log.head, h);
+    log.claim_log_and_clear_rollback_info(o);
     missing.clear();
     mark_dirty_to(eversion_t::max());
   }