mds: always make getattr wait for xlock to be released by the previous client

When the previous client's setattr request is still holding the xlock
for the linklock/authlock/xattrlock/filelock locks, if the same client
send a getattr request it will use the projected inode to fill the
reply, while for other clients the getattr requests will use the
non-projected inode to fill replies. This causes inconsistent file
mode across multiple clients.

This will just skip batching the ops when any of the xlock is held.

Fixes: https://tracker.ceph.com/issues/63906
Signed-off-by: Xiubo Li <[email protected]>

Author: lxbsz

src/mds/Server.cc

@@ -4085,6 +4085,7 @@ CDir* Server::try_open_auth_dirfrag(CInode *diri, frag_t fg, const MDRequestRef&
 void Server::handle_client_getattr(const MDRequestRef& mdr, bool is_lookup)
 {
   const cref_t<MClientRequest> &req = mdr->client_request;
+  client_t client = mdr->get_client();
 
   if (req->get_filepath().depth() == 0 && is_lookup) {
     // refpath can't be empty for lookup but it can for
@@ -4106,6 +4107,17 @@ void Server::handle_client_getattr(const MDRequestRef& mdr, bool is_lookup)
     if (r > 0)
       return; // delayed
 
+    // Do not batch if any xlock is held
+    if (!r) {
+      CInode *in = mdr->in[0];
+      if (((mask & CEPH_CAP_LINK_SHARED) && (in->linklock.is_xlocked_by_client(client))) ||
+          ((mask & CEPH_CAP_AUTH_SHARED) && (in->authlock.is_xlocked_by_client(client))) ||
+          ((mask & CEPH_CAP_XATTR_SHARED) && (in->xattrlock.is_xlocked_by_client(client))) ||
+          ((mask & CEPH_CAP_FILE_SHARED) && (in->filelock.is_xlocked_by_client(client)))) {
+	r = -1;
+      }
+    }
+
     if (r < 0) {
       // fall-thru. let rdlock_path_pin_ref() check again.
     } else if (is_lookup) {
@@ -4147,7 +4159,6 @@ void Server::handle_client_getattr(const MDRequestRef& mdr, bool is_lookup)
    * handling this case here is easier than weakening rdlock
    * semantics... that would cause problems elsewhere.
    */
-  client_t client = mdr->get_client();
   int issued = 0;
   Capability *cap = ref->get_client_cap(client);
   if (cap && (mdr->snapid == CEPH_NOSNAP ||