diff options
Diffstat (limited to '')
| -rw-r--r-- | src/test/crimson/test_messenger.cc | 3874 |
1 files changed, 3874 insertions, 0 deletions
diff --git a/src/test/crimson/test_messenger.cc b/src/test/crimson/test_messenger.cc new file mode 100644 index 000000000..a42572246 --- /dev/null +++ b/src/test/crimson/test_messenger.cc @@ -0,0 +1,3874 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "common/ceph_argparse.h" +#include "common/ceph_time.h" +#include "messages/MPing.h" +#include "messages/MCommand.h" +#include "messages/MCommandReply.h" +#include "messages/MOSDOp.h" +#include "messages/MOSDOpReply.h" +#include "crimson/auth/DummyAuth.h" +#include "crimson/common/log.h" +#include "crimson/net/Connection.h" +#include "crimson/net/Dispatcher.h" +#include "crimson/net/Messenger.h" +#include "crimson/net/Interceptor.h" + +#include <map> +#include <random> +#include <boost/program_options.hpp> +#include <fmt/format.h> +#include <fmt/ostream.h> +#include <seastar/core/app-template.hh> +#include <seastar/core/do_with.hh> +#include <seastar/core/future-util.hh> +#include <seastar/core/gate.hh> +#include <seastar/core/reactor.hh> +#include <seastar/core/sleep.hh> +#include <seastar/core/with_timeout.hh> + +#include "test_messenger.h" + +using namespace std::chrono_literals; +namespace bpo = boost::program_options; +using crimson::common::local_conf; + +namespace { + +seastar::logger& logger() { + return crimson::get_logger(ceph_subsys_test); +} + +static std::random_device rd; +static std::default_random_engine rng{rd()}; +static bool verbose = false; + +static entity_addr_t get_server_addr() { + static int port = 9030; + ++port; + entity_addr_t saddr; + saddr.parse("127.0.0.1", nullptr); + saddr.set_port(port); + return saddr; +} + +template <typename T, typename... Args> +seastar::future<T*> create_sharded(Args... args) { + // we should only construct/stop shards on #0 + return seastar::smp::submit_to(0, [=] { + auto sharded_obj = seastar::make_lw_shared<seastar::sharded<T>>(); + return sharded_obj->start(args... + ).then([sharded_obj] { + seastar::engine().at_exit([sharded_obj] { + return sharded_obj->stop().then([sharded_obj] {}); + }); + return sharded_obj.get(); + }); + }).then([](seastar::sharded<T> *ptr_shard) { + return &ptr_shard->local(); + }); +} + +class ShardedGates + : public seastar::peering_sharded_service<ShardedGates> { +public: + ShardedGates() = default; + ~ShardedGates() { + assert(gate.is_closed()); + } + + template <typename Func> + void dispatch_in_background(const char *what, Func &&f) { + std::ignore = seastar::with_gate( + container().local().gate, std::forward<Func>(f) + ).handle_exception([what](std::exception_ptr eptr) { + try { + std::rethrow_exception(eptr); + } catch (std::exception &e) { + logger().error("ShardedGates::dispatch_in_background: " + "{} got exxception {}", what, e.what()); + } + }); + } + + seastar::future<> close() { + return container().invoke_on_all([](auto &local) { + return local.gate.close(); + }); + } + + static seastar::future<ShardedGates*> create() { + return create_sharded<ShardedGates>(); + } + + // seastar::future<> stop() is intentially not implemented + +private: + seastar::gate gate; +}; + +static seastar::future<> test_echo(unsigned rounds, + double keepalive_ratio) +{ + struct test_state { + struct Server final + : public crimson::net::Dispatcher { + ShardedGates &gates; + crimson::net::MessengerRef msgr; + crimson::auth::DummyAuthClientServer dummy_auth; + + Server(ShardedGates &gates) : gates{gates} {} + + void ms_handle_accept( + crimson::net::ConnectionRef conn, + seastar::shard_id prv_shard, + bool is_replace) override { + logger().info("server accepted {}", *conn); + ceph_assert(prv_shard == seastar::this_shard_id()); + ceph_assert(!is_replace); + } + + std::optional<seastar::future<>> ms_dispatch( + crimson::net::ConnectionRef c, MessageRef m) override { + if (verbose) { + logger().info("server got {}", *m); + } + // reply with a pong + gates.dispatch_in_background("echo_send_pong", [c] { + return c->send(crimson::make_message<MPing>()); + }); + return {seastar::now()}; + } + + seastar::future<> init(const entity_name_t& name, + const std::string& lname, + const uint64_t nonce, + const entity_addr_t& addr) { + msgr = crimson::net::Messenger::create( + name, lname, nonce, false); + msgr->set_default_policy(crimson::net::SocketPolicy::stateless_server(0)); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + return msgr->bind(entity_addrvec_t{addr}).safe_then([this] { + return msgr->start({this}); + }, crimson::net::Messenger::bind_ertr::all_same_way( + [addr] (const std::error_code& e) { + logger().error("test_echo(): " + "there is another instance running at {}", addr); + ceph_abort(); + })); + } + seastar::future<> shutdown() { + ceph_assert(msgr); + msgr->stop(); + return msgr->shutdown(); + } + }; + + class Client final + : public crimson::net::Dispatcher, + public seastar::peering_sharded_service<Client> { + public: + Client(seastar::shard_id primary_sid, + unsigned rounds, + double keepalive_ratio, + ShardedGates *gates) + : primary_sid{primary_sid}, + keepalive_dist(std::bernoulli_distribution{keepalive_ratio}), + rounds(rounds), + gates{*gates} {} + + seastar::future<> init(const entity_name_t& name, + const std::string& lname, + const uint64_t nonce) { + assert(seastar::this_shard_id() == primary_sid); + msgr = crimson::net::Messenger::create( + name, lname, nonce, false); + msgr->set_default_policy(crimson::net::SocketPolicy::lossy_client(0)); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + return msgr->start({this}); + } + + seastar::future<> shutdown() { + assert(seastar::this_shard_id() == primary_sid); + ceph_assert(msgr); + msgr->stop(); + return msgr->shutdown(); + } + + seastar::future<> dispatch_pingpong(const entity_addr_t& peer_addr) { + assert(seastar::this_shard_id() == primary_sid); + mono_time start_time = mono_clock::now(); + auto conn = msgr->connect(peer_addr, entity_name_t::TYPE_OSD); + return seastar::futurize_invoke([this, conn] { + return do_dispatch_pingpong(conn); + }).then([] { + // 500ms should be enough to establish the connection + return seastar::sleep(500ms); + }).then([this, conn, start_time] { + return container().invoke_on( + conn->get_shard_id(), + [pconn=&*conn, start_time](auto &local) { + assert(pconn->is_connected()); + auto session = local.find_session(pconn); + std::chrono::duration<double> dur_handshake = session->connected_time - start_time; + std::chrono::duration<double> dur_pingpong = session->finish_time - session->connected_time; + logger().info("{}: handshake {}, pingpong {}", + *pconn, dur_handshake.count(), dur_pingpong.count()); + }).then([conn] {}); + }); + } + + static seastar::future<Client*> create( + unsigned rounds, + double keepalive_ratio, + ShardedGates *gates) { + return create_sharded<Client>( + seastar::this_shard_id(), + rounds, + keepalive_ratio, + gates); + } + + private: + struct PingSession : public seastar::enable_shared_from_this<PingSession> { + unsigned count = 0u; + mono_time connected_time; + mono_time finish_time; + }; + using PingSessionRef = seastar::shared_ptr<PingSession>; + + void ms_handle_connect( + crimson::net::ConnectionRef conn, + seastar::shard_id prv_shard) override { + auto &local = container().local(); + assert(prv_shard == seastar::this_shard_id()); + auto session = seastar::make_shared<PingSession>(); + auto [i, added] = local.sessions.emplace(&*conn, session); + std::ignore = i; + ceph_assert(added); + session->connected_time = mono_clock::now(); + } + + std::optional<seastar::future<>> ms_dispatch( + crimson::net::ConnectionRef c, MessageRef m) override { + auto &local = container().local(); + auto session = local.find_session(&*c); + ++(session->count); + if (verbose) { + logger().info("client ms_dispatch {}", session->count); + } + + if (session->count > rounds) { + logger().error("{}: got {} pongs, more than expected {}", *c, session->count, rounds); + ceph_abort(); + } else if (session->count == rounds) { + logger().info("{}: finished receiving {} pongs", *c, session->count); + session->finish_time = mono_clock::now(); + gates.dispatch_in_background("echo_notify_done", [c, this] { + return container().invoke_on(primary_sid, [pconn=&*c](auto &local) { + auto found = local.pending_conns.find(pconn); + ceph_assert(found != local.pending_conns.end()); + found->second.set_value(); + }).then([c] {}); + }); + } + return {seastar::now()}; + } + + PingSessionRef find_session(crimson::net::Connection *c) { + auto found = sessions.find(c); + if (found == sessions.end()) { + ceph_assert(false); + } + return found->second; + } + + seastar::future<> do_dispatch_pingpong(crimson::net::ConnectionRef conn) { + auto [i, added] = pending_conns.emplace(&*conn, seastar::promise<>()); + std::ignore = i; + ceph_assert(added); + return seastar::do_with(0u, 0u, + [this, conn](auto &count_ping, auto &count_keepalive) { + return seastar::do_until( + [this, conn, &count_ping, &count_keepalive] { + bool stop = (count_ping == rounds); + if (stop) { + logger().info("{}: finished sending {} pings with {} keepalives", + *conn, count_ping, count_keepalive); + } + return stop; + }, + [this, conn, &count_ping, &count_keepalive] { + return seastar::repeat([this, conn, &count_ping, &count_keepalive] { + if (keepalive_dist(rng)) { + return conn->send_keepalive( + ).then([&count_keepalive] { + count_keepalive += 1; + return seastar::make_ready_future<seastar::stop_iteration>( + seastar::stop_iteration::no); + }); + } else { + return conn->send(crimson::make_message<MPing>() + ).then([&count_ping] { + count_ping += 1; + return seastar::make_ready_future<seastar::stop_iteration>( + seastar::stop_iteration::yes); + }); + } + }); + }).then([this, conn] { + auto found = pending_conns.find(&*conn); + assert(found != pending_conns.end()); + return found->second.get_future(); + } + ); + }); + } + + private: + // primary shard only + const seastar::shard_id primary_sid; + std::bernoulli_distribution keepalive_dist; + crimson::net::MessengerRef msgr; + std::map<crimson::net::Connection*, seastar::promise<>> pending_conns; + crimson::auth::DummyAuthClientServer dummy_auth; + + // per shard + const unsigned rounds; + std::map<crimson::net::Connection*, PingSessionRef> sessions; + ShardedGates &gates; + }; + }; + + logger().info("test_echo(rounds={}, keepalive_ratio={}):", + rounds, keepalive_ratio); + return ShardedGates::create( + ).then([rounds, keepalive_ratio](auto *gates) { + return seastar::when_all_succeed( + test_state::Client::create(rounds, keepalive_ratio, gates), + test_state::Client::create(rounds, keepalive_ratio, gates), + seastar::make_ready_future<ShardedGates*>(gates)); + }).then_unpack([](auto *client1, auto *client2, auto *gates) { + auto server1 = seastar::make_shared<test_state::Server>(*gates); + auto server2 = seastar::make_shared<test_state::Server>(*gates); + // start servers and clients + auto addr1 = get_server_addr(); + auto addr2 = get_server_addr(); + addr1.set_type(entity_addr_t::TYPE_MSGR2); + addr2.set_type(entity_addr_t::TYPE_MSGR2); + return seastar::when_all_succeed( + server1->init(entity_name_t::OSD(0), "server1", 1, addr1), + server2->init(entity_name_t::OSD(1), "server2", 2, addr2), + client1->init(entity_name_t::OSD(2), "client1", 3), + client2->init(entity_name_t::OSD(3), "client2", 4) + // dispatch pingpoing + ).then_unpack([client1, client2, server1, server2] { + return seastar::when_all_succeed( + // test connecting in parallel, accepting in parallel + client1->dispatch_pingpong(server1->msgr->get_myaddr()), + client1->dispatch_pingpong(server2->msgr->get_myaddr()), + client2->dispatch_pingpong(server1->msgr->get_myaddr()), + client2->dispatch_pingpong(server2->msgr->get_myaddr())); + // shutdown + }).then_unpack([client1] { + logger().info("client1 shutdown..."); + return client1->shutdown(); + }).then([client2] { + logger().info("client2 shutdown..."); + return client2->shutdown(); + }).then([server1] { + logger().info("server1 shutdown..."); + return server1->shutdown(); + }).then([server2] { + logger().info("server2 shutdown..."); + return server2->shutdown(); + }).then([] { + logger().info("test_echo() done!\n"); + }).handle_exception([](auto eptr) { + logger().error("test_echo() failed: got exception {}", eptr); + throw; + }).finally([gates, server1, server2] { + return gates->close(); + }); + }); +} + +seastar::future<> test_preemptive_shutdown() { + struct test_state { + class Server final + : public crimson::net::Dispatcher { + crimson::net::MessengerRef msgr; + crimson::auth::DummyAuthClientServer dummy_auth; + + std::optional<seastar::future<>> ms_dispatch( + crimson::net::ConnectionRef c, MessageRef m) override { + std::ignore = c->send(crimson::make_message<MPing>()); + return {seastar::now()}; + } + + public: + seastar::future<> init(const entity_name_t& name, + const std::string& lname, + const uint64_t nonce, + const entity_addr_t& addr) { + msgr = crimson::net::Messenger::create( + name, lname, nonce, true); + msgr->set_default_policy(crimson::net::SocketPolicy::stateless_server(0)); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + return msgr->bind(entity_addrvec_t{addr}).safe_then([this] { + return msgr->start({this}); + }, crimson::net::Messenger::bind_ertr::all_same_way( + [addr] (const std::error_code& e) { + logger().error("test_preemptive_shutdown(): " + "there is another instance running at {}", addr); + ceph_abort(); + })); + } + entity_addr_t get_addr() const { + return msgr->get_myaddr(); + } + seastar::future<> shutdown() { + msgr->stop(); + return msgr->shutdown(); + } + }; + + class Client final + : public crimson::net::Dispatcher { + crimson::net::MessengerRef msgr; + crimson::auth::DummyAuthClientServer dummy_auth; + + bool stop_send = false; + seastar::promise<> stopped_send_promise; + + std::optional<seastar::future<>> ms_dispatch( + crimson::net::ConnectionRef, MessageRef m) override { + return {seastar::now()}; + } + + public: + seastar::future<> init(const entity_name_t& name, + const std::string& lname, + const uint64_t nonce) { + msgr = crimson::net::Messenger::create( + name, lname, nonce, true); + msgr->set_default_policy(crimson::net::SocketPolicy::lossy_client(0)); + msgr->set_auth_client(&dummy_auth); + msgr->set_auth_server(&dummy_auth); + return msgr->start({this}); + } + void send_pings(const entity_addr_t& addr) { + auto conn = msgr->connect(addr, entity_name_t::TYPE_OSD); + // forwarded to stopped_send_promise + (void) seastar::do_until( + [this] { return stop_send; }, + [conn] { + return conn->send(crimson::make_message<MPing>()).then([] { + return seastar::sleep(0ms); + }); + } + ).then_wrapped([this, conn] (auto fut) { + fut.forward_to(std::move(stopped_send_promise)); + }); + } + seastar::future<> shutdown() { + msgr->stop(); + return msgr->shutdown().then([this] { + stop_send = true; + return stopped_send_promise.get_future(); + }); + } + }; + }; + + logger().info("test_preemptive_shutdown():"); + auto server = seastar::make_shared<test_state::Server>(); + auto client = seastar::make_shared<test_state::Client>(); + auto addr = get_server_addr(); + addr.set_type(entity_addr_t::TYPE_MSGR2); + addr.set_family(AF_INET); + return seastar::when_all_succeed( + server->init(entity_name_t::OSD(6), "server4", 7, addr), + client->init(entity_name_t::OSD(7), "client4", 8) + ).then_unpack([server, client] { + client->send_pings(server->get_addr()); + return seastar::sleep(100ms); + }).then([client] { + logger().info("client shutdown..."); + return client->shutdown(); + }).then([server] { + logger().info("server shutdown..."); + return server->shutdown(); + }).then([] { + logger().info("test_preemptive_shutdown() done!\n"); + }).handle_exception([server, client] (auto eptr) { + logger().error("test_preemptive_shutdown() failed: got exception {}", eptr); + throw; + }); +} + +using ceph::msgr::v2::Tag; +using crimson::net::bp_action_t; +using crimson::net::bp_type_t; +using crimson::net::Breakpoint; +using crimson::net::Connection; +using crimson::net::ConnectionRef; +using crimson::net::custom_bp_t; +using crimson::net::Dispatcher; +using crimson::net::Interceptor; +using crimson::net::Messenger; +using crimson::net::MessengerRef; +using crimson::net::SocketPolicy; +using crimson::net::tag_bp_t; +using namespace ceph::net::test; + +struct counter_t { unsigned counter = 0; }; + +enum class conn_state_t { + unknown = 0, + established, + closed, + replaced, +}; + +std::ostream& operator<<(std::ostream& out, const conn_state_t& state) { + switch(state) { + case conn_state_t::unknown: + return out << "unknown"; + case conn_state_t::established: + return out << "established"; + case conn_state_t::closed: + return out << "closed"; + case conn_state_t::replaced: + return out << "replaced"; + default: + ceph_abort(); + } +} + +} // anonymous namespace + +#if FMT_VERSION >= 90000 +template<> +struct fmt::formatter<conn_state_t> : fmt::ostream_formatter {}; +#endif + +namespace { + +struct ConnResult { + ConnectionRef conn; + unsigned index; + conn_state_t state = conn_state_t::unknown; + + unsigned connect_attempts = 0; + unsigned client_connect_attempts = 0; + unsigned client_reconnect_attempts = 0; + unsigned cnt_connect_dispatched = 0; + + unsigned accept_attempts = 0; + unsigned server_connect_attempts = 0; + unsigned server_reconnect_attempts = 0; + unsigned cnt_accept_dispatched = 0; + + unsigned cnt_reset_dispatched = 0; + unsigned cnt_remote_reset_dispatched = 0; + + ConnResult(ConnectionRef conn, unsigned index) + : conn(conn), index(index) {} + + template <typename T> + void _assert_eq(const char* expr_actual, T actual, + const char* expr_expected, T expected) const { + if (actual != expected) { + throw std::runtime_error(fmt::format( + "[{}] {} '{}' is actually {}, not the expected '{}' {}", + index, *conn, expr_actual, actual, expr_expected, expected)); + } + } + +#define ASSERT_EQUAL(actual, expected) \ + _assert_eq(#actual, actual, #expected, expected) + + void assert_state_at(conn_state_t expected) const { + ASSERT_EQUAL(state, expected); + } + + void assert_connect(unsigned attempts, + unsigned connects, + unsigned reconnects, + unsigned dispatched) const { + ASSERT_EQUAL(connect_attempts, attempts); + ASSERT_EQUAL(client_connect_attempts, connects); + ASSERT_EQUAL(client_reconnect_attempts, reconnects); + ASSERT_EQUAL(cnt_connect_dispatched, dispatched); + } + + void assert_connect(unsigned attempts, + unsigned dispatched) const { + ASSERT_EQUAL(connect_attempts, attempts); + ASSERT_EQUAL(cnt_connect_dispatched, dispatched); + } + + void assert_accept(unsigned attempts, + unsigned accepts, + unsigned reaccepts, + unsigned dispatched) const { + ASSERT_EQUAL(accept_attempts, attempts); + ASSERT_EQUAL(server_connect_attempts, accepts); + ASSERT_EQUAL(server_reconnect_attempts, reaccepts); + ASSERT_EQUAL(cnt_accept_dispatched, dispatched); + } + + void assert_accept(unsigned attempts, + unsigned dispatched) const { + ASSERT_EQUAL(accept_attempts, attempts); + ASSERT_EQUAL(cnt_accept_dispatched, dispatched); + } + + void assert_reset(unsigned local, unsigned remote) const { + ASSERT_EQUAL(cnt_reset_dispatched, local); + ASSERT_EQUAL(cnt_remote_reset_dispatched, remote); + } + + void dump() const { + logger().info("\nResult({}):\n" + " conn: [{}] {}:\n" + " state: {}\n" + " connect_attempts: {}\n" + " client_connect_attempts: {}\n" + " client_reconnect_attempts: {}\n" + " cnt_connect_dispatched: {}\n" + " accept_attempts: {}\n" + " server_connect_attempts: {}\n" + " server_reconnect_attempts: {}\n" + " cnt_accept_dispatched: {}\n" + " cnt_reset_dispatched: {}\n" + " cnt_remote_reset_dispatched: {}\n", + static_cast<const void*>(this), + index, *conn, + state, + connect_attempts, + client_connect_attempts, + client_reconnect_attempts, + cnt_connect_dispatched, + accept_attempts, + server_connect_attempts, + server_reconnect_attempts, + cnt_accept_dispatched, + cnt_reset_dispatched, + cnt_remote_reset_dispatched); + } +}; +using ConnResults = std::vector<ConnResult>; + +struct TestInterceptor : public Interceptor { + std::map<Breakpoint, std::map<unsigned, bp_action_t>> breakpoints; + std::map<Breakpoint, counter_t> breakpoints_counter; + std::map<Connection*, unsigned> conns; + ConnResults results; + std::optional<seastar::abort_source> signal; + const seastar::shard_id primary_sid; + + TestInterceptor() : primary_sid{seastar::this_shard_id()} {} + + // only used for copy breakpoint configurations + TestInterceptor(const TestInterceptor& other) : primary_sid{other.primary_sid} { + assert(other.breakpoints_counter.empty()); + assert(other.conns.empty()); + assert(other.results.empty()); + breakpoints = other.breakpoints; + assert(!other.signal); + assert(seastar::this_shard_id() == primary_sid); + } + + void make_fault(Breakpoint bp, unsigned round = 1) { + assert(round >= 1); + breakpoints[bp][round] = bp_action_t::FAULT; + } + + void make_block(Breakpoint bp, unsigned round = 1) { + assert(round >= 1); + breakpoints[bp][round] = bp_action_t::BLOCK; + } + + void make_stall(Breakpoint bp, unsigned round = 1) { + assert(round >= 1); + breakpoints[bp][round] = bp_action_t::STALL; + } + + ConnResult* find_result(Connection *conn) { + assert(seastar::this_shard_id() == primary_sid); + auto it = conns.find(conn); + if (it == conns.end()) { + return nullptr; + } else { + return &results[it->second]; + } + } + + seastar::future<> wait() { + assert(seastar::this_shard_id() == primary_sid); + assert(!signal); + signal = seastar::abort_source(); + return seastar::sleep_abortable(10s, *signal).then([] { + throw std::runtime_error("Timeout (10s) in TestInterceptor::wait()"); + }).handle_exception_type([] (const seastar::sleep_aborted& e) { + // wait done! + }); + } + + void notify() { + assert(seastar::this_shard_id() == primary_sid); + if (signal) { + signal->request_abort(); + signal = std::nullopt; + } + } + + private: + void register_conn(ConnectionRef conn) override { + auto result = find_result(&*conn); + if (result != nullptr) { + logger().error("The connection [{}] {} already exists when register {}", + result->index, *result->conn, *conn); + ceph_abort(); + } + unsigned index = results.size(); + results.emplace_back(conn, index); + conns[&*conn] = index; + notify(); + logger().info("[{}] {} new connection registered", index, *conn); + } + + void register_conn_closed(ConnectionRef conn) override { + auto result = find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked closed connection: {}", *conn); + ceph_abort(); + } + + if (result->state != conn_state_t::replaced) { + result->state = conn_state_t::closed; + } + notify(); + logger().info("[{}] {} closed({})", result->index, *conn, result->state); + } + + void register_conn_ready(ConnectionRef conn) override { + auto result = find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked ready connection: {}", *conn); + ceph_abort(); + } + + ceph_assert(conn->is_protocol_ready()); + notify(); + logger().info("[{}] {} ready", result->index, *conn); + } + + void register_conn_replaced(ConnectionRef conn) override { + auto result = find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked replaced connection: {}", *conn); + ceph_abort(); + } + + result->state = conn_state_t::replaced; + logger().info("[{}] {} {}", result->index, *conn, result->state); + } + + seastar::future<bp_action_t> + intercept(Connection &_conn, std::vector<Breakpoint> bps) override { + assert(bps.size() >= 1); + Connection *conn = &_conn; + + return seastar::smp::submit_to(primary_sid, [conn, bps, this] { + std::vector<bp_action_t> actions; + for (const Breakpoint &bp : bps) { + ++breakpoints_counter[bp].counter; + + auto result = find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked intercepted connection: {}, at breakpoint {}({})", + *conn, bp, breakpoints_counter[bp].counter); + ceph_abort(); + } + + if (bp == custom_bp_t::SOCKET_CONNECTING) { + ++result->connect_attempts; + logger().info("[Test] connect_attempts={}", result->connect_attempts); + } else if (bp == tag_bp_t{Tag::CLIENT_IDENT, bp_type_t::WRITE}) { + ++result->client_connect_attempts; + logger().info("[Test] client_connect_attempts={}", result->client_connect_attempts); + } else if (bp == tag_bp_t{Tag::SESSION_RECONNECT, bp_type_t::WRITE}) { + ++result->client_reconnect_attempts; + logger().info("[Test] client_reconnect_attempts={}", result->client_reconnect_attempts); + } else if (bp == custom_bp_t::SOCKET_ACCEPTED) { + ++result->accept_attempts; + logger().info("[Test] accept_attempts={}", result->accept_attempts); + } else if (bp == tag_bp_t{Tag::CLIENT_IDENT, bp_type_t::READ}) { + ++result->server_connect_attempts; + logger().info("[Test] server_connect_attemps={}", result->server_connect_attempts); + } else if (bp == tag_bp_t{Tag::SESSION_RECONNECT, bp_type_t::READ}) { + ++result->server_reconnect_attempts; + logger().info("[Test] server_reconnect_attempts={}", result->server_reconnect_attempts); + } + + auto it_bp = breakpoints.find(bp); + if (it_bp != breakpoints.end()) { + auto it_cnt = it_bp->second.find(breakpoints_counter[bp].counter); + if (it_cnt != it_bp->second.end()) { + logger().info("[{}] {} intercepted {}({}) => {}", + result->index, *conn, bp, + breakpoints_counter[bp].counter, it_cnt->second); + actions.emplace_back(it_cnt->second); + continue; + } + } + logger().info("[{}] {} intercepted {}({})", + result->index, *conn, bp, breakpoints_counter[bp].counter); + actions.emplace_back(bp_action_t::CONTINUE); + } + + bp_action_t action = bp_action_t::CONTINUE; + for (bp_action_t &a : actions) { + if (a != bp_action_t::CONTINUE) { + if (action == bp_action_t::CONTINUE) { + action = a; + } else { + ceph_abort("got multiple incompatible actions"); + } + } + } + return seastar::make_ready_future<bp_action_t>(action); + }); + } +}; + +SocketPolicy to_socket_policy(policy_t policy) { + switch (policy) { + case policy_t::stateful_server: + return SocketPolicy::stateful_server(0); + case policy_t::stateless_server: + return SocketPolicy::stateless_server(0); + case policy_t::lossless_peer: + return SocketPolicy::lossless_peer(0); + case policy_t::lossless_peer_reuse: + return SocketPolicy::lossless_peer_reuse(0); + case policy_t::lossy_client: + return SocketPolicy::lossy_client(0); + case policy_t::lossless_client: + return SocketPolicy::lossless_client(0); + default: + logger().error("unexpected policy type"); + ceph_abort(); + } +} + +class FailoverSuite : public Dispatcher { + crimson::auth::DummyAuthClientServer dummy_auth; + MessengerRef test_msgr; + const entity_addr_t test_peer_addr; + TestInterceptor interceptor; + + unsigned tracked_index = 0; + Connection *tracked_conn = nullptr; + unsigned pending_send = 0; + unsigned pending_peer_receive = 0; + unsigned pending_receive = 0; + + ShardedGates &gates; + const seastar::shard_id primary_sid; + + std::optional<seastar::future<>> ms_dispatch( + ConnectionRef conn_ref, MessageRef m) override { + ceph_assert(m->get_type() == CEPH_MSG_OSD_OP); + Connection *conn = &*conn_ref; + gates.dispatch_in_background("TestSuite_ms_dispatch", + [this, conn, conn_ref] { + return seastar::smp::submit_to(primary_sid, [this, conn] { + auto result = interceptor.find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked ms dispatched connection: {}", *conn); + ceph_abort(); + } + + if (tracked_conn != &*conn) { + logger().warn("[{}] {} got op, but doesn't match tracked_conn [{}] {}", + result->index, *conn, tracked_index, *tracked_conn); + } else { + ceph_assert(result->index == tracked_index); + } + + ceph_assert(pending_receive > 0); + --pending_receive; + if (pending_receive == 0) { + interceptor.notify(); + } + logger().info("[Test] got op, left {} ops -- [{}] {}", + pending_receive, result->index, *conn); + }).then([conn_ref] {}); + }); + return {seastar::now()}; + } + + void ms_handle_accept( + ConnectionRef conn_ref, + seastar::shard_id prv_shard, + bool is_replace) override { + Connection *conn = &*conn_ref; + gates.dispatch_in_background("TestSuite_ms_dispatch", + [this, conn, conn_ref] { + return seastar::smp::submit_to(primary_sid, [this, conn] { + auto result = interceptor.find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked accepted connection: {}", *conn); + ceph_abort(); + } + + if (tracked_conn && + !tracked_conn->is_protocol_closed() && + tracked_conn != &*conn) { + logger().error("[{}] {} got accepted, but there's already a valid traced_conn [{}] {}", + result->index, *conn, tracked_index, *tracked_conn); + ceph_abort(); + } + + tracked_index = result->index; + tracked_conn = &*conn; + ++result->cnt_accept_dispatched; + logger().info("[Test] got accept (cnt_accept_dispatched={}), track [{}] {}", + result->cnt_accept_dispatched, result->index, *conn); + return flush_pending_send(); + }).then([conn_ref] {}); + }); + } + + void ms_handle_connect( + ConnectionRef conn_ref, + seastar::shard_id prv_shard) override { + Connection *conn = &*conn_ref; + gates.dispatch_in_background("TestSuite_ms_dispatch", + [this, conn, conn_ref] { + return seastar::smp::submit_to(primary_sid, [this, conn] { + auto result = interceptor.find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked connected connection: {}", *conn); + ceph_abort(); + } + + if (tracked_conn && + !tracked_conn->is_protocol_closed() && + tracked_conn != &*conn) { + logger().error("[{}] {} got connected, but there's already a avlid tracked_conn [{}] {}", + result->index, *conn, tracked_index, *tracked_conn); + ceph_abort(); + } + + if (tracked_conn == &*conn) { + ceph_assert(result->index == tracked_index); + } + + ++result->cnt_connect_dispatched; + logger().info("[Test] got connected (cnt_connect_dispatched={}) -- [{}] {}", + result->cnt_connect_dispatched, result->index, *conn); + }).then([conn_ref] {}); + }); + } + + void ms_handle_reset( + ConnectionRef conn_ref, + bool is_replace) override { + Connection *conn = &*conn_ref; + gates.dispatch_in_background("TestSuite_ms_dispatch", + [this, conn, conn_ref] { + return seastar::smp::submit_to(primary_sid, [this, conn] { + auto result = interceptor.find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked reset connection: {}", *conn); + ceph_abort(); + } + + if (tracked_conn != &*conn) { + logger().warn("[{}] {} got reset, but doesn't match tracked_conn [{}] {}", + result->index, *conn, tracked_index, *tracked_conn); + } else { + ceph_assert(result->index == tracked_index); + tracked_index = 0; + tracked_conn = nullptr; + } + + ++result->cnt_reset_dispatched; + logger().info("[Test] got reset (cnt_reset_dispatched={}), untrack [{}] {}", + result->cnt_reset_dispatched, result->index, *conn); + }).then([conn_ref] {}); + }); + } + + void ms_handle_remote_reset( + ConnectionRef conn_ref) override { + Connection *conn = &*conn_ref; + gates.dispatch_in_background("TestSuite_ms_dispatch", + [this, conn, conn_ref] { + return seastar::smp::submit_to(primary_sid, [this, conn] { + auto result = interceptor.find_result(&*conn); + if (result == nullptr) { + logger().error("Untracked remotely reset connection: {}", *conn); + ceph_abort(); + } + + if (tracked_conn != &*conn) { + logger().warn("[{}] {} got remotely reset, but doesn't match tracked_conn [{}] {}", + result->index, *conn, tracked_index, *tracked_conn); + } else { + ceph_assert(result->index == tracked_index); + } + + ++result->cnt_remote_reset_dispatched; + logger().info("[Test] got remote reset (cnt_remote_reset_dispatched={}) -- [{}] {}", + result->cnt_remote_reset_dispatched, result->index, *conn); + }).then([conn_ref] {}); + }); + } + + private: + seastar::future<> init(entity_addr_t test_addr, SocketPolicy policy) { + test_msgr->set_default_policy(policy); + test_msgr->set_auth_client(&dummy_auth); + test_msgr->set_auth_server(&dummy_auth); + test_msgr->set_interceptor(&interceptor); + return test_msgr->bind(entity_addrvec_t{test_addr}).safe_then([this] { + return test_msgr->start({this}); + }, Messenger::bind_ertr::all_same_way([test_addr] (const std::error_code& e) { + logger().error("FailoverSuite: " + "there is another instance running at {}", test_addr); + ceph_abort(); + })); + } + + seastar::future<> send_op(bool expect_reply=true) { + ceph_assert(tracked_conn); + ceph_assert(!tracked_conn->is_protocol_closed()); + if (expect_reply) { + ++pending_peer_receive; + } + pg_t pgid; + object_locator_t oloc; + hobject_t hobj(object_t(), oloc.key, CEPH_NOSNAP, pgid.ps(), + pgid.pool(), oloc.nspace); + spg_t spgid(pgid); + return tracked_conn->send(crimson::make_message<MOSDOp>(0, 0, hobj, spgid, 0, 0, 0)); + } + + seastar::future<> flush_pending_send() { + if (pending_send != 0) { + logger().info("[Test] flush sending {} ops", pending_send); + } + ceph_assert(tracked_conn); + ceph_assert(!tracked_conn->is_protocol_closed()); + return seastar::do_until( + [this] { return pending_send == 0; }, + [this] { + --pending_send; + return send_op(); + }); + } + + seastar::future<> wait_ready(unsigned num_ready_conns, + unsigned num_replaced, + bool wait_received) { + assert(seastar::this_shard_id() == primary_sid); + unsigned pending_conns = 0; + unsigned pending_establish = 0; + unsigned replaced_conns = 0; + for (auto& result : interceptor.results) { + if (result.conn->is_protocol_closed_clean()) { + if (result.state == conn_state_t::replaced) { + ++replaced_conns; + } + } else if (result.conn->is_protocol_ready()) { + if (pending_send == 0 && pending_peer_receive == 0 && pending_receive == 0) { + result.state = conn_state_t::established; + } else { + ++pending_establish; + } + } else { + ++pending_conns; + } + } + + bool do_wait = false; + if (num_ready_conns > 0) { + if (interceptor.results.size() > num_ready_conns) { + throw std::runtime_error(fmt::format( + "{} connections, more than expected: {}", + interceptor.results.size(), num_ready_conns)); + } else if (interceptor.results.size() < num_ready_conns || pending_conns > 0) { + logger().info("[Test] wait_ready(): wait for connections," + " currently {} out of {}, pending {} ready ...", + interceptor.results.size(), num_ready_conns, pending_conns); + do_wait = true; + } + } + if (wait_received) { + if (pending_send || pending_peer_receive || pending_receive) { + if (pending_conns || pending_establish) { + logger().info("[Test] wait_ready(): wait for pending_send={}," + " pending_peer_receive={}, pending_receive={}," + " pending {}/{} ready/establish connections ...", + pending_send, pending_peer_receive, pending_receive, + pending_conns, pending_establish); + do_wait = true; + } else { + // If there are pending messages, stop waiting if there are + // no longer pending connections. + } + } else { + // Stop waiting if there are no pending messages. Pending connections + // should not be important. + } + } + if (num_replaced > 0) { + if (replaced_conns > num_replaced) { + throw std::runtime_error(fmt::format( + "{} replaced connections, more than expected: {}", + replaced_conns, num_replaced)); + } + if (replaced_conns < num_replaced) { + logger().info("[Test] wait_ready(): wait for {} replaced connections," + " currently {} ...", + num_replaced, replaced_conns); + do_wait = true; + } + } + + if (do_wait) { + return interceptor.wait( + ).then([this, num_ready_conns, num_replaced, wait_received] { + return wait_ready(num_ready_conns, num_replaced, wait_received); + }); + } else { + logger().info("[Test] wait_ready(): wait done!"); + return seastar::now(); + } + } + + // called by FailoverTest + public: + FailoverSuite(MessengerRef test_msgr, + entity_addr_t test_peer_addr, + const TestInterceptor& interceptor, + ShardedGates &gates) + : test_msgr(test_msgr), + test_peer_addr(test_peer_addr), + interceptor(interceptor), + gates{gates}, + primary_sid{seastar::this_shard_id()} { } + + entity_addr_t get_addr() const { + return test_msgr->get_myaddr(); + } + + seastar::future<> shutdown() { + test_msgr->stop(); + return test_msgr->shutdown(); + } + + void needs_receive() { + ++pending_receive; + } + + void notify_peer_reply() { + ceph_assert(pending_peer_receive > 0); + --pending_peer_receive; + logger().info("[Test] TestPeer said got op, left {} ops", + pending_peer_receive); + if (pending_peer_receive == 0) { + interceptor.notify(); + } + } + + void post_check() const { + // make sure all breakpoints were hit + for (auto& kv : interceptor.breakpoints) { + auto it = interceptor.breakpoints_counter.find(kv.first); + if (it == interceptor.breakpoints_counter.end()) { + throw std::runtime_error(fmt::format("{} was missed", kv.first)); + } + auto expected = kv.second.rbegin()->first; + if (expected > it->second.counter) { + throw std::runtime_error(fmt::format( + "{} only triggered {} times, not the expected {}", + kv.first, it->second.counter, expected)); + } + } + } + + void dump_results() const { + for (auto& result : interceptor.results) { + result.dump(); + } + } + + static seastar::future<std::unique_ptr<FailoverSuite>> + create(entity_addr_t test_addr, + SocketPolicy test_policy, + entity_addr_t test_peer_addr, + const TestInterceptor& interceptor, + ShardedGates &gates) { + auto suite = std::make_unique<FailoverSuite>( + Messenger::create( + entity_name_t::OSD(TEST_OSD), + "Test", + TEST_NONCE, + false), + test_peer_addr, + interceptor, + gates); + return suite->init(test_addr, test_policy + ).then([suite = std::move(suite)] () mutable { + return std::move(suite); + }); + } + + // called by tests + public: + seastar::future<> connect_peer() { + logger().info("[Test] connect_peer({})", test_peer_addr); + assert(seastar::this_shard_id() == primary_sid); + auto conn = test_msgr->connect(test_peer_addr, entity_name_t::TYPE_OSD); + auto result = interceptor.find_result(&*conn); + ceph_assert(result != nullptr); + + if (tracked_conn) { + if (tracked_conn->is_protocol_closed()) { + logger().info("[Test] this is a new session" + " replacing an closed one"); + ceph_assert(tracked_conn != &*conn); + } else { + logger().info("[Test] this is not a new session"); + ceph_assert(tracked_index == result->index); + ceph_assert(tracked_conn == &*conn); + } + } else { + logger().info("[Test] this is a new session"); + } + tracked_index = result->index; + tracked_conn = &*conn; + + return flush_pending_send(); + } + + seastar::future<> send_peer() { + assert(seastar::this_shard_id() == primary_sid); + if (tracked_conn) { + logger().info("[Test] send_peer()"); + ceph_assert(!tracked_conn->is_protocol_closed()); + ceph_assert(!pending_send); + return send_op(); + } else { + ++pending_send; + logger().info("[Test] send_peer() (pending {})", pending_send); + return seastar::now(); + } + } + + seastar::future<> keepalive_peer() { + logger().info("[Test] keepalive_peer()"); + assert(seastar::this_shard_id() == primary_sid); + ceph_assert(tracked_conn); + ceph_assert(!tracked_conn->is_protocol_closed()); + return tracked_conn->send_keepalive(); + } + + seastar::future<> try_send_peer() { + logger().info("[Test] try_send_peer()"); + assert(seastar::this_shard_id() == primary_sid); + ceph_assert(tracked_conn); + ceph_assert(!tracked_conn->is_protocol_closed()); + return send_op(false); + } + + seastar::future<> markdown() { + logger().info("[Test] markdown() in 100ms ..."); + assert(seastar::this_shard_id() == primary_sid); + ceph_assert(tracked_conn); + // sleep to propagate potential remaining acks + return seastar::sleep(50ms + ).then([this] { + return seastar::smp::submit_to( + tracked_conn->get_shard_id(), [tracked_conn=tracked_conn] { + assert(tracked_conn->get_shard_id() == seastar::this_shard_id()); + tracked_conn->mark_down(); + }); + }).then([] { + // sleep to wait for markdown propagate to the primary sid + return seastar::sleep(100ms); + }); + } + + seastar::future<> wait_blocked() { + logger().info("[Test] wait_blocked() ..."); + assert(seastar::this_shard_id() == primary_sid); + return interceptor.blocker.wait_blocked(); + } + + void unblock() { + logger().info("[Test] unblock()"); + assert(seastar::this_shard_id() == primary_sid); + return interceptor.blocker.unblock(); + } + + seastar::future<> wait_replaced(unsigned count) { + logger().info("[Test] wait_replaced({}) ...", count); + return wait_ready(0, count, false); + } + + seastar::future<> wait_established() { + logger().info("[Test] wait_established() ..."); + return wait_ready(0, 0, true); + } + + seastar::future<std::reference_wrapper<ConnResults>> + wait_results(unsigned count) { + logger().info("[Test] wait_result({}) ...", count); + return wait_ready(count, 0, true).then([this] { + return std::reference_wrapper<ConnResults>(interceptor.results); + }); + } + + bool is_standby() { + assert(seastar::this_shard_id() == primary_sid); + ceph_assert(tracked_conn); + return tracked_conn->is_protocol_standby(); + } +}; + +class FailoverTest : public Dispatcher { + crimson::auth::DummyAuthClientServer dummy_auth; + MessengerRef cmd_msgr; + ConnectionRef cmd_conn; + const entity_addr_t test_addr; + const entity_addr_t test_peer_addr; + + std::optional<seastar::promise<>> recv_pong; + std::optional<seastar::promise<>> recv_cmdreply; + + std::unique_ptr<FailoverSuite> test_suite; + + std::optional<seastar::future<>> ms_dispatch(ConnectionRef c, MessageRef m) override { + switch (m->get_type()) { + case CEPH_MSG_PING: + ceph_assert(recv_pong); + recv_pong->set_value(); + recv_pong = std::nullopt; + break; + case MSG_COMMAND_REPLY: + ceph_assert(recv_cmdreply); + recv_cmdreply->set_value(); + recv_cmdreply = std::nullopt; + break; + case MSG_COMMAND: { + auto m_cmd = boost::static_pointer_cast<MCommand>(m); + ceph_assert(static_cast<cmd_t>(m_cmd->cmd[0][0]) == cmd_t::suite_recv_op); + ceph_assert(test_suite); + test_suite->notify_peer_reply(); + break; + } + default: + logger().error("{} got unexpected msg from cmd server: {}", *c, *m); + ceph_abort(); + } + return {seastar::now()}; + } + + private: + seastar::future<> prepare_cmd( + cmd_t cmd, + std::function<void(MCommand&)> + f_prepare = [] (auto& m) { return; }) { + assert(!recv_cmdreply); + recv_cmdreply = seastar::promise<>(); + auto fut = recv_cmdreply->get_future(); + auto m = crimson::make_message<MCommand>(); + m->cmd.emplace_back(1, static_cast<char>(cmd)); + f_prepare(*m); + return cmd_conn->send(std::move(m)).then([fut = std::move(fut)] () mutable { + return std::move(fut); + }); + } + + seastar::future<> start_peer(policy_t peer_policy) { + return prepare_cmd(cmd_t::suite_start, + [peer_policy] (auto& m) { + m.cmd.emplace_back(1, static_cast<char>(peer_policy)); + }); + } + + seastar::future<> stop_peer() { + return prepare_cmd(cmd_t::suite_stop); + } + + seastar::future<> pingpong() { + assert(!recv_pong); + recv_pong = seastar::promise<>(); + auto fut = recv_pong->get_future(); + return cmd_conn->send(crimson::make_message<MPing>() + ).then([fut = std::move(fut)] () mutable { + return std::move(fut); + }); + } + + seastar::future<> init(entity_addr_t cmd_peer_addr) { + cmd_msgr->set_default_policy(SocketPolicy::lossy_client(0)); + cmd_msgr->set_auth_client(&dummy_auth); + cmd_msgr->set_auth_server(&dummy_auth); + return cmd_msgr->start({this}).then([this, cmd_peer_addr] { + logger().info("CmdCli connect to CmdSrv({}) ...", cmd_peer_addr); + cmd_conn = cmd_msgr->connect(cmd_peer_addr, entity_name_t::TYPE_OSD); + return pingpong(); + }); + } + + public: + FailoverTest(MessengerRef cmd_msgr, + entity_addr_t test_addr, + entity_addr_t test_peer_addr) + : cmd_msgr(cmd_msgr), + test_addr(test_addr), + test_peer_addr(test_peer_addr) { } + + seastar::future<> shutdown() { + logger().info("CmdCli shutdown..."); + assert(!recv_cmdreply); + auto m = crimson::make_message<MCommand>(); + m->cmd.emplace_back(1, static_cast<char>(cmd_t::shutdown)); + return cmd_conn->send(std::move(m)).then([] { + return seastar::sleep(200ms); + }).then([this] { + cmd_msgr->stop(); + return cmd_msgr->shutdown(); + }); + } + + static seastar::future<seastar::lw_shared_ptr<FailoverTest>> + create(entity_addr_t test_addr, + entity_addr_t cmd_peer_addr, + entity_addr_t test_peer_addr) { + auto test = seastar::make_lw_shared<FailoverTest>( + Messenger::create( + entity_name_t::OSD(CMD_CLI_OSD), + "CmdCli", + CMD_CLI_NONCE, + true), + test_addr, test_peer_addr); + return test->init(cmd_peer_addr).then([test] { + logger().info("CmdCli ready"); + return test; + }); + } + + // called by tests + public: + seastar::future<> run_suite( + std::string name, + const TestInterceptor& interceptor, + policy_t test_policy, + policy_t peer_policy, + std::function<seastar::future<>(FailoverSuite&)>&& f) { + logger().info("\n\n[{}]", name); + ceph_assert(!test_suite); + SocketPolicy test_policy_ = to_socket_policy(test_policy); + return ShardedGates::create( + ).then([this, test_policy_, peer_policy, interceptor, + f=std::move(f)](auto *gates) mutable { + return FailoverSuite::create( + test_addr, test_policy_, test_peer_addr, interceptor, *gates + ).then([this, peer_policy, f = std::move(f)](auto suite) mutable { + ceph_assert(suite->get_addr() == test_addr); + test_suite.swap(suite); + return start_peer(peer_policy + ).then([this, f = std::move(f)] { + return f(*test_suite); + }).then([this] { + test_suite->post_check(); + logger().info("\n[SUCCESS]"); + }).handle_exception([this](auto eptr) { + logger().info("\n[FAIL: {}]", eptr); + test_suite->dump_results(); + throw; + }).then([this] { + return stop_peer(); + }).then([this] { + return test_suite->shutdown( + ).then([this] { + test_suite.reset(); + }); + }); + }).then([gates] { + return gates->close(); + }); + }); + } + + seastar::future<> peer_connect_me() { + logger().info("[Test] peer_connect_me({})", test_addr); + return prepare_cmd(cmd_t::suite_connect_me, + [this] (auto& m) { + m.cmd.emplace_back(fmt::format("{}", test_addr)); + }); + } + + seastar::future<> peer_send_me() { + logger().info("[Test] peer_send_me()"); + ceph_assert(test_suite); + test_suite->needs_receive(); + return prepare_cmd(cmd_t::suite_send_me); + } + + seastar::future<> try_peer_send_me() { + logger().info("[Test] try_peer_send_me()"); + ceph_assert(test_suite); + return prepare_cmd(cmd_t::suite_send_me); + } + + seastar::future<> send_bidirectional() { + ceph_assert(test_suite); + return test_suite->send_peer().then([this] { + return peer_send_me(); + }); + } + + seastar::future<> peer_keepalive_me() { + logger().info("[Test] peer_keepalive_me()"); + ceph_assert(test_suite); + return prepare_cmd(cmd_t::suite_keepalive_me); + } + + seastar::future<> markdown_peer() { + logger().info("[Test] markdown_peer() in 150ms ..."); + // sleep to propagate potential remaining acks + return seastar::sleep(50ms + ).then([this] { + return prepare_cmd(cmd_t::suite_markdown); + }).then([] { + // sleep awhile for peer markdown propagated + return seastar::sleep(100ms); + }); + } +}; + +class FailoverSuitePeer : public Dispatcher { + using cb_t = std::function<seastar::future<>()>; + crimson::auth::DummyAuthClientServer dummy_auth; + MessengerRef peer_msgr; + cb_t op_callback; + + ConnectionRef tracked_conn; + unsigned pending_send = 0; + + std::optional<seastar::future<>> ms_dispatch(ConnectionRef conn, MessageRef m) override { + logger().info("[TestPeer] got op from Test"); + ceph_assert(m->get_type() == CEPH_MSG_OSD_OP); + std::ignore = op_callback(); + return {seastar::now()}; + } + + void ms_handle_accept( + ConnectionRef conn, + seastar::shard_id prv_shard, + bool is_replace) override { + assert(prv_shard == seastar::this_shard_id()); + logger().info("[TestPeer] got accept from Test"); + + if (tracked_conn && + !tracked_conn->is_protocol_closed() && + tracked_conn != conn) { + logger().error("[TestPeer] {} got accepted, but there's already a valid traced_conn {}", + *conn, *tracked_conn); + } + tracked_conn = conn; + std::ignore = flush_pending_send(); + } + + void ms_handle_reset(ConnectionRef conn, bool is_replace) override { + logger().info("[TestPeer] got reset from Test"); + } + + private: + seastar::future<> init(entity_addr_t test_peer_addr, SocketPolicy policy) { + peer_msgr->set_default_policy(policy); + peer_msgr->set_auth_client(&dummy_auth); + peer_msgr->set_auth_server(&dummy_auth); + return peer_msgr->bind(entity_addrvec_t{test_peer_addr}).safe_then([this] { + return peer_msgr->start({this}); + }, Messenger::bind_ertr::all_same_way([test_peer_addr] (const std::error_code& e) { + logger().error("FailoverSuitePeer: " + "there is another instance running at {}", test_peer_addr); + ceph_abort(); + })); + } + + seastar::future<> send_op() { + ceph_assert(tracked_conn); + if (tracked_conn->is_protocol_closed()) { + logger().error("[TestPeer] send op but the connection is closed -- {}", + *tracked_conn); + } + + pg_t pgid; + object_locator_t oloc; + hobject_t hobj(object_t(), oloc.key, CEPH_NOSNAP, pgid.ps(), + pgid.pool(), oloc.nspace); + spg_t spgid(pgid); + return tracked_conn->send(crimson::make_message<MOSDOp>(0, 0, hobj, spgid, 0, 0, 0)); + } + + seastar::future<> flush_pending_send() { + if (pending_send != 0) { + logger().info("[TestPeer] flush sending {} ops", pending_send); + } + ceph_assert(tracked_conn); + return seastar::do_until( + [this] { return pending_send == 0; }, + [this] { + --pending_send; + return send_op(); + }); + } + + public: + FailoverSuitePeer(MessengerRef peer_msgr, cb_t op_callback) + : peer_msgr(peer_msgr), + op_callback(op_callback) { } + + seastar::future<> shutdown() { + peer_msgr->stop(); + return peer_msgr->shutdown(); + } + + seastar::future<> connect_peer(entity_addr_t test_addr_decoded) { + logger().info("[TestPeer] connect_peer({})", test_addr_decoded); + auto conn = peer_msgr->connect(test_addr_decoded, entity_name_t::TYPE_OSD); + + if (tracked_conn) { + if (tracked_conn->is_protocol_closed()) { + logger().info("[TestPeer] this is a new session" + " replacing an closed one"); + ceph_assert(tracked_conn != conn); + } else { + logger().info("[TestPeer] this is not a new session"); + ceph_assert(tracked_conn == conn); + } + } else { + logger().info("[TestPeer] this is a new session"); + } + tracked_conn = conn; + + return flush_pending_send(); + } + + seastar::future<> send_peer() { + if (tracked_conn) { + logger().info("[TestPeer] send_peer()"); + ceph_assert(!pending_send); + return send_op(); + } else { + ++pending_send; + logger().info("[TestPeer] send_peer() (pending {})", pending_send); + return seastar::now(); + } + } + + seastar::future<> keepalive_peer() { + logger().info("[TestPeer] keepalive_peer()"); + ceph_assert(tracked_conn); + return tracked_conn->send_keepalive(); + } + + seastar::future<> markdown() { + logger().info("[TestPeer] markdown()"); + ceph_assert(tracked_conn); + tracked_conn->mark_down(); + return seastar::now(); + } + + static seastar::future<std::unique_ptr<FailoverSuitePeer>> + create(entity_addr_t test_peer_addr, const SocketPolicy& policy, cb_t op_callback) { + auto suite = std::make_unique<FailoverSuitePeer>( + Messenger::create( + entity_name_t::OSD(TEST_PEER_OSD), + "TestPeer", + TEST_PEER_NONCE, + true), + op_callback + ); + return suite->init(test_peer_addr, policy + ).then([suite = std::move(suite)] () mutable { + return std::move(suite); + }); + } +}; + +class FailoverTestPeer : public Dispatcher { + crimson::auth::DummyAuthClientServer dummy_auth; + MessengerRef cmd_msgr; + ConnectionRef cmd_conn; + const entity_addr_t test_peer_addr; + std::unique_ptr<FailoverSuitePeer> test_suite; + + std::optional<seastar::future<>> ms_dispatch(ConnectionRef c, MessageRef m) override { + ceph_assert(cmd_conn == c); + switch (m->get_type()) { + case CEPH_MSG_PING: + std::ignore = c->send(crimson::make_message<MPing>()); + break; + case MSG_COMMAND: { + auto m_cmd = boost::static_pointer_cast<MCommand>(m); + auto cmd = static_cast<cmd_t>(m_cmd->cmd[0][0]); + if (cmd == cmd_t::shutdown) { + logger().info("CmdSrv shutdown..."); + // forwarded to FailoverTestPeer::wait() + cmd_msgr->stop(); + std::ignore = cmd_msgr->shutdown(); + } else { + std::ignore = handle_cmd(cmd, m_cmd).then([c] { + return c->send(crimson::make_message<MCommandReply>()); + }); + } + break; + } + default: + logger().error("{} got unexpected msg from cmd client: {}", *c, *m); + ceph_abort(); + } + return {seastar::now()}; + } + + void ms_handle_accept( + ConnectionRef conn, + seastar::shard_id prv_shard, + bool is_replace) override { + assert(prv_shard == seastar::this_shard_id()); + cmd_conn = conn; + } + + private: + seastar::future<> notify_recv_op() { + ceph_assert(cmd_conn); + auto m = crimson::make_message<MCommand>(); + m->cmd.emplace_back(1, static_cast<char>(cmd_t::suite_recv_op)); + return cmd_conn->send(std::move(m)); + } + + seastar::future<> handle_cmd(cmd_t cmd, MRef<MCommand> m_cmd) { + switch (cmd) { + case cmd_t::suite_start: { + ceph_assert(!test_suite); + auto policy = to_socket_policy(static_cast<policy_t>(m_cmd->cmd[1][0])); + return FailoverSuitePeer::create( + test_peer_addr, policy, [this] { return notify_recv_op(); } + ).then([this] (auto suite) { + test_suite.swap(suite); + }); + } + case cmd_t::suite_stop: + ceph_assert(test_suite); + return test_suite->shutdown().then([this] { + test_suite.reset(); + }); + case cmd_t::suite_connect_me: { + ceph_assert(test_suite); + entity_addr_t test_addr_decoded = entity_addr_t(); + test_addr_decoded.parse(m_cmd->cmd[1].c_str(), nullptr); + return test_suite->connect_peer(test_addr_decoded); + } + case cmd_t::suite_send_me: + ceph_assert(test_suite); + return test_suite->send_peer(); + case cmd_t::suite_keepalive_me: + ceph_assert(test_suite); + return test_suite->keepalive_peer(); + case cmd_t::suite_markdown: + ceph_assert(test_suite); + return test_suite->markdown(); + default: + logger().error("TestPeer got unexpected command {} from Test", + fmt::ptr(m_cmd.get())); + ceph_abort(); + return seastar::now(); + } + } + + seastar::future<> init(entity_addr_t cmd_peer_addr) { + cmd_msgr->set_default_policy(SocketPolicy::stateless_server(0)); + cmd_msgr->set_auth_client(&dummy_auth); + cmd_msgr->set_auth_server(&dummy_auth); + return cmd_msgr->bind(entity_addrvec_t{cmd_peer_addr}).safe_then([this] { + return cmd_msgr->start({this}); + }, Messenger::bind_ertr::all_same_way([cmd_peer_addr] (const std::error_code& e) { + logger().error("FailoverTestPeer: " + "there is another instance running at {}", cmd_peer_addr); + ceph_abort(); + })); + } + + public: + FailoverTestPeer(MessengerRef cmd_msgr, + entity_addr_t test_peer_addr) + : cmd_msgr(cmd_msgr), + test_peer_addr(test_peer_addr) { } + + seastar::future<> wait() { + return cmd_msgr->wait(); + } + + static seastar::future<std::unique_ptr<FailoverTestPeer>> + create(entity_addr_t cmd_peer_addr, entity_addr_t test_peer_addr) { + auto test_peer = std::make_unique<FailoverTestPeer>( + Messenger::create( + entity_name_t::OSD(CMD_SRV_OSD), + "CmdSrv", + CMD_SRV_NONCE, + true), + test_peer_addr); + return test_peer->init(cmd_peer_addr + ).then([test_peer = std::move(test_peer)] () mutable { + logger().info("CmdSrv ready"); + return std::move(test_peer); + }); + } +}; + +seastar::future<> +test_v2_lossy_early_connect_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {custom_bp_t::SOCKET_CONNECTING}, + {custom_bp_t::BANNER_WRITE}, + {custom_bp_t::BANNER_READ}, + {custom_bp_t::BANNER_PAYLOAD_READ}, + {Tag::HELLO, bp_type_t::WRITE}, + {Tag::HELLO, bp_type_t::READ}, + {Tag::AUTH_REQUEST, bp_type_t::WRITE}, + {Tag::AUTH_DONE, bp_type_t::READ}, + {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_early_connect_fault -- {}", bp), + interceptor, + policy_t::lossy_client, + policy_t::stateless_server, + [] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossy_connect_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::CLIENT_IDENT, bp_type_t::WRITE}, + {Tag::SERVER_IDENT, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_connect_fault -- {}", bp), + interceptor, + policy_t::lossy_client, + policy_t::stateless_server, + [] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 2, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossy_connected_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::MESSAGE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_connected_fault -- {}", bp), + interceptor, + policy_t::lossy_client, + policy_t::stateless_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(1, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossy_early_accept_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {custom_bp_t::BANNER_WRITE}, + {custom_bp_t::BANNER_READ}, + {custom_bp_t::BANNER_PAYLOAD_READ}, + {Tag::HELLO, bp_type_t::WRITE}, + {Tag::HELLO, bp_type_t::READ}, + {Tag::AUTH_REQUEST, bp_type_t::READ}, + {Tag::AUTH_DONE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_early_accept_fault -- {}", bp), + interceptor, + policy_t::stateless_server, + policy_t::lossy_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossy_accept_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::CLIENT_IDENT, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_accept_fault -- {}", bp), + interceptor, + policy_t::stateless_server, + policy_t::lossy_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossy_establishing_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::SERVER_IDENT, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_establishing_fault -- {}", bp), + interceptor, + policy_t::stateless_server, + policy_t::lossy_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(1, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossy_accepted_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::MESSAGE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossy_accepted_fault -- {}", bp), + interceptor, + policy_t::stateless_server, + policy_t::lossy_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(1, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_connect_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::CLIENT_IDENT, bp_type_t::WRITE}, + {Tag::SERVER_IDENT, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_connect_fault -- {}", bp), + interceptor, + policy_t::lossless_client, + policy_t::stateful_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 2, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_connected_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::MESSAGE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_connected_fault -- {}", bp), + interceptor, + policy_t::lossless_client, + policy_t::stateful_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1, 1, 2); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_connected_fault2(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::ACK, bp_type_t::READ}, + {Tag::ACK, bp_type_t::WRITE}, + {Tag::KEEPALIVE2, bp_type_t::READ}, + {Tag::KEEPALIVE2, bp_type_t::WRITE}, + {Tag::KEEPALIVE2_ACK, bp_type_t::READ}, + {Tag::KEEPALIVE2_ACK, bp_type_t::WRITE}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_connected_fault2 -- {}", bp), + interceptor, + policy_t::lossless_client, + policy_t::stateful_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.keepalive_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_keepalive_me(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1, 1, 2); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_reconnect_fault(FailoverTest& test) { + return seastar::do_with(std::vector<std::pair<Breakpoint, Breakpoint>>{ + {{Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::SESSION_RECONNECT, bp_type_t::WRITE}}, + {{Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::SESSION_RECONNECT_OK, bp_type_t::READ}}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp_pair) { + TestInterceptor interceptor; + interceptor.make_fault(bp_pair.first); + interceptor.make_fault(bp_pair.second); + return test.run_suite( + fmt::format("test_v2_lossless_reconnect_fault -- {}, {}", + bp_pair.first, bp_pair.second), + interceptor, + policy_t::lossless_client, + policy_t::stateful_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(3, 1, 2, 2); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_accept_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::CLIENT_IDENT, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_accept_fault -- {}", bp), + interceptor, + policy_t::stateful_server, + policy_t::lossless_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossless_establishing_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::SERVER_IDENT, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_establishing_fault -- {}", bp), + interceptor, + policy_t::stateful_server, + policy_t::lossless_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossless_accepted_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::MESSAGE, bp_type_t::WRITE}, + {Tag::MESSAGE, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_lossless_accepted_fault -- {}", bp), + interceptor, + policy_t::stateful_server, + policy_t::lossless_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_lossless_reaccept_fault(FailoverTest& test) { + return seastar::do_with(std::vector<std::pair<Breakpoint, Breakpoint>>{ + {{Tag::MESSAGE, bp_type_t::READ}, + {Tag::SESSION_RECONNECT, bp_type_t::READ}}, + {{Tag::MESSAGE, bp_type_t::READ}, + {Tag::SESSION_RECONNECT_OK, bp_type_t::WRITE}}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp_pair) { + TestInterceptor interceptor; + interceptor.make_fault(bp_pair.first); + interceptor.make_fault(bp_pair.second); + return test.run_suite( + fmt::format("test_v2_lossless_reaccept_fault -- {}, {}", + bp_pair.first, bp_pair.second), + interceptor, + policy_t::stateful_server, + policy_t::lossless_client, + [&test, bp = bp_pair.second] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.send_bidirectional(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(3); + }).then([bp] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + if (bp == Breakpoint{Tag::SESSION_RECONNECT, bp_type_t::READ}) { + results[0].assert_accept(1, 1, 0, 2); + } else { + results[0].assert_accept(1, 1, 0, 3); + } + results[0].assert_reset(0, 0); + if (bp == Breakpoint{Tag::SESSION_RECONNECT, bp_type_t::READ}) { + results[1].assert_state_at(conn_state_t::closed); + } else { + results[1].assert_state_at(conn_state_t::replaced); + } + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::replaced); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 0, 1, 0); + results[2].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_peer_connect_fault(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {Tag::CLIENT_IDENT, bp_type_t::WRITE}, + {Tag::SERVER_IDENT, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_peer_connect_fault -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 2, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_peer_accept_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::CLIENT_IDENT, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_peer_accept_fault -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_peer_establishing_fault(FailoverTest& test) { + auto bp = Breakpoint{Tag::SERVER_IDENT, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_peer_establishing_fault -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_peer_connected_fault_reconnect(FailoverTest& test) { + auto bp = Breakpoint{Tag::MESSAGE, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_peer_connected_fault_reconnect -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1, 1, 2); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_peer_connected_fault_reaccept(FailoverTest& test) { + auto bp = Breakpoint{Tag::MESSAGE, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_fault(bp); + return test.run_suite( + fmt::format("test_v2_peer_connected_fault_reaccept -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<bool> +check_peer_wins(FailoverTest& test) { + return seastar::do_with(bool(), [&test] (auto& ret) { + return test.run_suite("check_peer_wins", + TestInterceptor(), + policy_t::lossy_client, + policy_t::stateless_server, + [&ret] (FailoverSuite& suite) { + return suite.connect_peer().then([&suite] { + return suite.wait_results(1); + }).then([&ret] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + ret = results[0].conn->peer_wins(); + logger().info("check_peer_wins: {}", ret); + }); + }).then([&ret] { + return ret; + }); + }); +} + +seastar::future<> +test_v2_racing_reconnect_acceptor_lose(FailoverTest& test) { + return seastar::do_with(std::vector<std::pair<unsigned, Breakpoint>>{ + {1, {Tag::SESSION_RECONNECT, bp_type_t::READ}}, + {2, {custom_bp_t::BANNER_WRITE}}, + {2, {custom_bp_t::BANNER_READ}}, + {2, {custom_bp_t::BANNER_PAYLOAD_READ}}, + {2, {Tag::HELLO, bp_type_t::WRITE}}, + {2, {Tag::HELLO, bp_type_t::READ}}, + {2, {Tag::AUTH_REQUEST, bp_type_t::READ}}, + {2, {Tag::AUTH_DONE, bp_type_t::WRITE}}, + {2, {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}}, + {2, {Tag::AUTH_SIGNATURE, bp_type_t::READ}}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + // fault acceptor + interceptor.make_fault({Tag::MESSAGE, bp_type_t::READ}); + // block acceptor + interceptor.make_block(bp.second, bp.first); + return test.run_suite( + fmt::format("test_v2_racing_reconnect_acceptor_lose -- {}({})", + bp.second, bp.first), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 0, 1, 1); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_racing_reconnect_acceptor_win(FailoverTest& test) { + return seastar::do_with(std::vector<std::pair<unsigned, Breakpoint>>{ + {1, {Tag::SESSION_RECONNECT, bp_type_t::WRITE}}, + {2, {custom_bp_t::SOCKET_CONNECTING}}, + {2, {custom_bp_t::BANNER_WRITE}}, + {2, {custom_bp_t::BANNER_READ}}, + {2, {custom_bp_t::BANNER_PAYLOAD_READ}}, + {2, {Tag::HELLO, bp_type_t::WRITE}}, + {2, {Tag::HELLO, bp_type_t::READ}}, + {2, {Tag::AUTH_REQUEST, bp_type_t::WRITE}}, + {2, {Tag::AUTH_DONE, bp_type_t::READ}}, + {2, {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}}, + {2, {Tag::AUTH_SIGNATURE, bp_type_t::READ}}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + // fault connector + interceptor.make_fault({Tag::MESSAGE, bp_type_t::WRITE}); + // block connector + interceptor.make_block(bp.second, bp.first); + return test.run_suite( + fmt::format("test_v2_racing_reconnect_acceptor_win -- {}({})", + bp.second, bp.first), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_racing_connect_acceptor_lose(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {custom_bp_t::BANNER_WRITE}, + {custom_bp_t::BANNER_READ}, + {custom_bp_t::BANNER_PAYLOAD_READ}, + {Tag::HELLO, bp_type_t::WRITE}, + {Tag::HELLO, bp_type_t::READ}, + {Tag::AUTH_REQUEST, bp_type_t::READ}, + {Tag::AUTH_DONE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::READ}, + {Tag::CLIENT_IDENT, bp_type_t::READ}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + // block acceptor + interceptor.make_block(bp); + return test.run_suite( + fmt::format("test_v2_racing_connect_acceptor_lose -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_racing_connect_acceptor_win(FailoverTest& test) { + return seastar::do_with(std::vector<Breakpoint>{ + {custom_bp_t::SOCKET_CONNECTING}, + {custom_bp_t::BANNER_WRITE}, + {custom_bp_t::BANNER_READ}, + {custom_bp_t::BANNER_PAYLOAD_READ}, + {Tag::HELLO, bp_type_t::WRITE}, + {Tag::HELLO, bp_type_t::READ}, + {Tag::AUTH_REQUEST, bp_type_t::WRITE}, + {Tag::AUTH_DONE, bp_type_t::READ}, + {Tag::AUTH_SIGNATURE, bp_type_t::WRITE}, + {Tag::AUTH_SIGNATURE, bp_type_t::READ}, + {Tag::CLIENT_IDENT, bp_type_t::WRITE}, + }, [&test] (auto& failure_cases) { + return seastar::do_for_each(failure_cases, [&test] (auto bp) { + TestInterceptor interceptor; + // block connector + interceptor.make_block(bp); + return test.run_suite( + fmt::format("test_v2_racing_connect_acceptor_win -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 0); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }); + }); + }); + }); +} + +seastar::future<> +test_v2_racing_connect_reconnect_lose(FailoverTest& test) { + TestInterceptor interceptor; + interceptor.make_fault({Tag::SERVER_IDENT, bp_type_t::READ}); + interceptor.make_block({Tag::CLIENT_IDENT, bp_type_t::WRITE}, 2); + return test.run_suite("test_v2_racing_connect_reconnect_lose", + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 2, 0, 0); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_racing_connect_reconnect_win(FailoverTest& test) { + TestInterceptor interceptor; + interceptor.make_fault({Tag::SERVER_IDENT, bp_type_t::READ}); + interceptor.make_block({Tag::SESSION_RECONNECT, bp_type_t::READ}); + return test.run_suite("test_v2_racing_connect_reconnect_win", + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 2, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stale_connect(FailoverTest& test) { + auto bp = Breakpoint{Tag::SERVER_IDENT, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_stall(bp); + return test.run_suite( + fmt::format("test_v2_stale_connect -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 0); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stale_reconnect(FailoverTest& test) { + auto bp = Breakpoint{Tag::SESSION_RECONNECT_OK, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_fault({Tag::MESSAGE, bp_type_t::WRITE}); + interceptor.make_stall(bp); + return test.run_suite( + fmt::format("test_v2_stale_reconnect -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(2, 1, 1, 1); + results[0].assert_accept(0, 0, 0, 1); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stale_accept(FailoverTest& test) { + auto bp = Breakpoint{Tag::CLIENT_IDENT, bp_type_t::READ}; + TestInterceptor interceptor; + interceptor.make_stall(bp); + return test.run_suite( + fmt::format("test_v2_stale_accept -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_established(); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stale_establishing(FailoverTest& test) { + auto bp = Breakpoint{Tag::SERVER_IDENT, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_stall(bp); + return test.run_suite( + fmt::format("test_v2_stale_establishing -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_replaced(1); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0); + results[1].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stale_reaccept(FailoverTest& test) { + auto bp = Breakpoint{Tag::SESSION_RECONNECT_OK, bp_type_t::WRITE}; + TestInterceptor interceptor; + interceptor.make_fault({Tag::MESSAGE, bp_type_t::READ}); + interceptor.make_stall(bp); + return test.run_suite( + fmt::format("test_v2_stale_reaccept -- {}", bp), + interceptor, + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + return test.peer_send_me(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&suite] { + return suite.wait_blocked(); + }).then([] { + logger().info("[Test] block the broken REPLACING for 210ms..."); + return seastar::sleep(210ms); + }).then([&suite] { + suite.unblock(); + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 3); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 0, 1, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::replaced); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 0); + results[2].assert_reset(0, 0); + ceph_assert(results[2].server_reconnect_attempts >= 1); + }); + }); +} + +seastar::future<> +test_v2_lossy_client(FailoverTest& test) { + return test.run_suite( + "test_v2_lossy_client", + TestInterceptor(), + policy_t::lossy_client, + policy_t::stateless_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return suite.connect_peer(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 1 --"); + logger().info("[Test] client markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 2 --"); + logger().info("[Test] server markdown..."); + return test.markdown_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(1, 0); + }).then([&suite] { + logger().info("-- 3 --"); + logger().info("[Test] client reconnect..."); + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(1, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(1, 1, 0, 1); + results[2].assert_accept(0, 0, 0, 0); + results[2].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_stateless_server(FailoverTest& test) { + return test.run_suite( + "test_v2_stateless_server", + TestInterceptor(), + policy_t::stateless_server, + policy_t::lossy_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return test.peer_connect_me(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 1 --"); + logger().info("[Test] client markdown..."); + return test.markdown_peer(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(1, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 2 --"); + logger().info("[Test] server markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(1, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 3 --"); + logger().info("[Test] client reconnect..."); + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(1, 0); + results[1].assert_state_at(conn_state_t::closed); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 1); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 1, 0, 1); + results[2].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossless_client(FailoverTest& test) { + return test.run_suite( + "test_v2_lossless_client", + TestInterceptor(), + policy_t::lossless_client, + policy_t::stateful_server, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return suite.connect_peer(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 1 --"); + logger().info("[Test] client markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 2 --"); + logger().info("[Test] server markdown..."); + return test.markdown_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(2, 2, 1, 2); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 1); + }).then([&suite] { + logger().info("-- 3 --"); + logger().info("[Test] client reconnect..."); + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(2, 2, 1, 2); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 1); + }); + }); +} + +seastar::future<> +test_v2_stateful_server(FailoverTest& test) { + return test.run_suite( + "test_v2_stateful_server", + TestInterceptor(), + policy_t::stateful_server, + policy_t::lossless_client, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return test.peer_connect_me(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 1 --"); + logger().info("[Test] client markdown..."); + return test.markdown_peer(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 2 --"); + logger().info("[Test] server markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 1, 1, 1); + results[2].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 3 --"); + logger().info("[Test] client reconnect..."); + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 1, 1, 1); + results[2].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_peer_reuse_connector(FailoverTest& test) { + return test.run_suite( + "test_v2_peer_reuse_connector", + TestInterceptor(), + policy_t::lossless_peer_reuse, + policy_t::lossless_peer_reuse, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return suite.connect_peer(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 1 --"); + logger().info("[Test] connector markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 2 --"); + logger().info("[Test] acceptor markdown..."); + return test.markdown_peer(); + }).then([&suite] { + ceph_assert(suite.is_standby()); + logger().info("-- 3 --"); + logger().info("[Test] connector reconnect..."); + return suite.connect_peer(); + }).then([&suite] { + return suite.try_send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(2, 2, 1, 2); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 1); + }); + }); +} + +seastar::future<> +test_v2_peer_reuse_acceptor(FailoverTest& test) { + return test.run_suite( + "test_v2_peer_reuse_acceptor", + TestInterceptor(), + policy_t::lossless_peer_reuse, + policy_t::lossless_peer_reuse, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return test.peer_connect_me(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 1 --"); + logger().info("[Test] connector markdown..."); + return test.markdown_peer(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 2 --"); + logger().info("[Test] acceptor markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 3 --"); + logger().info("[Test] connector reconnect..."); + return test.peer_connect_me(); + }).then([&test] { + return test.try_peer_send_me(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 1); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 1, 1, 1); + results[2].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_lossless_peer_connector(FailoverTest& test) { + return test.run_suite( + "test_v2_lossless_peer_connector", + TestInterceptor(), + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&suite] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return suite.connect_peer(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 1 --"); + logger().info("[Test] connector markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.connect_peer(); + }).then([&suite] { + return suite.send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(1, 1, 0, 1); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 2 --"); + logger().info("[Test] acceptor markdown..."); + return test.markdown_peer(); + }).then([&suite] { + ceph_assert(suite.is_standby()); + logger().info("-- 3 --"); + logger().info("[Test] connector reconnect..."); + return suite.connect_peer(); + }).then([&suite] { + return suite.try_send_peer(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(1, 1, 0, 1); + results[0].assert_accept(0, 0, 0, 0); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::established); + results[1].assert_connect(2, 2, 1, 2); + results[1].assert_accept(0, 0, 0, 0); + results[1].assert_reset(0, 1); + }); + }); +} + +seastar::future<> +test_v2_lossless_peer_acceptor(FailoverTest& test) { + return test.run_suite( + "test_v2_lossless_peer_acceptor", + TestInterceptor(), + policy_t::lossless_peer, + policy_t::lossless_peer, + [&test] (FailoverSuite& suite) { + return seastar::futurize_invoke([&test] { + logger().info("-- 0 --"); + logger().info("[Test] setup connection..."); + return test.peer_connect_me(); + }).then([&test] { + return test.send_bidirectional(); + }).then([&suite] { + return suite.wait_results(1); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 1); + results[0].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 1 --"); + logger().info("[Test] connector markdown..."); + return test.markdown_peer(); + }).then([&test] { + return test.peer_connect_me(); + }).then([&test] { + return test.peer_send_me(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::established); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }).then([&suite] { + logger().info("-- 2 --"); + logger().info("[Test] acceptor markdown..."); + return suite.markdown(); + }).then([&suite] { + return suite.wait_results(2); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + }).then([&test] { + logger().info("-- 3 --"); + logger().info("[Test] connector reconnect..."); + return test.peer_connect_me(); + }).then([&test] { + return test.try_peer_send_me(); + }).then([&suite] { + return suite.wait_results(3); + }).then([] (ConnResults& results) { + results[0].assert_state_at(conn_state_t::closed); + results[0].assert_connect(0, 0, 0, 0); + results[0].assert_accept(1, 1, 0, 2); + results[0].assert_reset(0, 0); + results[1].assert_state_at(conn_state_t::replaced); + results[1].assert_connect(0, 0, 0, 0); + results[1].assert_accept(1, 1, 0, 0); + results[1].assert_reset(0, 0); + results[2].assert_state_at(conn_state_t::established); + results[2].assert_connect(0, 0, 0, 0); + results[2].assert_accept(1, 1, 1, 1); + results[2].assert_reset(0, 0); + }); + }); +} + +seastar::future<> +test_v2_protocol(entity_addr_t test_addr, + entity_addr_t cmd_peer_addr, + entity_addr_t test_peer_addr, + bool test_peer_islocal, + bool peer_wins) { + ceph_assert_always(test_addr.is_msgr2()); + ceph_assert_always(cmd_peer_addr.is_msgr2()); + ceph_assert_always(test_peer_addr.is_msgr2()); + + if (test_peer_islocal) { + // initiate crimson test peer locally + logger().info("test_v2_protocol: start local TestPeer at {}...", cmd_peer_addr); + return FailoverTestPeer::create(cmd_peer_addr, test_peer_addr + ).then([test_addr, cmd_peer_addr, test_peer_addr, peer_wins](auto peer) { + return test_v2_protocol( + test_addr, + cmd_peer_addr, + test_peer_addr, + false, + peer_wins + ).then([peer = std::move(peer)] () mutable { + return peer->wait().then([peer = std::move(peer)] {}); + }); + }).handle_exception([] (auto eptr) { + logger().error("FailoverTestPeer failed: got exception {}", eptr); + throw; + }); + } + + return FailoverTest::create(test_addr, cmd_peer_addr, test_peer_addr + ).then([peer_wins](auto test) { + return seastar::futurize_invoke([test] { + return test_v2_lossy_early_connect_fault(*test); + }).then([test] { + return test_v2_lossy_connect_fault(*test); + }).then([test] { + return test_v2_lossy_connected_fault(*test); + }).then([test] { + return test_v2_lossy_early_accept_fault(*test); + }).then([test] { + return test_v2_lossy_accept_fault(*test); + }).then([test] { + return test_v2_lossy_establishing_fault(*test); + }).then([test] { + return test_v2_lossy_accepted_fault(*test); + }).then([test] { + return test_v2_lossless_connect_fault(*test); + }).then([test] { + return test_v2_lossless_connected_fault(*test); + }).then([test] { + return test_v2_lossless_connected_fault2(*test); + }).then([test] { + return test_v2_lossless_reconnect_fault(*test); + }).then([test] { + return test_v2_lossless_accept_fault(*test); + }).then([test] { + return test_v2_lossless_establishing_fault(*test); + }).then([test] { + return test_v2_lossless_accepted_fault(*test); + }).then([test] { + return test_v2_lossless_reaccept_fault(*test); + }).then([test] { + return test_v2_peer_connect_fault(*test); + }).then([test] { + return test_v2_peer_accept_fault(*test); + }).then([test] { + return test_v2_peer_establishing_fault(*test); + }).then([test] { + return test_v2_peer_connected_fault_reconnect(*test); + }).then([test] { + return test_v2_peer_connected_fault_reaccept(*test); + }).then([test] { + return check_peer_wins(*test); + }).then([test, peer_wins](bool ret_peer_wins) { + ceph_assert(peer_wins == ret_peer_wins); + if (ret_peer_wins) { + return seastar::futurize_invoke([test] { + return test_v2_racing_connect_acceptor_win(*test); + }).then([test] { + return test_v2_racing_reconnect_acceptor_win(*test); + }); + } else { + return seastar::futurize_invoke([test] { + return test_v2_racing_connect_acceptor_lose(*test); + }).then([test] { + return test_v2_racing_reconnect_acceptor_lose(*test); + }); + } + }).then([test] { + return test_v2_racing_connect_reconnect_win(*test); + }).then([test] { + return test_v2_racing_connect_reconnect_lose(*test); + }).then([test] { + return test_v2_stale_connect(*test); + }).then([test] { + return test_v2_stale_reconnect(*test); + }).then([test] { + return test_v2_stale_accept(*test); + }).then([test] { + return test_v2_stale_establishing(*test); + }).then([test] { + return test_v2_stale_reaccept(*test); + }).then([test] { + return test_v2_lossy_client(*test); + }).then([test] { + return test_v2_stateless_server(*test); + }).then([test] { + return test_v2_lossless_client(*test); + }).then([test] { + return test_v2_stateful_server(*test); + }).then([test] { + return test_v2_peer_reuse_connector(*test); + }).then([test] { + return test_v2_peer_reuse_acceptor(*test); + }).then([test] { + return test_v2_lossless_peer_connector(*test); + }).then([test] { + return test_v2_lossless_peer_acceptor(*test); + }).then([test] { + return test->shutdown().then([test] {}); + }); + }).handle_exception([] (auto eptr) { + logger().error("FailoverTest failed: got exception {}", eptr); + throw; + }); +} + +} + +seastar::future<int> do_test(seastar::app_template& app) +{ + std::vector<const char*> args; + std::string cluster; + std::string conf_file_list; + auto init_params = ceph_argparse_early_args(args, + CEPH_ENTITY_TYPE_CLIENT, + &cluster, + &conf_file_list); + return crimson::common::sharded_conf().start( + init_params.name, cluster + ).then([] { + return local_conf().start(); + }).then([conf_file_list] { + return local_conf().parse_config_files(conf_file_list); + }).then([&app] { + auto&& config = app.configuration(); + verbose = config["verbose"].as<bool>(); + auto rounds = config["rounds"].as<unsigned>(); + auto keepalive_ratio = config["keepalive-ratio"].as<double>(); + auto testpeer_islocal = config["testpeer-islocal"].as<bool>(); + + entity_addr_t test_addr; + ceph_assert(test_addr.parse( + config["test-addr"].as<std::string>().c_str(), nullptr)); + test_addr.set_nonce(TEST_NONCE); + + entity_addr_t cmd_peer_addr; + ceph_assert(cmd_peer_addr.parse( + config["testpeer-addr"].as<std::string>().c_str(), nullptr)); + cmd_peer_addr.set_nonce(CMD_SRV_NONCE); + + entity_addr_t test_peer_addr = get_test_peer_addr(cmd_peer_addr); + bool peer_wins = (test_addr > test_peer_addr); + + logger().info("test configuration: verbose={}, rounds={}, keepalive_ratio={}, " + "test_addr={}, cmd_peer_addr={}, test_peer_addr={}, " + "testpeer_islocal={}, peer_wins={}, smp={}", + verbose, rounds, keepalive_ratio, + test_addr, cmd_peer_addr, test_peer_addr, + testpeer_islocal, peer_wins, + seastar::smp::count); + return test_echo(rounds, keepalive_ratio + ).then([] { + return test_preemptive_shutdown(); + }).then([test_addr, cmd_peer_addr, test_peer_addr, testpeer_islocal, peer_wins] { + return test_v2_protocol( + test_addr, + cmd_peer_addr, + test_peer_addr, + testpeer_islocal, + peer_wins); + }).then([] { + logger().info("All tests succeeded"); + // Seastar has bugs to have events undispatched during shutdown, + // which will result in memory leak and thus fail LeakSanitizer. + return seastar::sleep(100ms); + }); + }).then([] { + return crimson::common::sharded_conf().stop(); + }).then([] { + return 0; + }).handle_exception([] (auto eptr) { + logger().error("Test failed: got exception {}", eptr); + return 1; + }); +} + +int main(int argc, char** argv) +{ + seastar::app_template app; + app.add_options() + ("verbose,v", bpo::value<bool>()->default_value(false), + "chatty if true") + ("rounds", bpo::value<unsigned>()->default_value(512), + "number of pingpong rounds") + ("keepalive-ratio", bpo::value<double>()->default_value(0.1), + "ratio of keepalive in ping messages") + ("test-addr", bpo::value<std::string>()->default_value("v2:127.0.0.1:9014"), + "address of v2 failover tests") + ("testpeer-addr", bpo::value<std::string>()->default_value("v2:127.0.0.1:9012"), + "addresses of v2 failover testpeer" + " (This is CmdSrv address, and TestPeer address is at port+=1)") + ("testpeer-islocal", bpo::value<bool>()->default_value(true), + "create a local crimson testpeer, or connect to a remote testpeer"); + return app.run(argc, argv, [&app] { + // This test normally succeeds within 60 seconds, so kill it after 300 + // seconds in case it is blocked forever due to unaddressed bugs. + return seastar::with_timeout(seastar::lowres_clock::now() + 300s, do_test(app)) + .handle_exception_type([](seastar::timed_out_error&) { + logger().error("test_messenger timeout after 300s, abort! " + "Consider to extend the period if the test is still running."); + // use the retcode of timeout(1) + return 124; + }); + }); +} |