Merge PR ceph#55144 into main

* refs/pull/55144/head:
	client: fix file cache cap leak which can stall async read call
	test/client: test contiguous read for a non-contiguous write

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Xiubo Li <[email protected]>

Author: vshankar

src/client/Client.cc

@@ -11110,16 +11110,24 @@ int Client::_read_async(Fh *f, uint64_t off, uint64_t len, bufferlist *bl,
 
   ldout(cct, 10) << __func__ << " " << *in << " " << off << "~" << len << dendl;
 
+  // get Fc cap ref before commencing read
+  get_cap_ref(in, CEPH_CAP_FILE_CACHE);
+
   if (onfinish != nullptr) {
     io_finish.reset(new C_Read_Async_Finisher(this, onfinish, f, in,
                                               f->pos, off, len));
   }
 
   // trim read based on file size?
   if ((off >= in->size) || (len == 0)) {
+    // read is requested at the EOF or the read len is zero, therefore release
+    // Fc cap first before proceeding further
+    put_cap_ref(in, CEPH_CAP_FILE_CACHE);
+
     // If not async, immediate return of 0 bytes
-    if (onfinish == nullptr) 
+    if (onfinish == nullptr) {
       return 0;
+    }
 
     // Release C_Read_Async_Finisher from managed pointer, we need to complete
     // immediately. The C_Read_Async_Finisher is safely handled and won't be
@@ -11152,6 +11160,8 @@ int Client::_read_async(Fh *f, uint64_t off, uint64_t len, bufferlist *bl,
 			      off, len, bl, 0, io_finish.get());
 
   if (onfinish != nullptr) {
+    // put the cap ref since we're releasing C_Read_Async_Finisher
+    put_cap_ref(in, CEPH_CAP_FILE_CACHE);
     // Release C_Read_Async_Finisher from managed pointer, either
     // file_read will result in non-blocking complete, or we need to complete
     // immediately. In either case, the C_Read_Async_Finisher is safely
@@ -11160,19 +11170,19 @@ int Client::_read_async(Fh *f, uint64_t off, uint64_t len, bufferlist *bl,
     if (r != 0) {
       // need to do readahead, so complete the crf
       crf->complete(r);
-    } else {
-      get_cap_ref(in, CEPH_CAP_FILE_CACHE);
     }
     return 0;
   }
 
+  // Wait for the blocking read to complete and then do readahead
   if (r == 0) {
-    get_cap_ref(in, CEPH_CAP_FILE_CACHE);
     client_lock.unlock();
     r = io_finish_cond->wait();
     client_lock.lock();
     put_cap_ref(in, CEPH_CAP_FILE_CACHE);
     update_read_io_size(bl->length());
+  } else {
+    put_cap_ref(in, CEPH_CAP_FILE_CACHE);
   }
 
   do_readahead(f, in, off, len);

src/test/client/nonblocking.cc

@@ -611,3 +611,104 @@ TEST_F(TestClient, LlreadvLlwritevInvalidFileHandle) {
   ASSERT_EQ(bytes_read, -CEPHFS_EBADF);
   ASSERT_EQ(bl.length(), 0);
 }
+
+TEST_F(TestClient, LlreadvContiguousLlwritevNonContiguous) {
+  /* Test writing at non-contiguous memory locations, and make sure
+  contiguous read returns bytes requested. */
+
+  int mypid = getpid();
+  char filename[256];
+
+  client->unmount();
+  TearDown();
+  SetUp();
+
+  sprintf(filename, "test_llreadvcontiguousllwritevnoncontiguousfile%u", mypid);
+
+  Inode *root, *file;
+  root = client->get_root();
+  ASSERT_NE(root, (Inode *)NULL);
+
+  Fh *fh;
+  struct ceph_statx stx;
+
+  ASSERT_EQ(0, client->ll_createx(root, filename, 0666,
+				  O_RDWR | O_CREAT | O_TRUNC,
+				  &file, &fh, &stx, 0, 0, myperm));
+
+  const int NUM_BUF = 5;
+  char out_buf_0[] = "hello ";
+  char out_buf_1[] = "world\n";
+  char out_buf_2[] = "Ceph - ";
+  char out_buf_3[] = "a scalable distributed ";
+  char out_buf_4[] = "storage system\n";
+
+  struct iovec iov_out_non_contiguous[NUM_BUF] = {
+    {out_buf_0, sizeof(out_buf_0)},
+    {out_buf_1, sizeof(out_buf_1)},
+    {out_buf_2, sizeof(out_buf_2)},
+    {out_buf_3, sizeof(out_buf_3)},
+    {out_buf_4, sizeof(out_buf_4)}
+  };
+
+  char in_buf_0[sizeof(out_buf_0)];
+  char in_buf_1[sizeof(out_buf_1)];
+  char in_buf_2[sizeof(out_buf_2)];
+  char in_buf_3[sizeof(out_buf_3)];
+  char in_buf_4[sizeof(out_buf_4)];
+
+  struct iovec iov_in_contiguous[NUM_BUF] = {
+    {in_buf_0, sizeof(in_buf_0)},
+    {in_buf_1, sizeof(in_buf_1)},
+    {in_buf_2, sizeof(in_buf_2)},
+    {in_buf_3, sizeof(in_buf_3)},
+    {in_buf_4, sizeof(in_buf_4)}
+  };
+
+  ssize_t bytes_to_write = 0, total_bytes_written = 0, total_bytes_read = 0;
+  for(int i = 0; i < NUM_BUF; ++i) {
+    bytes_to_write += iov_out_non_contiguous[i].iov_len;
+  }
+
+  std::unique_ptr<C_SaferCond> writefinish = nullptr;
+  std::unique_ptr<C_SaferCond> readfinish = nullptr;
+
+  int64_t rc;
+  bufferlist bl;
+
+  struct iovec *current_iov = iov_out_non_contiguous;
+
+  for(int i = 0; i < NUM_BUF; ++i) {
+    writefinish.reset(new C_SaferCond("test-nonblocking-writefinish-non-contiguous"));
+    rc = client->ll_preadv_pwritev(fh, current_iov++, 1, i * NUM_BUF * 100,
+                                   true, writefinish.get(), nullptr);
+    ASSERT_EQ(rc, 0);
+    total_bytes_written += writefinish->wait();
+  }
+  ASSERT_EQ(total_bytes_written, bytes_to_write);
+
+  readfinish.reset(new C_SaferCond("test-nonblocking-readfinish-contiguous"));
+  rc = client->ll_preadv_pwritev(fh, iov_in_contiguous, NUM_BUF, 0, false,
+                                 readfinish.get(), &bl);
+  ASSERT_EQ(rc, 0);
+  total_bytes_read = readfinish->wait();
+  ASSERT_EQ(total_bytes_read, bytes_to_write);
+  ASSERT_EQ(bl.length(), bytes_to_write);
+
+  copy_bufferlist_to_iovec(iov_in_contiguous, NUM_BUF, &bl,
+                           total_bytes_read);
+  /* since the iovec structures are written at gaps of 100, only the first
+  iovec structure content should match when reading contiguously while rest
+  of the read buffers should just be 0s(holes filled with zeros) */
+  ASSERT_EQ(0, strncmp((const char*)iov_in_contiguous[0].iov_base,
+                       (const char*)iov_out_non_contiguous[0].iov_base,
+                       iov_out_non_contiguous[0].iov_len));
+  for(int i = 1; i < NUM_BUF; ++i) {
+    ASSERT_NE(0, strncmp((const char*)iov_in_contiguous[i].iov_base,
+                         (const char*)iov_out_non_contiguous[i].iov_base,
+                         iov_out_non_contiguous[i].iov_len));
+  }
+
+  client->ll_release(fh);
+  ASSERT_EQ(0, client->ll_unlink(root, filename, myperm));        
+}