rgw: RGWSI_Notify spawns a coroutine on null_yield

init_watch() and finalize_watch() use spawn_throttle for concurrent
operations, so need to spawn a parent coroutine unless init_watch()
already has one

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

diff --git a/src/rgw/services/svc_notify.cc b/src/rgw/services/svc_notify.cc
index 0fb2c18f3b936ba54608c1c01cfc23c5de0ec934..e26bce3a1a10b009d35a36458a93925b5e508c3a 100644 (file)
--- a/src/rgw/services/svc_notify.cc
+++ b/src/rgw/services/svc_notify.cc
@@ -167,7 +167,8 @@ rgw_rados_ref RGWSI_Notify::pick_control_obj(const string& key)
   return watchers[i].get_obj();
 }
 
-int RGWSI_Notify::init_watch(const DoutPrefixProvider *dpp, optional_yield y)
+int RGWSI_Notify::init_watch(const DoutPrefixProvider *dpp,
+                             boost::asio::yield_context yield)
 {
   num_watchers = cct->_conf->rgw_num_control_oids;
 
@@ -178,7 +179,7 @@ int RGWSI_Notify::init_watch(const DoutPrefixProvider *dpp, optional_yield y)
 
   const size_t max_aio = cct->_conf.get_val<int64_t>("rgw_max_control_aio");
   auto throttle = ceph::async::spawn_throttle{
-      y, max_aio, ceph::async::cancel_on_error::all};
+      yield, max_aio, ceph::async::cancel_on_error::all};
   watchers.reserve(num_watchers);
 
   for (int i=0; i < num_watchers; i++) {
@@ -230,11 +231,11 @@ int RGWSI_Notify::init_watch(const DoutPrefixProvider *dpp, optional_yield y)
   return 0;
 }
 
-void RGWSI_Notify::finalize_watch()
+void RGWSI_Notify::finalize_watch(boost::asio::yield_context yield)
 {
   const size_t max_aio = cct->_conf.get_val<int64_t>("rgw_max_control_aio");
   auto throttle = ceph::async::spawn_throttle{
-      null_yield, max_aio, ceph::async::cancel_on_error::all};
+      yield, max_aio, ceph::async::cancel_on_error::all};
   for (int i = 0; i < num_watchers; i++) {
     if (!watchers_set.contains(i)) {
       continue;
@@ -268,7 +269,26 @@ int RGWSI_Notify::do_start(optional_yield y, const DoutPrefixProvider *dpp)
 
   control_pool = zone_svc->get_zone_params().control_pool;
 
-  int ret = init_watch(dpp, y);
+  int ret = 0;
+
+  // if we're not running in a coroutine, spawn one
+  if (!y) {
+    boost::asio::io_context context;
+    boost::asio::spawn(context,
+        [this, dpp] (boost::asio::yield_context yield) {
+          return init_watch(dpp, yield);
+        },
+        [&ret] (std::exception_ptr eptr, int result) {
+          if (eptr) {
+            std::rethrow_exception(eptr);
+          } else {
+            ret = result;
+          }
+        });
+    context.run();
+  } else {
+    ret = init_watch(dpp, y.get_yield_context());
+  }
   if (ret < 0) {
     ldpp_dout(dpp, -1) << "ERROR: failed to initialize watch: " << cpp_strerror(-ret) << dendl;
     return ret;
@@ -291,7 +311,17 @@ void RGWSI_Notify::shutdown()
   if (finisher_handle) {
     finisher_svc->unregister_caller(*finisher_handle);
   }
-  finalize_watch();
+
+  // we're not running in a coroutine, so spawn one
+  boost::asio::io_context context;
+  boost::asio::spawn(context,
+      [this] (boost::asio::yield_context yield) {
+        finalize_watch(yield);
+      },
+      [] (std::exception_ptr eptr) {
+        if (eptr) std::rethrow_exception(eptr);
+      });
+  context.run();
 
   delete shutdown_cb;
 

diff --git a/src/rgw/services/svc_notify.h b/src/rgw/services/svc_notify.h
index a13219c684b09186e37ff4b6d7cf7f9c11099768..c033e499c10ce5307b6b0d2dd75190f1b39063bd 100644 (file)
--- a/src/rgw/services/svc_notify.h
+++ b/src/rgw/services/svc_notify.h
@@ -53,8 +53,9 @@ private:
 
   bool finalized{false};
 
-  int init_watch(const DoutPrefixProvider *dpp, optional_yield y);
-  void finalize_watch();
+  int init_watch(const DoutPrefixProvider *dpp,
+                 boost::asio::yield_context yield);
+  void finalize_watch(boost::asio::yield_context yield);
 
   void init(RGWSI_Zone *_zone_svc,
             librados::Rados* rados_,