From f6c3fd9994a11b7f8e588524d08857049ef17f89 Mon Sep 17 00:00:00 2001
From: Haomai Wang <haomai@xsky.com>
Date: Mon, 18 Jul 2016 17:27:31 +0800
Subject: [PATCH] test_msgr: ensure wait for ms_dispatch executed

Otherwise in slow runner, ms_dispatch may not executed when assert

Signed-off-by: Haomai Wang <haomai@xsky.com>
---
 src/msg/async/AsyncConnection.cc |  2 +-
 src/test/msgr/test_msgr.cc       | 30 +++++++++++++++++++++++++-----
 2 files changed, 26 insertions(+), 6 deletions(-)

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 498cb03ff665e..96142c3b661e5 100644
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -2006,8 +2006,8 @@ void AsyncConnection::fault()
   }
 
   write_lock.lock();
-  shutdown_socket();
   can_write = WriteStatus::NOWRITE;
+  shutdown_socket();
   open_write = false;
 
   // queue delayed items immediately
diff --git a/src/test/msgr/test_msgr.cc b/src/test/msgr/test_msgr.cc
index a94c77125b294..05fcbbe4ba437 100644
--- a/src/test/msgr/test_msgr.cc
+++ b/src/test/msgr/test_msgr.cc
@@ -175,6 +175,7 @@ class FakeDispatcher : public Dispatcher {
       s->put();
     }
     got_remote_reset = true;
+    cond.Signal();
   }
   void ms_fast_dispatch(Message *m) {
     Mutex::Locker l(lock);
@@ -450,6 +451,7 @@ TEST_P(MessengerTest, StatefulTest) {
   // don't lose state
   ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);
 
+  srv_dispatcher.got_new = false;
   conn = client_msgr->get_connection(server_msgr->get_myinst());
   {
     m = new MPing();
@@ -461,14 +463,26 @@ TEST_P(MessengerTest, StatefulTest) {
   }
   ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
   server_conn = server_msgr->get_connection(client_msgr->get_myinst());
-  ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);
+  {
+    Mutex::Locker l(srv_dispatcher.lock);
+    while (!srv_dispatcher.got_remote_reset)
+      srv_dispatcher.cond.Wait(srv_dispatcher.lock);
+  }
 
   // 2. test for client reconnect
   ASSERT_FALSE(cli_dispatcher.got_remote_reset);
   cli_dispatcher.got_connect = false;
+  cli_dispatcher.got_new = false;
+  cli_dispatcher.got_remote_reset = false;
   server_conn->mark_down();
   ASSERT_FALSE(server_conn->is_connected());
   // ensure client detect server socket closed
+  {
+    Mutex::Locker l(cli_dispatcher.lock);
+    while (!cli_dispatcher.got_remote_reset)
+      cli_dispatcher.cond.Wait(cli_dispatcher.lock);
+    cli_dispatcher.got_remote_reset = false;
+  }
   {
     Mutex::Locker l(cli_dispatcher.lock);
     while (!cli_dispatcher.got_connect)
@@ -477,6 +491,7 @@ TEST_P(MessengerTest, StatefulTest) {
   }
   CHECK_AND_WAIT_TRUE(conn->is_connected());
   ASSERT_TRUE(conn->is_connected());
+
   {
     m = new MPing();
     ASSERT_EQ(conn->send_message(m), 0);
@@ -487,10 +502,9 @@ TEST_P(MessengerTest, StatefulTest) {
     cli_dispatcher.got_new = false;
   }
   // resetcheck happen
-  ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
+  ASSERT_EQ(static_cast<Session*>(conn->get_priv())->get_count(), 1);
   server_conn = server_msgr->get_connection(client_msgr->get_myinst());
-  ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);
-  ASSERT_TRUE(cli_dispatcher.got_remote_reset);
+  ASSERT_EQ(static_cast<Session*>(server_conn->get_priv())->get_count(), 1);
   cli_dispatcher.got_remote_reset = false;
 
   server_msgr->shutdown();
@@ -529,6 +543,7 @@ TEST_P(MessengerTest, StatelessTest) {
   conn->mark_down();
   ASSERT_FALSE(conn->is_connected());
 
+  srv_dispatcher.got_new = false;
   conn = client_msgr->get_connection(server_msgr->get_myinst());
   {
     m = new MPing();
@@ -541,7 +556,12 @@ TEST_P(MessengerTest, StatelessTest) {
   ASSERT_TRUE(static_cast<Session*>(conn->get_priv())->get_count() == 1);
   ConnectionRef server_conn = server_msgr->get_connection(client_msgr->get_myinst());
   // server lose state
-  ASSERT_TRUE(static_cast<Session*>(server_conn->get_priv())->get_count() == 1);
+  {
+    Mutex::Locker l(srv_dispatcher.lock);
+    while (!srv_dispatcher.got_new)
+      srv_dispatcher.cond.Wait(srv_dispatcher.lock);
+  }
+  ASSERT_EQ(static_cast<Session*>(server_conn->get_priv())->get_count(), 1);
 
   // 2. test for client lossy
   server_conn->mark_down();
-- 
2.39.5