From ce0b74e55e6647ccecf01aa001849bf1eb3f4185 Mon Sep 17 00:00:00 2001
From: "Yan, Zheng" <zheng.z.yan@intel.com>
Date: Wed, 13 Mar 2013 20:58:26 +0800
Subject: [PATCH] mds: encode dirfrag base in cache rejoin ack

Cache rejoin ack message already encodes inode base, make it also encode
dirfrag base. This allowes the message to replicate stray dentries like
MDentryUnlink message. The function will be used by later patch.

Signed-off-by: Yan, Zheng <zheng.z.yan@intel.com>
Reviewed-by: Greg Farnum <greg@inktank.com>
---
 src/mds/CDir.h                 | 20 +++++++++++++-------
 src/mds/MDCache.cc             | 15 ++++++++++++++-
 src/messages/MMDSCacheRejoin.h | 10 +++++++++-
 3 files changed, 36 insertions(+), 9 deletions(-)

diff --git a/src/mds/CDir.h b/src/mds/CDir.h
index 79946f15633..f4a3a3d4496 100644
--- a/src/mds/CDir.h
+++ b/src/mds/CDir.h
@@ -437,23 +437,29 @@ private:
     ::encode(dist, bl);
   }
 
-  void encode_replica(int who, bufferlist& bl) {
-    __u32 nonce = add_replica(who);
-    ::encode(nonce, bl);
+  void _encode_base(bufferlist& bl) {
     ::encode(first, bl);
     ::encode(fnode, bl);
     ::encode(dir_rep, bl);
     ::encode(dir_rep_by, bl);
   }
-  void decode_replica(bufferlist::iterator& p) {
-    __u32 nonce;
-    ::decode(nonce, p);
-    replica_nonce = nonce;
+  void _decode_base(bufferlist::iterator& p) {
     ::decode(first, p);
     ::decode(fnode, p);
     ::decode(dir_rep, p);
     ::decode(dir_rep_by, p);
   }
+  void encode_replica(int who, bufferlist& bl) {
+    __u32 nonce = add_replica(who);
+    ::encode(nonce, bl);
+    _encode_base(bl);
+  }
+  void decode_replica(bufferlist::iterator& p) {
+    __u32 nonce;
+    ::decode(nonce, p);
+    replica_nonce = nonce;
+    _decode_base(p);
+  }
 
 
 
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 3ca6ada4cf5..8edb11cd60e 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -4514,6 +4514,17 @@ void MDCache::handle_cache_rejoin_ack(MMDSCacheRejoin *ack)
     }
   }
 
+  // full dirfrags
+  for (map<dirfrag_t, bufferlist>::iterator p = ack->dirfrag_bases.begin();
+       p != ack->dirfrag_bases.end();
+       ++p) {
+    CDir *dir = get_dirfrag(p->first);
+    assert(dir);
+    bufferlist::iterator q = p->second.begin();
+    dir->_decode_base(q);
+    dout(10) << " got dir replica " << *dir << dendl;
+  }
+
   // full inodes
   bufferlist::iterator p = ack->inode_base.begin();
   while (!p.end()) {
@@ -5182,8 +5193,10 @@ void MDCache::rejoin_send_acks()
       // dir
       for (map<int,int>::iterator r = dir->replicas_begin();
 	   r != dir->replicas_end();
-	   ++r) 
+	   ++r) {
 	ack[r->first]->add_strong_dirfrag(dir->dirfrag(), ++r->second, dir->dir_rep);
+	ack[r->first]->add_dirfrag_base(dir);
+      }
 	   
       for (CDir::map_t::iterator q = dir->items.begin();
 	   q != dir->items.end();
diff --git a/src/messages/MMDSCacheRejoin.h b/src/messages/MMDSCacheRejoin.h
index eb565e3a665..c6a22585aa4 100644
--- a/src/messages/MMDSCacheRejoin.h
+++ b/src/messages/MMDSCacheRejoin.h
@@ -20,6 +20,7 @@
 #include "include/types.h"
 
 #include "mds/CInode.h"
+#include "mds/CDir.h"
 
 // sent from replica to auth
 
@@ -173,6 +174,7 @@ class MMDSCacheRejoin : public Message {
   // full
   bufferlist inode_base;
   bufferlist inode_locks;
+  map<dirfrag_t, bufferlist> dirfrag_bases;
 
   // authpins, xlocks
   struct slave_reqid {
@@ -264,7 +266,11 @@ public:
   void add_strong_dirfrag(dirfrag_t df, int n, int dr) {
     strong_dirfrags[df] = dirfrag_strong(n, dr);
   }
-   
+  void add_dirfrag_base(CDir *dir) {
+    bufferlist& bl = dirfrag_bases[dir->dirfrag()];
+    dir->_encode_base(bl);
+  }
+
   // dentries
   void add_weak_dirfrag(dirfrag_t df) {
     weak_dirfrags.insert(df);
@@ -300,6 +306,7 @@ public:
     ::encode(wrlocked_inodes, payload);
     ::encode(cap_export_bl, payload);
     ::encode(strong_dirfrags, payload);
+    ::encode(dirfrag_bases, payload);
     ::encode(weak, payload);
     ::encode(weak_dirfrags, payload);
     ::encode(weak_inodes, payload);
@@ -325,6 +332,7 @@ public:
       ::decode(cap_export_paths, q);
     }
     ::decode(strong_dirfrags, p);
+    ::decode(dirfrag_bases, p);
     ::decode(weak, p);
     ::decode(weak_dirfrags, p);
     ::decode(weak_inodes, p);