librgw: fix RGWReadRequest send_response_data

In hindsight, it seems clear how to consume the available bytes and
account for the bytes written into the caller's buffer.

Verified by unit test on data from S3, with a 6M file read in 1M
chunks, the preferred max read in modern Linux kernel NFS.

Signed-off-by: Matt Benjamin <mbenjamin@redhat.com>

diff --git a/src/rgw/rgw_file.cc b/src/rgw/rgw_file.cc
index 72fb00e7805ef5c4d777375f2ba9ece41fbe0299..a4060a7f9209efc040547e0c1709b54c4c48bf5a 100644 (file)
--- a/src/rgw/rgw_file.cc
+++ b/src/rgw/rgw_file.cc
@@ -1140,6 +1140,7 @@ int rgw_write(struct rgw_fs *rgw_fs,
              uint32_t flags)
 {
   RGWFileHandle* rgw_fh = get_rgwfh(fh);
+  int rc;
 
   if (! rgw_fh->is_file())
     return -EISDIR;
@@ -1147,20 +1148,24 @@ int rgw_write(struct rgw_fs *rgw_fs,
   if (! rgw_fh->is_open())
     return -EPERM;
 
-  char tbuf[256];
-  size_t tlen = std::min(length, 255UL);
-  memset(tbuf, 0, 256);
-  memcpy(tbuf, buffer, tlen);
-  std::cout << __func__ << " " << length << " bytes"
-           << " at offset " << offset
-           << " {{" << tbuf << "}}" << std::endl;
-#if 1 /* XXXX buffer::create_static w/buffer corrupts data  */
-  string xxx_bogus{tbuf};
-  return rgw_fh->write(0, xxx_bogus.size(), bytes_written,
-                      const_cast<char*>(xxx_bogus.c_str()));
-#else
-  return rgw_fh->write(offset, length+1, bytes_written, tbuf);
-#endif
+  std::cout << __func__ << " before write of "
+           << length << " bytes at offset " << offset
+           << std::endl;
+
+  rc = rgw_fh->write(offset, length, bytes_written, buffer);
+
+  std::cout << __func__ << " after write of "
+           << length << " bytes at offset " << offset
+           << " wrote " << *bytes_written
+           << " rc " << rc
+           << std::endl;
+
+  std::string str;
+  str.reserve(length+1);
+  str.assign(static_cast<char*>(buffer), uint32_t(length));
+  str += '\0';
+
+  return rc;
 }
 
 /*

diff --git a/src/rgw/rgw_file.h b/src/rgw/rgw_file.h
index 79ab1c0cad9b5f4ffcc88bc20d7c6f69c96969bf..3f3d375c62478ff93afa286789f7a99a88357206 100644 (file)
--- a/src/rgw/rgw_file.h
+++ b/src/rgw/rgw_file.h
@@ -1334,14 +1334,14 @@ public:
   RGWFileHandle* rgw_fh;
   void *ulp_buffer;
   size_t nread;
-  size_t read_len;
+  size_t read_resid; /* initialize to len, <= sizeof(ulp_buffer) */
   bool do_hexdump = false;
 
   RGWReadRequest(CephContext* _cct, RGWUserInfo *_user,
                 RGWFileHandle* _rgw_fh, uint64_t off, uint64_t len,
                 void *_ulp_buffer)
     : RGWLibRequest(_cct, _user), rgw_fh(_rgw_fh), ulp_buffer(_ulp_buffer),
-      nread(0), read_len(len) {
+      nread(0), read_resid(len) {
     magic = 76;
     op = this;
 
@@ -1390,25 +1390,25 @@ public:
     return 0;
   }
 
-  virtual int send_response_data(ceph::buffer::list& bl, off_t s_off,
-                               off_t e_off) {
-    if (do_hexdump) {
-      dout(15) << __func__ << " s_off " << s_off
-              << " e_off " << e_off << " len " << bl.length()
-              << " ";
-      bl.hexdump(*_dout);
-      *_dout << dendl;
-    }
-    uint64_t off = 0;
+  virtual int send_response_data(ceph::buffer::list& bl, off_t bl_off,
+                               off_t bl_len) {
+    size_t bytes;
     for (auto& bp : bl.buffers()) {
-      if (nread >= read_len)
-       break;
-      size_t bytes = std::min(std::min(read_len, size_t(bp.length())),
-                             size_t(e_off));
-      memcpy(static_cast<char*>(ulp_buffer)+off, bp.c_str()+s_off, bytes);
+      /* if for some reason bl_off indicates the start-of-data is not at
+       * the current buffer::ptr, skip it and account */
+      if (bl_off > bp.length()) {
+       bl_off -= bp.length();
+       continue;
+      }
+      /* read no more than read_resid */
+      bytes = std::min(read_resid, size_t(bp.length()-bl_off));
+      memcpy(static_cast<char*>(ulp_buffer)+nread, bp.c_str()+bl_off, bytes);
+      read_resid -= bytes; /* reduce read_resid by bytes read */
       nread += bytes;
-      off += bytes;
-      s_off -= bytes;
+      bl_off = 0;
+      /* stop if we have no residual ulp_buffer */
+      if (! read_resid)
+       break;
     }
     return 0;
   }
@@ -1823,8 +1823,8 @@ public:
 
   virtual int get_data(buffer::list& _bl) {
     /* XXX for now, use sharing semantics */
+    uint32_t len = data.length();
     _bl.claim(data);
-    uint32_t len = _bl.length();
     bytes_written += len;
     return len;
   }

diff --git a/src/test/librgw_file_nfsns.cc b/src/test/librgw_file_nfsns.cc
index 27723d0928a5ffa10812f0d8cfdb8542dff011ef..f5970c826abd2850e987196d491e6a3f10ca087a 100644 (file)
--- a/src/test/librgw_file_nfsns.cc
+++ b/src/test/librgw_file_nfsns.cc
@@ -15,6 +15,7 @@
 #include <stdint.h>
 #include <tuple>
 #include <iostream>
+#include <fstream>
 #include <stack>
 
 #include "include/rados/librgw.h"
@@ -45,6 +46,8 @@ namespace {
 
   string bucket_name("nfsroot");
   string dirs1_bucket_name("bdirs1");
+  string readf_name("toyland");
+
   int n_dirs1_dirs = 3;
   int n_dirs1_objs = 2;
 
@@ -131,6 +134,7 @@ namespace {
 
   bool do_hier1 = false;
   bool do_dirs1 = false;
+  bool do_readf = false;
   bool do_marker1 = false;
   bool do_create = false;
   bool do_delete = false;
@@ -448,8 +452,7 @@ TEST(LibRGW, BAD_DELETES_DIRS1) {
       ASSERT_NE(rc, 0);
     }
     /* try to unlink a non-empty directory (non-bucket) */
-    obj_rec& sdir_0 = get<1>(dirs_vec[0])[0];
-    ASSERT_EQ(sdir_0.name, "sdir_0");
+    obj_rec& sdir_0 = get<1>(dirs_vec[0])[0];    ASSERT_EQ(sdir_0.name, "sdir_0");
     ASSERT_TRUE(sdir_0.rgw_fh->is_dir());
     /* XXX we can't enforce this currently */
 #if 0
@@ -533,6 +536,39 @@ TEST(LibRGW, READ_DIRS1)
   }
 }
 
+TEST(LibRGW, READF_DIRS1) {
+  if (do_dirs1) {
+    if (do_readf) {
+      obj_rec fobj{readf_name, nullptr, dirs1_b.fh, nullptr};
+
+      int rc = rgw_lookup(fs, dirs1_b.fh, fobj.name.c_str(), &fobj.fh,
+                         RGW_LOOKUP_FLAG_NONE);
+      ASSERT_EQ(rc, 0);
+      ASSERT_NE(fobj.fh, nullptr);
+      fobj.sync();
+
+      ofstream of;
+      of.open("rgwlib_readf.out", ios::out|ios::app|ios::binary);
+      int bufsz = 1024 * 1024 * sizeof(char);
+      char *buffer = (char*) malloc(bufsz);
+
+      uint64_t offset = 0;
+      uint64_t length = bufsz;
+      for (int ix = 0; ix < 6; ++ix) {
+       uint64_t nread = 0;
+       memset(buffer, 0, length); // XXX
+       rc = rgw_read(fs, fobj.fh, offset, length, &nread, buffer,
+                     RGW_READ_FLAG_NONE);
+       ASSERT_EQ(rc, 0);
+       ASSERT_EQ(nread, length);
+       of.write(buffer, length);
+       offset += nread;
+      }
+      of.close();
+    }
+  }
+}
+
 TEST(LibRGW, RELEASE_DIRS1) {
   if (do_dirs1) {
     /* force release of handles for children of dirs1--force subsequent
@@ -858,6 +894,9 @@ int main(int argc, char *argv[])
     } else if (ceph_argparse_flag(args, arg_iter, "--rename",
                                            (char*) nullptr)) {
       do_rename = true;
+    } else if (ceph_argparse_flag(args, arg_iter, "--readf",
+                                           (char*) nullptr)) {
+      do_readf = true;
     } else if (ceph_argparse_flag(args, arg_iter, "--verbose",
                                            (char*) nullptr)) {
       verbose = true;