OSD: Add rep_scrub_wq

Previously, replica scrubs would be handled in sub_op_scrub in the op
queue.  Replica scrubs will now be processed by rep_scrub_wq using the
disk tp.

Signed-off-by: Samuel Just <samuel.just@dreamhost.com>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index bf2da5a2193403497b2c23c2c17fabb8bf884e21..c7e549b5031e0a0d73a188729660f0a8c3f5de65 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -384,6 +384,7 @@ OSD::OSD(int id, Messenger *internal_messenger, Messenger *external_messenger, M
   scrubs_pending(0),
   scrubs_active(0),
   scrub_wq(this, &disk_tp),
+  rep_scrub_wq(this, &disk_tp),
   remove_wq(this, &disk_tp)
 {
   monc->set_messenger(client_messenger);

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 429d02f8c3e34767edbd0802cb5bda0391064a9a..6d8f0826dc87e684f1c3940aaead0f88a345fac7 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -34,6 +34,7 @@
 #include "include/CompatSet.h"
 
 #include "auth/KeyRing.h"
+#include "messages/MOSDRepScrub.h"
 
 #include <map>
 #include <memory>
@@ -873,6 +874,54 @@ protected:
     }
   } scrub_wq;
 
+  struct RepScrubWQ : public ThreadPool::WorkQueue<MOSDRepScrub> {
+  private: 
+    OSD *osd;
+    list<MOSDRepScrub*> rep_scrub_queue;
+
+  public:
+    RepScrubWQ(OSD *o, ThreadPool *tp) : 
+      ThreadPool::WorkQueue<MOSDRepScrub>("OSD::RepScrubWQ", tp), osd(o) {}
+
+    bool _empty() {
+      return rep_scrub_queue.empty();
+    }
+    bool _enqueue(MOSDRepScrub *msg) {
+      rep_scrub_queue.push_back(msg);
+      return true;
+    }
+    void _dequeue(MOSDRepScrub *msg) {
+      assert(0); // Not applicable for this wq
+      return;
+    }
+    MOSDRepScrub *_dequeue() {
+      if (rep_scrub_queue.empty())
+       return NULL;
+      MOSDRepScrub *msg = rep_scrub_queue.front();
+      rep_scrub_queue.pop_front();
+      return msg;
+    }
+    void _process(MOSDRepScrub *msg) {
+      osd->osd_lock.Lock();
+      if (osd->_have_pg(msg->pgid)) {
+       PG *pg = osd->_lookup_lock_pg(msg->pgid);
+       osd->osd_lock.Unlock();
+       pg->replica_scrub(msg);
+       pg->unlock();
+      } else {
+       msg->put();
+       osd->osd_lock.Unlock();
+      }
+    }
+    void _clear() {
+      while (!rep_scrub_queue.empty()) {
+       MOSDRepScrub *msg = rep_scrub_queue.front();
+       rep_scrub_queue.pop_front();
+       msg->put();
+      }
+    }
+  } rep_scrub_wq;
+
   // -- removing --
   xlist<PG*> remove_queue;