rgw: safe disposal of async processors

the async processor might out its container cr, need to refcount.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>

diff --git a/src/rgw/rgw_cr_rados.cc b/src/rgw/rgw_cr_rados.cc
index 90a168a22f77435630919fd7971f2d44ae870794..83b472cd7809e992583a1643cb9b60cf645fe73a 100644 (file)
--- a/src/rgw/rgw_cr_rados.cc
+++ b/src/rgw/rgw_cr_rados.cc
@@ -8,6 +8,7 @@
 #define dout_subsys ceph_subsys_rgw
 
 bool RGWAsyncRadosProcessor::RGWWQ::_enqueue(RGWAsyncRadosRequest *req) {
+  req->get();
   processor->m_req_queue.push_back(req);
   dout(20) << "enqueued request req=" << hex << req << dec << dendl;
   _dump_queue();
@@ -62,10 +63,14 @@ void RGWAsyncRadosProcessor::start() {
 void RGWAsyncRadosProcessor::stop() {
   m_tp.drain(&req_wq);
   m_tp.stop();
+  for (auto iter = m_req_queue.begin(); iter != m_req_queue.end(); ++iter) {
+    (*iter)->put();
+  }
 }
 
 void RGWAsyncRadosProcessor::handle_request(RGWAsyncRadosRequest *req) {
   req->send_request();
+  req->put();
 }
 
 void RGWAsyncRadosProcessor::queue(RGWAsyncRadosRequest *req) {
@@ -296,7 +301,9 @@ RGWSimpleRadosLockCR::RGWSimpleRadosLockCR(RGWAsyncRadosProcessor *_async_rados,
 
 RGWSimpleRadosLockCR::~RGWSimpleRadosLockCR()
 {
-  delete req;
+  if (req) {
+    req->finish();
+  }
 }
 
 int RGWSimpleRadosLockCR::send_request()
@@ -330,7 +337,9 @@ RGWSimpleRadosUnlockCR::RGWSimpleRadosUnlockCR(RGWAsyncRadosProcessor *_async_ra
 
 RGWSimpleRadosUnlockCR::~RGWSimpleRadosUnlockCR()
 {
-  delete req;
+  if (req) {
+    req->finish();
+  }
 }
 
 int RGWSimpleRadosUnlockCR::send_request()

diff --git a/src/rgw/rgw_cr_rados.h b/src/rgw/rgw_cr_rados.h
index dc1371021aeb1da305acf37bab7da81a87a10d8b..6f6c47438a45c9a005dad32a86d3deda00641265 100644 (file)
--- a/src/rgw/rgw_cr_rados.h
+++ b/src/rgw/rgw_cr_rados.h
@@ -5,24 +5,41 @@
 #include "common/WorkQueue.h"
 #include "common/Throttle.h"
 
-class RGWAsyncRadosRequest {
+class RGWAsyncRadosRequest : public RefCountedObject {
   RGWAioCompletionNotifier *notifier;
 
   void *user_info;
   int retcode;
 
+  bool done;
+
+  Mutex lock;
+
 protected:
   virtual int _send_request() = 0;
 public:
-  RGWAsyncRadosRequest(RGWAioCompletionNotifier *_cn) : notifier(_cn) {}
+  RGWAsyncRadosRequest(RGWAioCompletionNotifier *_cn) : notifier(_cn), done(false), lock("RGWAsyncRadosRequest::lock") {}
   virtual ~RGWAsyncRadosRequest() {}
 
   void send_request() {
     retcode = _send_request();
-    notifier->cb();
+    {
+      Mutex::Locker l(lock);
+      if (!done) {
+        notifier->cb();
+      }
+    }
   }
 
   int get_ret_status() { return retcode; }
+
+  void finish() {
+    {
+      Mutex::Locker l(lock);
+      done = true;
+    }
+    put();
+  }
 };
 
 
@@ -168,7 +185,9 @@ public:
                                                 req(NULL) { }
                                                          
   ~RGWSimpleRadosReadCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request();
@@ -241,7 +260,9 @@ public:
                                                 req(NULL) { }
                                                          
   ~RGWSimpleRadosReadAttrsCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request();
@@ -271,7 +292,9 @@ public:
   }
 
   ~RGWSimpleRadosWriteCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {
@@ -305,11 +328,13 @@ public:
                                                 async_rados(_async_rados),
                                                store(_store),
                                                pool(_pool), oid(_oid),
-                                                attrs(_attrs) {
+                                                attrs(_attrs), req(NULL) {
   }
 
   ~RGWSimpleRadosWriteAttrsCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {
@@ -478,12 +503,14 @@ public:
   RGWWaitCR(RGWAsyncRadosProcessor *_async_rados, CephContext *_cct,
            Mutex *_lock, Cond *_cond,
             int _secs) : RGWSimpleCoroutine(cct), cct(_cct),
-                         async_rados(_async_rados), lock(_lock), cond(_cond), secs(_secs) {
+                         async_rados(_async_rados), lock(_lock), cond(_cond), secs(_secs), req(NULL) {
   }
 
   ~RGWWaitCR() {
     wakeup();
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {
@@ -578,7 +605,9 @@ public:
                                                        bucket_name(_bucket_name), bucket_id(_bucket_id),
                                                        bucket_info(_bucket_info), req(NULL) {}
   ~RGWGetBucketInstanceInfoCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {
@@ -653,7 +682,9 @@ public:
 
 
   ~RGWFetchRemoteObjCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {
@@ -750,7 +781,7 @@ public:
                                        key(_key),
                                        versioned(_versioned),
                                        versioned_epoch(_versioned_epoch),
-                                       delete_marker(_delete_marker) {
+                                       delete_marker(_delete_marker), req(NULL) {
     del_if_older = (_timestamp != NULL);
     if (_timestamp) {
       timestamp = *_timestamp;
@@ -766,7 +797,9 @@ public:
   }
 
   ~RGWRemoveObjCR() {
-    delete req;
+    if (req) {
+      req->finish();
+    }
   }
 
   int send_request() {