From 8ac1851049bf56736d2e91af59ddca75741071e7 Mon Sep 17 00:00:00 2001
From: sageweil <sageweil@29311d96-e01e-0410-9327-a35deaab8ce9>
Date: Sun, 11 Mar 2007 02:04:36 +0000
Subject: [PATCH] filepath cleanup.  major discover code cleanup. 
 thrashing+log trimming is stable!

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@1202 29311d96-e01e-0410-9327-a35deaab8ce9
---
 branches/sage/cephmds2/TODO                   |   5 +
 .../sage/cephmds2/client/SyntheticClient.cc   |   2 +-
 .../sage/cephmds2/client/SyntheticClient.h    |  10 +-
 branches/sage/cephmds2/include/filepath.h     | 133 ++++-----
 branches/sage/cephmds2/mds/CDir.h             |   6 -
 branches/sage/cephmds2/mds/CInode.cc          |   6 +
 branches/sage/cephmds2/mds/CInode.h           |   1 +
 branches/sage/cephmds2/mds/Locker.cc          |   6 +-
 branches/sage/cephmds2/mds/Locker.h           |   4 +-
 branches/sage/cephmds2/mds/MDBalancer.cc      |   3 +-
 branches/sage/cephmds2/mds/MDCache.cc         | 275 +++++++++---------
 branches/sage/cephmds2/mds/Server.cc          |  42 +--
 branches/sage/cephmds2/mds/Server.h           |   4 +-
 .../sage/cephmds2/messages/MClientRequest.h   |   2 +-
 branches/sage/cephmds2/messages/MDiscover.h   |   2 +-
 .../sage/cephmds2/messages/MDiscoverReply.h   |   3 +
 branches/sage/cephmds2/messages/MLock.h       |   2 +-
 17 files changed, 258 insertions(+), 248 deletions(-)

diff --git a/branches/sage/cephmds2/TODO b/branches/sage/cephmds2/TODO
index b06dcc1ac9c4d..7d6393c5258a5 100644
--- a/branches/sage/cephmds2/TODO
+++ b/branches/sage/cephmds2/TODO
@@ -34,6 +34,11 @@ mds
 - dir version/committed/etc versus migration, log expires.  
   - DOCUMENT.
 
+- discover
+  - dir_discovers should be dirfrag_t's?
+  - hard link dentries
+  - base_dir thing is broken, i think, bc of dirfrags.
+
 - mds failure vs clients
   - clean up client op redirection
   - idempotent ops
diff --git a/branches/sage/cephmds2/client/SyntheticClient.cc b/branches/sage/cephmds2/client/SyntheticClient.cc
index e443c09a139ea..8a3fc4e9d79ad 100644
--- a/branches/sage/cephmds2/client/SyntheticClient.cc
+++ b/branches/sage/cephmds2/client/SyntheticClient.cc
@@ -1087,7 +1087,7 @@ int SyntheticClient::random_walk(int num_req)
     // descend?
     if (.9*roll_die(::pow((double).9,(double)cwd.depth())) && subdirs.size()) {
       string s = get_random_subdir();
-      cwd.add_dentry( s );
+      cwd.push_dentry( s );
       dout(DBL) << "cd " << s << " -> " << cwd << endl;
       clear_dir();
       continue;
diff --git a/branches/sage/cephmds2/client/SyntheticClient.h b/branches/sage/cephmds2/client/SyntheticClient.h
index ebf96386be95c..a83f4bc12ae21 100644
--- a/branches/sage/cephmds2/client/SyntheticClient.h
+++ b/branches/sage/cephmds2/client/SyntheticClient.h
@@ -99,22 +99,22 @@ class SyntheticClient {
     while (r--) it++;
 
     n1 = cwd;
-    n1.add_dentry( *it );
+    n1.push_dentry( *it );
     return n1.get_path().c_str();
   }
   filepath n2;
   const char *get_random_sub() {
     assert(!contents.empty());
     int r = ((rand() % contents.size()) + (rand() % contents.size())) / 2;  // non-uniform distn
-    if (cwd.depth() && cwd.last_bit().length()) 
-      r += cwd.last_bit().c_str()[0];                                         // slightly permuted
+    if (cwd.depth() && cwd.last_dentry().length()) 
+      r += cwd.last_dentry().c_str()[0];                                         // slightly permuted
     r %= contents.size();
 
     map<string,inode_t>::iterator it = contents.begin();
     while (r--) it++;
 
     n2 = cwd;
-    n2.add_dentry( it->first );
+    n2.push_dentry( it->first );
     return n2.get_path().c_str();
   }
   
@@ -124,7 +124,7 @@ class SyntheticClient {
     sprintf(sub_s, "%s.%d", base, rand() % 100);
     string f = sub_s;
     sub = cwd;
-    sub.add_dentry(f);
+    sub.push_dentry(f);
     return sub.c_str();
   }
 
diff --git a/branches/sage/cephmds2/include/filepath.h b/branches/sage/cephmds2/include/filepath.h
index 11874ea9e61ec..013f3b36704c3 100644
--- a/branches/sage/cephmds2/include/filepath.h
+++ b/branches/sage/cephmds2/include/filepath.h
@@ -32,10 +32,19 @@ using namespace std;
 
 
 class filepath {
+  /** path
+   * can be relative "a/b/c" or absolute "/a/b/c".
+   */
   string path;
-  vector<string> bits;
 
-  void rebuild() {
+  /** bits - path segemtns
+   * this is ['a', 'b', 'c'] for both the aboslute and relative case.
+   *
+   * NOTE: this value is LAZILY maintained... i.e. it's a cache
+   */
+  mutable vector<string> bits;
+
+  void rebuild_path() {
     if (absolute()) 
       path = "/";
     else 
@@ -45,7 +54,7 @@ class filepath {
       path += bits[i];
     }
   }
-  void parse() {
+  void parse_bits() const {
     bits.clear();
     int off = 0;
     while (off < (int)path.length()) {
@@ -71,103 +80,97 @@ class filepath {
   filepath(const char* s) {
     set_path(s);
   }
-
-  bool absolute() { return path[0] == '/'; }
-  bool relative() { return !absolute(); }
-  
-  void set_path(const string& s) {
-    path = s;
-    parse();
-  }
-  void set_path(const char *s) {
-    path = s;
-    parse();
+  filepath(const filepath& o) {
+    set_path(o.get_path());
   }
 
-  string& get_path() {
+
+  // accessors
+  const string& get_path() const {
     return path;
   }
+  const char *c_str() const {
+    return path.c_str();
+  }
+
   int length() const {
     return path.length();
   }
+  unsigned depth() const {
+    if (bits.empty() && path.length() > 0) parse_bits();
+    return bits.size();
+  }
+  bool empty() const {
+    return path.length() == 0;
+  }
 
-  const char *c_str() const {
-    return path.c_str();
+  // FIXME: const-edness
+  bool absolute() { return path.length() && path[0] == '/'; }
+  bool relative() { return !absolute(); }
+  
+  const string& operator[](int i) const {
+    if (bits.empty() && path.length() > 0) parse_bits();
+    return bits[i];
   }
 
+  const string& last_dentry() const {
+    if (bits.empty() && path.length() > 0) parse_bits();
+    return bits[ bits.size()-1 ];
+  }
 
   filepath prefixpath(int s) const {
     filepath t;
     for (int i=0; i<s; i++)
-      t.add_dentry(bits[i]);
+      t.push_dentry(bits[i]);
     return t;
   }
   filepath postfixpath(int s) const {
     filepath t;
     for (unsigned i=s; i<bits.size(); i++)
-      t.add_dentry(bits[i]);
+      t.push_dentry(bits[i]);
     return t;
   }
-  void add_dentry(const string& s) {
-    bits.push_back(s);
-    if (path.length())
-      path += "/";
-    path += s;
-  }
-  void append(const filepath& a) {
-    for (unsigned i=0; i<a.depth(); i++) 
-      add_dentry(a[i]);
-  }
 
-  void pop_dentry() {
-    bits.pop_back();
-    rebuild();
-  }    
-    
 
 
-  void clear() {
-    path = "";
+  // modifiers
+  void set_path(const string& s) {
+    path = s;
     bits.clear();
   }
-
-  const string& operator[](int i) const {
-    return bits[i];
+  void set_path(const char *s) {
+    path = s;
+    bits.clear();
   }
-
-  const string& last_bit() const {
-    return bits[ bits.size()-1 ];
+  void clear() {
+    path = "";
+    bits.clear();
   }
 
-  unsigned depth() const {
-    return bits.size();
+  void pop_dentry() {
+    if (bits.empty() && path.length() > 0) parse_bits();
+    bits.pop_back();
+    rebuild_path();
+  }    
+  void push_dentry(const string& s) {
+    if (bits.empty() && path.length() > 0) parse_bits();
+    bits.push_back(s);
+    if (path.length() && path[path.length()-1] != '/')
+      path += "/";
+    path += s;
   }
-  bool empty() {
-    return bits.size() == 0;
+  void append(const filepath& a) {
+    for (unsigned i=0; i<a.depth(); i++) 
+      push_dentry(a[i]);
   }
 
-
+  // encoding
   void _encode(bufferlist& bl) {
-    char n = bits.size();
-    bl.append((char*)&n, sizeof(char));
-    for (vector<string>::iterator it = bits.begin();
-         it != bits.end();
-         it++) { 
-      bl.append((*it).c_str(), (*it).length()+1);
-    }
+    ::_encode(path, bl);
   }
-
   void _decode(bufferlist& bl, int& off) {
-    clear();
-
-    char n;
-    bl.copy(off, sizeof(char), (char*)&n);
-    off += sizeof(char);
-    for (int i=0; i<n; i++) {
-      string s = bl.c_str() + off;
-      off += s.length() + 1;
-      add_dentry(s);
-    }
+    ::_decode(path, bl, off);
+    bits.clear();
   }
 
 };
diff --git a/branches/sage/cephmds2/mds/CDir.h b/branches/sage/cephmds2/mds/CDir.h
index 79ce1d9d38a45..511854a26557f 100644
--- a/branches/sage/cephmds2/mds/CDir.h
+++ b/branches/sage/cephmds2/mds/CDir.h
@@ -491,7 +491,6 @@ class CDir : public MDSCacheObject {
 class CDirDiscover {
   dirfrag_t dirfrag;
   int       nonce;
-  //int       dir_auth2;
   int       dir_rep;
   set<int>  rep_by;
 
@@ -500,7 +499,6 @@ class CDirDiscover {
   CDirDiscover(CDir *dir, int nonce) {
     dirfrag = dir->dirfrag();
     this->nonce = nonce;
-    //dir_auth2 = dir->dir_auth.second;
     dir_rep = dir->dir_rep;
     rep_by = dir->dir_rep_by;
   }
@@ -510,7 +508,6 @@ class CDirDiscover {
     assert(!dir->is_auth());
 
     dir->replica_nonce = nonce;
-    //dir->set_dir_auth( dir_auth );
     dir->dir_rep = dir_rep;
     dir->dir_rep_by = rep_by;
   }
@@ -521,7 +518,6 @@ class CDirDiscover {
   void _encode(bufferlist& bl) {
     bl.append((char*)&dirfrag, sizeof(dirfrag));
     bl.append((char*)&nonce, sizeof(nonce));
-    //bl.append((char*)&dir_auth, sizeof(dir_auth));
     bl.append((char*)&dir_rep, sizeof(dir_rep));
     ::_encode(rep_by, bl);
   }
@@ -531,8 +527,6 @@ class CDirDiscover {
     off += sizeof(dirfrag);
     bl.copy(off, sizeof(nonce), (char*)&nonce);
     off += sizeof(nonce);
-    //bl.copy(off, sizeof(dir_auth), (char*)&dir_auth);
-    //off += sizeof(dir_auth);
     bl.copy(off, sizeof(dir_rep), (char*)&dir_rep);
     off += sizeof(dir_rep);
     ::_decode(rep_by, bl, off);
diff --git a/branches/sage/cephmds2/mds/CInode.cc b/branches/sage/cephmds2/mds/CInode.cc
index e1cbe25bacdff..ffe133a37349c 100644
--- a/branches/sage/cephmds2/mds/CInode.cc
+++ b/branches/sage/cephmds2/mds/CInode.cc
@@ -581,6 +581,12 @@ pair<int,int> CInode::authority()
   return CDIR_AUTH_UNDEF;
 }
 
+bool CInode::auth_is_ambiguous()
+{
+  return authority().second != CDIR_AUTH_UNKNOWN;
+}
+
+
 CInodeDiscover* CInode::replicate_to( int rep )
 {
   assert(is_auth());
diff --git a/branches/sage/cephmds2/mds/CInode.h b/branches/sage/cephmds2/mds/CInode.h
index 0a289af22d659..c2bec5b4104d9 100644
--- a/branches/sage/cephmds2/mds/CInode.h
+++ b/branches/sage/cephmds2/mds/CInode.h
@@ -407,6 +407,7 @@ protected:
 
   // -- authority --
   pair<int,int> authority();
+  bool auth_is_ambiguous();
 
 
   // -- auth pins --
diff --git a/branches/sage/cephmds2/mds/Locker.cc b/branches/sage/cephmds2/mds/Locker.cc
index 69f1b79f2b7be..9c925b7312edd 100644
--- a/branches/sage/cephmds2/mds/Locker.cc
+++ b/branches/sage/cephmds2/mds/Locker.cc
@@ -1872,7 +1872,7 @@ class C_MDC_XlockRequest : public Context {
   Context *finisher;
 public:
   C_MDC_XlockRequest(Locker *mdc, 
-                     CDir *dir, string& dname, 
+                     CDir *dir, const string& dname, 
                      Message *req,
                      Context *finisher) {
     this->mdc = mdc;
@@ -1888,7 +1888,7 @@ public:
 };
 
 void Locker::dentry_xlock_request_finish(int r, 
-					  CDir *dir, string& dname, 
+					  CDir *dir, const string& dname, 
 					  Message *req,
 					  Context *finisher) 
 {
@@ -1910,7 +1910,7 @@ void Locker::dentry_xlock_request_finish(int r,
   delete finisher;
 }
 
-void Locker::dentry_xlock_request(CDir *dir, string& dname, bool create,
+void Locker::dentry_xlock_request(CDir *dir, const string& dname, bool create,
                                    Message *req, Context *onfinish)
 {
   dout(10) << "dentry_xlock_request on dn " << dname << " create=" << create << " in " << *dir << endl; 
diff --git a/branches/sage/cephmds2/mds/Locker.h b/branches/sage/cephmds2/mds/Locker.h
index d8dcb2c541a37..6146a6ccff27a 100644
--- a/branches/sage/cephmds2/mds/Locker.h
+++ b/branches/sage/cephmds2/mds/Locker.h
@@ -113,10 +113,10 @@ private:
                           Message *m, CInode *ref);
   void dentry_xlock_finish(CDentry *dn, bool quiet=false);
   void handle_lock_dn(MLock *m);
-  void dentry_xlock_request(CDir *dir, string& dname, bool create,
+  void dentry_xlock_request(CDir *dir, const string& dname, bool create,
                             Message *req, Context *onfinish);
   void dentry_xlock_request_finish(int r,
-				   CDir *dir, string& dname, 
+				   CDir *dir, const string& dname, 
 				   Message *req,
 				   Context *finisher);
 
diff --git a/branches/sage/cephmds2/mds/MDBalancer.cc b/branches/sage/cephmds2/mds/MDBalancer.cc
index 002d17c9f6b8a..11652bfa2395c 100644
--- a/branches/sage/cephmds2/mds/MDBalancer.cc
+++ b/branches/sage/cephmds2/mds/MDBalancer.cc
@@ -432,7 +432,8 @@ void MDBalancer::do_rebalance(int beat)
 
     double pop = im->popularity[MDS_POP_CURDOM].meta_load();
     if (pop < g_conf.mds_bal_idle_threshold &&
-        im->inode != mds->mdcache->get_root()) {
+        im->inode != mds->mdcache->get_root() &&
+	im->inode->authority().first != mds->get_nodeid()) {
       dout(-5) << " exporting idle import " << *im
                << " back to mds" << im->inode->authority().first
                << endl;
diff --git a/branches/sage/cephmds2/mds/MDCache.cc b/branches/sage/cephmds2/mds/MDCache.cc
index 8e08876bc8ed0..1864fb61f0b11 100644
--- a/branches/sage/cephmds2/mds/MDCache.cc
+++ b/branches/sage/cephmds2/mds/MDCache.cc
@@ -239,7 +239,7 @@ int MDCache::open_root(Context *c)
 
       filepath want;
       MDiscover *req = new MDiscover(whoami,
-                                     0,
+                                     1,
                                      want,
                                      false);  // there _is_ no base dir for the root inode
       mds->send_message_mds(req, 0, MDS_PORT_CACHE);
@@ -2719,7 +2719,7 @@ void MDCache::open_remote_ino_2(inodeno_t ino,
   // construct path
   filepath path;
   for (unsigned i=0; i<anchortrace.size(); i++) 
-    path.add_dentry(anchortrace[i]->ref_dn);
+    path.push_dentry(anchortrace[i]->ref_dn);
 
   dout(7) << " path is " << path << endl;
 
@@ -3110,30 +3110,20 @@ void MDCache::handle_discover(MDiscover *dis)
 {
   int whoami = mds->get_nodeid();
 
-  // from me to me?
-  if (dis->get_asker() == whoami) {
-    dout(7) << "discover for " << dis->get_want().get_path() << " bounced back to me, dropping." << endl;
-    delete dis;
-    return;
-  }
+  assert(dis->get_asker() != whoami);
 
   CInode *cur = 0;
-  MDiscoverReply *reply = 0;
-  //filepath fullpath;
+  MDiscoverReply *reply = new MDiscoverReply(dis->get_base_ino());
 
   // get started.
-  if (dis->get_base_ino() == 0) {
+  if (dis->get_base_ino() == 1) {
     // wants root
-    dout(7) << "discover from mds" << dis->get_asker() << " wants root + " << dis->get_want().get_path() << endl;
+    dout(7) << "handle_discover from mds" << dis->get_asker() << " wants root + " << dis->get_want().get_path() << endl;
 
     assert(mds->get_nodeid() == 0);
     assert(root->is_auth());
 
-    //fullpath = dis->get_want();
-
-
     // add root
-    reply = new MDiscoverReply(0);
     reply->add_inode( root->replicate_to( dis->get_asker() ) );
     dout(10) << "added root " << *root << endl;
 
@@ -3142,114 +3132,99 @@ void MDCache::handle_discover(MDiscover *dis)
   } else {
     // there's a base inode
     cur = get_inode(dis->get_base_ino());
-    assert(cur);
-
-    if (dis->wants_base_dir()) {
-      dout(7) << "discover from mds" << dis->get_asker() << " has " << *cur << " wants dir+" << dis->get_want().get_path() << endl;
-    } else {
-      dout(7) << "discover from mds" << dis->get_asker() << " has " << *cur->dir << " wants " << dis->get_want().get_path() << endl;
-    }
     
-    assert(cur->is_dir());
-    
-    // crazyness?
-    if (!cur->dir && !cur->is_auth()) {
-      int iauth = cur->authority().first;
-      dout(7) << "no dir and not inode auth; fwd to auth " << iauth << endl;
-      mds->send_message_mds( dis, iauth, MDS_PORT_CACHE);
+    if (!cur) {
+      dout(7) << "handle_discover mds" << dis->get_asker() 
+	      << " don't have base ino " << dis->get_base_ino() 
+	      << ", dropping" << endl;
+      delete reply;
       return;
     }
 
-    // frozen_dir?
-    if (!cur->dir && cur->is_frozen_dir()) {
-      dout(7) << "is frozen_dir, waiting" << endl;
-      cur->get_parent_dir()->add_waiter(CDir::WAIT_UNFREEZE, 
-                                        new C_MDS_RetryMessage(mds, dis));
-      return;
+    if (dis->wants_base_dir()) {
+      dout(7) << "handle_discover mds" << dis->get_asker() 
+	      << " has " << *cur 
+	      << " wants basedir+" << dis->get_want().get_path() 
+	      << endl;
+    } else {
+      dout(7) << "handle_discover mds" << dis->get_asker() 
+	      << " has " << *cur
+	      << " wants " << dis->get_want().get_path()
+	      << endl;
     }
-
-    if (!cur->dir) 
-      cur->get_or_open_dir(this);
-    assert(cur->dir);
-
-    dout(10) << "dir is " << *cur->dir << endl;
-    
-    // create reply
-    reply = new MDiscoverReply(cur->ino());
   }
 
   assert(reply);
   assert(cur);
   
-  /*
-  // first traverse and make sure we won't have to do any waiting
-  dout(10) << "traversing full discover path = " << fullpath << endl;
-  vector<CInode*> trav;
-  int r = path_traverse(fullpath, trav, dis, MDS_TRAVERSE_FAIL);
-  if (r > 0) 
-    return;  // fw or delay
-  dout(10) << "traverse finish w/o blocking, continuing" << endl;
-  // ok, now we know we won't block on dentry locks or readdir.
-  */
-
-
   // add content
   // do some fidgeting to include a dir if they asked for the base dir, or just root.
   for (unsigned i = 0; i < dis->get_want().depth() || dis->get_want().depth() == 0; i++) {
-    // add dir
-    if (reply->is_empty() && !dis->wants_base_dir()) {
-      dout(7) << "they don't want the base dir" << endl;
-    } else {
-      // is it actaully a dir at all?
-      if (!cur->is_dir()) {
-        dout(7) << "not a dir " << *cur << endl;
-        reply->set_flag_error_dir();
-        break;
-      }
+    
+    // -- figure out the dir
 
-      // add dir
-      if (!cur->dir_is_auth()) {
-        dout(7) << *cur << " dir auth is someone else, i'm done" << endl;
-        break;
-      }
-      
-      // did we hit a frozen_dir?
-      if (!cur->dir && cur->is_frozen_dir()) {
-        dout(7) << *cur << " is frozen_dir, stopping" << endl;
-        break;
+    // is *cur even a dir at all?
+    if (!cur->is_dir()) {
+      dout(7) << *cur << " not a dir" << endl;
+      reply->set_flag_error_dir();
+      break;
+    }
+
+    // pick frag
+    frag_t fg = cur->pick_dirfrag(dis->get_dentry(i));
+    CDir *curdir = cur->get_dirfrag(fg);
+
+    // am i dir auth (or if no dir, at least the inode auth)
+    if ((!curdir && !cur->is_auth()) ||
+	(curdir && !curdir->is_auth())) {
+      if (curdir) {
+	dout(7) << *curdir << " not dirfrag auth, setting dir_auth_hint" << endl;
+      } else {
+	dout(7) << *cur << " dirfrag not open, not inode auth, setting dir_auth_hint" << endl;
       }
       
-      if (!cur->dir) cur->get_or_open_dir(this);
-      assert(cur->dir);
+      // set hint (+ dentry, if there is one)
+      reply->set_dir_auth_hint(curdir->authority().first);
+      if (dis->get_want().depth() > i)
+	reply->set_error_dentry(dis->get_dentry(i));
+      break;
+    }
 
-      // is dir frozen?
-      if (cur->dir->is_frozen()) {
-	dout(7) << *cur->dir << " is frozen, stopping" << endl;
+    // open dir?
+    if (!curdir) {
+      assert(cur->is_auth());
+      curdir = cur->get_or_open_dirfrag(this, fg);
+    }
+    assert(curdir);
+    assert(curdir->is_auth());
+    
+    // is dir frozen?
+    if (curdir->is_frozen()) {
+      if (reply->is_empty()) {
+	dout(7) << *curdir << " is frozen, empty reply, waiting" << endl;
+	curdir->add_waiter(CDir::WAIT_UNFREEZE,
+			   new C_MDS_RetryMessage(mds, dis));
+	return;
+      } else {
+	dout(7) << *curdir << " is frozen, non-empty reply, stopping" << endl;
 	break;
       }
-
+    }
+    
+    // add dir
+    if (reply->is_empty() && !dis->wants_base_dir()) {
+      dout(7) << "they don't want the base dir" << endl;
+    } else {
       // add the dir.
-      reply->add_dir( new CDirDiscover( cur->dir, 
-                                        cur->dir->add_replica( dis->get_asker() ) ) );
-      dout(7) << "added dir " << *cur->dir << endl;
+      assert(!curdir->auth_is_ambiguous());
+      reply->add_dir( new CDirDiscover( curdir, 
+                                        curdir->add_replica( dis->get_asker() ) ) );
+      dout(7) << "added dir " << *curdir << endl;
     }
     if (dis->get_want().depth() == 0) break;
-
-    // is dir frozen?
-    if (cur->dir->is_frozen()) {
-      dout(7) << *cur->dir << " is frozen, stopping" << endl;
-      break;
-    }
     
     // lookup dentry
-    int dentry_auth = cur->dir->dentry_authority( dis->get_dentry(i) ).first;
-    if (dentry_auth != mds->get_nodeid()) {
-      dout(7) << *cur->dir << " dentry " << dis->get_dentry(i) << " auth " << dentry_auth << ", i'm done." << endl;
-      break;      // that's it for us!
-    }
-
-    // get inode
-    CDentry *dn = cur->dir->lookup( dis->get_dentry(i) );
+    CDentry *dn = curdir->lookup( dis->get_dentry(i) );
     
     /*
     if (dn && !dn->can_read()) { // xlocked?
@@ -3263,7 +3238,7 @@ void MDCache::handle_discover(MDiscover *dis)
     
     if (dn) {
       if (!dn->inode && dn->is_sync()) {
-        dout(7) << "mds" << whoami << " dentry " << dis->get_dentry(i) << " null in " << *cur->dir << ", returning error" << endl;
+        dout(7) << "mds" << whoami << " dentry " << dis->get_dentry(i) << " null in " << *curdir << ", returning error" << endl;
         reply->set_flag_error_dn( dis->get_dentry(i) );
         break;   // don't replicate null but non-locked dentries.
       }
@@ -3274,6 +3249,7 @@ void MDCache::handle_discover(MDiscover *dis)
       if (!dn->inode) break;  // we're done.
     }
 
+    // FIXME: hard links
     if (dn && dn->inode) {
         CInode *next = dn->inode;
         assert(next->is_auth());
@@ -3287,37 +3263,36 @@ void MDCache::handle_discover(MDiscover *dis)
         cur = next;
     } else {
       // don't have inode?
-      if (cur->dir->is_complete()) {
+      if (curdir->is_complete()) {
         // set error flag in reply
-        dout(7) << "mds" << whoami << " dentry " << dis->get_dentry(i) << " not found in " << *cur->dir << ", returning error" << endl;
+        dout(7) << "dname " << dis->get_dentry(i) << " dne in " << *curdir << ", returning error" << endl;
         reply->set_flag_error_dn( dis->get_dentry(i) );
         break;
       } else {
         // readdir
-        dout(7) << "mds" << whoami << " incomplete dir contents for " << *cur->dir << ", fetching" << endl;
+        dout(7) << "incomplete dir contents for " << *curdir << ", fetching" << endl;
 
-        //mds->mdstore->fetch_dir(cur->dir, NULL); //new C_MDS_RetryMessage(mds, dis));
-        //break; // send what we have so far
-
-        cur->dir->fetch(new C_MDS_RetryMessage(mds, dis));
-        return;
+	if (reply->is_empty()) {
+	  // fetch and wait
+	  curdir->fetch(new C_MDS_RetryMessage(mds, dis));
+	  return;
+	} else {
+	  // fetch, but send what we have so far
+	  curdir->fetch(0);
+	  break;
+	}
       }
     }
   }
 
-  // set dir_auth hint?
-  if (cur->is_dir() && cur->dir && 
-      !cur->dir->is_auth()) {
-    dout(7) << "setting dir_auth_hint for " << *cur->dir << endl;
-    reply->set_dir_auth_hint(cur->dir->authority().first);
-  }
-       
   // how did we do.
   if (reply->is_empty()) {
 
     // discard empty reply
     delete reply;
 
+    dout(7) << "dropping this empty reply)." << endl;
+    
     /*
     if (cur->is_auth() && !cur->dir->is_auth()) {
       // set hint
@@ -3334,6 +3309,7 @@ void MDCache::handle_discover(MDiscover *dis)
       return;
       }*/
     
+    /*
     // wait for frozen dir?
     if (cur->dir->is_frozen()) {
       dout(7) << "waiting for frozen " << *cur->dir << endl;
@@ -3342,6 +3318,7 @@ void MDCache::handle_discover(MDiscover *dis)
     } else {
       dout(7) << "i'm not auth, dropping request (+this empty reply)." << endl;
     }
+    */
     //assert(0);
     
   } else {
@@ -3406,47 +3383,55 @@ void MDCache::handle_discover_reply(MDiscoverReply *m)
   for (int i=m->has_root(); i<m->get_depth(); i++) {
     dout(10) << "discover_reply i=" << i << " cur " << *cur << endl;
 
+    frag_t fg;
+    CDir *curdir = 0;
+
     // dir
     if ((i >  0) ||
         (i == 0 && m->has_base_dir())) {
-      if (cur->dir) {
+      assert(m->get_dir(i).get_dirfrag().ino == cur->ino());
+      fg = m->get_dir(i).get_dirfrag().frag;
+      curdir = cur->get_dirfrag(fg);
+
+      if (curdir) {
         // had it
         /* this is strange, but it happens when:
            we discover multiple dentries under a dir.
            bc, no flag to indicate a dir discover is underway, (as there is w/ a dentry one).
            this is actually good, since (dir aside) they're asking for different information.
         */
-        dout(7) << "had " << *cur->dir << endl;
-        m->get_dir(i).update_dir(cur->dir);
-        dout2(7) << "now " << *cur->dir << endl;
+        dout(7) << "had " << *curdir << endl;
+        m->get_dir(i).update_dir(curdir);
+        dout2(7) << "now " << *curdir << endl;
       } else {
         // add it (_replica_)
-	CDir *ndir = cur->add_dirfrag( new CDir(cur, frag_t(), this, false) );  // FIXME dirfrag_t
-        m->get_dir(i).update_dir(ndir);
+	curdir = cur->add_dirfrag( new CDir(cur, fg, this, false) );
+        m->get_dir(i).update_dir(curdir);
 
 	// is this a dir_auth delegation boundary?
 	if (m->get_source().num() != cur->authority().first ||
+	    cur->auth_is_ambiguous() ||
 	    cur->ino() == 1)
-	  adjust_subtree_auth(ndir, m->get_source().num());
+	  adjust_subtree_auth(curdir, m->get_source().num());
+	
+        dout(7) << "added " << *curdir << " nonce " << curdir->replica_nonce << endl;
 	
-        dout(7) << "added " << *ndir << " nonce " << ndir->replica_nonce << endl;
-
         // get waiters
         cur->take_waiting(CInode::WAIT_DIR, finished);
 	dir_discovers.erase(cur->ino());
       }
     }    
-
+    
     // dentry error?
     if (i == m->get_depth()-1 && 
         m->is_flag_error_dn()) {
       // error!
       assert(cur->is_dir());
-      if (cur->dir) {
+      if (curdir) {
         dout(7) << " flag_error on dentry " << m->get_error_dentry() << ", triggering dentry?" << endl;
-        cur->dir->take_waiting(CDir::WAIT_DENTRY,
-                               m->get_error_dentry(),
-                               error);
+        curdir->take_waiting(CDir::WAIT_DENTRY,
+			     m->get_error_dentry(),
+			     error);
       } else {
         dout(7) << " flag_error on dentry " << m->get_error_dentry() << ", triggering dir?" << endl;
         cur->take_waiting(CInode::WAIT_DIR, error);
@@ -3460,23 +3445,29 @@ void MDCache::handle_discover_reply(MDiscoverReply *m)
     // dentry
     dout(7) << "i = " << i << " dentry is " << m->get_dentry(i).get_dname() << endl;
 
+    if (!curdir) {
+      fg = cur->pick_dirfrag(m->get_dentry(i).get_dname());
+      curdir = cur->get_dirfrag(fg);
+    }
+    assert(curdir);
+
     CDentry *dn = 0;
     if (i > 0 || 
         m->has_base_dentry()) {
-      dn = cur->dir->lookup( m->get_dentry(i).get_dname() );
+      dn = curdir->lookup( m->get_dentry(i).get_dname() );
       
       if (dn) {
         dout(7) << "had " << *dn << endl;
 	dn->replica_nonce = m->get_dentry(i).get_nonce();  // fix nonce.
       } else {
-        dn = cur->dir->add_dentry( m->get_dentry(i).get_dname(), 0, false );
+        dn = curdir->add_dentry( m->get_dentry(i).get_dname(), 0, false );
 	m->get_dentry(i).update_dentry(dn);
         dout(7) << "added " << *dn << endl;
       }
 
-      cur->dir->take_waiting(CDir::WAIT_DENTRY,
-                             m->get_dentry(i).get_dname(),
-                             finished);
+      curdir->take_waiting(CDir::WAIT_DENTRY,
+			   m->get_dentry(i).get_dname(),
+			   finished);
     }
     
     if (i >= m->get_num_inodes()) break;
@@ -3523,15 +3514,21 @@ void MDCache::handle_discover_reply(MDiscoverReply *m)
   }
 
   // dir_auth hint?
-  if (m->get_dir_auth_hint() != CDIR_AUTH_UNKNOWN) {
+  if (m->get_dir_auth_hint() != CDIR_AUTH_UNKNOWN &&
+      m->get_dir_auth_hint() != mds->get_nodeid()) {
     dout(7) << " dir_auth_hint is " << m->get_dir_auth_hint() << endl;
-    // let's just open it.
+    // let's try again.
     int hint = m->get_dir_auth_hint();
-    filepath want;  // no dentries, i just want the dir open
+
+    // include any path fragment we were looking for at the time
+    filepath want;
+    if (m->get_error_dentry().length() > 0)
+      want.push_dentry(m->get_error_dentry());
+    
     mds->send_message_mds(new MDiscover(mds->get_nodeid(),
 					cur->ino(),
 					want,
-					true),  // need the dir open
+					true),  // being conservative here.
 			  hint, MDS_PORT_CACHE);
     
     // note the dangling discover
diff --git a/branches/sage/cephmds2/mds/Server.cc b/branches/sage/cephmds2/mds/Server.cc
index 6ab130907ef60..48411e0c2500a 100644
--- a/branches/sage/cephmds2/mds/Server.cc
+++ b/branches/sage/cephmds2/mds/Server.cc
@@ -374,7 +374,7 @@ void Server::handle_client_request(MClientRequest *req)
       // <HACK>
       // is this a special debug command?
       if (refpath.depth() - 1 == trace.size() &&
-	  refpath.last_bit().find(".ceph.") == 0) {
+	  refpath.last_dentry().find(".ceph.") == 0) {
 	/*
 FIXME dirfrag
 	CDir *dir = 0;
@@ -383,14 +383,14 @@ FIXME dirfrag
 	else
 	  dir = trace[trace.size()-1]->get_inode()->dir;
 
-	dout(1) << "** POSSIBLE CEPH DEBUG COMMAND '" << refpath.last_bit() << "' in " << *dir << endl;
+	dout(1) << "** POSSIBLE CEPH DEBUG COMMAND '" << refpath.last_dentry() << "' in " << *dir << endl;
 
-	if (refpath.last_bit() == ".ceph.hash" &&
+	if (refpath.last_dentry() == ".ceph.hash" &&
 	    refpath.depth() > 1) {
 	  dout(1) << "got explicit hash command " << refpath << endl;
 	  /// ....
 	}
-	else if (refpath.last_bit() == ".ceph.commit") {
+	else if (refpath.last_dentry() == ".ceph.commit") {
 	  dout(1) << "got explicit commit command on  " << *dir << endl;
 	  dir->commit(0, 0);
 	}
@@ -1040,7 +1040,7 @@ int Server::prepare_mknod(MClientRequest *req, CInode *diri,
   
   // get containing directory (without last bit)
   filepath dirpath = req->get_filepath().prefixpath(req->get_filepath().depth() - 1);
-  string name = req->get_filepath().last_bit();
+  string name = req->get_filepath().last_dentry();
   
   CDir *dir = *pdir = validate_new_dentry_dir(req, diri, name);
   if (!dir) return 0;
@@ -1218,7 +1218,7 @@ public:
 void Server::handle_client_link(MClientRequest *req, CInode *ref)
 {
   // figure out name
-  string dname = req->get_filepath().last_bit();
+  string dname = req->get_filepath().last_dentry();
   dout(7) << "handle_client_link dname is " << dname << endl;
   
   // validate dir
@@ -1305,7 +1305,7 @@ void Server::handle_client_link_2(int r, MClientRequest *req, CInode *diri, vect
   }
   
   // what was the new dentry again?
-  string dname = req->get_filepath().last_bit();
+  string dname = req->get_filepath().last_dentry();
   frag_t fg = diri->pick_dirfrag(dname);
   CDir *dir = diri->get_dirfrag(fg);
   assert(dir);
@@ -1383,7 +1383,7 @@ void Server::handle_client_unlink(MClientRequest *req,
     reply_request(req, -EINVAL);
     return;
   }
-  string name = req->get_filepath().last_bit();
+  string name = req->get_filepath().last_dentry();
   
   // make sure parent is a dir?
   if (!diri->is_dir()) {
@@ -1627,7 +1627,7 @@ void Server::handle_client_rename(MClientRequest *req,
  -> so, re-traverse path.  and make sure we request_finish in the case of a forward!
    */
   filepath refpath = req->get_filepath();
-  string srcname = refpath.last_bit();
+  string srcname = refpath.last_dentry();
   refpath = refpath.prefixpath(refpath.depth()-1);
 
   dout(7) << "handle_client_rename src traversing to srcdir " << refpath << endl;
@@ -1779,12 +1779,12 @@ void Server::handle_client_rename_2(MClientRequest *req,
       // mv /some/thing /to/some/dir 
       destdir = try_open_dir(d, dfg, req);        // /to/some/dir
       if (!destdir) return;
-      destname = req->get_filepath().last_bit();  // thing
-      destpath.add_dentry(destname);
+      destname = req->get_filepath().last_dentry();  // thing
+      destpath.push_dentry(destname);
     } else {
       // mv /some/thing /to/some/existing_filename
       destdir = trace[trace.size()-1]->dir;       // /to/some
-      destname = destpath.last_bit();             // existing_filename
+      destname = destpath.last_dentry();             // existing_filename
     }
   }
   else if (trace.size() == destpath.depth()-1) {
@@ -1792,7 +1792,7 @@ void Server::handle_client_rename_2(MClientRequest *req,
       // mv /some/thing /to/some/place_that_maybe_dne     (we might be replica)
       destdir = try_open_dir(d, dfg, req); // /to/some
       if (!destdir) return;
-      destname = destpath.last_bit();    // place_that_MAYBE_dne
+      destname = destpath.last_dentry();    // place_that_MAYBE_dne
     } else {
       dout(7) << "dest dne" << endl;
       reply_request(req, -EINVAL);
@@ -1892,14 +1892,14 @@ void Server::handle_client_rename_2(MClientRequest *req,
 
 
 void Server::handle_client_rename_local(MClientRequest *req,
-                                     CInode *ref,
-                                     string& srcpath,
-                                     CInode *srcdiri,
-                                     CDentry *srcdn,
-                                     string& destpath,
-                                     CDir *destdir,
-                                     CDentry *destdn,
-                                     string& destname)
+					CInode *ref,
+					const string& srcpath,
+					CInode *srcdiri,
+					CDentry *srcdn,
+					const string& destpath,
+					CDir *destdir,
+					CDentry *destdn,
+					const string& destname)
 {
   //bool everybody = false;
   //if (true || srcdn->inode->is_dir()) {
diff --git a/branches/sage/cephmds2/mds/Server.h b/branches/sage/cephmds2/mds/Server.h
index 719bdf11aa03c..86bd11577c026 100644
--- a/branches/sage/cephmds2/mds/Server.h
+++ b/branches/sage/cephmds2/mds/Server.h
@@ -106,8 +106,8 @@ public:
                               vector<CDentry*>& trace,
                               int r);
   void handle_client_rename_local(MClientRequest *req, CInode *ref,
-                                  string& srcpath, CInode *srcdiri, CDentry *srcdn, 
-                                  string& destpath, CDir *destdir, CDentry *destdn, string& name);
+                                  const string& srcpath, CInode *srcdiri, CDentry *srcdn, 
+                                  const string& destpath, CDir *destdir, CDentry *destdn, const string& name);
 
   void handle_client_mkdir(MClientRequest *req, CInode *ref);
   void handle_client_rmdir(MClientRequest *req, CInode *ref);
diff --git a/branches/sage/cephmds2/messages/MClientRequest.h b/branches/sage/cephmds2/messages/MClientRequest.h
index 0115d0793c842..8b9e262723fdf 100644
--- a/branches/sage/cephmds2/messages/MClientRequest.h
+++ b/branches/sage/cephmds2/messages/MClientRequest.h
@@ -111,7 +111,7 @@ class MClientRequest : public Message {
   int get_caller_uid() { return st.caller_uid; }
   int get_caller_gid() { return st.caller_gid; }
   inodeno_t get_ino() { return st.ino; }
-  string& get_path() { return path.get_path(); }
+  const string& get_path() { return path.get_path(); }
   filepath& get_filepath() { return path; }
   int get_iarg() { return st.iarg; }
   int get_iarg2() { return st.iarg2; }
diff --git a/branches/sage/cephmds2/messages/MDiscover.h b/branches/sage/cephmds2/messages/MDiscover.h
index 7cb35d34a9cc3..08e3fff1aac61 100644
--- a/branches/sage/cephmds2/messages/MDiscover.h
+++ b/branches/sage/cephmds2/messages/MDiscover.h
@@ -26,7 +26,7 @@ using namespace std;
 
 class MDiscover : public Message {
   int             asker;
-  inodeno_t       base_ino;          // 0 -> none, want root
+  inodeno_t       base_ino;          // 1 -> root
   bool            want_base_dir;
   bool            want_root_inode;
   
diff --git a/branches/sage/cephmds2/messages/MDiscoverReply.h b/branches/sage/cephmds2/messages/MDiscoverReply.h
index db95d862b3e5b..c1b3f75d9d4dd 100644
--- a/branches/sage/cephmds2/messages/MDiscoverReply.h
+++ b/branches/sage/cephmds2/messages/MDiscoverReply.h
@@ -173,6 +173,9 @@ class MDiscoverReply : public Message {
   void set_dir_auth_hint(int a) {
     dir_auth_hint = a;
   }
+  void set_error_dentry(const string& dn) {
+    error_dentry = dn;
+  }
 
 
   // ...
diff --git a/branches/sage/cephmds2/messages/MLock.h b/branches/sage/cephmds2/messages/MLock.h
index 1d22d297d79d4..df3f413f76303 100644
--- a/branches/sage/cephmds2/messages/MLock.h
+++ b/branches/sage/cephmds2/messages/MLock.h
@@ -87,7 +87,7 @@ class MLock : public Message {
     otype = LOCK_OTYPE_DIR;
     this->ino = ino;
   }
-  void set_dn(inodeno_t dirino, string& dn) {
+  void set_dn(inodeno_t dirino, const string& dn) {
     otype = LOCK_OTYPE_DN;
     this->ino = dirino;
     this->dn = dn;
-- 
2.39.5