librgw: try-implement rgw_readv/rgw_writev

The model in the rgw_readv call is the caller owns the struct rgw_uio,
and uses it to pass uio_offset.

The caller returns the rgw_uio filled out, and provides indirectly
rgw_uio->uio_vio (the full private data is at uio->uio_p1 on return,
the caller (p==private) must not touch it.  The caller returns
private data in rgw_readv_release() when finished.

Meanwhile rgw_writev is atomic.

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

1 files changed, 85 insertions, 12 deletions

diff --git a/src/test/librgw_file_gp.cc b/src/test/librgw_file_gp.cc
index 4b9fa9b28cf..6de26aaa36e 100644
--- a/src/test/librgw_file_gp.cc
+++ b/src/test/librgw_file_gp.cc
@@ -40,6 +40,8 @@ namespace {
   bool do_pre_list = false;
   bool do_put = false;
   bool do_bulk = false;
+  bool do_writev = false;
+  bool do_readv = false;
   bool do_get = false;
   bool do_delete = false;
 
@@ -54,10 +56,13 @@ namespace {
 
   std::uniform_int_distribution<uint8_t> uint_dist;
   std::mt19937 rng;
-  
+
+  constexpr int iovcnt = 16;
+  constexpr int page_size = 65536;
+
   struct ZPage
   {
-    char data[65536];
+    char data[page_size];
     uint64_t cksum;
   }; /* ZPage */
   
@@ -71,15 +76,15 @@ namespace {
       iovs = (struct iovec*) calloc(n, sizeof(struct iovec));
       for (int page_ix = 0; page_ix < n; ++page_ix) {
 	ZPage* p = new ZPage();
-	for (int data_ix = 0; data_ix < 65536; ++data_ix) {
+	for (int data_ix = 0; data_ix < page_size; ++data_ix) {
 	  p->data[data_ix] = uint_dist(rng);
 	} // data_ix
-	p->cksum = XXH64(p->data, 65536, 8675309);
+	p->cksum = XXH64(p->data, page_size, 8675309);
 	pages.emplace_back(p);
 	// and iovs
 	struct iovec* iov = &iovs[page_ix];
 	iov->iov_base = p->data;
-	iov->iov_len = 65536;
+	iov->iov_len = page_size;
       } // page_ix
     }
 
@@ -102,7 +107,7 @@ namespace {
       int n = size();
       for (int page_ix = 0; page_ix < n; ++page_ix) {
 	ZPage* p = pages[page_ix];
-	p->cksum = XXH64(p->data, 65536, 8675309);
+	p->cksum = XXH64(p->data, page_size, 8675309);
       }
     }
 
@@ -112,7 +117,7 @@ namespace {
 	ZPage* p = pages[page_ix];
 	struct iovec* iov = &iovs[page_ix];
 	iov->iov_base = p->data;
-	iov->iov_len = 65536;
+	iov->iov_len = page_size;
       }
     }
 
@@ -216,23 +221,85 @@ TEST(LibRGW, GET_OBJECT) {
 
 TEST(LibRGW, WRITE_READ_VERIFY)
 {
-  if (do_bulk) {
-    const int iovcnt = 16;
+  if (do_bulk && do_put) {
     ZPageSet zp_set1{iovcnt}; // 1M random data in 16 64K pages
     struct iovec *iovs = zp_set1.get_iovs();
 
     /* read after write POSIX-style */
     size_t nbytes, off = 0;
-    for (int ix = 0; ix < 16; ++ix, off += 65536) {
+    for (int ix = 0; ix < 16; ++ix, off += page_size) {
       struct iovec *iov = &iovs[ix];
-      int ret = rgw_write(fs, object_fh, off, 65536, &nbytes, iov->iov_base);
+      int ret = rgw_write(fs, object_fh, off, page_size, &nbytes,
+			  iov->iov_base);
       ASSERT_EQ(ret, 0);
-      ASSERT_EQ(nbytes, size_t(65536));
+      ASSERT_EQ(nbytes, size_t(page_size));
     }
     zp_set1.reset_iovs();
   }
 }
 
+/* "functions that call alloca are not inlined"
+ * --alexandre oliva
+ * http://gcc.gnu.org/ml/gcc-help/2004-04/msg00158.html
+ */
+#define alloca_uio()				\
+  do {\
+    int uiosz = sizeof(rgw_uio) + iovcnt*sizeof(rgw_vio);		\
+    uio = static_cast<rgw_uio*>(alloca(uiosz));				\
+    memset(uio, 0, uiosz);						\
+    uio->uio_vio = reinterpret_cast<rgw_vio*>(uio+sizeof(rgw_uio));	\
+  } while (0);								\
+
+TEST(LibRGW, WRITEV)
+{
+  if (do_writev && do_put) {
+    rgw_uio* uio;
+    ZPageSet zp_set1{iovcnt}; // 1M random data in 16 64K pages
+    struct iovec *iovs = zp_set1.get_iovs();
+    alloca_uio();
+    ASSERT_NE(uio, nullptr);
+
+    for (int ix = 0; ix < iovcnt; ++ix) {
+      struct iovec *iov = &iovs[ix];
+      rgw_vio *vio = &(uio->uio_vio[ix]);
+      vio->vio_base = iov->iov_base;
+      vio->vio_len = iov->iov_len;
+      vio->vio_u1 = iov; // private data
+    }
+    uio->uio_cnt = iovcnt;
+    uio->uio_offset = iovcnt * page_size;
+
+    int ret = rgw_writev(fs, object_fh, uio);
+    ASSERT_EQ(ret, 0);
+    //zp_set1.reset_iovs();
+  }
+}
+
+TEST(LibRGW, READV)
+{
+  if (do_readv && do_get) {
+    rgw_uio uio[1];
+    memset(uio, 0, sizeof(rgw_uio));
+    uio->uio_offset = 0; // ok, it was already 0
+    int ret = rgw_readv(fs, object_fh, uio);
+    ASSERT_EQ(ret, 0);
+    //buffer::list bl;
+    buffer::list& bl = *(new buffer::list());
+    for (unsigned int ix = 0; ix < uio->uio_cnt; ++ix) {
+      rgw_vio *vio = &(uio->uio_vio[ix]);
+      bl.push_back(
+	buffer::create_static(vio->vio_len,
+			      static_cast<char*>(vio->vio_base)));
+    }
+    bl.hexdump(std::cout);
+    // release resources
+    ASSERT_NE(uio->uio_rele, nullptr);
+    if (uio->uio_rele) {
+      uio->uio_rele(uio, RGW_UIO_NONE);
+    }
+  }
+}
+
 TEST(LibRGW, DELETE_OBJECT) {
   if (do_delete) {
     int ret = rgw_unlink(fs, bucket_fh, object_name.c_str());
@@ -302,6 +369,12 @@ int main(int argc, char *argv[])
     } else if (ceph_argparse_flag(args, arg_iter, "--bulk",
 					    (char*) nullptr)) {
       do_bulk = true;
+    } else if (ceph_argparse_flag(args, arg_iter, "--writev",
+					    (char*) nullptr)) {
+      do_writev = true;
+    } else if (ceph_argparse_flag(args, arg_iter, "--readv",
+					    (char*) nullptr)) {
+      do_readv = true;
     } else if (ceph_argparse_flag(args, arg_iter, "--delete",
 					    (char*) nullptr)) {
       do_delete = true;