Merge PR ceph#59300 into main

* refs/pull/59300/head:
	client: calls to _ll_fh_exists() should hold client_lock

Reviewed-by: Xiubo Li <xiubli@redhat.com>
Reviewed-by: Dhairya Parmar <dparmar@redhat.com>

Author: vshankar

src/client/Client.cc

@@ -15836,6 +15836,10 @@ int Client::ll_read(Fh *fh, loff_t off, loff_t len, bufferlist *bl)
     return -CEPHFS_ENOTCONN;
   }
 
+  /* We can't return bytes written larger than INT_MAX, clamp len to that */
+  len = std::min(len, (loff_t)INT_MAX);
+
+  std::scoped_lock lock(client_lock);
   if (fh == NULL || !_ll_fh_exists(fh)) {
     ldout(cct, 3) << "(fh)" << fh << " is invalid" << dendl;
     return -CEPHFS_EBADF;
@@ -15847,10 +15851,6 @@ int Client::ll_read(Fh *fh, loff_t off, loff_t len, bufferlist *bl)
   tout(cct) << off << std::endl;
   tout(cct) << len << std::endl;
 
-  /* We can't return bytes written larger than INT_MAX, clamp len to that */
-  len = std::min(len, (loff_t)INT_MAX);
-  std::scoped_lock lock(client_lock);
-
   int r = _read(fh, off, len, bl);
   ldout(cct, 3) << "ll_read " << fh << " " << off << "~" << len << " = " << r
 		<< dendl;
@@ -15981,6 +15981,10 @@ int Client::ll_write(Fh *fh, loff_t off, loff_t len, const char *data)
     return -CEPHFS_ENOTCONN;
   }
 
+  /* We can't return bytes written larger than INT_MAX, clamp len to that */
+  len = std::min(len, (loff_t)INT_MAX);
+
+  std::scoped_lock lock(client_lock);
   if (fh == NULL || !_ll_fh_exists(fh)) {
     ldout(cct, 3) << "(fh)" << fh << " is invalid" << dendl;
     return -CEPHFS_EBADF;
@@ -15993,10 +15997,6 @@ int Client::ll_write(Fh *fh, loff_t off, loff_t len, const char *data)
   tout(cct) << off << std::endl;
   tout(cct) << len << std::endl;
 
-  /* We can't return bytes written larger than INT_MAX, clamp len to that */
-  len = std::min(len, (loff_t)INT_MAX);
-  std::scoped_lock lock(client_lock);
-
   int r = _write(fh, off, len, data, NULL, 0);
   ldout(cct, 3) << "ll_write " << fh << " " << off << "~" << len << " = " << r
 		<< dendl;
@@ -16010,12 +16010,11 @@ int64_t Client::ll_writev(struct Fh *fh, const struct iovec *iov, int iovcnt, in
     return -CEPHFS_ENOTCONN;
   }
 
+  std::scoped_lock cl(client_lock);
   if (fh == NULL || !_ll_fh_exists(fh)) {
     ldout(cct, 3) << "(fh)" << fh << " is invalid" << dendl;
     return -CEPHFS_EBADF;
   }
-
-  std::scoped_lock cl(client_lock);
   return _preadv_pwritev_locked(fh, iov, iovcnt, off, true, false);
 }
 
@@ -16026,12 +16025,11 @@ int64_t Client::ll_readv(struct Fh *fh, const struct iovec *iov, int iovcnt, int
     return -CEPHFS_ENOTCONN;
   }
 
+  std::scoped_lock cl(client_lock);
   if (fh == NULL || !_ll_fh_exists(fh)) {
     ldout(cct, 3) << "(fh)" << fh << " is invalid" << dendl;
     return -CEPHFS_EBADF;
   }
-
-  std::scoped_lock cl(client_lock);
   return _preadv_pwritev_locked(fh, iov, iovcnt, off, false, false);
 }
 
@@ -16054,18 +16052,24 @@ int64_t Client::ll_preadv_pwritev(struct Fh *fh, const struct iovec *iov,
       return retval;
     }
 
+    retval = 0;
+    std::unique_lock cl(client_lock);
+
     if(fh == NULL || !_ll_fh_exists(fh)) {
       ldout(cct, 3) << "(fh)" << fh << " is invalid" << dendl;
       retval = -CEPHFS_EBADF;
+    }
+
+    if (retval != 0) {
       if (onfinish != nullptr) {
+        cl.unlock();
         onfinish->complete(retval);
+        cl.lock();
         retval = 0;
       }
       return retval;
     }
 
-    std::scoped_lock cl(client_lock);
-
     retval = _preadv_pwritev_locked(fh, iov, iovcnt, offset, write, true,
                                     onfinish, bl, do_fsync, syncdataonly);
     /* There are two scenarios with each having two cases to handle here
@@ -16086,9 +16090,9 @@ int64_t Client::ll_preadv_pwritev(struct Fh *fh, const struct iovec *iov,
     if (retval < 0) {
       if (onfinish != nullptr) {
         //async io failed
-        client_lock.unlock();
+        cl.unlock();
         onfinish->complete(retval);
-        client_lock.lock();
+        cl.lock();
         /* async call should always return zero to caller and allow the
         caller to wait on callback for the actual errno/retval. */
         retval = 0;