Merge pull request ceph#65775 from cbodley/wip-73361

rgw: RGWSI_Notify drains the finisher before deleting RGWWatchers

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

Author: cbodley

src/rgw/CMakeLists.txt

@@ -152,7 +152,6 @@ list(APPEND librgw_common_srcs
 
 if(WITH_RADOSGW_RADOS)
   list(APPEND librgw_common_srcs
-          services/svc_finisher.cc
           services/svc_bi_rados.cc
           services/svc_bilog_rados.cc
           services/svc_bucket.cc

src/rgw/driver/rados/rgw_service.cc

@@ -3,7 +3,6 @@
 
 #include "rgw_service.h"
 
-#include "services/svc_finisher.h"
 #include "services/svc_bi_rados.h"
 #include "services/svc_bilog_rados.h"
 #include "services/svc_bucket_sobj.h"
@@ -57,7 +56,6 @@ int RGWServices_Def::init(CephContext *cct,
                           rgw::sal::ConfigStore* cfgstore,
                           const rgw::SiteConfig* site)
 {
-  finisher = std::make_unique<RGWSI_Finisher>(cct);
   bucket_sobj = std::make_unique<RGWSI_Bucket_SObj>(cct);
   bucket_sync_sobj = std::make_unique<RGWSI_Bucket_Sync_SObj>(cct);
   bi_rados = std::make_unique<RGWSI_BucketIndex_RADOS>(cct);
@@ -66,7 +64,9 @@ int RGWServices_Def::init(CephContext *cct,
   config_key_rados = std::make_unique<RGWSI_ConfigKey_RADOS>(cct);
   datalog_rados = std::make_unique<RGWDataChangesLog>(driver);
   mdlog = std::make_unique<RGWSI_MDLog>(cct, run_sync, cfgstore);
-  notify = std::make_unique<RGWSI_Notify>(cct);
+  if (have_cache) {
+    notify = std::make_unique<RGWSI_Notify>(cct);
+  }
   zone = std::make_unique<RGWSI_Zone>(cct, cfgstore, site);
   zone_utils = std::make_unique<RGWSI_ZoneUtils>(cct);
   quota = std::make_unique<RGWSI_Quota>(cct);
@@ -82,7 +82,6 @@ int RGWServices_Def::init(CephContext *cct,
   }
 
   async_processor->start();
-  finisher->init();
   bi_rados->init(zone.get(), driver->getRados()->get_rados_handle(),
 		 bilog_rados.get(), datalog_rados.get());
   bilog_rados->init(bi_rados.get());
@@ -97,8 +96,9 @@ int RGWServices_Def::init(CephContext *cct,
   config_key_rados->init(driver->getRados()->get_rados_handle());
   mdlog->init(driver->getRados()->get_rados_handle(), zone.get(), sysobj.get(),
 	      cls.get(), async_processor.get());
-  notify->init(zone.get(), driver->getRados()->get_rados_handle(),
-	       finisher.get());
+  if (notify) {
+    notify->init(zone.get(), driver->getRados()->get_rados_handle());
+  }
   zone->init(sysobj.get(), driver->getRados()->get_rados_handle(),
 	     sync_modules.get(), bucket_sync_sobj.get());
   zone_utils->init(driver->getRados()->get_rados_handle(), zone.get());
@@ -116,13 +116,9 @@ int RGWServices_Def::init(CephContext *cct,
 
   can_shutdown = true;
 
-  int r = finisher->start(y, dpp);
-  if (r < 0) {
-    ldpp_dout(dpp, 0) << "ERROR: failed to start finisher service (" << cpp_strerror(-r) << dendl;
-    return r;
-  }
+  int r = 0;
 
-  if (!raw) {
+  if (notify) {
     r = notify->start(y, dpp);
     if (r < 0) {
       ldpp_dout(dpp, 0) << "ERROR: failed to start notify service (" << cpp_strerror(-r) << dendl;
@@ -240,19 +236,19 @@ void RGWServices_Def::shutdown()
   datalog_rados.reset();
   user_rados->shutdown();
   sync_modules->shutdown();
-  notify->shutdown();
+  if (notify) {
+    notify->shutdown();
+  }
   mdlog->shutdown();
   config_key_rados->shutdown();
   cls->shutdown();
   bilog_rados->shutdown();
   bi_rados->shutdown();
   bucket_sync_sobj->shutdown();
   bucket_sobj->shutdown();
-  finisher->shutdown();
 
   sysobj->shutdown();
   sysobj_core->shutdown();
-  notify->shutdown();
   if (sysobj_cache) {
     sysobj_cache->shutdown();
   }
@@ -274,7 +270,6 @@ int RGWServices::do_init(CephContext *_cct, rgw::sal::RadosStore* driver, bool h
     return r;
   }
 
-  finisher = _svc.finisher.get();
   bi_rados = _svc.bi_rados.get();
   bi = bi_rados;
   bilog_rados = _svc.bilog_rados.get();
@@ -287,7 +282,6 @@ int RGWServices::do_init(CephContext *_cct, rgw::sal::RadosStore* driver, bool h
   config_key = config_key_rados;
   datalog_rados = _svc.datalog_rados.get();
   mdlog = _svc.mdlog.get();
-  notify = _svc.notify.get();
   zone = _svc.zone.get();
   zone_utils = _svc.zone_utils.get();
   quota = _svc.quota.get();

src/rgw/driver/rados/rgw_service.h

@@ -53,7 +53,6 @@ class RGWServiceInstance
   }
 };
 
-class RGWSI_Finisher;
 class RGWSI_Bucket;
 class RGWSI_Bucket_SObj;
 class RGWSI_Bucket_Sync;
@@ -84,7 +83,6 @@ struct RGWServices_Def
   bool can_shutdown{false};
   bool has_shutdown{false};
 
-  std::unique_ptr<RGWSI_Finisher> finisher;
   std::unique_ptr<RGWSI_Bucket_SObj> bucket_sobj;
   std::unique_ptr<RGWSI_Bucket_Sync_SObj> bucket_sync_sobj;
   std::unique_ptr<RGWSI_BucketIndex_RADOS> bi_rados;
@@ -122,7 +120,6 @@ struct RGWServices
   CephContext *cct;
   const rgw::SiteConfig* site{nullptr};
 
-  RGWSI_Finisher *finisher{nullptr};
   RGWSI_Bucket *bucket{nullptr};
   RGWSI_Bucket_SObj *bucket_sobj{nullptr};
   RGWSI_Bucket_Sync *bucket_sync{nullptr};
@@ -135,7 +132,6 @@ struct RGWServices
   RGWSI_ConfigKey *config_key{nullptr};
   RGWDataChangesLog *datalog_rados{nullptr};
   RGWSI_MDLog *mdlog{nullptr};
-  RGWSI_Notify *notify{nullptr};
   RGWSI_Zone *zone{nullptr};
   RGWSI_ZoneUtils *zone_utils{nullptr};
   RGWSI_Quota *quota{nullptr};

src/rgw/services/svc_finisher.cc

@@ -9,7 +9,6 @@
 
 #include "rgw_cache.h"
 #include "svc_notify.h"
-#include "svc_finisher.h"
 #include "svc_zone.h"
 
 #include "rgw_zone.h"
@@ -134,22 +133,15 @@ class RGWWatcher : public DoutPrefixProvider , public librados::WatchCtx2 {
   }
 };
 
-RGWSI_Notify::RGWSI_Notify(CephContext *cct) : RGWServiceInstance(cct) {}
+RGWSI_Notify::RGWSI_Notify(CephContext *cct)
+  : RGWServiceInstance(cct), finisher(cct)
+{
+}
 RGWSI_Notify::~RGWSI_Notify()
 {
   shutdown();
 }
 
-class RGWSI_Notify_ShutdownCB : public RGWSI_Finisher::ShutdownCB
-{
-  RGWSI_Notify *svc;
-public:
-  RGWSI_Notify_ShutdownCB(RGWSI_Notify *_svc) : svc(_svc) {}
-  void call() override {
-    svc->shutdown();
-  }
-};
-
 string RGWSI_Notify::get_control_oid(int i)
 {
   char buf[notify_oid_prefix.size() + 16];
@@ -245,8 +237,6 @@ void RGWSI_Notify::finalize_watch(boost::asio::yield_context yield)
         });
   }
   throttle.wait();
-
-  watchers.clear();
 }
 
 int RGWSI_Notify::do_start(optional_yield y, const DoutPrefixProvider *dpp)
@@ -258,10 +248,7 @@ int RGWSI_Notify::do_start(optional_yield y, const DoutPrefixProvider *dpp)
 
   assert(zone_svc->is_started()); /* otherwise there's an ordering problem */
 
-  r = finisher_svc->start(y, dpp);
-  if (r < 0) {
-    return r;
-  }
+  finisher.start();
 
   inject_notify_timeout_probability =
     cct->_conf.get_val<double>("rgw_inject_notify_timeout_probability");
@@ -294,11 +281,6 @@ int RGWSI_Notify::do_start(optional_yield y, const DoutPrefixProvider *dpp)
     return ret;
   }
 
-  shutdown_cb = new RGWSI_Notify_ShutdownCB(this);
-  int handle;
-  finisher_svc->register_caller(shutdown_cb, &handle);
-  finisher_handle = handle;
-
   return 0;
 }
 
@@ -308,10 +290,6 @@ void RGWSI_Notify::shutdown()
     return;
   }
 
-  if (finisher_handle) {
-    finisher_svc->unregister_caller(*finisher_handle);
-  }
-
   // we're not running in a coroutine, so spawn one
   boost::asio::io_context context;
   boost::asio::spawn(context,
@@ -323,7 +301,12 @@ void RGWSI_Notify::shutdown()
       });
   context.run();
 
-  delete shutdown_cb;
+  // wait for any racing C_ReinitWatch calls on the finisher thread
+  // before destroying the RGWWatchers
+  finisher.wait_for_empty();
+  finisher.stop();
+
+  watchers.clear();
 
   finalized = true;
 }
@@ -525,5 +508,5 @@ void RGWSI_Notify::register_watch_cb(CB *_cb)
 
 void RGWSI_Notify::schedule_context(Context *c)
 {
-  finisher_svc->schedule_context(c);
+  finisher.queue(c);
 }