FileJournal: add committed_up_to to header

header_t::committed_up_to provides a lower bound for safetly committed
journal entries.  If read_entry fails prior to committed_up_to, we
know we have a corrupt jorunal entry.  Furthermore, if
journal_write_header_frequency is not 0, we will write out the
journal header once every journal_write_header_frequency
journal writes.

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

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 5d3c64e45f20618c0b5c1eb34808046650e2d4dd..a79c4d19e6758e09e34b042845a7797caa80dd42 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -475,6 +475,7 @@ OPTION(filestore_debug_verify_split, OPT_BOOL, false)
 OPTION(journal_dio, OPT_BOOL, true)
 OPTION(journal_aio, OPT_BOOL, false)
 OPTION(journal_block_align, OPT_BOOL, true)
+OPTION(journal_write_header_frequency, OPT_U64, 0)
 OPTION(journal_max_write_bytes, OPT_INT, 10 << 20)
 OPTION(journal_max_write_entries, OPT_INT, 100)
 OPTION(journal_queue_max_ops, OPT_INT, 300)

diff --git a/src/os/FileJournal.cc b/src/os/FileJournal.cc
index 0056f1f23b447326f46be836eb83931b5cc119dc..ad84b0dbf80d5aeb7def709298571df90fb8ec1e 100644 (file)
--- a/src/os/FileJournal.cc
+++ b/src/os/FileJournal.cc
@@ -698,6 +698,10 @@ int FileJournal::read_header()
 bufferptr FileJournal::prepare_header()
 {
   bufferlist bl;
+  {
+    Mutex::Locker l(finisher_lock);
+    header.committed_up_to = journaled_seq;
+  }
   ::encode(header, bl);
   bufferptr bp = buffer::create_page_aligned(get_top());
   bp.zero();
@@ -961,6 +965,12 @@ void FileJournal::do_write(bufferlist& bl)
     return;
 
   buffer::ptr hbp;
+  if (g_conf->journal_write_header_frequency &&
+      (((++journaled_since_start) %
+       g_conf->journal_write_header_frequency) == 0)) {
+    must_write_header = true;
+  }
+
   if (must_write_header) {
     must_write_header = false;
     hbp = prepare_header();
@@ -1170,6 +1180,13 @@ void FileJournal::write_thread_entry()
 #ifdef HAVE_LIBAIO
 void FileJournal::do_aio_write(bufferlist& bl)
 {
+
+  if (g_conf->journal_write_header_frequency &&
+      (((++journaled_since_start) %
+       g_conf->journal_write_header_frequency) == 0)) {
+    must_write_header = true;
+  }
+
   // nothing to do?
   if (bl.length() == 0 && !must_write_header) 
     return;
@@ -1611,6 +1628,13 @@ bool FileJournal::read_entry(bufferlist& bl, uint64_t& seq)
   h = (entry_header_t *)hbl.c_str();
 
   if (!h->check_magic(read_pos, header.get_fsid64())) {
+    if (header.committed_up_to && seq <= header.committed_up_to) {
+      derr << "ERROR: header claims we are committed up through "
+          << header.committed_up_to << " however, we have failed to read "
+          << "entry " << seq
+          << " journal is likely corrupt" << dendl;
+      assert(0 == "FileJournal::read_entry(): corrupt journal");
+    }
     dout(2) << "read_entry " << read_pos << " : bad header magic, end of journal" << dendl;
     return false;
   }

diff --git a/src/os/FileJournal.h b/src/os/FileJournal.h
index 0dd880bdfea1f06a0b9e58c20376a62bffee9f9a..ed6ca10dc3113e4814415c8c0d409e7eb9cf531b 100644 (file)
--- a/src/os/FileJournal.h
+++ b/src/os/FileJournal.h
@@ -108,8 +108,11 @@ public:
     __u32 alignment;
     int64_t max_size;   // max size of journal ring buffer
     int64_t start;      // offset of first entry
+    uint64_t committed_up_to; // committed up to
 
-    header_t() : flags(0), block_size(0), alignment(0), max_size(0), start(0) {}
+    header_t() :
+      flags(0), block_size(0), alignment(0), max_size(0), start(0),
+      committed_up_to(0) {}
 
     void clear() {
       start = block_size;
@@ -120,7 +123,7 @@ public:
     }
 
     void encode(bufferlist& bl) const {
-      __u32 v = 2;
+      __u32 v = 3;
       ::encode(v, bl);
       bufferlist em;
       {
@@ -130,6 +133,7 @@ public:
        ::encode(alignment, em);
        ::encode(max_size, em);
        ::encode(start, em);
+       ::encode(committed_up_to, em);
       }
       ::encode(em, bl);
     }
@@ -159,6 +163,10 @@ public:
       ::decode(alignment, t);
       ::decode(max_size, t);
       ::decode(start, t);
+      if (v > 2)
+       ::decode(committed_up_to, t);
+      else
+       committed_up_to = 0;
     }
   } header;
 
@@ -224,6 +232,7 @@ private:
 #endif
 
   uint64_t last_committed_seq;
+  uint64_t journaled_since_start;
 
   /*
    * full states cycle at the beginnging of each commit epoch, when commit_start()
@@ -330,6 +339,7 @@ private:
     aio_num(0), aio_bytes(0),
 #endif
     last_committed_seq(0), 
+    journaled_since_start(0),
     full_state(FULL_NOTFULL),
     fd(-1),
     writing_seq(0),