osd: preserve ordering when ops are requeued

Requeue ops under osd_lock to preserve ordering wrt incoming messages.
Also drain the waiter queue when ms_dispatch takes the lock before calling
_dispatch(m).

Fixes: #743
Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 4cdd8eff1a30092d44ec09e40a9792ae351d2121..95e09df931e7dbd15c9ecd363dc4e973617f3988 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -2048,6 +2048,7 @@ bool OSD::ms_dispatch(Message *m)
 {
   // lock!
   osd_lock.Lock();
+  do_waiters();
   ++dispatch_running;
   _dispatch(m);
   --dispatch_running;
@@ -2520,7 +2521,12 @@ void OSD::handle_osd_map(MOSDMap *m)
   osd_lock.Unlock();
 
   op_tp.pause();
+
+  // requeue under osd_lock to preserve ordering of _dispatch() wrt incoming messages
+  osd_lock.Lock();  
+
   op_wq.lock();
+
   list<Message*> rq;
   while (!op_queue.empty()) {
     PG *pg = op_queue.back();
@@ -2534,9 +2540,8 @@ void OSD::handle_osd_map(MOSDMap *m)
     dout(15) << " will requeue " << *mess << dendl;
     rq.push_front(mess);
   }
+  push_waiters(rq);  // requeue under osd_lock!
   op_wq.unlock();
-  push_waiters(rq);
-  osd_lock.Lock();
 
   recovery_tp.pause();
   disk_tp.pause_new();   // _process() may be waiting for a replica message

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 5961add8e9a631095eb8c5ca6bbe15fbb9d7f495..5e9e16b3f0aa2e79d00392f000be9d33551338a4 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -369,6 +369,7 @@ private:
     finished_lock.Unlock();
   }
   void push_waiters(list<class Message*>& ls) {
+    assert(osd_lock.is_locked());   // currently, at least.  be careful if we change this (see #743)
     finished_lock.Lock();
     finished.splice(finished.begin(), ls);
     finished_lock.Unlock();