Merge pull request ceph#60379 from cbodley/wip-librados-cancellation

librados/asio: forward asio cancellations to AioCompletion::cancel()

Reviewed-by: Adam Emerson <[email protected]>
Reviewed-by: Ilya Dryomov <[email protected]>

Author: cbodley

PendingReleaseNotes

@@ -34,6 +34,8 @@
   (--yes-i-really-mean-it). This has been added as a precaution to tell the
   users that modifying "max_mds" may not help with troubleshooting or recovery
   effort. Instead, it might further destabilize the cluster.
+* RADOS: Added convenience function `librados::AioCompletion::cancel()` with
+  the same behavior as `librados::IoCtx::aio_cancel()`.
 
 * mgr/restful, mgr/zabbix: both modules, already deprecated since 2020, have been
   finally removed. They have not been actively maintenance in the last years,

src/include/rados/librados.hpp

@@ -202,6 +202,8 @@ inline namespace v14_2_0 {
     int set_complete_callback(void *cb_arg, callback_t cb);
     int set_safe_callback(void *cb_arg, callback_t cb)
       __attribute__ ((deprecated));
+    /// Request immediate cancellation as if by IoCtx::aio_cancel().
+    int cancel();
     int wait_for_complete();
     int wait_for_safe() __attribute__ ((deprecated));
     int wait_for_complete_and_cb();
@@ -772,17 +774,30 @@ inline namespace v14_2_0 {
     void tier_evict();
   };
 
-  /* IoCtx : This is a context in which we can perform I/O.
-   * It includes a Pool,
+  /**
+   * @brief A handle to a RADOS pool used to perform I/O operations.
    *
    * Typical use (error checking omitted):
-   *
+   * @code
    * IoCtx p;
    * rados.ioctx_create("my_pool", p);
-   * p->stat(&stats);
-   * ... etc ...
+   * p.stat("my_object", &size, &mtime);
+   * @endcode
+   *
+   * IoCtx holds a pointer to its underlying implementation. The dup()
+   * method performs a deep copy of this implementation, but the copy
+   * construction and assignment operations perform shallow copies by
+   * sharing that pointer.
+   *
+   * Function names starting with aio_ are asynchronous operations that
+   * return immediately after submitting a request, and whose completions
+   * are managed by the given AioCompletion pointer. The IoCtx's underlying
+   * implementation is involved in the delivery of these completions, so
+   * the caller must guarantee that its lifetime is preserved until then -
+   * if not by preserving the IoCtx instance that submitted the request,
+   * then by a copied/moved instance that shares the same implementation.
    *
-   * NOTE: be sure to call watch_flush() prior to destroying any IoCtx
+   * @note Be sure to call watch_flush() prior to destroying any IoCtx
    * that is used for watch events to ensure that racing callbacks
    * have completed.
    */
@@ -791,9 +806,13 @@ inline namespace v14_2_0 {
   public:
     IoCtx();
     static void from_rados_ioctx_t(rados_ioctx_t p, IoCtx &pool);
+    /// Construct a shallow copy of rhs, sharing its underlying implementation.
     IoCtx(const IoCtx& rhs);
+    /// Assign a shallow copy of rhs, sharing its underlying implementation.
     IoCtx& operator=(const IoCtx& rhs);
+    /// Move construct from rhs, transferring its underlying implementation.
     IoCtx(IoCtx&& rhs) noexcept;
+    /// Move assign from rhs, transferring its underlying implementation.
     IoCtx& operator=(IoCtx&& rhs) noexcept;
 
     ~IoCtx();
@@ -1150,7 +1169,8 @@ inline namespace v14_2_0 {
     int aio_stat2(const std::string& oid, AioCompletion *c, uint64_t *psize, struct timespec *pts);
 
     /**
-     * Cancel aio operation
+     * Request immediate cancellation with error code -ECANCELED
+     * if the operation hasn't already completed.
      *
      * @param c completion handle
      * @returns 0 on success, negative error code on failure

src/librados/librados_asio.h

@@ -14,6 +14,9 @@
 #ifndef LIBRADOS_ASIO_H
 #define LIBRADOS_ASIO_H
 
+#include <boost/asio/associated_cancellation_slot.hpp>
+#include <boost/asio/cancellation_type.hpp>
+
 #include "include/rados/librados.hpp"
 #include "common/async/completion.h"
 #include "librados/AioCompletionImpl.h"
@@ -74,6 +77,7 @@ struct Invoker<void> {
 template <typename Result>
 struct AsyncOp : Invoker<Result> {
   unique_aio_completion_ptr aio_completion;
+  boost::asio::cancellation_slot slot;
 
   using Signature = typename Invoker<Result>::Signature;
   using Completion = ceph::async::Completion<Signature, AsyncOp<Result>>;
@@ -83,6 +87,7 @@ struct AsyncOp : Invoker<Result> {
     auto p = std::unique_ptr<Completion>{static_cast<Completion*>(arg)};
     // move result out of Completion memory being freed
     auto op = std::move(p->user_data);
+    op.slot.clear(); // clear our cancellation handler
     // access AioCompletionImpl directly to avoid locking
     const librados::AioCompletionImpl* pc = op.aio_completion->pc;
     const int ret = pc->rval;
@@ -94,11 +99,46 @@ struct AsyncOp : Invoker<Result> {
     op.dispatch(std::move(p), ec, ver);
   }
 
+  struct op_cancellation {
+    AioCompletion* completion = nullptr;
+    bool is_read = false;
+
+    void operator()(boost::asio::cancellation_type type) {
+      if (completion == nullptr) {
+        return; // no AioCompletion attached
+      } else if (type == boost::asio::cancellation_type::none) {
+        return; // no cancellation requested
+      } else if (is_read) {
+        // read operations produce no side effects, so can satisfy the
+        // requirements of 'total' cancellation. the weaker requirements
+        // of 'partial' and 'terminal' are also satisfied
+        completion->cancel();
+      } else if (type == boost::asio::cancellation_type::terminal) {
+        // write operations only support 'terminal' cancellation because we
+        // can't guarantee that no osd has succeeded (or will succeed) in
+        // applying the write
+        completion->cancel();
+      }
+    }
+  };
+
   template <typename Executor1, typename CompletionHandler>
-  static auto create(const Executor1& ex1, CompletionHandler&& handler) {
+  static auto create(const Executor1& ex1, bool is_read,
+                     CompletionHandler&& handler) {
+    op_cancellation* cancel_handler = nullptr;
+    auto slot = boost::asio::get_associated_cancellation_slot(handler);
+    if (slot.is_connected()) {
+      cancel_handler = &slot.template emplace<op_cancellation>();
+    }
+
     auto p = Completion::create(ex1, std::move(handler));
     p->user_data.aio_completion.reset(
         Rados::aio_create_completion(p.get(), aio_dispatch));
+    if (cancel_handler) {
+      cancel_handler->completion = p->user_data.aio_completion.get();
+      cancel_handler->is_read = is_read;
+      p->user_data.slot = std::move(slot);
+    }
     return p;
   }
 };
@@ -108,6 +148,9 @@ struct AsyncOp : Invoker<Result> {
 
 /// Calls IoCtx::aio_read() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t, bufferlist).
+///
+/// The given IoCtx reference is not required to remain valid, but some IoCtx
+/// instance must preserve its underlying implementation until completion.
 template <typename ExecutionContext, typename CompletionToken>
 auto async_read(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
                 size_t len, uint64_t off, CompletionToken&& token)
@@ -117,7 +160,8 @@ auto async_read(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
   return boost::asio::async_initiate<CompletionToken, Signature>(
       [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
           size_t len, uint64_t off) {
-        auto p = Op::create(ex, std::move(handler));
+        constexpr bool is_read = true;
+        auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
         int ret = io.aio_read(oid, op.aio_completion.get(), &op.result, len, off);
@@ -132,6 +176,9 @@ auto async_read(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
 
 /// Calls IoCtx::aio_write() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t).
+///
+/// The given IoCtx reference is not required to remain valid, but some IoCtx
+/// instance must preserve its underlying implementation until completion.
 template <typename ExecutionContext, typename CompletionToken>
 auto async_write(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
                  const bufferlist &bl, size_t len, uint64_t off,
@@ -142,7 +189,8 @@ auto async_write(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
   return boost::asio::async_initiate<CompletionToken, Signature>(
       [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
           const bufferlist &bl, size_t len, uint64_t off) {
-        auto p = Op::create(ex, std::move(handler));
+        constexpr bool is_read = false;
+        auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
         int ret = io.aio_write(oid, op.aio_completion.get(), bl, len, off);
@@ -157,6 +205,9 @@ auto async_write(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
 
 /// Calls IoCtx::aio_operate() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t, bufferlist).
+///
+/// The given IoCtx reference is not required to remain valid, but some IoCtx
+/// instance must preserve its underlying implementation until completion.
 template <typename ExecutionContext, typename CompletionToken>
 auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
                    ObjectReadOperation *read_op, int flags,
@@ -167,7 +218,8 @@ auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
   return boost::asio::async_initiate<CompletionToken, Signature>(
       [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
           ObjectReadOperation *read_op, int flags) {
-        auto p = Op::create(ex, std::move(handler));
+        constexpr bool is_read = true;
+        auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
         int ret = io.aio_operate(oid, op.aio_completion.get(), read_op,
@@ -183,6 +235,9 @@ auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
 
 /// Calls IoCtx::aio_operate() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t).
+///
+/// The given IoCtx reference is not required to remain valid, but some IoCtx
+/// instance must preserve its underlying implementation until completion.
 template <typename ExecutionContext, typename CompletionToken>
 auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
                    ObjectWriteOperation *write_op, int flags,
@@ -194,7 +249,8 @@ auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
       [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
           ObjectWriteOperation *write_op, int flags,
           const jspan_context* trace_ctx) {
-        auto p = Op::create(ex, std::move(handler));
+        constexpr bool is_read = false;
+        auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
         int ret = io.aio_operate(oid, op.aio_completion.get(), write_op, flags, trace_ctx);
@@ -209,6 +265,9 @@ auto async_operate(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
 
 /// Calls IoCtx::aio_notify() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t, bufferlist).
+///
+/// The given IoCtx reference is not required to remain valid, but some IoCtx
+/// instance must preserve its underlying implementation until completion.
 template <typename ExecutionContext, typename CompletionToken>
 auto async_notify(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
                   bufferlist& bl, uint64_t timeout_ms, CompletionToken &&token)
@@ -218,7 +277,8 @@ auto async_notify(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
   return boost::asio::async_initiate<CompletionToken, Signature>(
       [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
           bufferlist& bl, uint64_t timeout_ms) {
-        auto p = Op::create(ex, std::move(handler));
+        constexpr bool is_read = false;
+        auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
         int ret = io.aio_notify(oid, op.aio_completion.get(),

src/librados/librados_cxx.cc

@@ -1103,6 +1103,14 @@ void librados::AioCompletion::release()
   delete this;
 }
 
+int librados::AioCompletion::cancel()
+{
+  if (!pc->io) {
+    return 0; // no operation was started
+  }
+  return pc->io->aio_cancel(pc);
+}
+
 ///////////////////////////// IoCtx //////////////////////////////
 librados::IoCtx::IoCtx() : io_ctx_impl(NULL)
 {

src/test/librados/aio.cc

@@ -1722,3 +1722,59 @@ TEST(LibRadosAioEC, MultiWrite) {
   rados_aio_release(my_completion2);
   rados_aio_release(my_completion3);
 }
+
+TEST(LibRadosAio, CancelBeforeSubmit) {
+  AioTestData test_data;
+  ASSERT_EQ("", test_data.init());
+
+  rados_completion_t completion;
+  ASSERT_EQ(0, rados_aio_create_completion2(nullptr, nullptr, &completion));
+
+  ASSERT_EQ(0, rados_aio_cancel(test_data.m_ioctx, completion));
+  rados_aio_release(completion);
+}
+
+TEST(LibRadosAio, CancelBeforeComplete) {
+  AioTestData test_data;
+  ASSERT_EQ("", test_data.init());
+
+  // cancellation tests are racy, so retry if completion beats the cancellation
+  int ret = 0;
+  int tries = 10;
+  do {
+    rados_completion_t completion;
+    ASSERT_EQ(0, rados_aio_create_completion2(nullptr, nullptr, &completion));
+    char buf[128];
+    ASSERT_EQ(0, rados_aio_read(test_data.m_ioctx, "nonexistent",
+                                completion, buf, sizeof(buf), 0));
+
+    ASSERT_EQ(0, rados_aio_cancel(test_data.m_ioctx, completion));
+    {
+      TestAlarm alarm;
+      ASSERT_EQ(0, rados_aio_wait_for_complete(completion));
+    }
+    ret = rados_aio_get_return_value(completion);
+    rados_aio_release(completion);
+  } while (ret == -ENOENT && --tries);
+
+  ASSERT_EQ(-ECANCELED, ret);
+}
+
+TEST(LibRadosAio, CancelAfterComplete) {
+  AioTestData test_data;
+  rados_completion_t completion;
+  ASSERT_EQ("", test_data.init());
+
+  ASSERT_EQ(0, rados_aio_create_completion2(nullptr, nullptr, &completion));
+  char buf[128];
+  ASSERT_EQ(0, rados_aio_read(test_data.m_ioctx, "nonexistent",
+                              completion, buf, sizeof(buf), 0));
+
+  {
+    TestAlarm alarm;
+    ASSERT_EQ(0, rados_aio_wait_for_complete(completion));
+  }
+  ASSERT_EQ(0, rados_aio_cancel(test_data.m_ioctx, completion));
+  ASSERT_EQ(-ENOENT, rados_aio_get_return_value(completion));
+  rados_aio_release(completion);
+}