Merge pull request ceph#62538 from cbodley/wip-rgw-control-watch-par

rgw: send concurrent watch/unwatch operations

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

Author: cbodley

src/common/async/detail/spawn_throttle_impl.h

@@ -169,7 +169,7 @@ struct spawn_throttle_handler {
   }
 };
 
-spawn_throttle_handler spawn_throttle_impl::get()
+inline spawn_throttle_handler spawn_throttle_impl::get()
 {
   report_exception(); // throw unreported exception
 
@@ -345,8 +345,8 @@ class async_spawn_throttle_impl final :
   }
 };
 
-auto spawn_throttle_impl::create(optional_yield y, size_t limit,
-                                 cancel_on_error on_error)
+inline auto spawn_throttle_impl::create(optional_yield y, size_t limit,
+                                        cancel_on_error on_error)
     -> boost::intrusive_ptr<spawn_throttle_impl>
 {
   if (y) {

src/common/options/rgw.yaml.in

@@ -1133,7 +1133,23 @@ options:
   default: 8
   services:
   - rgw
+  see_also:
+  - rgw_cache_enabled
+  - rgw_max_control_aio
   with_legacy: true
+- name: rgw_max_control_aio
+  type: int
+  level: advanced
+  desc: Maximum number of concurrent operations over control objects.
+  long_desc: When metadata caching is enabled, a watch operation is sent to each
+    control object on startup, with a corresponding unwatch on shutdown. To
+    accelerate startup/shutdown, allow several concurrent operations to be sent
+    at once.
+  default: 8
+  services:
+  - rgw
+  see_also:
+  - rgw_num_control_oids
 - name: rgw_verify_ssl
   type: bool
   level: advanced

src/librados/librados_asio.h

@@ -268,6 +268,65 @@ auto async_operate(IoExecutor ex, IoCtx& io, const std::string& oid,
       }, token, ex, io, oid, std::move(write_op), flags, trace_ctx);
 }
 
+/// Calls IoCtx::aio_watch2() 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 <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_watch(IoExecutor ex, IoCtx& io, const std::string& oid,
+                 uint64_t* handle, librados::WatchCtx2* ctx,
+                 uint32_t timeout_ms, CompletionToken &&token)
+{
+  using Op = detail::AsyncOp<void>;
+  using Signature = typename Op::Signature;
+  return boost::asio::async_initiate<CompletionToken, Signature>(
+      [] (auto handler, IoExecutor ex, const IoCtx& i, const std::string& oid,
+          uint64_t* handle, librados::WatchCtx2* ctx, uint32_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);
+        int ret = io.aio_watch2(oid, op.aio_completion.get(),
+                                handle, ctx, timeout_ms);
+        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, ex, io, oid, handle, ctx, timeout_ms);
+}
+
+/// Calls IoCtx::aio_unwatch() 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 <boost::asio::execution::executor IoExecutor, typename CompletionToken>
+auto async_unwatch(IoExecutor ex, IoCtx& io, uint64_t handle,
+                   CompletionToken &&token)
+{
+  using Op = detail::AsyncOp<void>;
+  using Signature = typename Op::Signature;
+  return boost::asio::async_initiate<CompletionToken, Signature>(
+      [] (auto handler, IoExecutor ex, const IoCtx& i, uint64_t handle) {
+        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_unwatch(handle, op.aio_completion.get());
+        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, ex, io, handle);
+}
+
 /// Calls IoCtx::aio_notify() and arranges for the AioCompletion to call a
 /// given handler with signature (error_code, version_t, bufferlist).
 ///

src/rgw/driver/rados/rgw_tools.cc

@@ -117,6 +117,34 @@ int rgw_get_rados_ref(const DoutPrefixProvider* dpp, librados::Rados* rados,
   return 0;
 }
 
+int rgw_rados_ref::watch(const DoutPrefixProvider* dpp, uint64_t* handle,
+                         librados::WatchCtx2* ctx, optional_yield y)
+{
+  if (y) {
+    auto& yield = y.get_yield_context();
+    boost::system::error_code ec;
+    librados::async_watch(yield.get_executor(), ioctx, obj.oid,
+                          handle, ctx, 0, yield[ec]);
+    return ceph::from_error_code(ec);
+  } else {
+    maybe_warn_about_blocking(dpp);
+    return ioctx.watch2(obj.oid, handle, ctx);
+  }
+}
+
+int rgw_rados_ref::unwatch(const DoutPrefixProvider* dpp, uint64_t handle,
+                           optional_yield y)
+{
+  if (y) {
+    auto& yield = y.get_yield_context();
+    boost::system::error_code ec;
+    librados::async_unwatch(yield.get_executor(), ioctx, handle, yield[ec]);
+    return ceph::from_error_code(ec);
+  } else {
+    maybe_warn_about_blocking(dpp);
+    return ioctx.unwatch2(handle);
+  }
+}
 
 map<string, bufferlist>* no_change_attrs() {
   static map<string, bufferlist> no_change;

src/rgw/driver/rados/rgw_tools.h

@@ -145,18 +145,10 @@ struct rgw_rados_ref {
     return ioctx.aio_operate(obj.oid, c, op, pbl);
   }
 
-  int watch(uint64_t* handle, librados::WatchCtx2* ctx) {
-    return ioctx.watch2(obj.oid, handle, ctx);
-  }
-
-  int aio_watch(librados::AioCompletion* c, uint64_t* handle,
-		librados::WatchCtx2 *ctx) {
-    return ioctx.aio_watch(obj.oid, c, handle, ctx);
-  }
+  int watch(const DoutPrefixProvider* dpp, uint64_t* handle,
+            librados::WatchCtx2* ctx, optional_yield y);
 
-  int unwatch(uint64_t handle) {
-    return ioctx.unwatch2(handle);
-  }
+  int unwatch(const DoutPrefixProvider* dpp, uint64_t handle, optional_yield y);
 
   int notify(const DoutPrefixProvider* dpp, bufferlist& bl, uint64_t timeout_ms,
 	     bufferlist* pbl, optional_yield y) {