librados: add methods to check and wait for safe/complete -and- callback

Allow user to check for (safe|complete) -and- callback return, and wait
for that condition.

Fixes: #2301
Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index fe6e17109517f5db68c4caff47be54325264325f..61bdf6d157e8e5d2fe5a981606dd9cae9901379b 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -1226,6 +1226,46 @@ int rados_aio_is_complete(rados_completion_t c);
  */
 int rados_aio_is_safe(rados_completion_t c);
 
+/**
+ * Block until an operation completes and callback completes
+ *
+ * This means it is in memory on all replicas and can be read.
+ *
+ * @note BUG: this should be void
+ *
+ * @param c operation to wait for
+ * @returns 0
+ */
+int rados_aio_wait_for_complete_and_cb(rados_completion_t c);
+
+/**
+ * Block until an operation is safe and callback has completed
+ *
+ * This means it is on stable storage on all replicas.
+ *
+ * @note BUG: this should be void
+ *
+ * @param c operation to wait for
+ * @returns 0
+ */
+int rados_aio_wait_for_safe_and_cb(rados_completion_t c);
+
+/**
+ * Has an asynchronous operation and callback completed
+ *
+ * @param c async operation to inspect
+ * @returns whether c is complete
+ */
+int rados_aio_is_complete_and_cb(rados_completion_t c);
+
+/**
+ * Is an asynchronous operation safe and has the callback completed
+ *
+ * @param c async operation to inspect
+ * @returns whether c is safe
+ */
+int rados_aio_is_safe_and_cb(rados_completion_t c);
+
 /**
  * Get the return value of an asychronous operation
  *

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index 89d3ea1404b4bd2f451675e9ca380b83c4d4ef58..64101481b63c42da4d477d32a8dd2055836fc61e 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -88,8 +88,12 @@ namespace librados
     int set_safe_callback(void *cb_arg, callback_t cb);
     int wait_for_complete();
     int wait_for_safe();
+    int wait_for_complete_and_cb();
+    int wait_for_safe_and_cb();
     bool is_complete();
     bool is_safe();
+    bool is_complete_and_cb();
+    bool is_safe_and_cb();
     int get_return_value();
     int get_version();
     void release();

diff --git a/src/librados/AioCompletionImpl.h b/src/librados/AioCompletionImpl.h
index a6fda7d08c4ad8cd2995e27a63a582ba4ae149d0..9db4f72e734546267b9400e71912db35ffcbac0b 100644 (file)
--- a/src/librados/AioCompletionImpl.h
+++ b/src/librados/AioCompletionImpl.h
@@ -92,6 +92,32 @@ struct librados::AioCompletionImpl {
     lock.Unlock();
     return r;
   }
+  int wait_for_complete_and_cb() {
+    lock.Lock();
+    while (!ack || callback_complete)
+      cond.Wait(lock);
+    lock.Unlock();
+    return 0;
+  }
+  int wait_for_safe_and_cb() {
+    lock.Lock();
+    while (!safe || callback_safe)
+      cond.Wait(lock);
+    lock.Unlock();
+    return 0;
+  }
+  int is_complete_and_cb() {
+    lock.Lock();
+    int r = ack && !callback_complete;
+    lock.Unlock();
+    return r;
+  }
+  int is_safe_and_cb() {
+    lock.Lock();
+    int r = safe && !callback_safe;
+    lock.Unlock();
+    return r;
+  }
   int get_return_value() {
     lock.Lock();
     int r = rval;
@@ -142,7 +168,11 @@ struct C_AioComplete : public Context {
     rados_callback_t cb = c->callback_complete;
     void *cb_arg = c->callback_arg;
     cb(c, cb_arg);
-    c->put();
+
+    c->lock.Lock();
+    c->callback_complete = NULL;
+    c->cond.Signal();
+    c->put_unlock();
   }
 };
 
@@ -157,7 +187,11 @@ struct C_AioSafe : public Context {
     rados_callback_t cb = c->callback_safe;
     void *cb_arg = c->callback_arg;
     cb(c, cb_arg);
-    c->put();
+
+    c->lock.Lock();
+    c->callback_safe = NULL;
+    c->cond.Signal();
+    c->put_unlock();
   }
 };
 

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index 0505842874aa1c2a4f74a197a9f7776e6ae29d4d..9fc2f4f2e5c175cabc1e7c8ec763c0bb1ab17a0d 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -460,6 +460,30 @@ bool librados::AioCompletion::AioCompletion::is_safe()
   return c->is_safe();
 }
 
+int librados::AioCompletion::AioCompletion::wait_for_complete_and_cb()
+{
+  AioCompletionImpl *c = (AioCompletionImpl *)pc;
+  return c->wait_for_complete_and_cb();
+}
+
+int librados::AioCompletion::AioCompletion::wait_for_safe_and_cb()
+{
+  AioCompletionImpl *c = (AioCompletionImpl *)pc;
+  return c->wait_for_safe_and_cb();
+}
+
+bool librados::AioCompletion::AioCompletion::is_complete_and_cb()
+{
+  AioCompletionImpl *c = (AioCompletionImpl *)pc;
+  return c->is_complete_and_cb();
+}
+
+bool librados::AioCompletion::AioCompletion::is_safe_and_cb()
+{
+  AioCompletionImpl *c = (AioCompletionImpl *)pc;
+  return c->is_safe_and_cb();
+}
+
 int librados::AioCompletion::AioCompletion::get_return_value()
 {
   AioCompletionImpl *c = (AioCompletionImpl *)pc;
@@ -1973,6 +1997,26 @@ extern "C" int rados_aio_is_safe(rados_completion_t c)
   return ((librados::AioCompletionImpl*)c)->is_safe();
 }
 
+extern "C" int rados_aio_wait_for_complete_and_cb(rados_completion_t c)
+{
+  return ((librados::AioCompletionImpl*)c)->wait_for_complete_and_cb();
+}
+
+extern "C" int rados_aio_wait_for_safe_and_cb(rados_completion_t c)
+{
+  return ((librados::AioCompletionImpl*)c)->wait_for_safe_and_cb();
+}
+
+extern "C" int rados_aio_is_complete_and_cb(rados_completion_t c)
+{
+  return ((librados::AioCompletionImpl*)c)->is_complete_and_cb();
+}
+
+extern "C" int rados_aio_is_safe_and_cb(rados_completion_t c)
+{
+  return ((librados::AioCompletionImpl*)c)->is_safe_and_cb();
+}
+
 extern "C" int rados_aio_get_return_value(rados_completion_t c)
 {
   return ((librados::AioCompletionImpl*)c)->get_return_value();

diff --git a/src/test/rados-api/misc.cc b/src/test/rados-api/misc.cc
index 53694761e458abeea0ef3ebd65c9ec2b8d9cb0c6..a4342231758a9aed2989b23713b0b2708546543d 100644 (file)
--- a/src/test/rados-api/misc.cc
+++ b/src/test/rados-api/misc.cc
@@ -275,7 +275,7 @@ TEST(LibRadosMisc, AioOperatePP) {
     o.append(bl2);
   }
   ASSERT_EQ(0, ioctx.aio_operate("foo", my_completion, &o));
-  ASSERT_EQ(0, my_completion->wait_for_complete());
+  ASSERT_EQ(0, my_completion->wait_for_complete_and_cb());
   ASSERT_EQ(my_aio_complete, true);
 
   uint64_t size;