librados: add read to c object operation api

Do the usual bufferlist to buffer conversion in a callback from the
objecter before the librados user gets called.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index 20bb6bf5fdb79e27b42aa37a9574cbb375f30129..bd97bc71be01d64e308b6909a1666f5bebd88b31 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -240,6 +240,7 @@ typedef void *rados_write_op_t;
  * executed atomically. For usage, see:
  * - Creation and deletion: rados_create_read_op() rados_release_read_op()
  * - Object properties: rados_read_op_stat(), rados_read_op_assert_exists()
+ * - IO on objects: rados_read_op_read()
  * - Request properties: rados_read_op_set_flags()
  * - Performing the operation: rados_read_op_operate(),
  *   rados_aio_read_op_operate()
@@ -1964,6 +1965,28 @@ void rados_read_op_stat(rados_read_op_t read_op,
                        time_t *pmtime,
                        int *prval);
 
+/**
+ * Read bytes from offset into buffer.
+ *
+ * prlen will be filled with the number of bytes read if successful.
+ * A short read can only occur if the read reaches the end of the
+ * object.
+ *
+ * @param read_op operation to add this action to
+ * @param offset offset to read from
+ * @param buffer where to put the data
+ * @param len length of buffer
+ * @param prval where to store the return value of this action
+ * @param bytes_read where to store the number of bytes read by this action
+ */
+void rados_read_op_read(rados_read_op_t read_op,
+                       uint64_t offset,
+                       size_t len,
+                       char *buf,
+                       size_t *bytes_read,
+                       int *prval);
+
+/**
 
 /**
  * Perform a write operation synchronously

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index 90fe2927f80ba8e70bd9c37891daf92481de5790..8dfc91315afaf774ceeb7e64bf3b035109f3e2d6 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -3208,6 +3208,45 @@ extern "C" void rados_read_op_stat(rados_read_op_t read_op,
   ((::ObjectOperation *)read_op)->stat(psize, pmtime, prval);
 }
 
+class C_bl_to_buf : public Context {
+  char *out_buf;
+  size_t out_len;
+  size_t *bytes_read;
+  int *prval;
+public:
+  bufferlist out_bl;
+  C_bl_to_buf(char *out_buf,
+             size_t out_len,
+             size_t *bytes_read,
+             int *prval) : out_buf(out_buf), out_len(out_len),
+                           bytes_read(bytes_read), prval(prval) {}
+  void finish(int r) {
+    if (out_bl.length() > out_len) {
+      if (prval)
+       *prval = -ERANGE;
+      if (bytes_read)
+       *bytes_read = 0;
+      return;
+    }
+    if (bytes_read)
+      *bytes_read = out_bl.length();
+    if (out_buf && out_bl.c_str() != out_buf)
+      out_bl.copy(0, out_bl.length(), out_buf);
+  }
+};
+
+extern "C" void rados_read_op_read(rados_read_op_t read_op,
+                                  uint64_t offset,
+                                  size_t len,
+                                  char *buf,
+                                  size_t *bytes_read,
+                                  int *prval)
+{
+  C_bl_to_buf *ctx = new C_bl_to_buf(buf, len, bytes_read, prval);
+  ctx->out_bl.push_back(buffer::create_static(len, buf));
+  ((::ObjectOperation *)read_op)->read(offset, len, &ctx->out_bl, prval, ctx);
+}
+
 extern "C" int rados_read_op_operate(rados_read_op_t read_op,
                                     rados_ioctx_t io,
                                     const char *oid,

diff --git a/src/test/librados/c_read_operations.cc b/src/test/librados/c_read_operations.cc
index 13c03b486b41e386d13c6f5b47b0b0c734fc36da..bec72f5a7ead79c2935e171b2aba26e5cbc22ce1 100644 (file)
--- a/src/test/librados/c_read_operations.cc
+++ b/src/test/librados/c_read_operations.cc
@@ -75,6 +75,104 @@ TEST_F(CReadOpsTest, AssertExists) {
   remove_object();
 }
 
+TEST_F(CReadOpsTest, Read) {
+  write_object();
+
+  char buf[len];
+  // check that using read_ops returns the same data with
+  // or without bytes_read and rval out params
+  {
+    rados_read_op_t op = rados_create_read_op();
+    rados_read_op_read(op, 0, len, buf, NULL, NULL);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    int rval;
+    rados_read_op_read(op, 0, len, buf, NULL, &rval);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(0, rval);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    size_t bytes_read = 0;
+    rados_read_op_read(op, 0, len, buf, &bytes_read, NULL);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(len, (int)bytes_read);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    size_t bytes_read = 0;
+    int rval;
+    rados_read_op_read(op, 0, len, buf, &bytes_read, &rval);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(len, (int)bytes_read);
+    ASSERT_EQ(0, rval);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  remove_object();
+}
+
+TEST_F(CReadOpsTest, ShortRead) {
+  write_object();
+
+  char buf[len * 2];
+  // check that using read_ops returns the same data with
+  // or without bytes_read and rval out params
+  {
+    rados_read_op_t op = rados_create_read_op();
+    rados_read_op_read(op, 0, len * 2, buf, NULL, NULL);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    int rval;
+    rados_read_op_read(op, 0, len * 2, buf, NULL, &rval);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(0, rval);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    size_t bytes_read = 0;
+    rados_read_op_read(op, 0, len * 2, buf, &bytes_read, NULL);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(len, (int)bytes_read);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  {
+    rados_read_op_t op = rados_create_read_op();
+    size_t bytes_read = 0;
+    int rval;
+    rados_read_op_read(op, 0, len * 2, buf, &bytes_read, &rval);
+    ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+    ASSERT_EQ(len, (int)bytes_read);
+    ASSERT_EQ(0, rval);
+    ASSERT_EQ(0, memcmp(data, buf, len));
+    rados_release_read_op(op);
+  }
+
+  remove_object();
+}
+
 TEST_F(CReadOpsTest, Stat) {
   rados_read_op_t op = rados_create_read_op();
   uint64_t size = 1;