mds: don't abort request that has already started slave operation

Access check and fragment size check can fail after start slave
operation. This causes mds to crash.

Signed-off-by: Yan, Zheng <zyan@redhat.com>

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index e95bdb3636c02a9ff351c914ab58006565ef26e9..c384a5a29786376b1b6a6cee269034056cf4178e 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1505,10 +1505,12 @@ void Server::dispatch_client_request(MDRequestRef& mdr)
         req->get_op() == CEPH_MDS_OP_SETXATTR ||
         req->get_op() == CEPH_MDS_OP_SETFILELOCK ||
         req->get_op() == CEPH_MDS_OP_CREATE ||
-        req->get_op() == CEPH_MDS_OP_LINK ||
-        req->get_op() == CEPH_MDS_OP_RENAME ||
         req->get_op() == CEPH_MDS_OP_SYMLINK ||
-        req->get_op() == CEPH_MDS_OP_MKSNAP) {
+        req->get_op() == CEPH_MDS_OP_MKSNAP ||
+       ((req->get_op() == CEPH_MDS_OP_LINK ||
+         req->get_op() == CEPH_MDS_OP_RENAME) &&
+        (!mdr->has_more() || mdr->more()->witnessed.empty())) // haven't started slave request
+       ) {
 
       dout(20) << __func__ << ": full, responding ENOSPC to op " << ceph_mds_op_name(req->get_op()) << dendl;
       respond_to_request(mdr, -ENOSPC);
@@ -4950,14 +4952,16 @@ void Server::handle_client_link(MDRequestRef& mdr)
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;
 
-  if (!check_access(mdr, targeti, MAY_WRITE))
-    return;
+  if ((!mdr->has_more() || mdr->more()->witnessed.empty())) {
+    if (!check_access(mdr, targeti, MAY_WRITE))
+      return;
 
-  if (!check_access(mdr, dir->get_inode(), MAY_WRITE))
-    return;
+    if (!check_access(mdr, dir->get_inode(), MAY_WRITE))
+      return;
 
-  if (!check_fragment_space(mdr, dir))
-    return;
+    if (!check_fragment_space(mdr, dir))
+      return;
+  }
 
   // go!
   assert(g_conf->mds_kill_link_at != 1);
@@ -5560,15 +5564,17 @@ void Server::handle_client_unlink(MDRequestRef& mdr)
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;
 
-  if (!check_access(mdr, diri, MAY_WRITE))
-    return;
-
   if (in->is_dir() &&
       _dir_is_nonempty(mdr, in)) {
     respond_to_request(mdr, -ENOTEMPTY);
     return;
   }
 
+  if ((!mdr->has_more() || mdr->more()->witnessed.empty())) {
+    if (!check_access(mdr, diri, MAY_WRITE))
+      return;
+  }
+
   // yay!
   if (in->is_dir() && in->has_subtree_root_dirfrag()) {
     // subtree root auths need to be witnesses
@@ -6401,17 +6407,19 @@ void Server::handle_client_rename(MDRequestRef& mdr)
                                  &remote_wrlocks, auth_pin_freeze))
     return;
 
-  if (!check_access(mdr, srcdn->get_dir()->get_inode(), MAY_WRITE))
-    return;
+  if ((!mdr->has_more() || mdr->more()->witnessed.empty())) {
+    if (!check_access(mdr, srcdn->get_dir()->get_inode(), MAY_WRITE))
+      return;
 
-  if (!check_access(mdr, destdn->get_dir()->get_inode(), MAY_WRITE))
-    return;
+    if (!check_access(mdr, destdn->get_dir()->get_inode(), MAY_WRITE))
+      return;
 
-  if (!check_fragment_space(mdr, destdn->get_dir()))
-    return;
+    if (!check_fragment_space(mdr, destdn->get_dir()))
+      return;
 
-  if (!check_access(mdr, srci, MAY_WRITE))
-    return;
+    if (!check_access(mdr, srci, MAY_WRITE))
+      return;
+  }
 
   // with read lock, really verify oldin is empty
   if (oldin &&