osd: pause/unpause recovery thread while processing map

Otherwise bad things happen (everyone assumes *osdmap is static and
readable).

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 7ce513aad3c12ddd9db814142085ead2529bf21c..e7b82e72330269d546e9cead8a668e0eed752182 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -275,7 +275,7 @@ OSD::OSD(int id, Messenger *m, MonMap *mm, const char *dev) :
   pg_stat_queue_lock("OSD::pg_stat_queue_lock"),
   tid_lock("OSD::tid_lock"),
   recovery_lock("OSD::recovery_lock"),
-  recovery_ops_active(0), recovery_stop(false),
+  recovery_ops_active(0), recovery_stop(false), recovery_pause(false),
   remove_list_lock("OSD::remove_list_lock"),
   recovery_thread(this)
 {
@@ -1494,6 +1494,7 @@ void OSD::handle_osd_map(MOSDMap *m)
   booting = boot_pending = false;
 
   wait_for_no_ops();
+  pause_recovery_thread();
   
   assert(osd_lock.is_locked());
 
@@ -1684,6 +1685,8 @@ void OSD::handle_osd_map(MOSDMap *m)
   store->apply_transaction(t);
   store->sync();
 
+  unpause_recovery_thread();
+  
   //if (osdmap->get_epoch() == 1) store->sync();     // in case of early death, blah
 
   delete m;
@@ -2937,7 +2940,7 @@ void OSD::recovery_entry()
   recovery_lock.Lock();
   dout(10) << "recovery_entry - start" << dendl;
   while (!recovery_stop) {
-    while (_recover_now())
+    if (!recovery_pause && _recover_now())
       _do_recovery();
     recovery_cond.Wait(recovery_lock);
   }

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 265cc2a0d2a3f4723193732e497595fd1b3ff058..b4ac9ed33827744dc57a60a4123188dc041706b9 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -430,6 +430,7 @@ private:
   utime_t defer_recovery_until;
   int recovery_ops_active;
   bool recovery_stop;
+  bool recovery_pause;
 
   Mutex remove_list_lock;
   map<epoch_t, map<int, vector<pg_t> > > remove_list;
@@ -454,6 +455,17 @@ private:
     recovery_thread.join();
     osd_lock.Lock();
   }
+  void pause_recovery_thread() {
+    recovery_lock.Lock();
+    recovery_pause = true;
+    recovery_lock.Unlock();
+  }
+  void unpause_recovery_thread() {
+    recovery_lock.Lock();
+    recovery_pause = false;
+    recovery_cond.Signal();
+    recovery_lock.Unlock();
+  }
 
   void queue_for_removal(int osd, pg_t pgid) {
     remove_list_lock.Lock();