msg/simple: wait dispatch_queue until all pipes closed

Otherwise accepter may continue to accept new connection which may queue
new item.

Fixes: http://tracker.ceph.com/issues/16472
Signed-off-by: Haomai Wang <haomai@xsky.com>

src/msg/simple/SimpleMessenger.cc

@@ -87,7 +87,6 @@ int SimpleMessenger::shutdown()
 {
   ldout(cct,10) << "shutdown " << get_myaddr() << dendl;
   mark_down_all();
-  dispatch_queue.shutdown();
 
   // break ref cycles on the loopback connection
   local_connection->set_priv(NULL);
@@ -527,13 +526,6 @@ void SimpleMessenger::wait()
   }
   lock.Unlock();
 
-  if (dispatch_queue.is_started()) {
-    ldout(cct,10) << "wait: waiting for dispatch queue" << dendl;
-    dispatch_queue.wait();
-    dispatch_queue.discard_local();
-    ldout(cct,10) << "wait: dispatch queue is stopped" << dendl;
-  }
-
   // done!  clean up.
   if (did_bind) {
     ldout(cct,20) << "wait: stopping accepter thread" << dendl;
@@ -542,6 +534,14 @@ void SimpleMessenger::wait()
     ldout(cct,20) << "wait: stopped accepter thread" << dendl;
   }
 
+  dispatch_queue.shutdown();
+  if (dispatch_queue.is_started()) {
+    ldout(cct,10) << "wait: waiting for dispatch queue" << dendl;
+    dispatch_queue.wait();
+    dispatch_queue.discard_local();
+    ldout(cct,10) << "wait: dispatch queue is stopped" << dendl;
+  }
+
   if (reaper_started) {
     ldout(cct,20) << "wait: stopping reaper thread" << dendl;
     lock.Lock();