rgw: add optional_yield to Aio::librados_op()

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/rgw/rgw_aio.cc b/src/rgw/rgw_aio.cc
index b311982872f1f7c21eb80ddc4271f297f2e4a2be..c42d8051e1d761ef823bbb2130cf61339b2c8dc2 100644 (file)
--- a/src/rgw/rgw_aio.cc
+++ b/src/rgw/rgw_aio.cc
@@ -15,13 +15,16 @@
 
 #include <type_traits>
 #include "include/rados/librados.hpp"
+#include "librados/librados_asio.h"
 
 #include "rgw_aio.h"
 
 namespace rgw {
+
 namespace {
+
 void cb(librados::completion_t, void* arg);
-}
+
 struct state {
   Aio* aio;
   librados::AioCompletion* c;
@@ -31,7 +34,6 @@ struct state {
       c(librados::Rados::aio_create_completion(&r, nullptr, &cb)) {}
 };
 
-namespace {
 void cb(librados::completion_t, void* arg) {
   static_assert(sizeof(AioResult::user_data) >= sizeof(state));
   static_assert(std::is_trivially_destructible_v<state>);
@@ -42,33 +44,80 @@ void cb(librados::completion_t, void* arg) {
   s->aio->put(r);
 }
 
-template<typename T,
-        bool read = std::is_same_v<std::decay_t<T>,
-                                   librados::ObjectReadOperation>,
-        typename = std::enable_if_t<std::is_base_of_v<librados::ObjectOperation,
-                                                      std::decay_t<T>> &&
-                                    !std::is_lvalue_reference_v<T> &&
-                                    !std::is_const_v<T>>>
-Aio::OpFunc aio_abstract(T&& op) {
-  return [op = std::move(op)](Aio* aio, AioResult& r) mutable {
-          auto s = new (&r.user_data) state(aio, r);
-          if constexpr (read) {
-            r.result = r.obj.aio_operate(s->c, &op, &r.data);
-          } else {
-            r.result = r.obj.aio_operate(s->c, &op);
-          }
-          if (r.result < 0) {
-            s->c->release();
-            aio->put(r);
-          }
-        };
+template <typename Op>
+Aio::OpFunc aio_abstract(Op&& op) {
+  return [op = std::move(op)] (Aio* aio, AioResult& r) mutable {
+      constexpr bool read = std::is_same_v<std::decay_t<Op>, librados::ObjectReadOperation>;
+      auto s = new (&r.user_data) state(aio, r);
+      if constexpr (read) {
+        r.result = r.obj.aio_operate(s->c, &op, &r.data);
+      } else {
+        r.result = r.obj.aio_operate(s->c, &op);
+      }
+      if (r.result < 0) {
+        s->c->release();
+        aio->put(r);
+      }
+    };
 }
+
+#ifdef HAVE_BOOST_CONTEXT
+struct Handler {
+  Aio* throttle = nullptr;
+  AioResult& r;
+  // write callback
+  void operator()(boost::system::error_code ec) const {
+    r.result = -ec.value();
+    throttle->put(r);
+  }
+  // read callback
+  void operator()(boost::system::error_code ec, bufferlist bl) const {
+    r.result = -ec.value();
+    r.data = std::move(bl);
+    throttle->put(r);
+  }
+};
+
+template <typename Op>
+Aio::OpFunc aio_abstract(Op&& op, boost::asio::io_context& context,
+                         boost::asio::yield_context yield) {
+  return [op = std::move(op), &context, yield] (Aio* aio, AioResult& r) mutable {
+      // arrange for the completion Handler to run on the yield_context's strand
+      // executor so it can safely call back into Aio without locking
+      using namespace boost::asio;
+      async_completion<yield_context, void()> init(yield);
+      auto ex = get_associated_executor(init.completion_handler);
+
+      auto& ref = r.obj.get_ref();
+      librados::async_operate(context, ref.ioctx, ref.obj.oid, &op, 0,
+                              bind_executor(ex, Handler{aio, r}));
+    };
 }
+#endif // HAVE_BOOST_CONTEXT
 
-Aio::OpFunc Aio::librados_op(librados::ObjectReadOperation&& op) {
-  return aio_abstract(std::move(op));
+template <typename Op>
+Aio::OpFunc aio_abstract(Op&& op, optional_yield y) {
+  static_assert(std::is_base_of_v<librados::ObjectOperation, std::decay_t<Op>>);
+  static_assert(!std::is_lvalue_reference_v<Op>);
+  static_assert(!std::is_const_v<Op>);
+#ifdef HAVE_BOOST_CONTEXT
+  if (y) {
+    return aio_abstract(std::forward<Op>(op), y.get_io_context(),
+                        y.get_yield_context());
+  }
+#endif
+  return aio_abstract(std::forward<Op>(op));
 }
-Aio::OpFunc Aio::librados_op(librados::ObjectWriteOperation&& op) {
-  return aio_abstract(std::move(op));
+
+} // anonymous namespace
+
+Aio::OpFunc Aio::librados_op(librados::ObjectReadOperation&& op,
+                             optional_yield y) {
+  return aio_abstract(std::move(op), y);
 }
+Aio::OpFunc Aio::librados_op(librados::ObjectWriteOperation&& op,
+                             optional_yield y) {
+  return aio_abstract(std::move(op), y);
 }
+
+} // namespace rgw

diff --git a/src/rgw/rgw_aio.h b/src/rgw/rgw_aio.h
index 4602be0a622a0612121f8581e19ac5ef634dca21..6c9eca5c67a9b9c56e0783b4758fc9d8fe868951 100644 (file)
--- a/src/rgw/rgw_aio.h
+++ b/src/rgw/rgw_aio.h
@@ -21,6 +21,7 @@
 
 #include <boost/intrusive/list.hpp>
 #include "include/rados/librados_fwd.hpp"
+#include "common/async/yield_context.h"
 
 #include "services/svc_rados.h" // cant forward declare RGWSI_RADOS::Obj
 
@@ -90,8 +91,10 @@ class Aio {
   // wait for all outstanding completions and return their results
   virtual AioResultList drain() = 0;
 
-  static OpFunc librados_op(librados::ObjectReadOperation&& op);
-  static OpFunc librados_op(librados::ObjectWriteOperation&& op);
+  static OpFunc librados_op(librados::ObjectReadOperation&& op,
+                            optional_yield y);
+  static OpFunc librados_op(librados::ObjectWriteOperation&& op,
+                            optional_yield y);
 };
 
 } // namespace rgw

diff --git a/src/rgw/rgw_putobj_processor.cc b/src/rgw/rgw_putobj_processor.cc
index 638d716a5a23d498f8d6c6a5c2502f4e06332608..6460d5218a9654455fa3949f1f1714684f20ce29 100644 (file)
--- a/src/rgw/rgw_putobj_processor.cc
+++ b/src/rgw/rgw_putobj_processor.cc
@@ -92,7 +92,7 @@ int RadosWriter::process(bufferlist&& bl, uint64_t offset)
     op.write(offset, data);
   }
   constexpr uint64_t id = 0; // unused
-  auto c = aio->get(stripe_obj, Aio::librados_op(std::move(op)), cost, id);
+  auto c = aio->get(stripe_obj, Aio::librados_op(std::move(op), y), cost, id);
   return process_completed(c, &written);
 }
 
@@ -105,7 +105,7 @@ int RadosWriter::write_exclusive(const bufferlist& data)
   op.write_full(data);
 
   constexpr uint64_t id = 0; // unused
-  auto c = aio->get(stripe_obj, Aio::librados_op(std::move(op)), cost, id);
+  auto c = aio->get(stripe_obj, Aio::librados_op(std::move(op), y), cost, id);
   auto d = aio->drain();
   c.splice(c.end(), d);
   return process_completed(c, &written);

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 00ee3f192e85fad24aa1bf20dd907c08787408e9..c354da706da71b08155f534811184cde6127edc3 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -6815,7 +6815,7 @@ int RGWRados::get_obj_iterate_cb(const rgw_raw_obj& read_obj, off_t obj_ofs,
   const uint64_t cost = len;
   const uint64_t id = obj_ofs; // use logical object offset for sorting replies
 
-  auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op)), cost, id);
+  auto completed = d->aio->get(obj, rgw::Aio::librados_op(std::move(op), d->yield), cost, id);
 
   return d->flush(std::move(completed));
 }