mds: allow explicit finisher context for path_traverse

Previously we could only path_traverse and retry a request or message.
This just allows an explicit context to be used as well.  It's the caller's
job to clean it up if we return <= 0.

Signed-off-by: Sage Weil <sage.weil@dreamhost.com>

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index a91e4df019367f925fd46e2551b115fb61f80f01..7e27980d7cc53e8941d7412c323f59e742bc2361 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -6196,14 +6196,16 @@ void MDCache::dispatch(Message *m)
  *  MDS_TRAVERSE_FAIL          - return an error
  */
 
-Context *MDCache::_get_waiter(MDRequest *mdr, Message *req)
+Context *MDCache::_get_waiter(MDRequest *mdr, Message *req, Context *fin)
 {
   if (mdr) {
     dout(20) << "_get_waiter retryrequest" << dendl;
     return new C_MDS_RetryRequest(this, mdr);
-  } else {
+  } else if (req) {
     dout(20) << "_get_waiter retrymessage" << dendl;
     return new C_MDS_RetryMessage(mds, req);
+  } else {
+    return fin;
   }
 }
 
@@ -6215,16 +6217,18 @@ Context *MDCache::_get_waiter(MDRequest *mdr, Message *req)
  * on failure, @pdnvec it is either the full trace, up to and
  *             including the final null dn, or empty.
  */
-int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
+int MDCache::path_traverse(MDRequest *mdr, Message *req, Context *fin,     // who
                           const filepath& path,                   // what
                            vector<CDentry*> *pdnvec,         // result
                           CInode **pin,
                            int onfail)
 {
-  assert(mdr || req);
   bool null_okay = (onfail == MDS_TRAVERSE_DISCOVERXLOCK);
   bool forward = (onfail == MDS_TRAVERSE_FORWARD);
 
+  assert(mdr || req || fin);
+  assert(!forward || mdr || req);  // forward requires a request
+
   snapid_t snapid = CEPH_NOSNAP;
   if (mdr)
     mdr->snapid = snapid;
@@ -6237,7 +6241,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
   CInode *cur = get_inode(path.get_ino());
   if (cur == NULL) {
     if (MDS_INO_IS_MDSDIR(path.get_ino())) 
-      open_foreign_mdsdir(path.get_ino(), _get_waiter(mdr, req));
+      open_foreign_mdsdir(path.get_ino(), _get_waiter(mdr, req, fin));
     else {
       //assert(0);  // hrm.. broken
       return -ESTALE;
@@ -6295,14 +6299,14 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
         // parent dir frozen_dir?
         if (cur->is_frozen_dir()) {
           dout(7) << "traverse: " << *cur->get_parent_dir() << " is frozen_dir, waiting" << dendl;
-          cur->get_parent_dn()->get_dir()->add_waiter(CDir::WAIT_UNFREEZE, _get_waiter(mdr, req));
+          cur->get_parent_dn()->get_dir()->add_waiter(CDir::WAIT_UNFREEZE, _get_waiter(mdr, req, fin));
           return 1;
         }
         curdir = cur->get_or_open_dirfrag(this, fg);
       } else {
         // discover?
        dout(10) << "traverse: need dirfrag " << fg << ", doing discover from " << *cur << dendl;
-       discover_path(cur, snapid, path.postfixpath(depth), _get_waiter(mdr, req),
+       discover_path(cur, snapid, path.postfixpath(depth), _get_waiter(mdr, req, fin),
                      onfail == MDS_TRAVERSE_DISCOVERXLOCK);
        if (mds->logger) mds->logger->inc(l_mds_tdis);
         return 1;
@@ -6321,7 +6325,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
     // doh!
       // FIXME: traverse is allowed?
       dout(7) << "traverse: " << *curdir << " is frozen, waiting" << dendl;
-      curdir->add_waiter(CDir::WAIT_UNFREEZE, _get_waiter(mdr, req));
+      curdir->add_waiter(CDir::WAIT_UNFREEZE, _get_waiter(mdr, req, fin));
       if (onfinish) delete onfinish;
       return 1;
     }
@@ -6332,7 +6336,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
     if (!noperm && 
        !mds->locker->rdlock_try(&cur->authlock, client, 0)) {
       dout(7) << "traverse: waiting on authlock rdlock on " << *cur << dendl;
-      cur->authlock.add_waiter(SimpleLock::WAIT_RD, _get_waiter(mdr, req));
+      cur->authlock.add_waiter(SimpleLock::WAIT_RD, _get_waiter(mdr, req, fin));
       return 1;
     }
 #endif
@@ -6364,7 +6368,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
        !dn->lock.can_read(client) &&
        (dnl->is_null() || forward)) {
       dout(10) << "traverse: xlocked dentry at " << *dn << dendl;
-      dn->lock.add_waiter(SimpleLock::WAIT_RD, _get_waiter(mdr, req));
+      dn->lock.add_waiter(SimpleLock::WAIT_RD, _get_waiter(mdr, req, fin));
       if (mds->logger) mds->logger->inc(l_mds_tlock);
       return 1;
     }
@@ -6383,7 +6387,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
        } else {
           dout(7) << "remote link to " << dnl->get_remote_ino() << ", which i don't have" << dendl;
          assert(mdr);  // we shouldn't hit non-primary dentries doing a non-mdr traversal!
-          open_remote_ino(dnl->get_remote_ino(), _get_waiter(mdr, req));
+          open_remote_ino(dnl->get_remote_ino(), _get_waiter(mdr, req, fin));
          if (mds->logger) mds->logger->inc(l_mds_trino);
           return 1;
         }        
@@ -6472,7 +6476,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
        // directory isn't complete; reload
         dout(7) << "traverse: incomplete dir contents for " << *cur << ", fetching" << dendl;
         touch_inode(cur);
-        curdir->fetch(_get_waiter(mdr, req), path[depth]);
+        curdir->fetch(_get_waiter(mdr, req, fin), path[depth]);
        if (mds->logger) mds->logger->inc(l_mds_tdirf);
         return 1;
       }
@@ -6492,7 +6496,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
       if ((onfail == MDS_TRAVERSE_DISCOVER ||
            onfail == MDS_TRAVERSE_DISCOVERXLOCK)) {
        dout(7) << "traverse: discover from " << path[depth] << " from " << *curdir << dendl;
-       discover_path(curdir, snapid, path.postfixpath(depth), _get_waiter(mdr, req),
+       discover_path(curdir, snapid, path.postfixpath(depth), _get_waiter(mdr, req, fin),
                      onfail == MDS_TRAVERSE_DISCOVERXLOCK);
        if (mds->logger) mds->logger->inc(l_mds_tdis);
         return 1;
@@ -6504,7 +6508,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
        if (curdir->is_ambiguous_auth()) {
          // wait
          dout(7) << "traverse: waiting for single auth in " << *curdir << dendl;
-         curdir->add_waiter(CDir::WAIT_SINGLEAUTH, _get_waiter(mdr, req));
+         curdir->add_waiter(CDir::WAIT_SINGLEAUTH, _get_waiter(mdr, req, fin));
          return 1;
        } 
 
@@ -6960,7 +6964,7 @@ void MDCache::handle_find_ino_reply(MMDSFindInoReply *m)
     if (!m->path.empty()) {
       // we got a path!
       vector<CDentry*> trace;
-      int r = path_traverse(NULL, m, m->path, &trace, NULL, MDS_TRAVERSE_DISCOVER);
+      int r = path_traverse(NULL, m, NULL, m->path, &trace, NULL, MDS_TRAVERSE_DISCOVER);
       if (r > 0)
        return; 
       dout(0) << "handle_find_ino_reply failed with " << r << " on " << m->path 
@@ -8765,7 +8769,7 @@ void MDCache::handle_dir_update(MDirUpdate *m)
       CInode *in;
       filepath path = m->get_path();
       dout(5) << "trying discover on dir_update for " << path << dendl;
-      int r = path_traverse(0, m, path, &trace, &in, MDS_TRAVERSE_DISCOVER);
+      int r = path_traverse(NULL, m, NULL, path, &trace, &in, MDS_TRAVERSE_DISCOVER);
       if (r > 0)
         return;
       assert(r == 0);

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 1ebf959683c4bf0d33b70a56f65de934bf065ac4..a128f1b7366a193d5bbf68a6ae86c1a5e56ebe97 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -985,8 +985,8 @@ public:
   void open_foreign_mdsdir(inodeno_t ino, Context *c);
   CDentry *get_or_create_stray_dentry(CInode *in);
 
-  Context *_get_waiter(MDRequest *mdr, Message *req);
-  int path_traverse(MDRequest *mdr, Message *req, const filepath& path,
+  Context *_get_waiter(MDRequest *mdr, Message *req, Context *fin);
+  int path_traverse(MDRequest *mdr, Message *req, Context *c, const filepath& path,
                    vector<CDentry*> *pdnvec, CInode **pin, int onfail);
   bool path_is_mine(filepath& path);
   bool path_is_mine(string& p) {

diff --git a/src/mds/Migrator.cc b/src/mds/Migrator.cc
index 592f3e5804d81826839b3f16324986dd253a7a7b..d4d18a7f6db7b6422668a2184f75948144b8887e 100644 (file)
--- a/src/mds/Migrator.cc
+++ b/src/mds/Migrator.cc
@@ -1536,7 +1536,7 @@ void Migrator::handle_export_discover(MExportDirDiscover *m)
     // must discover it!
     filepath fpath(m->get_path());
     vector<CDentry*> trace;
-    int r = cache->path_traverse(0, m, fpath, &trace, NULL, MDS_TRAVERSE_DISCOVER);
+    int r = cache->path_traverse(NULL, m, NULL, fpath, &trace, NULL, MDS_TRAVERSE_DISCOVER);
     if (r > 0) return;
     if (r < 0) {
       dout(7) << "handle_export_discover_2 failed to discover or not dir " << m->get_path() << ", NAK" << dendl;

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 01a5ada29a7c6e544e1d682e23f77954eea6d451..97fa764f2d9832b93277311851baad7d16cccd75 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1743,7 +1743,7 @@ CDir *Server::traverse_to_auth_dir(MDRequest *mdr, vector<CDentry*> &trace, file
 
   // traverse to parent dir
   CInode *diri;
-  int r = mdcache->path_traverse(mdr, 0, refpath, &trace, &diri, MDS_TRAVERSE_FORWARD);
+  int r = mdcache->path_traverse(mdr, NULL, NULL, refpath, &trace, &diri, MDS_TRAVERSE_FORWARD);
   if (r > 0) return 0; // delayed
   if (r < 0) {
     reply_request(mdr, r);
@@ -1778,7 +1778,7 @@ CInode* Server::rdlock_path_pin_ref(MDRequest *mdr, int n,
 
 
   // traverse
-  int r = mdcache->path_traverse(mdr, 0, refpath, &mdr->dn[n], &mdr->in[n], MDS_TRAVERSE_FORWARD);
+  int r = mdcache->path_traverse(mdr, NULL, NULL, refpath, &mdr->dn[n], &mdr->in[n], MDS_TRAVERSE_FORWARD);
   if (r > 0) return false; // delayed
   if (r < 0) {  // error
     if (r == -ENOENT && n == 0 && mdr->dn[n].size()) {
@@ -4066,7 +4066,7 @@ void Server::handle_client_unlink(MDRequest *mdr)
   // traverse to path
   vector<CDentry*> trace;
   CInode *in;
-  int r = mdcache->path_traverse(mdr, 0, req->get_filepath(), &trace, &in, MDS_TRAVERSE_FORWARD);
+  int r = mdcache->path_traverse(mdr, NULL, NULL, req->get_filepath(), &trace, &in, MDS_TRAVERSE_FORWARD);
   if (r > 0) return;
   if (r < 0) {
     reply_request(mdr, r);
@@ -4453,7 +4453,7 @@ void Server::handle_client_rename(MDRequest *mdr)
   CDir *destdir = destdn->get_dir();
   assert(destdir->is_auth());
 
-  int r = mdcache->path_traverse(mdr, 0, srcpath, &srctrace, NULL, MDS_TRAVERSE_DISCOVERXLOCK);
+  int r = mdcache->path_traverse(mdr, NULL, NULL, srcpath, &srctrace, NULL, MDS_TRAVERSE_DISCOVERXLOCK);
   if (r > 0)
     return; // delayed
   if (r < 0) {
@@ -5209,7 +5209,7 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
   filepath destpath(mdr->slave_request->destdnpath);
   dout(10) << " dest " << destpath << dendl;
   vector<CDentry*> trace;
-  int r = mdcache->path_traverse(mdr, 0, destpath, &trace, NULL, MDS_TRAVERSE_DISCOVERXLOCK);
+  int r = mdcache->path_traverse(mdr, NULL, NULL, destpath, &trace, NULL, MDS_TRAVERSE_DISCOVERXLOCK);
   if (r > 0) return;
   assert(r == 0);  // we shouldn't get an error here!
       
@@ -5222,7 +5222,7 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
   filepath srcpath(mdr->slave_request->srcdnpath);
   dout(10) << " src " << srcpath << dendl;
   CInode *srci;
-  r = mdcache->path_traverse(mdr, 0, srcpath, &trace, &srci, MDS_TRAVERSE_DISCOVERXLOCK);
+  r = mdcache->path_traverse(mdr, NULL, NULL, srcpath, &trace, &srci, MDS_TRAVERSE_DISCOVERXLOCK);
   if (r > 0) return;
   assert(r == 0);