AsyncConnection: eliminate reader vs writer ack race in process()

process() will potentially handle multiple messages, yet we are
scheduling the writer to write out acks only once, after the first
message is processed.  This effectively races the reader with the
writer and leaves the window for exiting the process() loop with
pending acks (ack_left > 0 but no kicked writer).

Close this window by moving dispatch_event_external() out of the loop.

Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 73ef39fd921f4c1c0faa39bd8375bfe18e76dc89..0afdbe66d38f2ff175ca531de3c027de9aa3b310 100644 (file)
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -324,7 +324,7 @@ void AsyncConnection::process()
 {
   ssize_t r = 0;
   int prev_state = state;
-  bool already_dispatch_writer = false;
+  bool need_dispatch_writer = false;
   std::lock_guard<std::mutex> l(lock);
   last_active = ceph::coarse_mono_clock::now();
   do {
@@ -746,13 +746,7 @@ void AsyncConnection::process()
                                    << " " << *message << dendl;
 
           ack_left.inc();
-          // if send_message always send inline, it may have no
-          // opportunity to send seq ack.
-          if (!already_dispatch_writer) {
-            center->dispatch_event_external(write_handler);
-            already_dispatch_writer = true;
-          }
-
+          need_dispatch_writer = true;
           state = STATE_OPEN;
 
           logger->inc(l_msgr_recv_messages);
@@ -821,6 +815,8 @@ void AsyncConnection::process()
     }
   } while (prev_state != state);
 
+  if (need_dispatch_writer && is_connected())
+    center->dispatch_event_external(write_handler);
   return;
 
  fail: