mds: fix ENOTEMPTY checking on rmdir/rename

We can't trust the inode rstat size without holding the locks.  We can
look at our auth frags and though without fear of a false positive
ENOTEMPTY, however.

Rename the function, introduce a helper for the locked check, update
comments, etc.

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 75ff6405a171f0451fe366950d2b4c1d32bc3cf1..8e61c67d517c7e33ccad94852a31face43da2ee0 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -3735,9 +3735,7 @@ void Server::handle_client_unlink(MDRequest *mdr)
   if (in->is_dir()) {
     if (rmdir) {
       // do empty directory checks
-      if (_dir_is_nonempty(mdr, in)) {
-       dout(7) << "handle_client_unlink on dir " << *in
-               << ", returning ENOTEMPTY" << dendl;
+      if (_dir_is_nonempty_unlocked(mdr, in)) {
        reply_request(mdr, -ENOTEMPTY);
        return;
       }
@@ -3795,9 +3793,7 @@ void Server::handle_client_unlink(MDRequest *mdr)
     return;
 
   if (in->is_dir() &&
-      in->get_projected_inode()->dirstat.size() != 0) {
-    dout(10) << " not empty, dirstat.size() = " << in->get_projected_inode()->dirstat.size()
-            << " on " << *in << dendl;
+      _dir_is_nonempty(mdr, in)) {
     reply_request(mdr, -ENOTEMPTY);
     return;
   }
@@ -3980,26 +3976,20 @@ void Server::_unlink_local_finish(MDRequest *mdr,
 
 
 
-/** _dir_is_not_empty
+/** _dir_is_nonempty[_unlocked]
  *
- * check if a directory is non-empty (i.e. we can rmdir it), and make
- * sure it is part of the same subtree (i.e. local) so that rmdir will
- * occur locally.
+ * check if a directory is non-empty (i.e. we can rmdir it).
  *
- * this is a fastpath check.  we can't really be sure until we rdlock
- * the filelock.
- *
- * @param in is the inode being rmdir'd.  return true if it is
- * definitely not empty.
+ * the unlocked varient this is a fastpath check.  we can't really be
+ * sure until we rdlock the filelock.
  */
-bool Server::_dir_is_nonempty(MDRequest *mdr, CInode *in)
+bool Server::_dir_is_nonempty_unlocked(MDRequest *mdr, CInode *in)
 {
-  dout(10) << "dir_is_nonempty " << *in << dendl;
+  dout(10) << "dir_is_nonempty_unlocked " << *in << dendl;
   assert(in->is_auth());
 
-  
   if (in->snaprealm && in->snaprealm->snaps.size())
-    return true; //in a snapshot!
+    return true; // in a snapshot!
 
   list<frag_t> frags;
   in->dirfragtree.get_leaves(frags);
@@ -4010,19 +4000,33 @@ bool Server::_dir_is_nonempty(MDRequest *mdr, CInode *in)
     CDir *dir = in->get_or_open_dirfrag(mdcache, *p);
     assert(dir);
 
-    // does the frag _look_ empty?
-    if (dir->inode->get_projected_inode()->dirstat.size() > 0) {       
-      dout(10) << "dir_is_nonempty projected dir size still "
-              << dir->inode->get_projected_inode()->dirstat.size()
-              << " on " << *dir->inode
-              << dendl;
-      return true;
+    // is the frag obviously non-empty?
+    if (dir->is_auth()) {
+      if (dir->get_projected_fnode()->fragstat.size()) {
+       dout(10) << "dir_is_nonempty_unlocked dirstat has " 
+                << dir->get_projected_fnode()->fragstat.size() << " items " << *dir << dendl;
+       return true;
+      }
     }
+  }
 
-    // not dir auth?
-    if (!dir->is_auth()) {
-      dout(10) << "dir_is_nonempty non-auth dirfrag for " << *dir << dendl;
-    }
+  return false;
+}
+
+bool Server::_dir_is_nonempty(MDRequest *mdr, CInode *in)
+{
+  dout(10) << "dir_is_nonempty " << *in << dendl;
+  assert(in->is_auth());
+
+  if (in->snaprealm && in->snaprealm->snaps.size())
+    return true; // in a snapshot!
+
+  if (in->get_projected_inode()->dirstat.size() > 0) { 
+    dout(10) << "dir_is_nonempty projected dir size still "
+            << in->get_projected_inode()->dirstat.size()
+            << " on " << *in
+            << dendl;
+    return true;
   }
 
   return false;
@@ -4117,7 +4121,7 @@ void Server::handle_client_rename(MDRequest *mdr)
     }
 
     // non-empty dir?
-    if (oldin->is_dir() && _dir_is_nonempty(mdr, oldin)) {
+    if (oldin->is_dir() && _dir_is_nonempty_unlocked(mdr, oldin)) {
       reply_request(mdr, -ENOTEMPTY);
       return;
     }
@@ -4277,9 +4281,7 @@ void Server::handle_client_rename(MDRequest *mdr)
   
   if (oldin &&
       oldin->is_dir() &&
-      oldin->get_projected_inode()->dirstat.size() != 0) {
-    dout(10) << " target inode dir not empty, dirstat.size() = " << oldin->get_projected_inode()->dirstat.size() 
-            << " on " << *oldin << dendl;
+      _dir_is_nonempty(mdr, oldin)) {
     reply_request(mdr, -ENOTEMPTY);
     return;
   }

diff --git a/src/mds/Server.h b/src/mds/Server.h
index 2eeb000db4d17ed9862014c12c2c8945440d9fc8..a0892eaccc253f943260072fcd2841a5aa8b86ed 100644 (file)
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -173,6 +173,7 @@ public:
 
   // unlink
   void handle_client_unlink(MDRequest *mdr);
+  bool _dir_is_nonempty_unlocked(MDRequest *mdr, CInode *rmdiri);
   bool _dir_is_nonempty(MDRequest *mdr, CInode *rmdiri);
   void _unlink_local(MDRequest *mdr, CDentry *dn, CDentry *straydn);
   void _unlink_local_finish(MDRequest *mdr,