mds: journal root inode changes

Journal root inode changes, and flush them when trimming the log segment.

4 files changed, 69 insertions, 17 deletions

diff --git a/src/mds/CInode.cc b/src/mds/CInode.cc
index 0ac6d25874f..142df442a56 100644
--- a/src/mds/CInode.cc
+++ b/src/mds/CInode.cc
@@ -532,10 +532,15 @@ Capability *CInode::add_client_cap(int client, Session *session,
 
 
 version_t CInode::pre_dirty()
-{    
-  assert(parent || projected_parent.size());
-  version_t pv = get_projected_parent_dn()->pre_dirty(get_projected_version());
-  dout(10) << "pre_dirty " << pv << " (current v " << inode.version << ")" << dendl;
+{
+  version_t pv; 
+  if (parent || projected_parent.size()) {
+    pv = get_projected_parent_dn()->pre_dirty(get_projected_version());
+    dout(10) << "pre_dirty " << pv << " (current v " << inode.version << ")" << dendl;
+  } else {
+    assert(is_root());
+    pv = get_projected_version() + 1;
+  }
   return pv;
 }
 
@@ -556,8 +561,6 @@ void CInode::mark_dirty(version_t pv, LogSegment *ls) {
   
   dout(10) << "mark_dirty " << *this << dendl;
 
-  assert(parent || projected_parent.size());
-
   /*
     NOTE: I may already be dirty, but this fn _still_ needs to be called so that
     the directory is (perhaps newly) dirtied, and so that parent_dir_version is 
@@ -574,7 +577,8 @@ void CInode::mark_dirty(version_t pv, LogSegment *ls) {
   _mark_dirty(ls);
 
   // mark dentry too
-  parent->mark_dirty(pv, ls);
+  if (parent)
+    parent->mark_dirty(pv, ls);
 }
 
 
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 8b5ecc7183a..577b1e489b8 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -1341,10 +1341,14 @@ void MDCache::journal_cow_inode(Mutation *mut, EMetaBlob *metablob, CInode *in,
 
 inode_t *MDCache::journal_dirty_inode(Mutation *mut, EMetaBlob *metablob, CInode *in, snapid_t follows)
 {
-  CDentry *dn = in->get_projected_parent_dn();
-  if (!dn->get_projected_linkage()->is_null())  // no need to cow a null dentry
-    journal_cow_dentry(mut, metablob, dn, follows);
-  return metablob->add_primary_dentry(dn, true, in, in->get_projected_inode());
+  if (in->is_root()) {
+    return metablob->add_root(true, in, in->get_projected_inode());
+  } else {
+    CDentry *dn = in->get_projected_parent_dn();
+    if (!dn->get_projected_linkage()->is_null())  // no need to cow a null dentry
+      journal_cow_dentry(mut, metablob, dn, follows);
+    return metablob->add_primary_dentry(dn, true, in, in->get_projected_inode());
+  }
 }
 
 
@@ -1578,6 +1582,9 @@ void MDCache::predirty_journal_parents(Mutation *mut, EMetaBlob *blob,
   if (mut->now == utime_t())
     mut->now = g_clock.real_now();
 
+  if (in->is_root())
+    return;
+
   dout(10) << "predirty_journal_parents"
 	   << (do_parent_mtime ? " do_parent_mtime":"")
 	   << " linkunlink=" <<  linkunlink
@@ -1680,10 +1687,6 @@ void MDCache::predirty_journal_parents(Mutation *mut, EMetaBlob *blob,
       cur->dirty_old_rstats.clear();
     }
 
-    // stop?
-    if (pin->is_root())
-      break;
-
     bool stop = false;
     if (!pin->is_auth() || pin->is_ambiguous_auth()) {
       dout(10) << "predirty_journal_parents !auth or ambig on " << *pin << dendl;
@@ -1742,7 +1745,12 @@ void MDCache::predirty_journal_parents(Mutation *mut, EMetaBlob *blob,
      * actually, no.  for now, silently drop rstats for old parents.  we need 
      * hard link backpointers to do the above properly.
      */
+
+    // stop?
+    if (pin->is_root())
+      break;
     parentdn = pin->get_projected_parent_dn();
+    assert(parentdn);
 
     // rstat
     if (primary_dn) {
diff --git a/src/mds/events/EMetaBlob.h b/src/mds/events/EMetaBlob.h
index dbd8ab51c37..569d1f8940c 100644
--- a/src/mds/events/EMetaBlob.h
+++ b/src/mds/events/EMetaBlob.h
@@ -320,6 +320,7 @@ private:
   // my lumps.  preserve the order we added them in a list.
   list<dirfrag_t>         lump_order;
   map<dirfrag_t, dirlump> lump_map;
+  fullbit *root;
 
   list<pair<__u8,version_t> > table_tids;  // tableclient transactions
 
@@ -345,6 +346,10 @@ private:
   void encode(bufferlist& bl) const {
     ::encode(lump_order, bl);
     ::encode(lump_map, bl);
+    bufferlist rootbl;
+    if (root)
+      root->encode(rootbl);
+    ::encode(rootbl, bl);
     ::encode(table_tids, bl);
     ::encode(opened_ino, bl);
     ::encode(allocated_ino, bl);
@@ -361,6 +366,12 @@ private:
   void decode(bufferlist::iterator &bl) {
     ::decode(lump_order, bl);
     ::decode(lump_map, bl);
+    bufferlist rootbl;
+    ::decode(rootbl, bl);
+    if (rootbl.length()) {
+      bufferlist::iterator p = rootbl.begin();
+      root = new fullbit(p);
+    }
     ::decode(table_tids, bl);
     ::decode(opened_ino, bl);
     ::decode(allocated_ino, bl);
@@ -384,6 +395,9 @@ private:
   //LogSegment *_segment;
 
   EMetaBlob(MDLog *mdl = 0);  // defined in journal.cc
+  ~EMetaBlob() {
+    delete root;
+  }
 
   void print(ostream& out) {
     for (list<dirfrag_t>::iterator p = lump_order.begin();
@@ -540,6 +554,27 @@ private:
     return add_primary_dentry(dn, dirty);
   }
 
+  inode_t *add_root(bool dirty, CInode *in, inode_t *pi=0, fragtree_t *pdft=0, bufferlist *psnapbl=0,
+		    map<string,bufferptr> *px=0) {
+    in->last_journaled = my_offset;
+    //cout << "journaling " << in->inode.ino << " at " << my_offset << std::endl;
+
+    bufferlist snapbl;
+    if (psnapbl)
+      snapbl = *psnapbl;
+    else
+      in->encode_snap_blob(snapbl);
+
+    nstring empty;
+    root = new fullbit(empty,
+	      in->first, in->last,
+	      0,
+	      in->inode, in->dirfragtree,
+	      px ? *px : in->xattrs,
+	      in->symlink, snapbl,
+	      dirty);
+    return &root->inode;
+  }
   
   dirlump& add_dir(CDir *dir, bool dirty, bool complete=false, bool isnew=false) {
     return add_dir(dir->dirfrag(), dir->get_projected_fnode(), dir->get_projected_version(),
diff --git a/src/mds/journal.cc b/src/mds/journal.cc
index f96e6189d8e..0f3b6ada1c0 100644
--- a/src/mds/journal.cc
+++ b/src/mds/journal.cc
@@ -98,7 +98,11 @@ C_Gather *LogSegment::try_to_expire(MDS *mds)
   for (xlist<CInode*>::iterator p = dirty_inodes.begin(); !p.end(); ++p) {
     dout(20) << " dirty_inode " << **p << dendl;
     assert((*p)->is_auth());
-    commit.insert((*p)->get_parent_dn()->get_dir());
+    if ((*p)->is_root()) {
+      if (!gather) gather = new C_Gather;
+      (*p)->store(gather->new_sub());
+    } else
+      commit.insert((*p)->get_parent_dn()->get_dir());
   }
 
   if (!commit.empty()) {
@@ -284,7 +288,8 @@ void EString::replay(MDS *mds)
 // -----------------------
 // EMetaBlob
 
-EMetaBlob::EMetaBlob(MDLog *mdlog) : opened_ino(0),
+EMetaBlob::EMetaBlob(MDLog *mdlog) : root(NULL),
+				     opened_ino(0),
 				     inotablev(0), sessionmapv(0),
 				     allocated_ino(0),
 				     last_subtree_map(mdlog ? mdlog->get_last_segment_offset() : 0),