diff options
Diffstat (limited to '')
| -rw-r--r-- | src/test/msgr/test_msgr.cc | 2365 |
1 files changed, 2365 insertions, 0 deletions
diff --git a/src/test/msgr/test_msgr.cc b/src/test/msgr/test_msgr.cc new file mode 100644 index 000000000..3ebe320be --- /dev/null +++ b/src/test/msgr/test_msgr.cc @@ -0,0 +1,2365 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2014 UnitedStack <haomai@unitedstack.com> + * + * Author: Haomai Wang <haomaiwang@gmail.com> + * + * This is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License version 2.1, as published by the Free Software + * Foundation. See file COPYING. + * + */ + +#include <atomic> +#include <iostream> +#include <memory> +#include <unistd.h> +#include <stdlib.h> +#include <time.h> +#include <set> +#include <list> +#include "common/ceph_mutex.h" +#include "common/ceph_argparse.h" +#include "global/global_init.h" +#include "msg/Dispatcher.h" +#include "msg/msg_types.h" +#include "msg/Message.h" +#include "msg/Messenger.h" +#include "msg/Connection.h" +#include "messages/MPing.h" +#include "messages/MCommand.h" + +#include <boost/random/mersenne_twister.hpp> +#include <boost/random/uniform_int.hpp> +#include <boost/random/binomial_distribution.hpp> +#include <gtest/gtest.h> + +typedef boost::mt11213b gen_type; + +#include "common/dout.h" +#include "include/ceph_assert.h" + +#include "auth/DummyAuth.h" + +#define dout_subsys ceph_subsys_ms +#undef dout_prefix +#define dout_prefix *_dout << " ceph_test_msgr " + + +#define CHECK_AND_WAIT_TRUE(expr) do { \ + int n = 1000; \ + while (--n) { \ + if (expr) \ + break; \ + usleep(1000); \ + } \ +} while(0); + +class MessengerTest : public ::testing::TestWithParam<const char*> { + public: + DummyAuthClientServer dummy_auth; + Messenger *server_msgr; + Messenger *client_msgr; + + MessengerTest() : dummy_auth(g_ceph_context), + server_msgr(NULL), client_msgr(NULL) { + dummy_auth.auth_registry.refresh_config(); + } + void SetUp() override { + lderr(g_ceph_context) << __func__ << " start set up " << GetParam() << dendl; + server_msgr = Messenger::create(g_ceph_context, string(GetParam()), entity_name_t::OSD(0), "server", getpid()); + client_msgr = Messenger::create(g_ceph_context, string(GetParam()), entity_name_t::CLIENT(-1), "client", getpid()); + server_msgr->set_default_policy(Messenger::Policy::stateless_server(0)); + client_msgr->set_default_policy(Messenger::Policy::lossy_client(0)); + server_msgr->set_auth_client(&dummy_auth); + server_msgr->set_auth_server(&dummy_auth); + client_msgr->set_auth_client(&dummy_auth); + client_msgr->set_auth_server(&dummy_auth); + server_msgr->set_require_authorizer(false); + } + void TearDown() override { + ASSERT_EQ(server_msgr->get_dispatch_queue_len(), 0); + ASSERT_EQ(client_msgr->get_dispatch_queue_len(), 0); + delete server_msgr; + delete client_msgr; + } + +}; + + +class FakeDispatcher : public Dispatcher { + public: + struct Session : public RefCountedObject { + atomic<uint64_t> count; + ConnectionRef con; + + explicit Session(ConnectionRef c): RefCountedObject(g_ceph_context), count(0), con(c) { + } + uint64_t get_count() { return count; } + }; + + ceph::mutex lock = ceph::make_mutex("FakeDispatcher::lock"); + ceph::condition_variable cond; + bool is_server; + bool got_new; + bool got_remote_reset; + bool got_connect; + bool loopback; + entity_addrvec_t last_accept; + ConnectionRef *last_accept_con_ptr = nullptr; + + explicit FakeDispatcher(bool s): Dispatcher(g_ceph_context), + is_server(s), got_new(false), got_remote_reset(false), + got_connect(false), loopback(false) { + } + bool ms_can_fast_dispatch_any() const override { return true; } + bool ms_can_fast_dispatch(const Message *m) const override { + switch (m->get_type()) { + case CEPH_MSG_PING: + return true; + default: + return false; + } + } + + void ms_handle_fast_connect(Connection *con) override { + std::scoped_lock l{lock}; + lderr(g_ceph_context) << __func__ << " " << con << dendl; + auto s = con->get_priv(); + if (!s) { + auto session = new Session(con); + con->set_priv(RefCountedPtr{session, false}); + lderr(g_ceph_context) << __func__ << " con: " << con + << " count: " << session->count << dendl; + } + got_connect = true; + cond.notify_all(); + } + void ms_handle_fast_accept(Connection *con) override { + last_accept = con->get_peer_addrs(); + if (last_accept_con_ptr) { + *last_accept_con_ptr = con; + } + if (!con->get_priv()) { + con->set_priv(RefCountedPtr{new Session(con), false}); + } + } + bool ms_dispatch(Message *m) override { + auto priv = m->get_connection()->get_priv(); + auto s = static_cast<Session*>(priv.get()); + if (!s) { + s = new Session(m->get_connection()); + priv.reset(s, false); + m->get_connection()->set_priv(priv); + } + s->count++; + lderr(g_ceph_context) << __func__ << " conn: " << m->get_connection() << " session " << s << " count: " << s->count << dendl; + if (is_server) { + reply_message(m); + } + std::lock_guard l{lock}; + got_new = true; + cond.notify_all(); + m->put(); + return true; + } + bool ms_handle_reset(Connection *con) override { + std::lock_guard l{lock}; + lderr(g_ceph_context) << __func__ << " " << con << dendl; + auto priv = con->get_priv(); + if (auto s = static_cast<Session*>(priv.get()); s) { + s->con.reset(); // break con <-> session ref cycle + con->set_priv(nullptr); // break ref <-> session cycle, if any + } + return true; + } + void ms_handle_remote_reset(Connection *con) override { + std::lock_guard l{lock}; + lderr(g_ceph_context) << __func__ << " " << con << dendl; + auto priv = con->get_priv(); + if (auto s = static_cast<Session*>(priv.get()); s) { + s->con.reset(); // break con <-> session ref cycle + con->set_priv(nullptr); // break ref <-> session cycle, if any + } + got_remote_reset = true; + cond.notify_all(); + } + bool ms_handle_refused(Connection *con) override { + return false; + } + void ms_fast_dispatch(Message *m) override { + auto priv = m->get_connection()->get_priv(); + auto s = static_cast<Session*>(priv.get()); + if (!s) { + s = new Session(m->get_connection()); + priv.reset(s, false); + m->get_connection()->set_priv(priv); + } + s->count++; + lderr(g_ceph_context) << __func__ << " conn: " << m->get_connection() << " session " << s << " count: " << s->count << dendl; + if (is_server) { + if (loopback) + ceph_assert(m->get_source().is_osd()); + else + reply_message(m); + } else if (loopback) { + ceph_assert(m->get_source().is_client()); + } + m->put(); + std::lock_guard l{lock}; + got_new = true; + cond.notify_all(); + } + + int ms_handle_authentication(Connection *con) override { + return 1; + } + + void reply_message(Message *m) { + MPing *rm = new MPing(); + m->get_connection()->send_message(rm); + } +}; + +typedef FakeDispatcher::Session Session; + +struct TestInterceptor : public Interceptor { + + bool step_waiting = false; + bool waiting = true; + std::map<Connection *, uint32_t> current_step; + std::map<Connection *, std::list<uint32_t>> step_history; + std::map<uint32_t, std::optional<ACTION>> decisions; + std::set<uint32_t> breakpoints; + + uint32_t count_step(Connection *conn, uint32_t step) { + uint32_t count = 0; + for (auto s : step_history[conn]) { + if (s == step) { + count++; + } + } + return count; + } + + void breakpoint(uint32_t step) { + breakpoints.insert(step); + } + + void remove_bp(uint32_t step) { + breakpoints.erase(step); + } + + Connection *wait(uint32_t step, Connection *conn=nullptr) { + std::unique_lock<std::mutex> l(lock); + while(true) { + if (conn) { + auto it = current_step.find(conn); + if (it != current_step.end()) { + if (it->second == step) { + break; + } + } + } else { + for (auto it : current_step) { + if (it.second == step) { + conn = it.first; + break; + } + } + if (conn) { + break; + } + } + step_waiting = true; + cond_var.wait(l); + } + step_waiting = false; + return conn; + } + + ACTION wait_for_decision(uint32_t step, std::unique_lock<std::mutex> &l) { + if (decisions[step]) { + return *(decisions[step]); + } + waiting = true; + cond_var.wait(l, [this] { return !waiting; }); + return *(decisions[step]); + } + + void proceed(uint32_t step, ACTION decision) { + std::unique_lock<std::mutex> l(lock); + decisions[step] = decision; + if (waiting) { + waiting = false; + cond_var.notify_one(); + } + } + + ACTION intercept(Connection *conn, uint32_t step) override { + lderr(g_ceph_context) << __func__ << " conn(" << conn + << ") intercept called on step=" << step << dendl; + + { + std::unique_lock<std::mutex> l(lock); + step_history[conn].push_back(step); + current_step[conn] = step; + if (step_waiting) { + cond_var.notify_one(); + } + } + + std::unique_lock<std::mutex> l(lock); + ACTION decision = ACTION::CONTINUE; + if (breakpoints.find(step) != breakpoints.end()) { + lderr(g_ceph_context) << __func__ << " conn(" << conn + << ") pausing on step=" << step << dendl; + decision = wait_for_decision(step, l); + } else { + if (decisions[step]) { + decision = *(decisions[step]); + } + } + lderr(g_ceph_context) << __func__ << " conn(" << conn + << ") resuming step=" << step << " with decision=" + << decision << dendl; + decisions[step].reset(); + return decision; + } + +}; + +/** + * Scenario: A connects to B, and B connects to A at the same time. + */ +TEST_P(MessengerTest, ConnectionRaceTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(false); + + TestInterceptor *cli_interceptor = new TestInterceptor(); + TestInterceptor *srv_interceptor = new TestInterceptor(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, Messenger::Policy::lossless_peer_reuse(0)); + server_msgr->interceptor = srv_interceptor; + + client_msgr->set_policy(entity_name_t::TYPE_OSD, Messenger::Policy::lossless_peer_reuse(0)); + client_msgr->interceptor = cli_interceptor; + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // pause before sending client_ident message + cli_interceptor->breakpoint(Interceptor::STEP::SEND_CLIENT_IDENTITY); + // pause before sending client_ident message + srv_interceptor->breakpoint(Interceptor::STEP::SEND_CLIENT_IDENTITY); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + MPing *m1 = new MPing(); + ASSERT_EQ(c2s->send_message(m1), 0); + + ConnectionRef s2c = server_msgr->connect_to(client_msgr->get_mytype(), + client_msgr->get_myaddrs()); + MPing *m2 = new MPing(); + ASSERT_EQ(s2c->send_message(m2), 0); + + cli_interceptor->wait(Interceptor::STEP::SEND_CLIENT_IDENTITY, c2s.get()); + srv_interceptor->wait(Interceptor::STEP::SEND_CLIENT_IDENTITY, s2c.get()); + + // at this point both connections (A->B, B->A) are paused just before sending + // the client_ident message. + + cli_interceptor->remove_bp(Interceptor::STEP::SEND_CLIENT_IDENTITY); + srv_interceptor->remove_bp(Interceptor::STEP::SEND_CLIENT_IDENTITY); + + cli_interceptor->proceed(Interceptor::STEP::SEND_CLIENT_IDENTITY, Interceptor::ACTION::CONTINUE); + srv_interceptor->proceed(Interceptor::STEP::SEND_CLIENT_IDENTITY, Interceptor::ACTION::CONTINUE); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + { + std::unique_lock l{srv_dispatcher.lock}; + srv_dispatcher.cond.wait(l, [&] { return srv_dispatcher.got_new; }); + srv_dispatcher.got_new = false; + } + + ASSERT_TRUE(s2c->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(s2c->get_priv().get())->get_count()); + ASSERT_TRUE(s2c->peer_is_client()); + + ASSERT_TRUE(c2s->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + ASSERT_TRUE(c2s->peer_is_osd()); + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); + + delete cli_interceptor; + delete srv_interceptor; +} + +/** + * Scenario: A connects to B, and B connects to A at the same time. + * The first (A -> B) connection gets to message flow handshake, the + * second (B -> A) connection is stuck waiting for a banner from A. + * After A sends client_ident to B, the first connection wins and B + * calls reuse_connection() to replace the second connection's socket + * while the second connection is still in BANNER_CONNECTING. + */ +TEST_P(MessengerTest, ConnectionRaceReuseBannerTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(false); + + auto cli_interceptor = std::make_unique<TestInterceptor>(); + auto srv_interceptor = std::make_unique<TestInterceptor>(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, + Messenger::Policy::lossless_peer_reuse(0)); + server_msgr->interceptor = srv_interceptor.get(); + + client_msgr->set_policy(entity_name_t::TYPE_OSD, + Messenger::Policy::lossless_peer_reuse(0)); + client_msgr->interceptor = cli_interceptor.get(); + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // pause before sending client_ident message + srv_interceptor->breakpoint(Interceptor::STEP::SEND_CLIENT_IDENTITY); + + ConnectionRef s2c = server_msgr->connect_to(client_msgr->get_mytype(), + client_msgr->get_myaddrs()); + MPing *m1 = new MPing(); + ASSERT_EQ(s2c->send_message(m1), 0); + + srv_interceptor->wait(Interceptor::STEP::SEND_CLIENT_IDENTITY); + srv_interceptor->remove_bp(Interceptor::STEP::SEND_CLIENT_IDENTITY); + + // pause before sending banner + cli_interceptor->breakpoint(Interceptor::STEP::BANNER_EXCHANGE_BANNER_CONNECTING); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + MPing *m2 = new MPing(); + ASSERT_EQ(c2s->send_message(m2), 0); + + cli_interceptor->wait(Interceptor::STEP::BANNER_EXCHANGE_BANNER_CONNECTING); + cli_interceptor->remove_bp(Interceptor::STEP::BANNER_EXCHANGE_BANNER_CONNECTING); + + // second connection is in BANNER_CONNECTING, ensure it stays so + // and send client_ident + srv_interceptor->breakpoint(Interceptor::STEP::BANNER_EXCHANGE); + srv_interceptor->proceed(Interceptor::STEP::SEND_CLIENT_IDENTITY, Interceptor::ACTION::CONTINUE); + + // handle client_ident -- triggers reuse_connection() with exproto + // in BANNER_CONNECTING + cli_interceptor->breakpoint(Interceptor::STEP::READY); + cli_interceptor->proceed(Interceptor::STEP::BANNER_EXCHANGE_BANNER_CONNECTING, Interceptor::ACTION::CONTINUE); + + cli_interceptor->wait(Interceptor::STEP::READY); + cli_interceptor->remove_bp(Interceptor::STEP::READY); + + // first connection is in READY + Connection *s2c_accepter = srv_interceptor->wait(Interceptor::STEP::BANNER_EXCHANGE); + srv_interceptor->remove_bp(Interceptor::STEP::BANNER_EXCHANGE); + + srv_interceptor->proceed(Interceptor::STEP::BANNER_EXCHANGE, Interceptor::ACTION::CONTINUE); + cli_interceptor->proceed(Interceptor::STEP::READY, Interceptor::ACTION::CONTINUE); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + { + std::unique_lock l{srv_dispatcher.lock}; + srv_dispatcher.cond.wait(l, [&] { return srv_dispatcher.got_new; }); + srv_dispatcher.got_new = false; + } + + EXPECT_TRUE(s2c->is_connected()); + EXPECT_EQ(1u, static_cast<Session*>(s2c->get_priv().get())->get_count()); + EXPECT_TRUE(s2c->peer_is_client()); + + EXPECT_TRUE(c2s->is_connected()); + EXPECT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + EXPECT_TRUE(c2s->peer_is_osd()); + + // closed in reuse_connection() -- EPIPE when writing banner/hello + EXPECT_FALSE(s2c_accepter->is_connected()); + + // established exactly once, never faulted and reconnected + EXPECT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::START_CLIENT_BANNER_EXCHANGE), 1u); + EXPECT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SEND_RECONNECT), 0u); + EXPECT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::READY), 1u); + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); +} + +/** + * Scenario: + * - A connects to B + * - A sends client_ident to B + * - B fails before sending server_ident to A + * - A reconnects + */ +TEST_P(MessengerTest, MissingServerIdenTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(false); + + TestInterceptor *cli_interceptor = new TestInterceptor(); + TestInterceptor *srv_interceptor = new TestInterceptor(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, Messenger::Policy::stateful_server(0)); + server_msgr->interceptor = srv_interceptor; + + client_msgr->set_policy(entity_name_t::TYPE_OSD, Messenger::Policy::lossy_client(0)); + client_msgr->interceptor = cli_interceptor; + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // pause before sending server_ident message + srv_interceptor->breakpoint(Interceptor::STEP::SEND_SERVER_IDENTITY); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + MPing *m1 = new MPing(); + ASSERT_EQ(c2s->send_message(m1), 0); + + Connection *c2s_accepter = srv_interceptor->wait(Interceptor::STEP::SEND_SERVER_IDENTITY); + srv_interceptor->remove_bp(Interceptor::STEP::SEND_SERVER_IDENTITY); + + // We inject a message from this side of the connection to force it to be + // in standby when we inject the failure below + MPing *m2 = new MPing(); + ASSERT_EQ(c2s_accepter->send_message(m2), 0); + + srv_interceptor->proceed(Interceptor::STEP::SEND_SERVER_IDENTITY, Interceptor::ACTION::FAIL); + + { + std::unique_lock l{srv_dispatcher.lock}; + srv_dispatcher.cond.wait(l, [&] { return srv_dispatcher.got_new; }); + srv_dispatcher.got_new = false; + } + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + ASSERT_TRUE(c2s->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + ASSERT_TRUE(c2s->peer_is_osd()); + + ASSERT_TRUE(c2s_accepter->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s_accepter->get_priv().get())->get_count()); + ASSERT_TRUE(c2s_accepter->peer_is_client()); + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); + + delete cli_interceptor; + delete srv_interceptor; +} + +/** + * Scenario: + * - A connects to B + * - A sends client_ident to B + * - B fails before sending server_ident to A + * - A goes to standby + * - B reconnects to A + */ +TEST_P(MessengerTest, MissingServerIdenTest2) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(false); + + TestInterceptor *cli_interceptor = new TestInterceptor(); + TestInterceptor *srv_interceptor = new TestInterceptor(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, Messenger::Policy::lossless_peer(0)); + server_msgr->interceptor = srv_interceptor; + + client_msgr->set_policy(entity_name_t::TYPE_OSD, Messenger::Policy::lossless_peer(0)); + client_msgr->interceptor = cli_interceptor; + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // pause before sending server_ident message + srv_interceptor->breakpoint(Interceptor::STEP::SEND_SERVER_IDENTITY); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + + Connection *c2s_accepter = srv_interceptor->wait(Interceptor::STEP::SEND_SERVER_IDENTITY); + srv_interceptor->remove_bp(Interceptor::STEP::SEND_SERVER_IDENTITY); + + // We inject a message from this side of the connection to force it to be + // in standby when we inject the failure below + MPing *m2 = new MPing(); + ASSERT_EQ(c2s_accepter->send_message(m2), 0); + + srv_interceptor->proceed(Interceptor::STEP::SEND_SERVER_IDENTITY, Interceptor::ACTION::FAIL); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + ASSERT_TRUE(c2s->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + ASSERT_TRUE(c2s->peer_is_osd()); + + ASSERT_TRUE(c2s_accepter->is_connected()); + ASSERT_EQ(0u, static_cast<Session*>(c2s_accepter->get_priv().get())->get_count()); + ASSERT_TRUE(c2s_accepter->peer_is_client()); + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); + + delete cli_interceptor; + delete srv_interceptor; +} + +/** + * Scenario: + * - A connects to B + * - A and B exchange messages + * - A fails + * - B goes into standby + * - A reconnects + */ +TEST_P(MessengerTest, ReconnectTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + + TestInterceptor *cli_interceptor = new TestInterceptor(); + TestInterceptor *srv_interceptor = new TestInterceptor(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, Messenger::Policy::stateful_server(0)); + server_msgr->interceptor = srv_interceptor; + + client_msgr->set_policy(entity_name_t::TYPE_OSD, Messenger::Policy::lossless_peer(0)); + client_msgr->interceptor = cli_interceptor; + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + + MPing *m1 = new MPing(); + ASSERT_EQ(c2s->send_message(m1), 0); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + ASSERT_TRUE(c2s->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + ASSERT_TRUE(c2s->peer_is_osd()); + + cli_interceptor->breakpoint(Interceptor::STEP::HANDLE_MESSAGE); + + MPing *m2 = new MPing(); + ASSERT_EQ(c2s->send_message(m2), 0); + + cli_interceptor->wait(Interceptor::STEP::HANDLE_MESSAGE, c2s.get()); + cli_interceptor->remove_bp(Interceptor::STEP::HANDLE_MESSAGE); + + // at this point client and server are connected together + + srv_interceptor->breakpoint(Interceptor::STEP::READY); + + // failing client + cli_interceptor->proceed(Interceptor::STEP::HANDLE_MESSAGE, Interceptor::ACTION::FAIL); + + MPing *m3 = new MPing(); + ASSERT_EQ(c2s->send_message(m3), 0); + + Connection *c2s_accepter = srv_interceptor->wait(Interceptor::STEP::READY); + // the srv end of theconnection is now paused at ready + // this means that the reconnect was successful + srv_interceptor->remove_bp(Interceptor::STEP::READY); + + ASSERT_TRUE(c2s_accepter->peer_is_client()); + // c2s_accepter sent 0 reconnect messages + ASSERT_EQ(srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT), 0u); + // c2s_accepter sent 1 reconnect_ok messages + ASSERT_EQ(srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT_OK), 1u); + // c2s sent 1 reconnect messages + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SEND_RECONNECT), 1u); + // c2s sent 0 reconnect_ok messages + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SEND_RECONNECT_OK), 0u); + + srv_interceptor->proceed(15, Interceptor::ACTION::CONTINUE); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); + + delete cli_interceptor; + delete srv_interceptor; +} + +/** + * Scenario: + * - A connects to B + * - A and B exchange messages + * - A fails + * - A reconnects // B reconnects + */ +TEST_P(MessengerTest, ReconnectRaceTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + + TestInterceptor *cli_interceptor = new TestInterceptor(); + TestInterceptor *srv_interceptor = new TestInterceptor(); + + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, Messenger::Policy::lossless_peer(0)); + server_msgr->interceptor = srv_interceptor; + + client_msgr->set_policy(entity_name_t::TYPE_OSD, Messenger::Policy::lossless_peer(0)); + client_msgr->interceptor = cli_interceptor; + + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:3300"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + bind_addr.parse("v2:127.0.0.1:3301"); + client_msgr->bind(bind_addr); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + ConnectionRef c2s = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + + MPing *m1 = new MPing(); + ASSERT_EQ(c2s->send_message(m1), 0); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + ASSERT_TRUE(c2s->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(c2s->get_priv().get())->get_count()); + ASSERT_TRUE(c2s->peer_is_osd()); + + cli_interceptor->breakpoint(Interceptor::STEP::HANDLE_MESSAGE); + + MPing *m2 = new MPing(); + ASSERT_EQ(c2s->send_message(m2), 0); + + cli_interceptor->wait(Interceptor::STEP::HANDLE_MESSAGE, c2s.get()); + cli_interceptor->remove_bp(Interceptor::STEP::HANDLE_MESSAGE); + + // at this point client and server are connected together + + // force both client and server to race on reconnect + cli_interceptor->breakpoint(Interceptor::STEP::SEND_RECONNECT); + srv_interceptor->breakpoint(Interceptor::STEP::SEND_RECONNECT); + + // failing client + // this will cause both client and server to reconnect at the same time + cli_interceptor->proceed(Interceptor::STEP::HANDLE_MESSAGE, Interceptor::ACTION::FAIL); + + MPing *m3 = new MPing(); + ASSERT_EQ(c2s->send_message(m3), 0); + + cli_interceptor->wait(Interceptor::STEP::SEND_RECONNECT, c2s.get()); + srv_interceptor->wait(Interceptor::STEP::SEND_RECONNECT); + + cli_interceptor->remove_bp(Interceptor::STEP::SEND_RECONNECT); + srv_interceptor->remove_bp(Interceptor::STEP::SEND_RECONNECT); + + // pause on "ready" + srv_interceptor->breakpoint(Interceptor::STEP::READY); + + cli_interceptor->proceed(Interceptor::STEP::SEND_RECONNECT, Interceptor::ACTION::CONTINUE); + srv_interceptor->proceed(Interceptor::STEP::SEND_RECONNECT, Interceptor::ACTION::CONTINUE); + + Connection *c2s_accepter = srv_interceptor->wait(Interceptor::STEP::READY); + + // the server has reconnected and is "ready" + srv_interceptor->remove_bp(Interceptor::STEP::READY); + + ASSERT_TRUE(c2s_accepter->peer_is_client()); + ASSERT_TRUE(c2s->peer_is_osd()); + + // the server should win the reconnect race + + // c2s_accepter sent 1 or 2 reconnect messages + ASSERT_LT(srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT), 3u); + ASSERT_GT(srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT), 0u); + // c2s_accepter sent 0 reconnect_ok messages + ASSERT_EQ(srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT_OK), 0u); + // c2s sent 1 reconnect messages + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SEND_RECONNECT), 1u); + // c2s sent 1 reconnect_ok messages + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SEND_RECONNECT_OK), 1u); + + if (srv_interceptor->count_step(c2s_accepter, Interceptor::STEP::SEND_RECONNECT) == 2) { + // if the server send the reconnect message two times then + // the client must have sent a session retry message to the server + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SESSION_RETRY), 1u); + } else { + ASSERT_EQ(cli_interceptor->count_step(c2s.get(), Interceptor::STEP::SESSION_RETRY), 0u); + } + + srv_interceptor->proceed(Interceptor::STEP::READY, Interceptor::ACTION::CONTINUE); + + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); + + delete cli_interceptor; + delete srv_interceptor; +} + +TEST_P(MessengerTest, SimpleTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. simple round trip + MPing *m = new MPing(); + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_TRUE(conn->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(conn->get_priv().get())->get_count()); + ASSERT_TRUE(conn->peer_is_osd()); + + // 2. test rebind port + set<int> avoid_ports; + for (int i = 0; i < 10 ; i++) { + for (auto a : server_msgr->get_myaddrs().v) { + avoid_ports.insert(a.get_port() + i); + } + } + server_msgr->rebind(avoid_ports); + for (auto a : server_msgr->get_myaddrs().v) { + ASSERT_TRUE(avoid_ports.count(a.get_port()) == 0); + } + + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + + // 3. test markdown connection + conn->mark_down(); + ASSERT_FALSE(conn->is_connected()); + + // 4. test failed connection + server_msgr->shutdown(); + server_msgr->wait(); + + m = new MPing(); + conn->send_message(m); + CHECK_AND_WAIT_TRUE(!conn->is_connected()); + ASSERT_FALSE(conn->is_connected()); + + // 5. loopback connection + srv_dispatcher.loopback = true; + conn = client_msgr->get_loopback_connection(); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + srv_dispatcher.loopback = false; + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); +} + +TEST_P(MessengerTest, SimpleMsgr2Test) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t legacy_addr; + legacy_addr.parse("v1:127.0.0.1"); + entity_addr_t msgr2_addr; + msgr2_addr.parse("v2:127.0.0.1"); + entity_addrvec_t bind_addrs; + bind_addrs.v.push_back(legacy_addr); + bind_addrs.v.push_back(msgr2_addr); + server_msgr->bindv(bind_addrs); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. simple round trip + MPing *m = new MPing(); + ConnectionRef conn = client_msgr->connect_to( + server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_TRUE(conn->is_connected()); + ASSERT_EQ(1u, static_cast<Session*>(conn->get_priv().get())->get_count()); + ASSERT_TRUE(conn->peer_is_osd()); + + // 2. test rebind port + set<int> avoid_ports; + for (int i = 0; i < 10 ; i++) { + for (auto a : server_msgr->get_myaddrs().v) { + avoid_ports.insert(a.get_port() + i); + } + } + server_msgr->rebind(avoid_ports); + for (auto a : server_msgr->get_myaddrs().v) { + ASSERT_TRUE(avoid_ports.count(a.get_port()) == 0); + } + + conn = client_msgr->connect_to( + server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + + // 3. test markdown connection + conn->mark_down(); + ASSERT_FALSE(conn->is_connected()); + + // 4. test failed connection + server_msgr->shutdown(); + server_msgr->wait(); + + m = new MPing(); + conn->send_message(m); + CHECK_AND_WAIT_TRUE(!conn->is_connected()); + ASSERT_FALSE(conn->is_connected()); + + // 5. loopback connection + srv_dispatcher.loopback = true; + conn = client_msgr->get_loopback_connection(); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + srv_dispatcher.loopback = false; + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + client_msgr->shutdown(); + client_msgr->wait(); + server_msgr->shutdown(); + server_msgr->wait(); +} + +TEST_P(MessengerTest, FeatureTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + uint64_t all_feature_supported, feature_required, feature_supported = 0; + for (int i = 0; i < 10; i++) + feature_supported |= 1ULL << i; + feature_supported |= CEPH_FEATUREMASK_MSG_ADDR2; + feature_supported |= CEPH_FEATUREMASK_SERVER_NAUTILUS; + feature_required = feature_supported | 1ULL << 13; + all_feature_supported = feature_required | 1ULL << 14; + + Messenger::Policy p = server_msgr->get_policy(entity_name_t::TYPE_CLIENT); + p.features_required = feature_required; + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + // 1. Suppose if only support less than required + p = client_msgr->get_policy(entity_name_t::TYPE_OSD); + p.features_supported = feature_supported; + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + MPing *m = new MPing(); + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + conn->send_message(m); + CHECK_AND_WAIT_TRUE(!conn->is_connected()); + // should failed build a connection + ASSERT_FALSE(conn->is_connected()); + + client_msgr->shutdown(); + client_msgr->wait(); + + // 2. supported met required + p = client_msgr->get_policy(entity_name_t::TYPE_OSD); + p.features_supported = all_feature_supported; + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + client_msgr->start(); + + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, TimeoutTest) { + g_ceph_context->_conf.set_val("ms_connection_idle_timeout", "1"); + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. build the connection + MPing *m = new MPing(); + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_TRUE(conn->is_connected()); + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + ASSERT_TRUE(conn->peer_is_osd()); + + // 2. wait for idle + usleep(2500*1000); + ASSERT_FALSE(conn->is_connected()); + + server_msgr->shutdown(); + server_msgr->wait(); + + client_msgr->shutdown(); + client_msgr->wait(); + g_ceph_context->_conf.set_val("ms_connection_idle_timeout", "900"); +} + +TEST_P(MessengerTest, StatefulTest) { + Message *m; + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + Messenger::Policy p = Messenger::Policy::stateful_server(0); + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + p = Messenger::Policy::lossless_client(0); + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. test for server standby + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + conn->mark_down(); + ASSERT_FALSE(conn->is_connected()); + ConnectionRef server_conn = server_msgr->connect_to( + client_msgr->get_mytype(), srv_dispatcher.last_accept); + // don't lose state + ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv().get())->get_count()); + + srv_dispatcher.got_new = false; + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + server_conn = server_msgr->connect_to(client_msgr->get_mytype(), + srv_dispatcher.last_accept); + { + std::unique_lock l{srv_dispatcher.lock}; + srv_dispatcher.cond.wait(l, [&] { return srv_dispatcher.got_remote_reset; }); + } + + // 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 + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_remote_reset; }); + cli_dispatcher.got_remote_reset = false; + } + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_connect; }); + cli_dispatcher.got_connect = false; + } + CHECK_AND_WAIT_TRUE(conn->is_connected()); + ASSERT_TRUE(conn->is_connected()); + + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + ASSERT_TRUE(conn->is_connected()); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + // resetcheck happen + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + server_conn = server_msgr->connect_to(client_msgr->get_mytype(), + srv_dispatcher.last_accept); + ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv().get())->get_count()); + cli_dispatcher.got_remote_reset = false; + + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, StatelessTest) { + Message *m; + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + Messenger::Policy p = Messenger::Policy::stateless_server(0); + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + p = Messenger::Policy::lossy_client(0); + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. test for server lose state + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + conn->mark_down(); + ASSERT_FALSE(conn->is_connected()); + + srv_dispatcher.got_new = false; + ConnectionRef server_conn; + srv_dispatcher.last_accept_con_ptr = &server_conn; + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + ASSERT_TRUE(server_conn); + + // server lose state + { + std::unique_lock l{srv_dispatcher.lock}; + srv_dispatcher.cond.wait(l, [&] { return srv_dispatcher.got_new; }); + } + ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv().get())->get_count()); + + // 2. test for client lossy + server_conn->mark_down(); + ASSERT_FALSE(server_conn->is_connected()); + conn->send_keepalive(); + CHECK_AND_WAIT_TRUE(!conn->is_connected()); + ASSERT_FALSE(conn->is_connected()); + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, AnonTest) { + Message *m; + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + Messenger::Policy p = Messenger::Policy::stateless_server(0); + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + p = Messenger::Policy::lossy_client(0); + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + ConnectionRef server_con_a, server_con_b; + + // a + srv_dispatcher.last_accept_con_ptr = &server_con_a; + ConnectionRef con_a = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs(), + true); + { + m = new MPing(); + ASSERT_EQ(con_a->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(con_a->get_priv().get())->get_count()); + + // b + srv_dispatcher.last_accept_con_ptr = &server_con_b; + ConnectionRef con_b = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs(), + true); + { + m = new MPing(); + ASSERT_EQ(con_b->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(con_b->get_priv().get())->get_count()); + + // these should be distinct + ASSERT_NE(con_a, con_b); + ASSERT_NE(server_con_a, server_con_b); + + // and both connected + { + m = new MPing(); + ASSERT_EQ(con_a->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + { + m = new MPing(); + ASSERT_EQ(con_b->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + + // clean up + con_a->mark_down(); + ASSERT_FALSE(con_a->is_connected()); + con_b->mark_down(); + ASSERT_FALSE(con_b->is_connected()); + + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, ClientStandbyTest) { + Message *m; + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + Messenger::Policy p = Messenger::Policy::stateful_server(0); + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + p = Messenger::Policy::lossless_peer(0); + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. test for client standby, resetcheck + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + ConnectionRef server_conn = server_msgr->connect_to( + client_msgr->get_mytype(), + srv_dispatcher.last_accept); + ASSERT_FALSE(cli_dispatcher.got_remote_reset); + cli_dispatcher.got_connect = false; + server_conn->mark_down(); + ASSERT_FALSE(server_conn->is_connected()); + // client should be standby + usleep(300*1000); + // client should be standby, so we use original connection + { + // Try send message to verify got remote reset callback + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + { + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_remote_reset; }); + cli_dispatcher.got_remote_reset = false; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_connect; }); + cli_dispatcher.got_connect = false; + } + CHECK_AND_WAIT_TRUE(conn->is_connected()); + ASSERT_TRUE(conn->is_connected()); + m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + server_conn = server_msgr->connect_to(client_msgr->get_mytype(), + srv_dispatcher.last_accept); + ASSERT_EQ(1U, static_cast<Session*>(server_conn->get_priv().get())->get_count()); + + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, AuthTest) { + g_ceph_context->_conf.set_val("auth_cluster_required", "cephx"); + g_ceph_context->_conf.set_val("auth_service_required", "cephx"); + g_ceph_context->_conf.set_val("auth_client_required", "cephx"); + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + // 1. simple auth round trip + MPing *m = new MPing(); + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_TRUE(conn->is_connected()); + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + + // 2. mix auth + g_ceph_context->_conf.set_val("auth_cluster_required", "none"); + g_ceph_context->_conf.set_val("auth_service_required", "none"); + g_ceph_context->_conf.set_val("auth_client_required", "none"); + conn->mark_down(); + ASSERT_FALSE(conn->is_connected()); + conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + MPing *m = new MPing(); + ASSERT_EQ(conn->send_message(m), 0); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + cli_dispatcher.got_new = false; + } + ASSERT_TRUE(conn->is_connected()); + ASSERT_EQ(1U, static_cast<Session*>(conn->get_priv().get())->get_count()); + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + +TEST_P(MessengerTest, MessageTest) { + FakeDispatcher cli_dispatcher(false), srv_dispatcher(true); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1"); + Messenger::Policy p = Messenger::Policy::stateful_server(0); + server_msgr->set_policy(entity_name_t::TYPE_CLIENT, p); + p = Messenger::Policy::lossless_peer(0); + client_msgr->set_policy(entity_name_t::TYPE_OSD, p); + + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->start(); + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->start(); + + + // 1. A very large "front"(as well as "payload") + // Because a external message need to invade Messenger::decode_message, + // here we only use existing message class(MCommand) + ConnectionRef conn = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + { + uuid_d uuid; + uuid.generate_random(); + vector<string> cmds; + string s("abcdefghijklmnopqrstuvwxyz"); + for (int i = 0; i < 1024*30; i++) + cmds.push_back(s); + MCommand *m = new MCommand(uuid); + m->cmd = cmds; + conn->send_message(m); + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait_for(l, 500s, [&] { return cli_dispatcher.got_new; }); + ASSERT_TRUE(cli_dispatcher.got_new); + cli_dispatcher.got_new = false; + } + + // 2. A very large "data" + { + bufferlist bl; + string s("abcdefghijklmnopqrstuvwxyz"); + for (int i = 0; i < 1024*30; i++) + bl.append(s); + MPing *m = new MPing(); + m->set_data(bl); + conn->send_message(m); + utime_t t; + t += 1000*1000*500; + std::unique_lock l{cli_dispatcher.lock}; + cli_dispatcher.cond.wait(l, [&] { return cli_dispatcher.got_new; }); + ASSERT_TRUE(cli_dispatcher.got_new); + cli_dispatcher.got_new = false; + } + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr->wait(); + client_msgr->wait(); +} + + +class SyntheticWorkload; + +struct Payload { + enum Who : uint8_t { + PING = 0, + PONG = 1, + }; + uint8_t who = 0; + uint64_t seq = 0; + bufferlist data; + + Payload(Who who, uint64_t seq, const bufferlist& data) + : who(who), seq(seq), data(data) + {} + Payload() = default; + DENC(Payload, v, p) { + DENC_START(1, 1, p); + denc(v.who, p); + denc(v.seq, p); + denc(v.data, p); + DENC_FINISH(p); + } +}; +WRITE_CLASS_DENC(Payload) + +ostream& operator<<(ostream& out, const Payload &pl) +{ + return out << "reply=" << pl.who << " i = " << pl.seq; +} + +class SyntheticDispatcher : public Dispatcher { + public: + ceph::mutex lock = ceph::make_mutex("SyntheticDispatcher::lock"); + ceph::condition_variable cond; + bool is_server; + bool got_new; + bool got_remote_reset; + bool got_connect; + map<ConnectionRef, list<uint64_t> > conn_sent; + map<uint64_t, bufferlist> sent; + atomic<uint64_t> index; + SyntheticWorkload *workload; + + SyntheticDispatcher(bool s, SyntheticWorkload *wl): + Dispatcher(g_ceph_context), is_server(s), got_new(false), + got_remote_reset(false), got_connect(false), index(0), workload(wl) { + } + bool ms_can_fast_dispatch_any() const override { return true; } + bool ms_can_fast_dispatch(const Message *m) const override { + switch (m->get_type()) { + case CEPH_MSG_PING: + case MSG_COMMAND: + return true; + default: + return false; + } + } + + void ms_handle_fast_connect(Connection *con) override { + std::lock_guard l{lock}; + list<uint64_t> c = conn_sent[con]; + for (list<uint64_t>::iterator it = c.begin(); + it != c.end(); ++it) + sent.erase(*it); + conn_sent.erase(con); + got_connect = true; + cond.notify_all(); + } + void ms_handle_fast_accept(Connection *con) override { + std::lock_guard l{lock}; + list<uint64_t> c = conn_sent[con]; + for (list<uint64_t>::iterator it = c.begin(); + it != c.end(); ++it) + sent.erase(*it); + conn_sent.erase(con); + cond.notify_all(); + } + bool ms_dispatch(Message *m) override { + ceph_abort(); + } + bool ms_handle_reset(Connection *con) override; + void ms_handle_remote_reset(Connection *con) override { + std::lock_guard l{lock}; + list<uint64_t> c = conn_sent[con]; + for (list<uint64_t>::iterator it = c.begin(); + it != c.end(); ++it) + sent.erase(*it); + conn_sent.erase(con); + got_remote_reset = true; + } + bool ms_handle_refused(Connection *con) override { + return false; + } + void ms_fast_dispatch(Message *m) override { + // MSG_COMMAND is used to disorganize regular message flow + if (m->get_type() == MSG_COMMAND) { + m->put(); + return ; + } + + Payload pl; + auto p = m->get_data().cbegin(); + decode(pl, p); + if (pl.who == Payload::PING) { + lderr(g_ceph_context) << __func__ << " conn=" << m->get_connection() << pl << dendl; + reply_message(m, pl); + m->put(); + std::lock_guard l{lock}; + got_new = true; + cond.notify_all(); + } else { + std::lock_guard l{lock}; + if (sent.count(pl.seq)) { + lderr(g_ceph_context) << __func__ << " conn=" << m->get_connection() << pl << dendl; + ASSERT_EQ(conn_sent[m->get_connection()].front(), pl.seq); + ASSERT_TRUE(pl.data.contents_equal(sent[pl.seq])); + conn_sent[m->get_connection()].pop_front(); + sent.erase(pl.seq); + } + m->put(); + got_new = true; + cond.notify_all(); + } + } + + int ms_handle_authentication(Connection *con) override { + return 1; + } + + void reply_message(const Message *m, Payload& pl) { + pl.who = Payload::PONG; + bufferlist bl; + encode(pl, bl); + MPing *rm = new MPing(); + rm->set_data(bl); + m->get_connection()->send_message(rm); + lderr(g_ceph_context) << __func__ << " conn=" << m->get_connection() << " reply m=" << m << " i=" << pl.seq << dendl; + } + + void send_message_wrap(ConnectionRef con, const bufferlist& data) { + Message *m = new MPing(); + Payload pl{Payload::PING, index++, data}; + bufferlist bl; + encode(pl, bl); + m->set_data(bl); + if (!con->get_messenger()->get_default_policy().lossy) { + std::lock_guard l{lock}; + sent[pl.seq] = pl.data; + conn_sent[con].push_back(pl.seq); + } + lderr(g_ceph_context) << __func__ << " conn=" << con.get() << " send m=" << m << " i=" << pl.seq << dendl; + ASSERT_EQ(0, con->send_message(m)); + } + + uint64_t get_pending() { + std::lock_guard l{lock}; + return sent.size(); + } + + void clear_pending(ConnectionRef con) { + std::lock_guard l{lock}; + + for (list<uint64_t>::iterator it = conn_sent[con].begin(); + it != conn_sent[con].end(); ++it) + sent.erase(*it); + conn_sent.erase(con); + } + + void print() { + for (auto && p : conn_sent) { + if (!p.second.empty()) { + lderr(g_ceph_context) << __func__ << " " << p.first << " wait " << p.second.size() << dendl; + } + } + } +}; + + +class SyntheticWorkload { + ceph::mutex lock = ceph::make_mutex("SyntheticWorkload::lock"); + ceph::condition_variable cond; + set<Messenger*> available_servers; + set<Messenger*> available_clients; + Messenger::Policy client_policy; + map<ConnectionRef, pair<Messenger*, Messenger*> > available_connections; + SyntheticDispatcher dispatcher; + gen_type rng; + vector<bufferlist> rand_data; + DummyAuthClientServer dummy_auth; + + public: + static const unsigned max_in_flight = 64; + static const unsigned max_connections = 128; + static const unsigned max_message_len = 1024 * 1024 * 4; + + SyntheticWorkload(int servers, int clients, string type, int random_num, + Messenger::Policy srv_policy, Messenger::Policy cli_policy) + : client_policy(cli_policy), + dispatcher(false, this), + rng(time(NULL)), + dummy_auth(g_ceph_context) { + dummy_auth.auth_registry.refresh_config(); + Messenger *msgr; + int base_port = 16800; + entity_addr_t bind_addr; + char addr[64]; + for (int i = 0; i < servers; ++i) { + msgr = Messenger::create(g_ceph_context, type, entity_name_t::OSD(0), + "server", getpid()+i); + snprintf(addr, sizeof(addr), "v2:127.0.0.1:%d", + base_port+i); + bind_addr.parse(addr); + msgr->bind(bind_addr); + msgr->add_dispatcher_head(&dispatcher); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + + ceph_assert(msgr); + msgr->set_default_policy(srv_policy); + available_servers.insert(msgr); + msgr->start(); + } + + for (int i = 0; i < clients; ++i) { + msgr = Messenger::create(g_ceph_context, type, entity_name_t::CLIENT(-1), + "client", getpid()+i+servers); + if (cli_policy.standby) { + snprintf(addr, sizeof(addr), "v2:127.0.0.1:%d", + base_port+i+servers); + bind_addr.parse(addr); + msgr->bind(bind_addr); + } + msgr->add_dispatcher_head(&dispatcher); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + + ceph_assert(msgr); + msgr->set_default_policy(cli_policy); + available_clients.insert(msgr); + msgr->start(); + } + + for (int i = 0; i < random_num; i++) { + bufferlist bl; + boost::uniform_int<> u(32, max_message_len); + uint64_t value_len = u(rng); + bufferptr bp(value_len); + bp.zero(); + for (uint64_t j = 0; j < value_len-sizeof(i); ) { + memcpy(bp.c_str()+j, &i, sizeof(i)); + j += 4096; + } + + bl.append(bp); + rand_data.push_back(bl); + } + } + + ConnectionRef _get_random_connection() { + while (dispatcher.get_pending() > max_in_flight) { + lock.unlock(); + usleep(500); + lock.lock(); + } + ceph_assert(ceph_mutex_is_locked(lock)); + boost::uniform_int<> choose(0, available_connections.size() - 1); + int index = choose(rng); + map<ConnectionRef, pair<Messenger*, Messenger*> >::iterator i = available_connections.begin(); + for (; index > 0; --index, ++i) ; + return i->first; + } + + bool can_create_connection() { + return available_connections.size() < max_connections; + } + + void generate_connection() { + std::lock_guard l{lock}; + if (!can_create_connection()) + return ; + + Messenger *server, *client; + { + boost::uniform_int<> choose(0, available_servers.size() - 1); + int index = choose(rng); + set<Messenger*>::iterator i = available_servers.begin(); + for (; index > 0; --index, ++i) ; + server = *i; + } + { + boost::uniform_int<> choose(0, available_clients.size() - 1); + int index = choose(rng); + set<Messenger*>::iterator i = available_clients.begin(); + for (; index > 0; --index, ++i) ; + client = *i; + } + + pair<Messenger*, Messenger*> p; + { + boost::uniform_int<> choose(0, available_servers.size() - 1); + if (server->get_default_policy().server) { + p = make_pair(client, server); + ConnectionRef conn = client->connect_to(server->get_mytype(), + server->get_myaddrs()); + available_connections[conn] = p; + } else { + ConnectionRef conn = client->connect_to(server->get_mytype(), + server->get_myaddrs()); + p = make_pair(client, server); + available_connections[conn] = p; + } + } + } + + void send_message() { + std::lock_guard l{lock}; + ConnectionRef conn = _get_random_connection(); + boost::uniform_int<> true_false(0, 99); + int val = true_false(rng); + if (val >= 95) { + uuid_d uuid; + uuid.generate_random(); + MCommand *m = new MCommand(uuid); + vector<string> cmds; + cmds.push_back("command"); + m->cmd = cmds; + m->set_priority(200); + conn->send_message(m); + } else { + boost::uniform_int<> u(0, rand_data.size()-1); + dispatcher.send_message_wrap(conn, rand_data[u(rng)]); + } + } + + void drop_connection() { + std::lock_guard l{lock}; + if (available_connections.size() < 10) + return; + ConnectionRef conn = _get_random_connection(); + dispatcher.clear_pending(conn); + conn->mark_down(); + if (!client_policy.server && + !client_policy.lossy && + client_policy.standby) { + // it's a lossless policy, so we need to mark down each side + pair<Messenger*, Messenger*> &p = available_connections[conn]; + if (!p.first->get_default_policy().server && !p.second->get_default_policy().server) { + ASSERT_EQ(conn->get_messenger(), p.first); + ConnectionRef peer = p.second->connect_to(p.first->get_mytype(), + p.first->get_myaddrs()); + peer->mark_down(); + dispatcher.clear_pending(peer); + available_connections.erase(peer); + } + } + ASSERT_EQ(available_connections.erase(conn), 1U); + } + + void print_internal_state(bool detail=false) { + std::lock_guard l{lock}; + lderr(g_ceph_context) << "available_connections: " << available_connections.size() + << " inflight messages: " << dispatcher.get_pending() << dendl; + if (detail && !available_connections.empty()) { + dispatcher.print(); + } + } + + void wait_for_done() { + int64_t tick_us = 1000 * 100; // 100ms + int64_t timeout_us = 5 * 60 * 1000 * 1000; // 5 mins + int i = 0; + while (dispatcher.get_pending()) { + usleep(tick_us); + timeout_us -= tick_us; + if (i++ % 50 == 0) + print_internal_state(true); + if (timeout_us < 0) + ceph_abort_msg(" loop time exceed 5 mins, it looks we stuck into some problems!"); + } + for (set<Messenger*>::iterator it = available_servers.begin(); + it != available_servers.end(); ++it) { + (*it)->shutdown(); + (*it)->wait(); + ASSERT_EQ((*it)->get_dispatch_queue_len(), 0); + delete (*it); + } + available_servers.clear(); + + for (set<Messenger*>::iterator it = available_clients.begin(); + it != available_clients.end(); ++it) { + (*it)->shutdown(); + (*it)->wait(); + ASSERT_EQ((*it)->get_dispatch_queue_len(), 0); + delete (*it); + } + available_clients.clear(); + } + + void handle_reset(Connection *con) { + std::lock_guard l{lock}; + available_connections.erase(con); + dispatcher.clear_pending(con); + } +}; + +bool SyntheticDispatcher::ms_handle_reset(Connection *con) { + workload->handle_reset(con); + return true; +} + +TEST_P(MessengerTest, SyntheticStressTest) { + SyntheticWorkload test_msg(8, 32, GetParam(), 100, + Messenger::Policy::stateful_server(0), + Messenger::Policy::lossless_client(0)); + for (int i = 0; i < 100; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 5000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + 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() % 1000 + 500); + } + } + test_msg.wait_for_done(); +} + +TEST_P(MessengerTest, SyntheticStressTest1) { + SyntheticWorkload test_msg(16, 32, GetParam(), 100, + Messenger::Policy::lossless_peer_reuse(0), + Messenger::Policy::lossless_peer_reuse(0)); + for (int i = 0; i < 10; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 10000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + test_msg.print_internal_state(); + } + boost::uniform_int<> true_false(0, 99); + int val = true_false(rng); + if (val > 80) { + test_msg.generate_connection(); + } else if (val > 60) { + test_msg.drop_connection(); + } else if (val > 10) { + test_msg.send_message(); + } else { + usleep(rand() % 1000 + 500); + } + } + test_msg.wait_for_done(); +} + + +TEST_P(MessengerTest, SyntheticInjectTest) { + uint64_t dispatch_throttle_bytes = g_ceph_context->_conf->ms_dispatch_throttle_bytes; + g_ceph_context->_conf.set_val("ms_inject_socket_failures", "30"); + g_ceph_context->_conf.set_val("ms_inject_internal_delays", "0.1"); + g_ceph_context->_conf.set_val("ms_dispatch_throttle_bytes", "16777216"); + SyntheticWorkload test_msg(8, 32, GetParam(), 100, + Messenger::Policy::stateful_server(0), + Messenger::Policy::lossless_client(0)); + for (int i = 0; i < 100; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 1000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + 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"); + g_ceph_context->_conf.set_val( + "ms_dispatch_throttle_bytes", std::to_string(dispatch_throttle_bytes)); +} + +TEST_P(MessengerTest, SyntheticInjectTest2) { + g_ceph_context->_conf.set_val("ms_inject_socket_failures", "30"); + g_ceph_context->_conf.set_val("ms_inject_internal_delays", "0.1"); + SyntheticWorkload test_msg(8, 16, GetParam(), 100, + Messenger::Policy::lossless_peer_reuse(0), + Messenger::Policy::lossless_peer_reuse(0)); + for (int i = 0; i < 100; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 1000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + 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"); +} + +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), + Messenger::Policy::lossy_client(0)); + for (int i = 0; i < 100; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 1000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + 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"); +} + + +TEST_P(MessengerTest, SyntheticInjectTest4) { + g_ceph_context->_conf.set_val("ms_inject_socket_failures", "30"); + g_ceph_context->_conf.set_val("ms_inject_internal_delays", "0.1"); + g_ceph_context->_conf.set_val("ms_inject_delay_probability", "1"); + g_ceph_context->_conf.set_val("ms_inject_delay_type", "client osd"); + g_ceph_context->_conf.set_val("ms_inject_delay_max", "5"); + SyntheticWorkload test_msg(16, 32, GetParam(), 100, + Messenger::Policy::lossless_peer(0), + Messenger::Policy::lossless_peer(0)); + for (int i = 0; i < 100; ++i) { + if (!(i % 10)) lderr(g_ceph_context) << "seeding connection " << i << dendl; + test_msg.generate_connection(); + } + gen_type rng(time(NULL)); + for (int i = 0; i < 1000; ++i) { + if (!(i % 10)) { + lderr(g_ceph_context) << "Op " << i << ": " << dendl; + test_msg.print_internal_state(); + } + boost::uniform_int<> true_false(0, 99); + int val = true_false(rng); + if (val > 95) { + 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"); + g_ceph_context->_conf.set_val("ms_inject_delay_probability", "0"); + g_ceph_context->_conf.set_val("ms_inject_delay_type", ""); + g_ceph_context->_conf.set_val("ms_inject_delay_max", "0"); +} + + +class MarkdownDispatcher : public Dispatcher { + ceph::mutex lock = ceph::make_mutex("MarkdownDispatcher::lock"); + set<ConnectionRef> conns; + bool last_mark; + public: + std::atomic<uint64_t> count = { 0 }; + explicit MarkdownDispatcher(bool s): Dispatcher(g_ceph_context), + last_mark(false) { + } + bool ms_can_fast_dispatch_any() const override { return false; } + bool ms_can_fast_dispatch(const Message *m) const override { + switch (m->get_type()) { + case CEPH_MSG_PING: + return true; + default: + return false; + } + } + + void ms_handle_fast_connect(Connection *con) override { + lderr(g_ceph_context) << __func__ << " " << con << dendl; + std::lock_guard l{lock}; + conns.insert(con); + } + void ms_handle_fast_accept(Connection *con) override { + std::lock_guard l{lock}; + conns.insert(con); + } + bool ms_dispatch(Message *m) override { + lderr(g_ceph_context) << __func__ << " conn: " << m->get_connection() << dendl; + std::lock_guard l{lock}; + count++; + conns.insert(m->get_connection()); + if (conns.size() < 2 && !last_mark) { + m->put(); + return true; + } + + last_mark = true; + usleep(rand() % 500); + for (set<ConnectionRef>::iterator it = conns.begin(); it != conns.end(); ++it) { + if ((*it) != m->get_connection().get()) { + (*it)->mark_down(); + conns.erase(it); + break; + } + } + if (conns.empty()) + last_mark = false; + m->put(); + return true; + } + bool ms_handle_reset(Connection *con) override { + lderr(g_ceph_context) << __func__ << " " << con << dendl; + std::lock_guard l{lock}; + conns.erase(con); + usleep(rand() % 500); + return true; + } + void ms_handle_remote_reset(Connection *con) override { + std::lock_guard l{lock}; + conns.erase(con); + lderr(g_ceph_context) << __func__ << " " << con << dendl; + } + bool ms_handle_refused(Connection *con) override { + return false; + } + void ms_fast_dispatch(Message *m) override { + ceph_abort(); + } + int ms_handle_authentication(Connection *con) override { + return 1; + } +}; + + +// Markdown with external lock +TEST_P(MessengerTest, MarkdownTest) { + Messenger *server_msgr2 = Messenger::create(g_ceph_context, string(GetParam()), entity_name_t::OSD(0), "server", getpid()); + MarkdownDispatcher cli_dispatcher(false), srv_dispatcher(true); + DummyAuthClientServer dummy_auth(g_ceph_context); + dummy_auth.auth_registry.refresh_config(); + entity_addr_t bind_addr; + bind_addr.parse("v2:127.0.0.1:16800"); + server_msgr->bind(bind_addr); + server_msgr->add_dispatcher_head(&srv_dispatcher); + server_msgr->set_auth_client(&dummy_auth); + server_msgr->set_auth_server(&dummy_auth); + server_msgr->start(); + bind_addr.parse("v2:127.0.0.1:16801"); + server_msgr2->bind(bind_addr); + server_msgr2->add_dispatcher_head(&srv_dispatcher); + server_msgr2->set_auth_client(&dummy_auth); + server_msgr2->set_auth_server(&dummy_auth); + server_msgr2->start(); + + client_msgr->add_dispatcher_head(&cli_dispatcher); + client_msgr->set_auth_client(&dummy_auth); + client_msgr->set_auth_server(&dummy_auth); + client_msgr->start(); + + int i = 1000; + uint64_t last = 0; + bool equal = false; + uint64_t equal_count = 0; + while (i--) { + ConnectionRef conn1 = client_msgr->connect_to(server_msgr->get_mytype(), + server_msgr->get_myaddrs()); + ConnectionRef conn2 = client_msgr->connect_to(server_msgr2->get_mytype(), + server_msgr2->get_myaddrs()); + MPing *m = new MPing(); + ASSERT_EQ(conn1->send_message(m), 0); + m = new MPing(); + ASSERT_EQ(conn2->send_message(m), 0); + CHECK_AND_WAIT_TRUE(srv_dispatcher.count > last + 1); + if (srv_dispatcher.count == last) { + lderr(g_ceph_context) << __func__ << " last is " << last << dendl; + equal = true; + equal_count++; + } else { + equal = false; + equal_count = 0; + } + last = srv_dispatcher.count; + if (equal_count) + usleep(1000*500); + ASSERT_FALSE(equal && equal_count > 3); + } + server_msgr->shutdown(); + client_msgr->shutdown(); + server_msgr2->shutdown(); + server_msgr->wait(); + client_msgr->wait(); + server_msgr2->wait(); + delete server_msgr2; +} + +INSTANTIATE_TEST_SUITE_P( + Messenger, + MessengerTest, + ::testing::Values( + "async+posix" + ) +); + +int main(int argc, char **argv) { + vector<const char*> args; + argv_to_vec(argc, (const char **)argv, args); + + auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, + CODE_ENVIRONMENT_UTILITY, + CINIT_FLAG_NO_DEFAULT_CONFIG_FILE); + g_ceph_context->_conf.set_val("auth_cluster_required", "none"); + g_ceph_context->_conf.set_val("auth_service_required", "none"); + g_ceph_context->_conf.set_val("auth_client_required", "none"); + g_ceph_context->_conf.set_val("keyring", "/dev/null"); + g_ceph_context->_conf.set_val("enable_experimental_unrecoverable_data_corrupting_features", "ms-type-async"); + g_ceph_context->_conf.set_val("ms_die_on_bad_msg", "true"); + g_ceph_context->_conf.set_val("ms_die_on_old_message", "true"); + g_ceph_context->_conf.set_val("ms_max_backoff", "1"); + common_init_finish(g_ceph_context); + + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} + +/* + * Local Variables: + * compile-command: "cd ../.. ; make -j4 ceph_test_msgr && valgrind --tool=memcheck ./ceph_test_msgr" + * End: + */ |