mds: follow snaps in path_traverse, returning resulting snapid to caller

diff --git a/src/include/ceph_fs.h b/src/include/ceph_fs.h
index 51e317c1f6cf271bbfbe32c2866c2e075e1b1a22..ac13a88a3f17d59eb998e99fd4e74f95d236a739 100644 (file)
--- a/src/include/ceph_fs.h
+++ b/src/include/ceph_fs.h
@@ -55,8 +55,9 @@ static inline int ceph_fsid_equal(const struct ceph_fsid *a, const struct ceph_f
 typedef __le64 ceph_ino_t;
 
 typedef __le64 ceph_snapid_t;
-#define CEPH_MAXSNAP 0xffffffffffffffull  /* 56 bits; see ceph_pg */
-#define CEPH_NOSNAP  ((__u64)(-1))
+#define CEPH_MAXSNAP ((1ull << 56)-1)  /* 56 bits; see ceph_pg */
+#define CEPH_SNAPDIR ((__u64)(-1))
+#define CEPH_NOSNAP  ((__u64)(-2))
 
 struct ceph_object {
        union {

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 594c02eb395bb3be233aa14f16d1fdbeff687339..dddd053cdb4fcd0aec33fe284abbd190832d0e57 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -4196,6 +4196,7 @@ Context *MDCache::_get_waiter(MDRequest *mdr, Message *req)
 int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
                           filepath& origpath,               // what
                            vector<CDentry*>& trace,          // result
+                          snapid_t *psnapid,
                            bool follow_trailing_symlink,     // how
                            int onfail)
 {
@@ -4207,6 +4208,8 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
   // keep a list of symlinks we touch to avoid loops
   set< pair<CInode*, string> > symlinks_resolved; 
 
+  snapid_t snapid = CEPH_NOSNAP;
+
   // root
   CInode *cur = get_inode(origpath.get_ino());
   if (cur == NULL) {
@@ -4239,6 +4242,25 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
       return -ENOTDIR;
     }
 
+    // snapdir?
+    if (path[depth].length() == 0) {
+      dout(10) << "traverse: snapdir" << dendl;
+      snapid = CEPH_SNAPDIR;
+      depth++;
+      continue;
+    }
+    if (snapid == CEPH_SNAPDIR) {
+      snapid = atoll(path[depth].c_str());
+      dout(10) << "traverse: snap " << path[depth] << " -> " << snapid << dendl;
+      if (!snapid)
+       return -ENOENT;
+      SnapRealm *realm = cur->find_snaprealm();
+      if (realm->get_snaps().count(snapid) == 0)
+       return -ENOENT;
+      depth++;
+      continue;
+    }
+
     // open dir
     frag_t fg = cur->pick_dirfrag(path[depth]);
     CDir *curdir = cur->get_dirfrag(fg);
@@ -4301,7 +4323,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
 
 
     // dentry
-    CDentry *dn = curdir->lookup(path[depth]);
+    CDentry *dn = curdir->lookup(path[depth], snapid);
 
     // null and last_bit and xlocked by me?
     if (dn && dn->is_null() && null_okay) {
@@ -4413,6 +4435,7 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
       continue;
     }
     
+
     // MISS.  dentry doesn't exist.
     dout(12) << "traverse: miss on dentry " << path[depth] << " in " << *curdir << dendl;
 
@@ -4478,6 +4501,8 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req,     // who
   }
   
   // success.
+  if (psnapid)
+    *psnapid = CEPH_NOSNAP;
   if (mds->logger) mds->logger->inc("thit");
   return 0;
 }
@@ -6227,7 +6252,7 @@ void MDCache::handle_dir_update(MDirUpdate *m)
       dout(5) << "trying discover on dir_update for " << path << dendl;
 
       int r = path_traverse(0, m,
-                           path, trace, true,
+                           path, trace, NULL, true,
                             MDS_TRAVERSE_DISCOVER);
       if (r > 0)
         return;

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 37b96f3481b008855cfb2112c7d7b836fd3eac2d..b4485dd5cb343f5bb92487d2c278daf69701bbc4 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -742,7 +742,8 @@ public:
 
   Context *_get_waiter(MDRequest *mdr, Message *req);
   int path_traverse(MDRequest *mdr, Message *req, filepath& path, 
-                   vector<CDentry*>& trace, bool follow_trailing_sym,
+                   vector<CDentry*>& trace, snapid_t *psnap,
+                   bool follow_trailing_sym,
                     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 0272be7f946aaaf0f113a4a9c0bfe4bca327ede9..aae86f38a6ed909ad27863db8653064237b09a58 100644 (file)
--- a/src/mds/Migrator.cc
+++ b/src/mds/Migrator.cc
@@ -1402,7 +1402,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, true, MDS_TRAVERSE_DISCOVER);
+    int r = cache->path_traverse(0, m, fpath, trace, NULL, true, MDS_TRAVERSE_DISCOVER);
     if (r > 0) return; // wait
     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 b07caf7b021cd5a1eb25e6349c64e74c2699991f..573471672ad31c6397557da940502e4c1a1a1653 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1247,8 +1247,9 @@ CDir *Server::traverse_to_auth_dir(MDRequest *mdr, vector<CDentry*> &trace, file
   dout(10) << "traverse_to_auth_dir dirpath " << refpath << " dname " << dname << dendl;
 
   // traverse to parent dir
+  snapid_t snapid;
   int r = mdcache->path_traverse(mdr, mdr->client_request,
-                                refpath, trace, false,
+                                refpath, trace, &snapid, false,
                                 MDS_TRAVERSE_FORWARD);
   if (r > 0) return 0; // delayed
   if (r < 0) {
@@ -1276,7 +1277,8 @@ CDir *Server::traverse_to_auth_dir(MDRequest *mdr, vector<CDentry*> &trace, file
 
 
 
-CInode* Server::rdlock_path_pin_ref(MDRequest *mdr, bool want_auth, bool rdlock_dft)
+CInode* Server::rdlock_path_pin_ref(MDRequest *mdr, snapid_t *psnapid,
+                                   bool want_auth, bool rdlock_dft)
 {
   dout(10) << "rdlock_path_pin_ref " << *mdr << dendl;
 
@@ -1297,7 +1299,7 @@ CInode* Server::rdlock_path_pin_ref(MDRequest *mdr, bool want_auth, bool rdlock_
   vector<CDentry*> trace;
   int r = mdcache->path_traverse(mdr, req,
                                 refpath, 
-                                trace, req->follow_trailing_symlink(),
+                                trace, psnapid, req->follow_trailing_symlink(),
                                 MDS_TRAVERSE_FORWARD);
   if (r > 0) return false; // delayed
   if (r < 0) {  // error
@@ -1513,7 +1515,8 @@ CDir* Server::try_open_auth_dirfrag(CInode *diri, frag_t fg, MDRequest *mdr)
 void Server::handle_client_stat(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *ref = rdlock_path_pin_ref(mdr, false);
+  snapid_t snapid;
+  CInode *ref = rdlock_path_pin_ref(mdr, &snapid, false);
   if (!ref) return;
 
   // which inode locks do I want?
@@ -1595,11 +1598,13 @@ public:
 void Server::handle_client_utime(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   __u32 mask;
   if (!cur) return;
 
-  if (cur->is_root()) {
+  if (snapid != CEPH_NOSNAP ||
+      cur->is_root()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
@@ -1642,10 +1647,12 @@ void Server::handle_client_utime(MDRequest *mdr)
 void Server::handle_client_chmod(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   if (!cur) return;
 
-  if (cur->is_root()) {
+  if (snapid != CEPH_NOSNAP ||
+      cur->is_root()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
@@ -1683,10 +1690,11 @@ void Server::handle_client_chmod(MDRequest *mdr)
 void Server::handle_client_chown(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   if (!cur) return;
 
-  if (cur->is_root()) {
+  if (snapid != CEPH_NOSNAP || cur->is_root()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
@@ -1724,10 +1732,11 @@ void Server::handle_client_chown(MDRequest *mdr)
 void Server::handle_client_setxattr(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   if (!cur) return;
 
-  if (cur->is_root()) {
+  if (snapid != CEPH_NOSNAP || cur->is_root()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
@@ -1780,10 +1789,11 @@ void Server::handle_client_setxattr(MDRequest *mdr)
 void Server::handle_client_removexattr(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   if (!cur) return;
 
-  if (cur->is_root()) {
+  if (snapid != CEPH_NOSNAP || cur->is_root()) {
     reply_request(mdr, -EINVAL);   // for now
     return;
   }
@@ -1834,7 +1844,8 @@ void Server::handle_client_readdir(MDRequest *mdr)
 {
   MClientRequest *req = mdr->client_request;
   int client = req->get_orig_source().num();
-  CInode *diri = rdlock_path_pin_ref(mdr, false, true);  // rdlock dirfragtreelock!
+  snapid_t snapid;
+  CInode *diri = rdlock_path_pin_ref(mdr, &snapid, false, true);  // rdlock dirfragtreelock!
   if (!diri) return;
 
   // it's a directory, right?
@@ -1892,6 +1903,8 @@ void Server::handle_client_readdir(MDRequest *mdr)
        it++) {
     CDentry *dn = it->second;
     if (dn->is_null()) continue;
+    if (dn->last < snapid || dn->first > snapid)
+      continue;
 
     CInode *in = dn->inode;
 
@@ -2138,7 +2151,7 @@ void Server::handle_client_link(MDRequest *mdr)
   dout(7) << "handle_client_link discovering target " << targetpath << dendl;
   vector<CDentry*> targettrace;
   int r = mdcache->path_traverse(mdr, req,
-                                targetpath, targettrace, false,
+                                targetpath, targettrace, NULL, false,
                                 MDS_TRAVERSE_DISCOVER);
   if (r > 0) return; // wait
   if (targettrace.empty()) r = -EINVAL; 
@@ -2677,7 +2690,7 @@ void Server::handle_client_unlink(MDRequest *mdr)
   // traverse to path
   vector<CDentry*> trace;
   int r = mdcache->path_traverse(mdr, req, 
-                                req->get_filepath(), trace, false,
+                                req->get_filepath(), trace, NULL, false,
                                 MDS_TRAVERSE_FORWARD);
   if (r > 0) return;
   if (trace.empty()) r = -EINVAL;   // can't unlink root
@@ -3026,7 +3039,7 @@ void Server::handle_client_rename(MDRequest *mdr)
   // traverse to src
   vector<CDentry*> srctrace;
   int r = mdcache->path_traverse(mdr, req,
-                                srcpath, srctrace, false,
+                                srcpath, srctrace, NULL, false,
                                 MDS_TRAVERSE_DISCOVER);
   if (r > 0) return;
   if (r < 0) {
@@ -3730,7 +3743,7 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
   dout(10) << " dest " << destpath << dendl;
   vector<CDentry*> trace;
   int r = mdcache->path_traverse(mdr, mdr->slave_request, 
-                                destpath, trace, false,
+                                destpath, trace, NULL, false,
                                 MDS_TRAVERSE_DISCOVERXLOCK);
   if (r > 0) return;
   assert(r == 0);  // we shouldn't get an error here!
@@ -3743,7 +3756,7 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
   filepath srcpath(mdr->slave_request->srcdnpath);
   dout(10) << " src " << srcpath << dendl;
   r = mdcache->path_traverse(mdr, mdr->slave_request,
-                            srcpath, trace, false,  
+                            srcpath, trace, NULL, false,  
                             MDS_TRAVERSE_DISCOVERXLOCK);
   if (r > 0) return;
   assert(r == 0);
@@ -4300,9 +4313,16 @@ void Server::handle_client_truncate(MDRequest *mdr)
     return;
   }
 
-  CInode *cur = rdlock_path_pin_ref(mdr, true);
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, true);
   if (!cur) return;
 
+  if (snapid != CEPH_NOSNAP) {
+    reply_request(mdr, -EINVAL);
+    return;
+  }
+    
+
   // check permissions?  
 
   // xlock inode
@@ -4355,8 +4375,9 @@ void Server::handle_client_open(MDRequest *mdr)
   bool need_auth = !file_mode_is_readonly(cmode) || (flags & O_TRUNC);
 
   dout(7) << "open on " << req->get_filepath() << dendl;
-
-  CInode *cur = rdlock_path_pin_ref(mdr, need_auth);
+  
+  snapid_t snapid;
+  CInode *cur = rdlock_path_pin_ref(mdr, &snapid, need_auth);
   if (!cur) return;
 
   // can only open a dir with mode FILE_MODE_PIN, at least for now.
@@ -4659,7 +4680,7 @@ void Server::handle_client_lssnap(MDRequest *mdr)
   // traverse to path
   vector<CDentry*> trace;
   int r = mdcache->path_traverse(mdr, req, 
-                                req->get_filepath(), trace, false,
+                                req->get_filepath(), trace, NULL, false,
                                 MDS_TRAVERSE_FORWARD);
   if (r > 0) return;
   if (trace.empty()) r = -EINVAL;   // can't snap root
@@ -4745,7 +4766,7 @@ void Server::handle_client_mksnap(MDRequest *mdr)
   // traverse to path
   vector<CDentry*> trace;
   int r = mdcache->path_traverse(mdr, req, 
-                                req->get_filepath(), trace, false,
+                                req->get_filepath(), trace, NULL, false,
                                 MDS_TRAVERSE_FORWARD);
   if (r > 0) return;
   if (trace.empty()) r = -EINVAL;   // can't snap root

diff --git a/src/mds/Server.h b/src/mds/Server.h
index bc32818fa6792decd5d0a037da29d5245d018112..cd3e93b4be999334d31f0f04ac950f267be64a24 100644 (file)
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -98,7 +98,7 @@ public:
   CDentry *prepare_null_dentry(MDRequest *mdr, CDir *dir, const string& dname, bool okexist=false);
   CInode* prepare_new_inode(MDRequest *mdr, CDir *dir);
 
-  CInode* rdlock_path_pin_ref(MDRequest *mdr, bool want_auth, bool rdlock_dft=false);
+  CInode* rdlock_path_pin_ref(MDRequest *mdr, snapid_t *psnapid, bool want_auth, bool rdlock_dft=false);
   CDentry* rdlock_path_xlock_dentry(MDRequest *mdr, bool okexist, bool mustexist);
 
   CDir* try_open_auth_dirfrag(CInode *diri, frag_t fg, MDRequest *mdr);