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>

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

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();