librados: add simple atomic read operations to c api

stat, assert_exists, and set_flags are simple and need no extra
infrastrucutre.

Signed-off-by: Josh Durgin <[email protected]>

src/include/rados/librados.h

@@ -233,6 +233,19 @@ struct rados_cluster_stat_t {
  */
 typedef void *rados_write_op_t;
 
+/**
+ * @typedef rados_read_op_t
+ *
+ * An object read operation stores a number of operations which can be
+ * 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()
+ * - Request properties: rados_read_op_set_flags()
+ * - Performing the operation: rados_read_op_operate(),
+ *   rados_aio_read_op_operate()
+ */
+typedef void *rados_read_op_t;
+
 /**
  * Get the version of librados.
  *
@@ -1909,6 +1922,74 @@ int rados_aio_write_op_operate(rados_write_op_t write_op,
                                time_t *mtime,
 			       int flags);
 
+/**
+ * Create a new rados_read_op_t write operation. This will store all
+ * actions to be performed atomically. You must call
+ * rados_release_read_op when you are finished with it (after it
+ * completes, or you decide not to send it in the first place).
+ *
+ * @returns non-NULL on success, NULL on memory allocation error.
+ */
+rados_read_op_t rados_create_read_op();
+
+/**
+ * Free a rados_read_op_t, must be called when you're done with it.
+ * @param read_op operation to deallocate, created with rados_create_read_op
+ */
+void rados_release_read_op(rados_read_op_t read_op);
+
+/**
+ * Set flags for the last operation added to this read_op.
+ * At least one op must have been added to the read_op.
+ * @param flags see librados.h constants beginning with LIBRADOS_OP_FLAG
+ */
+void rados_read_op_set_flags(rados_read_op_t read_op, int flags);
+
+/**
+ * Ensure that the object exists before reading
+ * @param read_op operation to add this action to
+ */
+void rados_read_op_assert_exists(rados_read_op_t read_op);
+
+
+/**
+ * Get object size and mtime
+ * @param read_op operation to add this action to
+ * @param psize where to store object size
+ * @param pmtime where to store modification time
+ * @param prval where to store the return value of this action
+ */
+void rados_read_op_stat(rados_read_op_t read_op,
+			uint64_t *psize,
+			time_t *pmtime,
+			int *prval);
+
+
+/**
+ * Perform a write operation synchronously
+ * @param read_op operation to perform
+ * @io the ioctx that the object is in
+ * @oid the object id
+ * @flags flags to apply to the entire operation (LIBRADOS_OPERATION_*)
+ */
+int rados_read_op_operate(rados_read_op_t read_op,
+			  rados_ioctx_t io,
+			  const char *oid,
+			  int flags);
+
+/**
+ * Perform a write operation asynchronously
+ * @param read_op operation to perform
+ * @io the ioctx that the object is in
+ * @param completion what to do when operation has been attempted
+ * @oid the object id
+ * @flags flags to apply to the entire operation (LIBRADOS_OPERATION_*)
+ */
+int rados_aio_read_op_operate(rados_read_op_t read_op,
+			      rados_ioctx_t io,
+			      rados_completion_t completion,
+			      const char *oid,
+			      int flags);
 
 /** @} Object Operations */
 

src/librados/librados.cc

@@ -3179,4 +3179,54 @@ extern "C" int rados_aio_write_op_operate(rados_write_op_t write_op,
   librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
   return ctx->aio_operate(obj, oo, c, ctx->snapc, flags);
 }
+
+extern "C" rados_read_op_t rados_create_read_op()
+{
+  return new (std::nothrow)::ObjectOperation;
+}
+
+extern "C" void rados_release_read_op(rados_read_op_t read_op)
+{
+  delete (::ObjectOperation *)read_op;
+}
+
+extern "C" void rados_read_op_set_flags(rados_read_op_t read_op, int flags)
+{
+  set_op_flags((::ObjectOperation *)read_op, flags);
+}
+
+extern "C" void rados_read_op_assert_exists(rados_read_op_t read_op)
+{
+  ((::ObjectOperation *)read_op)->stat(NULL, (utime_t *)NULL, NULL);
+}
+
+extern "C" void rados_read_op_stat(rados_read_op_t read_op,
+				   uint64_t *psize,
+				   time_t *pmtime,
+				   int *prval)
+{
+  ((::ObjectOperation *)read_op)->stat(psize, pmtime, prval);
+}
+
+extern "C" int rados_read_op_operate(rados_read_op_t read_op,
+				     rados_ioctx_t io,
+				     const char *oid,
+				     int flags)
+{
+  object_t obj(oid);
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  return ctx->operate_read(obj, (::ObjectOperation *)read_op, NULL, flags);
+}
+
+extern "C" int rados_aio_read_op_operate(rados_read_op_t read_op,
+					 rados_ioctx_t io,
+					 rados_completion_t completion,
+					 const char *oid,
+					 int flags)
+{
+  object_t obj(oid);
+  librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+  librados::AioCompletionImpl *c = (librados::AioCompletionImpl*)completion;
+  return ctx->aio_operate_read(obj, (::ObjectOperation *)read_op,
+			       c, flags, NULL);
 }

src/test/Makefile.am

@@ -703,6 +703,12 @@ ceph_test_rados_api_c_write_operations_LDADD = $(LIBRADOS) $(UNITTEST_LDADD) $(R
 ceph_test_rados_api_c_write_operations_CXXFLAGS = $(UNITTEST_CXXFLAGS)
 bin_DEBUGPROGRAMS += ceph_test_rados_api_c_write_operations
 
+ceph_test_rados_api_c_read_operations_SOURCES = \
+	test/librados/c_read_operations.cc
+ceph_test_rados_api_c_read_operations_LDADD = $(LIBRADOS) $(UNITTEST_LDADD) $(RADOS_TEST_LDADD)
+ceph_test_rados_api_c_read_operations_CXXFLAGS = $(UNITTEST_CXXFLAGS)
+bin_DEBUGPROGRAMS += ceph_test_rados_api_c_read_operations
+
 ceph_test_rados_api_aio_SOURCES = test/librados/aio.cc
 ceph_test_rados_api_aio_LDADD = $(LIBRADOS) $(UNITTEST_LDADD) $(RADOS_TEST_LDADD)
 ceph_test_rados_api_aio_CXXFLAGS = $(UNITTEST_CXXFLAGS)

src/test/librados/c_read_operations.cc

@@ -0,0 +1,108 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// Tests for the C API coverage of atomic read operations
+
+#include <errno.h>
+#include <string>
+
+#include "include/rados/librados.h"
+#include "test/librados/test.h"
+#include "test/librados/TestCase.h"
+
+const char *data = "testdata";
+const char *obj = "testobj";
+const int len = strlen(data);
+
+class CReadOpsTest : public RadosTest {
+protected:
+  void write_object() {
+    // Create an object and write to it
+    ASSERT_EQ(len, rados_write(ioctx, obj, data, len, 0));
+  }
+  void remove_object() {
+    ASSERT_EQ(0, rados_remove(ioctx, obj));
+  }
+};
+
+TEST_F(CReadOpsTest, NewDelete) {
+  rados_read_op_t op = rados_create_read_op();
+  ASSERT_TRUE(op);
+  rados_release_read_op(op);
+}
+
+TEST_F(CReadOpsTest, SetOpFlags) {
+  write_object();
+
+  rados_read_op_t op = rados_create_read_op();
+  size_t bytes_read = 0;
+  char *out = NULL;
+  int rval = 0;
+  rados_read_op_exec(op, "rbd", "get_id", NULL, 0, &out,
+		     &bytes_read, &rval);
+  rados_read_op_set_flags(op, LIBRADOS_OP_FLAG_FAILOK);
+  EXPECT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+  EXPECT_EQ(0, rval);
+  EXPECT_EQ(0u, bytes_read);
+  EXPECT_EQ((char*)NULL, out);
+  rados_release_read_op(op);
+
+  remove_object();
+}
+
+TEST_F(CReadOpsTest, AssertExists) {
+  rados_read_op_t op = rados_create_read_op();
+  rados_read_op_assert_exists(op);
+
+  ASSERT_EQ(-ENOENT, rados_read_op_operate(op, ioctx, obj, 0));
+  rados_release_read_op(op);
+
+  op = rados_create_read_op();
+  rados_read_op_assert_exists(op);
+
+  rados_completion_t completion;
+  ASSERT_EQ(0, rados_aio_create_completion(NULL, NULL, NULL, &completion));
+  ASSERT_EQ(0, rados_aio_read_op_operate(op, ioctx, completion, obj, 0));
+  rados_aio_wait_for_complete(completion);
+  ASSERT_EQ(-ENOENT, rados_aio_get_return_value(completion));
+  rados_release_read_op(op);
+
+  write_object();
+
+  op = rados_create_read_op();
+  rados_read_op_assert_exists(op);
+  ASSERT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+  rados_release_read_op(op);
+
+  remove_object();
+}
+
+TEST_F(CReadOpsTest, Stat) {
+  rados_read_op_t op = rados_create_read_op();
+  uint64_t size = 1;
+  int rval;
+  rados_read_op_stat(op, &size, NULL, &rval);
+  EXPECT_EQ(-ENOENT, rados_read_op_operate(op, ioctx, obj, 0));
+  EXPECT_EQ(-EIO, rval);
+  EXPECT_EQ(1u, size);
+  rados_release_read_op(op);
+
+  write_object();
+
+  op = rados_create_read_op();
+  rados_read_op_stat(op, &size, NULL, &rval);
+  EXPECT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+  EXPECT_EQ(0, rval);
+  EXPECT_EQ(len, (int)size);
+  rados_release_read_op(op);
+
+  op = rados_create_read_op();
+  rados_read_op_stat(op, NULL, NULL, NULL);
+  EXPECT_EQ(0, rados_read_op_operate(op, ioctx, obj, 0));
+  rados_release_read_op(op);
+
+  remove_object();
+
+  op = rados_create_read_op();
+  rados_read_op_stat(op, NULL, NULL, NULL);
+  EXPECT_EQ(-ENOENT, rados_read_op_operate(op, ioctx, obj, 0));
+  rados_release_read_op(op);
+}