OSD: process_peering_event check for new map on each pg

Signed-off-by: Samuel Just <sam.just@inktank.com>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 6b1789d9db062e5e482597e0d11b917360ce48be..32455d11a3246848064fa53ac2a64098598de266 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -5190,6 +5190,7 @@ void OSD::process_peering_events(const list<PG*> &pgs)
        ++i) {
     PG *pg = *i;
     pg->lock();
+    curmap = service.get_osdmap();
     if (pg->deleting) {
       pg->unlock();
       continue;

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index f897e5b5ff91b0532c14e5096dbf313933859492..9874e6ffd5e295c4a60a5a7afb6ff535bdd78038 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -569,9 +569,11 @@ private:
   struct PeeringWQ : public ThreadPool::BatchWorkQueue<PG> {
     list<PG*> peering_queue;
     OSD *osd;
+    set<PG*> in_use;
+    const size_t batch_size;
     PeeringWQ(OSD *o, time_t ti, ThreadPool *tp, size_t batch_size)
       : ThreadPool::BatchWorkQueue<PG>(
-       "OSD::PeeringWQ", ti, ti*10, tp, batch_size), osd(o) {}
+       "OSD::PeeringWQ", ti, ti*10, tp), osd(o), batch_size(batch_size) {}
 
     void _dequeue(PG *pg) {
       for (list<PG*>::iterator i = peering_queue.begin();
@@ -591,12 +593,20 @@ private:
     bool _empty() {
       return peering_queue.empty();
     }
-    PG *_dequeue() {
-      if (peering_queue.empty())
-       return 0;
-      PG *retval = peering_queue.front();
-      peering_queue.pop_front();
-      return retval;
+    void _dequeue(list<PG*> *out) {
+      set<PG*> got;
+      for (list<PG*>::iterator i = peering_queue.begin();
+          i != peering_queue.end() && out->size() < batch_size;
+          ) {
+       if (in_use.count(*i)) {
+         ++i;
+       } else {
+         out->push_back(*i);
+         got.insert(*i);
+         peering_queue.erase(i++);
+       }
+      }
+      in_use.insert(got.begin(), got.end());
     }
     void _process(const list<PG *> &pgs) {
       osd->process_peering_events(pgs);
@@ -606,6 +616,13 @@ private:
        (*i)->put();
       }
     }
+    void _process_finish(const list<PG *> &pgs) {
+      for (list<PG*>::const_iterator i = pgs.begin();
+          i != pgs.end();
+          ++i) {
+       in_use.erase(*i);
+      }
+    }
     void _clear() {
       assert(peering_queue.empty());
     }