AsyncConnection: Add SyntheticInjectTest for lossy connection

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 8db6849302b80ae5d4772f75957ecf7d3835b7d6..cbcb5dc04c5221089a45c8dea16c169798d0b0af 100644 (file)
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -1674,8 +1674,6 @@ int AsyncConnection::handle_connect_msg(ceph_msg_connect &connect, bufferlist &a
                             << existing << ".cseq " << existing->connect_seq
                             << " == " << connect.connect_seq << ", sending WAIT" << dendl;
         assert(peer_addr > async_msgr->get_myaddr());
-        // make sure our outgoing connection will follow through
-        existing->send_keepalive();
         existing->lock.Unlock();
         return _reply_accept(CEPH_MSGR_TAG_WAIT, connect, reply, authorizer_reply);
       }

diff --git a/src/test/msgr/test_msgr.cc b/src/test/msgr/test_msgr.cc
index 3e2fadbddfa1ac5b6ed1ac3c41397b047c99b08f..859b3887b8fa2d30edc4fac3f10e84044448da47 100644 (file)
--- a/src/test/msgr/test_msgr.cc
+++ b/src/test/msgr/test_msgr.cc
@@ -919,14 +919,10 @@ class SyntheticWorkload {
     pair<Messenger*, Messenger*> p;
     {
       boost::uniform_int<> choose(0, available_servers.size() - 1);
-      int index = choose(rng);
-      if (server->get_default_policy().server || index % 2) {
-        conn = client->get_connection(server->get_myinst());
+      if (server->get_default_policy().server || choose(rng) % 2)
         p = make_pair(client, server);
-      } else {
-        conn = server->get_connection(client->get_myinst());
+      else
         p = make_pair(server, client);
-      }
     }
     if (!available_connections.count(p)) {
       ConnectionRef conn = p.first->get_connection(p.second->get_myinst());
@@ -1113,6 +1109,39 @@ TEST_P(MessengerTest, SyntheticInjectTest2) {
   g_ceph_context->_conf->set_val("ms_inject_internal_delays", "0");
 }
 
+TEST_P(MessengerTest, SyntheticInjectTest3) {
+  g_ceph_context->_conf->set_val("ms_inject_socket_failures", "600");
+  g_ceph_context->_conf->set_val("ms_inject_internal_delays", "0.1");
+  SyntheticWorkload test_msg(8, 16, GetParam(), 100,
+                             Messenger::Policy::stateless_server(0, 0),
+                             Messenger::Policy::lossy_client(0, 0));
+  for (int i = 0; i < 100; ++i) {
+    if (!(i % 10)) cerr << "seeding connection " << i << std::endl;
+    test_msg.generate_connection();
+  }
+  gen_type rng(time(NULL));
+  for (int i = 0; i < 1000; ++i) {
+    if (!(i % 10)) {
+      cerr << "Op " << i << ": ";
+      test_msg.print_internal_state();
+    }
+    boost::uniform_int<> true_false(0, 99);
+    int val = true_false(rng);
+    if (val > 90) {
+      test_msg.generate_connection();
+    } else if (val > 80) {
+      test_msg.drop_connection();
+    } else if (val > 10) {
+      test_msg.send_message();
+    } else {
+      usleep(rand() % 500 + 100);
+    }
+  }
+  test_msg.wait_for_done();
+  g_ceph_context->_conf->set_val("ms_inject_socket_failures", "0");
+  g_ceph_context->_conf->set_val("ms_inject_internal_delays", "0");
+}
+
 
 class MarkdownDispatcher : public Dispatcher {
   Mutex lock;