Merge pull request ceph#60576 from cbodley/wip-librados-asio-awaitable

librados/asio: support 'deferred' and 'use_awaitable' completions

Author: yuriw

src/common/io_exerciser/RadosIo.cc

@@ -135,26 +135,28 @@ void RadosIo::applyIoOp(IoOp& op) {
           std::make_shared<AsyncOpInfo<1>>(std::array<uint64_t, 1>{0},
                                            std::array<uint64_t, 1>{opSize});
       op_info->bufferlist[0] = db->generate_data(0, opSize);
-      op_info->wop.write_full(op_info->bufferlist[0]);
+      librados::ObjectWriteOperation wop;
+      wop.write_full(op_info->bufferlist[0]);
       auto create_cb = [this](boost::system::error_code ec, version_t ver) {
         ceph_assert(ec == boost::system::errc::success);
         finish_io();
       };
-      librados::async_operate(asio, io, oid, &op_info->wop, 0, nullptr,
-                              create_cb);
+      librados::async_operate(asio.get_executor(), io, oid,
+                              std::move(wop), 0, nullptr, create_cb);
       break;
     }
 
     case OpType::Remove: {
       start_io();
       auto op_info = std::make_shared<AsyncOpInfo<0>>();
-      op_info->wop.remove();
+      librados::ObjectWriteOperation wop;
+      wop.remove();
       auto remove_cb = [this](boost::system::error_code ec, version_t ver) {
         ceph_assert(ec == boost::system::errc::success);
         finish_io();
       };
-      librados::async_operate(asio, io, oid, &op_info->wop, 0, nullptr,
-                              remove_cb);
+      librados::async_operate(asio.get_executor(), io, oid,
+                              std::move(wop), 0, nullptr, remove_cb);
       break;
     }
     case OpType::Read:
@@ -197,10 +199,11 @@ void RadosIo::applyReadWriteOp(IoOp& op) {
     auto op_info =
         std::make_shared<AsyncOpInfo<N>>(readOp.offset, readOp.length);
 
+    librados::ObjectReadOperation rop;
     for (int i = 0; i < N; i++) {
-      op_info->rop.read(readOp.offset[i] * block_size,
-                        readOp.length[i] * block_size, &op_info->bufferlist[i],
-                        nullptr);
+      rop.read(readOp.offset[i] * block_size,
+               readOp.length[i] * block_size, &op_info->bufferlist[i],
+               nullptr);
     }
     auto read_cb = [this, op_info](boost::system::error_code ec, version_t ver,
                                    bufferlist bl) {
@@ -211,44 +214,49 @@ void RadosIo::applyReadWriteOp(IoOp& op) {
       }
       finish_io();
     };
-    librados::async_operate(asio, io, oid, &op_info->rop, 0, nullptr, read_cb);
+    librados::async_operate(asio.get_executor(), io, oid,
+                            std::move(rop), 0, nullptr, read_cb);
     num_io++;
   };
 
   auto applyWriteOp = [this]<OpType opType, int N>(
                           ReadWriteOp<opType, N> writeOp) {
     auto op_info =
         std::make_shared<AsyncOpInfo<N>>(writeOp.offset, writeOp.length);
+    librados::ObjectWriteOperation wop;
     for (int i = 0; i < N; i++) {
       op_info->bufferlist[i] =
           db->generate_data(writeOp.offset[i], writeOp.length[i]);
-      op_info->wop.write(writeOp.offset[i] * block_size,
-                         op_info->bufferlist[i]);
+      wop.write(writeOp.offset[i] * block_size,
+                op_info->bufferlist[i]);
     }
     auto write_cb = [this](boost::system::error_code ec, version_t ver) {
       ceph_assert(ec == boost::system::errc::success);
       finish_io();
     };
-    librados::async_operate(asio, io, oid, &op_info->wop, 0, nullptr, write_cb);
+    librados::async_operate(asio.get_executor(), io, oid,
+                            std::move(wop), 0, nullptr, write_cb);
     num_io++;
   };
 
   auto applyFailedWriteOp = [this]<OpType opType, int N>(
                                 ReadWriteOp<opType, N> writeOp) {
     auto op_info =
         std::make_shared<AsyncOpInfo<N>>(writeOp.offset, writeOp.length);
+    librados::ObjectWriteOperation wop;
     for (int i = 0; i < N; i++) {
       op_info->bufferlist[i] =
           db->generate_data(writeOp.offset[i], writeOp.length[i]);
-      op_info->wop.write(writeOp.offset[i] * block_size,
-                         op_info->bufferlist[i]);
+      wop.write(writeOp.offset[i] * block_size,
+                op_info->bufferlist[i]);
     }
     auto write_cb = [this, writeOp](boost::system::error_code ec,
                                     version_t ver) {
       ceph_assert(ec != boost::system::errc::success);
       finish_io();
     };
-    librados::async_operate(asio, io, oid, &op_info->wop, 0, nullptr, write_cb);
+    librados::async_operate(asio.get_executor(), io, oid,
+                            std::move(wop), 0, nullptr, write_cb);
     num_io++;
   };
 
@@ -450,4 +458,4 @@ void RadosIo::applyInjectOp(IoOp& op) {
           fmt::format("Unsupported inject operation ({})", op.getOpType()));
       break;
   }
-}
+}

src/common/io_exerciser/RadosIo.h

@@ -51,8 +51,6 @@ class RadosIo : public Model {
   template <int N>
   class AsyncOpInfo {
    public:
-    librados::ObjectReadOperation rop;
-    librados::ObjectWriteOperation wop;
     std::array<ceph::bufferlist, N> bufferlist;
     std::array<uint64_t, N> offset;
     std::array<uint64_t, N> length;
@@ -71,4 +69,4 @@ class RadosIo : public Model {
   void applyInjectOp(IoOp& op);
 };
 }  // namespace io_exerciser
-}  // namespace ceph
+}  // namespace ceph

src/librados/librados_asio.h

@@ -16,6 +16,7 @@
 
 #include <boost/asio/associated_cancellation_slot.hpp>
 #include <boost/asio/cancellation_type.hpp>
+#include <boost/asio/execution/executor.hpp>
 
 #include "include/rados/librados.hpp"
 #include "common/async/completion.h"
@@ -151,136 +152,142 @@ struct AsyncOp : Invoker<Result> {
 ///
 /// 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,
+template <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_read(IoExecutor ex, IoCtx& io, const std::string& oid,
                 size_t len, uint64_t off, CompletionToken&& token)
 {
   using Op = detail::AsyncOp<bufferlist>;
   using Signature = typename Op::Signature;
   return boost::asio::async_initiate<CompletionToken, Signature>(
-      [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
-          size_t len, uint64_t off) {
+      [] (auto handler, IoExecutor ex, const IoCtx& i,
+          const std::string& oid, size_t len, uint64_t off) {
         constexpr bool is_read = true;
         auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
+        IoCtx& io = const_cast<IoCtx&>(i);
         int ret = io.aio_read(oid, op.aio_completion.get(), &op.result, len, off);
         if (ret < 0) {
           auto ec = boost::system::error_code{-ret, librados::detail::err_category()};
           ceph::async::post(std::move(p), ec, 0, bufferlist{});
         } else {
           p.release(); // release ownership until completion
         }
-      }, token, ctx.get_executor(), io, oid, len, off);
+      }, token, ex, io, oid, len, off);
 }
 
 /// 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,
+template <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_write(IoExecutor ex, IoCtx& io, const std::string& oid,
                  const bufferlist &bl, size_t len, uint64_t off,
                  CompletionToken&& token)
 {
   using Op = detail::AsyncOp<void>;
   using Signature = typename Op::Signature;
   return boost::asio::async_initiate<CompletionToken, Signature>(
-      [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
+      [] (auto handler, IoExecutor ex, const IoCtx& i, const std::string& oid,
           const bufferlist &bl, size_t len, uint64_t off) {
         constexpr bool is_read = false;
         auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
+        IoCtx& io = const_cast<IoCtx&>(i);
         int ret = io.aio_write(oid, op.aio_completion.get(), bl, len, off);
         if (ret < 0) {
           auto ec = boost::system::error_code{-ret, librados::detail::err_category()};
           ceph::async::post(std::move(p), ec, 0);
         } else {
           p.release(); // release ownership until completion
         }
-      }, token, ctx.get_executor(), io, oid, bl, len, off);
+      }, token, ex, io, oid, bl, len, off);
 }
 
 /// 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,
+template <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_operate(IoExecutor ex, IoCtx& io, const std::string& oid,
+                   ObjectReadOperation read_op, int flags,
                    const jspan_context* trace_ctx, CompletionToken&& token)
 {
   using Op = detail::AsyncOp<bufferlist>;
   using Signature = typename Op::Signature;
   return boost::asio::async_initiate<CompletionToken, Signature>(
-      [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
-          ObjectReadOperation *read_op, int flags) {
+      [] (auto handler, IoExecutor ex, const IoCtx& i, const std::string& oid,
+          ObjectReadOperation read_op, int flags) {
         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,
+        auto& io = const_cast<IoCtx&>(i);
+        int ret = io.aio_operate(oid, op.aio_completion.get(), &read_op,
                                  flags, &op.result);
         if (ret < 0) {
           auto ec = boost::system::error_code{-ret, librados::detail::err_category()};
           ceph::async::post(std::move(p), ec, 0, bufferlist{});
         } else {
           p.release(); // release ownership until completion
         }
-      }, token, ctx.get_executor(), io, oid, read_op, flags);
+      }, token, ex, io, oid, std::move(read_op), flags);
 }
 
 /// 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,
+template <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_operate(IoExecutor ex, IoCtx& io, const std::string& oid,
+                   ObjectWriteOperation write_op, int flags,
                    const jspan_context* trace_ctx, CompletionToken &&token)
 {
   using Op = detail::AsyncOp<void>;
   using Signature = typename Op::Signature;
   return boost::asio::async_initiate<CompletionToken, Signature>(
-      [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
-          ObjectWriteOperation *write_op, int flags,
+      [] (auto handler, IoExecutor ex, const IoCtx& i, const std::string& oid,
+          ObjectWriteOperation write_op, int flags,
           const jspan_context* trace_ctx) {
         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);
+        auto& io = const_cast<IoCtx&>(i);
+        int ret = io.aio_operate(oid, op.aio_completion.get(), &write_op, flags, trace_ctx);
         if (ret < 0) {
           auto ec = boost::system::error_code{-ret, librados::detail::err_category()};
           ceph::async::post(std::move(p), ec, 0);
         } else {
           p.release(); // release ownership until completion
         }
-      }, token, ctx.get_executor(), io, oid, write_op, flags, trace_ctx);
+      }, token, ex, io, oid, std::move(write_op), flags, trace_ctx);
 }
 
 /// 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,
+template <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_notify(IoExecutor ex, IoCtx& io, const std::string& oid,
                   bufferlist& bl, uint64_t timeout_ms, CompletionToken &&token)
 {
   using Op = detail::AsyncOp<bufferlist>;
   using Signature = typename Op::Signature;
   return boost::asio::async_initiate<CompletionToken, Signature>(
-      [] (auto handler, auto ex, IoCtx& io, const std::string& oid,
-          bufferlist& bl, uint64_t timeout_ms) {
+      [] (auto handler, IoExecutor ex, const IoCtx& i, const std::string& oid,
+          const bufferlist& b, uint64_t timeout_ms) {
         constexpr bool is_read = false;
         auto p = Op::create(ex, is_read, std::move(handler));
         auto& op = p->user_data;
 
+        IoCtx& io = const_cast<IoCtx&>(i);
+        bufferlist& bl = const_cast<bufferlist&>(b);
         int ret = io.aio_notify(oid, op.aio_completion.get(),
                                 bl, timeout_ms, &op.result);
         if (ret < 0) {
@@ -289,7 +296,7 @@ auto async_notify(ExecutionContext& ctx, IoCtx& io, const std::string& oid,
         } else {
           p.release(); // release ownership until completion
         }
-      }, token, ctx.get_executor(), io, oid, bl, timeout_ms);
+      }, token, ex, io, oid, bl, timeout_ms);
 }
 
 } // namespace librados

src/rgw/driver/rados/rgw_tools.cc

@@ -206,9 +206,10 @@ int rgw_rados_operate(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, con
   // of blocking
   if (y) {
     auto& yield = y.get_yield_context();
+    auto ex = yield.get_executor();
     boost::system::error_code ec;
-    auto [ver, bl] = librados::async_operate(
-      yield, ioctx, oid, op, flags, trace_info, yield[ec]);
+    auto [ver, bl] = librados::async_operate(ex, ioctx, oid, std::move(*op),
+                                             flags, trace_info, yield[ec]);
     if (pbl) {
       *pbl = std::move(bl);
     }
@@ -231,9 +232,10 @@ int rgw_rados_operate(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, con
 {
   if (y) {
     auto& yield = y.get_yield_context();
+    auto ex = yield.get_executor();
     boost::system::error_code ec;
-    version_t ver = librados::async_operate(yield, ioctx, oid, op, flags,
-                                            trace_info, yield[ec]);
+    version_t ver = librados::async_operate(ex, ioctx, oid, std::move(*op),
+                                            flags, trace_info, yield[ec]);
     if (pver) {
       *pver = ver;
     }
@@ -254,8 +256,8 @@ int rgw_rados_notify(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, cons
   if (y) {
     auto& yield = y.get_yield_context();
     boost::system::error_code ec;
-    auto [ver, reply] = librados::async_notify(yield, ioctx, oid,
-                                               bl, timeout_ms, yield[ec]);
+    auto [ver, reply] = librados::async_notify(
+        yield.get_executor(), ioctx, oid, bl, timeout_ms, yield[ec]);
     if (pbl) {
       *pbl = std::move(reply);
     }

src/rgw/rgw_aio.cc

@@ -97,7 +97,7 @@ Aio::OpFunc aio_abstract(librados::IoCtx ctx, Op&& op,
       // executor so it can safely call back into Aio without locking
       auto ex = yield.get_executor();
 
-      librados::async_operate(yield, ctx, r.obj.oid, &op, 0, trace_ctx,
+      librados::async_operate(ex, ctx, r.obj.oid, std::move(op), 0, trace_ctx,
                               bind_executor(ex, Handler{aio, ctx, r}));
     };
 }