ceph/src/test/crimson/test_messenger_thrash.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include <map>
   5 #include <random>
   6 #include <fmt/format.h>
   7 #include <fmt/ostream.h>
   8 #include <seastar/core/app-template.hh>
   9 #include <seastar/core/do_with.hh>
  10 #include <seastar/core/future-util.hh>
  11 #include <seastar/core/reactor.hh>
  12 #include <seastar/core/sleep.hh>
  13 #include <seastar/core/with_timeout.hh>
  14
  15 #include "common/ceph_argparse.h"
  16 #include "messages/MPing.h"
  17 #include "messages/MCommand.h"
  18 #include "crimson/auth/DummyAuth.h"
  19 #include "crimson/common/log.h"
  20 #include "crimson/net/Connection.h"
  21 #include "crimson/net/Dispatcher.h"
  22 #include "crimson/net/Messenger.h"
  23
  24 using namespace std::chrono_literals;
  25 namespace bpo = boost::program_options;
  26 using crimson::common::local_conf;
  27 using payload_seq_t = uint64_t;
  28
  29 struct Payload {
  30   enum Who : uint8_t {
  31     PING = 0,
  32     PONG = 1,
  33   };
  34   uint8_t who = 0;
  35   payload_seq_t seq = 0;
  36   bufferlist data;
  37
  38   Payload(Who who, uint64_t seq, const bufferlist& data)
  39     : who(who), seq(seq), data(data)
  40   {}
  41   Payload() = default;
  42   DENC(Payload, v, p) {
  43     DENC_START(1, 1, p);
  44     denc(v.who, p);
  45     denc(v.seq, p);
  46     denc(v.data, p);
  47     DENC_FINISH(p);
  48   }
  49 };
  50 WRITE_CLASS_DENC(Payload)
  51
  52 template<>
  53 struct fmt::formatter<Payload> : fmt::formatter<std::string_view> {
  54   template <typename FormatContext>
  55   auto format(const Payload& pl, FormatContext& ctx) const {
  56     return fmt::format_to(ctx.out(), "reply={} i={}", pl.who, pl.seq);
  57   }
  58 };
  59
  60 namespace {
  61
  62 seastar::logger& logger() {
  63   return crimson::get_logger(ceph_subsys_test);
  64 }
  65
  66 std::random_device rd;
  67 std::default_random_engine rng{rd()};
  68 std::uniform_int_distribution<> prob(0,99);
  69 bool verbose = false;
  70
  71 entity_addr_t get_server_addr() {
  72   static int port = 16800;
  73   ++port;
  74   entity_addr_t saddr;
  75   saddr.parse("127.0.0.1", nullptr);
  76   saddr.set_port(port);
  77   return saddr;
  78 }
  79
  80 uint64_t get_nonce() {
  81   static uint64_t nonce = 1;
  82   ++nonce;
  83   return nonce;
  84 }
  85
  86 struct thrash_params_t {
  87   std::size_t servers;
  88   std::size_t clients;
  89   std::size_t connections;
  90   std::size_t random_op;
  91 };
  92
  93 class SyntheticWorkload;
  94
  95 class SyntheticDispatcher final
  96     : public crimson::net::Dispatcher {
  97   public:
  98   std::map<crimson::net::Connection*, std::deque<payload_seq_t> > conn_sent;
  99   std::map<payload_seq_t, bufferlist> sent;
 100   unsigned index;
 101   SyntheticWorkload *workload;
 102
 103   SyntheticDispatcher(bool s, SyntheticWorkload *wl):
 104     index(0), workload(wl) {
 105   }
 106
 107   std::optional<seastar::future<>> ms_dispatch(crimson::net::ConnectionRef con,
 108                                                MessageRef m) {
 109     if (verbose) {
 110       logger().warn("{}: con = {}", __func__, *con);
 111     }
 112     // MSG_COMMAND is used to disorganize regular message flow
 113     if (m->get_type() == MSG_COMMAND) {
 114       return seastar::now();
 115     }
 116
 117     Payload pl;
 118     auto p = m->get_data().cbegin();
 119     decode(pl, p);
 120     if (pl.who == Payload::PING) {
 121       logger().info(" {} conn= {} {}", __func__, *con, pl);
 122       return reply_message(m, con, pl);
 123     } else {
 124       ceph_assert(pl.who == Payload::PONG);
 125       if (sent.count(pl.seq)) {
 126         logger().info(" {} conn= {} {}", __func__, *con, pl);
 127         ceph_assert(conn_sent[&*con].front() == pl.seq);
 128         ceph_assert(pl.data.contents_equal(sent[pl.seq]));
 129         conn_sent[&*con].pop_front();
 130         sent.erase(pl.seq);
 131       }
 132
 133       return seastar::now();
 134     }
 135   }
 136
 137   void ms_handle_accept(crimson::net::ConnectionRef conn) {
 138     logger().info("{} - Connection:{}", __func__, *conn);
 139   }
 140
 141   void ms_handle_connect(crimson::net::ConnectionRef conn) {
 142     logger().info("{} - Connection:{}", __func__, *conn);
 143   }
 144
 145   void ms_handle_reset(crimson::net::ConnectionRef con, bool is_replace);
 146
 147   void ms_handle_remote_reset(crimson::net::ConnectionRef con) {
 148     clear_pending(con);
 149   }
 150
 151   std::optional<seastar::future<>> reply_message(
 152       const MessageRef m,
 153       crimson::net::ConnectionRef con,
 154       Payload& pl) {
 155     pl.who = Payload::PONG;
 156     bufferlist bl;
 157     encode(pl, bl);
 158     auto rm = crimson::make_message<MPing>();
 159     rm->set_data(bl);
 160     if (verbose) {
 161       logger().info("{} conn= {} reply i= {}",
 162         __func__, *con, pl.seq);
 163     }
 164     return con->send(std::move(rm));
 165   }
 166
 167   seastar::future<> send_message_wrap(crimson::net::ConnectionRef con,
 168                                       const bufferlist& data) {
 169     auto m = crimson::make_message<MPing>();
 170     Payload pl{Payload::PING, index++, data};
 171     bufferlist bl;
 172     encode(pl, bl);
 173     m->set_data(bl);
 174     sent[pl.seq] = pl.data;
 175     conn_sent[&*con].push_back(pl.seq);
 176     logger().info("{} conn= {} send i= {}",
 177       __func__, *con, pl.seq);
 178
 179     return con->send(std::move(m));
 180   }
 181
 182   uint64_t get_num_pending_msgs() {
 183     return sent.size();
 184   }
 185
 186   void clear_pending(crimson::net::ConnectionRef con) {
 187     for (std::deque<uint64_t>::iterator it = conn_sent[&*con].begin();
 188          it != conn_sent[&*con].end(); ++it)
 189       sent.erase(*it);
 190     conn_sent.erase(&*con);
 191   }
 192
 193   void print() {
 194     for (auto && [connptr, list] : conn_sent) {
 195       if (!list.empty()) {
 196         logger().info("{} {} wait {}", __func__,
 197                       (void*)connptr, list.size());
 198       }
 199     }
 200   }
 201 };
 202
 203 class SyntheticWorkload {
 204   // messengers must be freed after its connections
 205   std::set<crimson::net::MessengerRef> available_servers;
 206   std::set<crimson::net::MessengerRef> available_clients;
 207
 208   crimson::net::SocketPolicy server_policy;
 209   crimson::net::SocketPolicy client_policy;
 210   std::map<crimson::net::ConnectionRef,
 211     std::pair<crimson::net::MessengerRef,
 212     crimson::net::MessengerRef>> available_connections;
 213   SyntheticDispatcher dispatcher;
 214   std::vector<bufferlist> rand_data;
 215   crimson::auth::DummyAuthClientServer dummy_auth;
 216
 217   seastar::future<crimson::net::ConnectionRef> get_random_connection() {
 218     return seastar::do_until(
 219       [this] { return dispatcher.get_num_pending_msgs() <= max_in_flight; },
 220       [] { return seastar::sleep(100ms); }
 221     ).then([this] {
 222       boost::uniform_int<> choose(0, available_connections.size() - 1);
 223       int index = choose(rng);
 224       std::map<crimson::net::ConnectionRef,
 225         std::pair<crimson::net::MessengerRef, crimson::net::MessengerRef>>::iterator i
 226         = available_connections.begin();
 227       for (; index > 0; --index, ++i) ;
 228       return seastar::make_ready_future<crimson::net::ConnectionRef>(i->first);
 229    });
 230   }
 231
 232  public:
 233    const unsigned min_connections = 10;
 234    const unsigned max_in_flight = 64;
 235    const unsigned max_connections = 128;
 236    const unsigned max_message_len = 1024 * 1024 * 4;
 237    const uint64_t  servers, clients;
 238
 239    SyntheticWorkload(int servers, int clients, int random_num,
 240                      crimson::net::SocketPolicy srv_policy,
 241                      crimson::net::SocketPolicy cli_policy)
 242      : server_policy(srv_policy),
 243        client_policy(cli_policy),
 244        dispatcher(false, this),
 245        servers(servers),
 246        clients(clients) {
 247
 248      for (int i = 0; i < random_num; i++) {
 249        bufferlist bl;
 250        boost::uniform_int<> u(32, max_message_len);
 251        uint64_t value_len = u(rng);
 252        bufferptr bp(value_len);
 253        bp.zero();
 254        for (uint64_t j = 0; j < value_len-sizeof(i); ) {
 255          memcpy(bp.c_str()+j, &i, sizeof(i));
 256          j += 4096;
 257        }
 258
 259        bl.append(bp);
 260        rand_data.push_back(bl);
 261      }
 262    }
 263
 264
 265    bool can_create_connection() {
 266      return available_connections.size() < max_connections;
 267    }
 268
 269    seastar::future<> maybe_generate_connection() {
 270      if (!can_create_connection()) {
 271        return seastar::now();
 272      }
 273      crimson::net::MessengerRef server, client;
 274      {
 275        boost::uniform_int<> choose(0, available_servers.size() - 1);
 276        int index = choose(rng);
 277        std::set<crimson::net::MessengerRef>::iterator i
 278          = available_servers.begin();
 279        for (; index > 0; --index, ++i) ;
 280        server = *i;
 281      }
 282      {
 283        boost::uniform_int<> choose(0, available_clients.size() - 1);
 284        int index = choose(rng);
 285        std::set<crimson::net::MessengerRef>::iterator i
 286          = available_clients.begin();
 287        for (; index > 0; --index, ++i) ;
 288        client = *i;
 289      }
 290
 291
 292      std::pair<crimson::net::MessengerRef, crimson::net::MessengerRef>
 293        connected_pair;
 294      {
 295        crimson::net::ConnectionRef conn = client->connect(
 296                                             server->get_myaddr(),
 297                                             entity_name_t::TYPE_OSD);
 298        connected_pair = std::make_pair(client, server);
 299        available_connections[conn] = connected_pair;
 300      }
 301      return seastar::now();
 302    }
 303
 304    seastar::future<> random_op (const uint64_t& iter) {
 305      return seastar::do_with(iter, [this] (uint64_t& iter) {
 306        return seastar::do_until(
 307          [&] { return iter == 0; },
 308          [&, this]
 309        {
 310          if (!(iter % 10)) {
 311            logger().info("{} Op {} : ", __func__ ,iter);
 312            print_internal_state();
 313          }
 314          --iter;
 315          int val = prob(rng);
 316          if(val > 90) {
 317            return maybe_generate_connection();
 318          } else if (val > 80) {
 319            return drop_connection();
 320          } else if (val > 10) {
 321            return send_message();
 322          } else {
 323            return seastar::sleep(
 324              std::chrono::milliseconds(rand() % 1000 + 500));
 325          }
 326        });
 327      });
 328    }
 329
 330    seastar::future<> generate_connections (const uint64_t& iter) {
 331      return seastar::do_with(iter, [this] (uint64_t& iter) {
 332        return seastar::do_until(
 333          [&] { return iter == 0; },
 334          [&, this]
 335        {
 336          --iter;
 337          if (!(connections_count() % 10)) {
 338           logger().info("seeding connection {}",
 339                         connections_count());
 340          }
 341          return maybe_generate_connection();
 342        });
 343      });
 344    }
 345
 346    seastar::future<> init_server(const entity_name_t& name,
 347                           const std::string& lname,
 348                           const uint64_t nonce,
 349                           const entity_addr_t& addr) {
 350      crimson::net::MessengerRef msgr =
 351        crimson::net::Messenger::create(name, lname, nonce);
 352      msgr->set_default_policy(server_policy);
 353      msgr->set_auth_client(&dummy_auth);
 354      msgr->set_auth_server(&dummy_auth);
 355      available_servers.insert(msgr);
 356      return msgr->bind(entity_addrvec_t{addr}).safe_then(
 357          [this, msgr] {
 358        return msgr->start({&dispatcher});
 359      }, crimson::net::Messenger::bind_ertr::all_same_way(
 360          [addr] (const std::error_code& e) {
 361        logger().error("{} test_messenger_thrash(): "
 362                       "there is another instance running at {}",
 363                        __func__, addr);
 364        ceph_abort();
 365      }));
 366    }
 367
 368    seastar::future<> init_client(const entity_name_t& name,
 369                           const std::string& lname,
 370                           const uint64_t nonce) {
 371      crimson::net::MessengerRef msgr =
 372        crimson::net::Messenger::create(name, lname, nonce);
 373      msgr->set_default_policy(client_policy);
 374      msgr->set_auth_client(&dummy_auth);
 375      msgr->set_auth_server(&dummy_auth);
 376      available_clients.insert(msgr);
 377      return msgr->start({&dispatcher});
 378    }
 379
 380    seastar::future<> send_message() {
 381      return get_random_connection()
 382      .then([this] (crimson::net::ConnectionRef conn) {
 383        boost::uniform_int<> true_false(0, 99);
 384        int val = true_false(rng);
 385        if (val >= 95) {
 386          uuid_d uuid;
 387          uuid.generate_random();
 388          auto m = crimson::make_message<MCommand>(uuid);
 389          std::vector<std::string> cmds;
 390          cmds.push_back("command");
 391          m->cmd = cmds;
 392          m->set_priority(200);
 393          return conn->send(std::move(m));
 394        } else {
 395          boost::uniform_int<> u(0, rand_data.size()-1);
 396          return dispatcher.send_message_wrap(conn, rand_data[u(rng)]);
 397        }
 398      });
 399    }
 400
 401    seastar::future<> drop_connection() {
 402      if (available_connections.size() < min_connections) {
 403        return seastar::now();
 404      }
 405
 406      return get_random_connection()
 407      .then([this] (crimson::net::ConnectionRef conn) {
 408        dispatcher.clear_pending(conn);
 409        conn->mark_down();
 410        if (!client_policy.server &&
 411            client_policy.standby) {
 412          // it's a lossless policy, so we need to mark down each side
 413          std::pair<crimson::net::MessengerRef, crimson::net::MessengerRef> &p =
 414            available_connections[conn];
 415          if (!p.first->get_default_policy().server &&
 416              !p.second->get_default_policy().server) {
 417              //verify that equal-to operator applies here
 418            ceph_assert(p.first->owns_connection(*conn));
 419            crimson::net::ConnectionRef peer = p.second->connect(
 420              p.first->get_myaddr(), p.first->get_mytype());
 421            peer->mark_down();
 422            dispatcher.clear_pending(peer);
 423            available_connections.erase(peer);
 424          }
 425        }
 426        ceph_assert(available_connections.erase(conn) == 1U);
 427        return seastar::now();
 428      });
 429    }
 430
 431    void print_internal_state(bool detail=false) {
 432      logger().info("available_connections: {} inflight messages: {}",
 433        available_connections.size(),
 434        dispatcher.get_num_pending_msgs());
 435      if (detail && !available_connections.empty()) {
 436        dispatcher.print();
 437      }
 438    }
 439
 440    seastar::future<> wait_for_done() {
 441      int i = 0;
 442      return seastar::do_until(
 443        [this] { return !dispatcher.get_num_pending_msgs(); },
 444        [this, &i]
 445      {
 446        if (i++ % 50 == 0){
 447          print_internal_state(true);
 448        }
 449        return seastar::sleep(100ms);
 450      }).then([this] {
 451        return seastar::do_for_each(available_servers, [] (auto server) {
 452          if (verbose) {
 453            logger().info("server {} shutdown" , server->get_myaddrs());
 454          }
 455          server->stop();
 456          return server->shutdown();
 457        });
 458      }).then([this] {
 459        return seastar::do_for_each(available_clients, [] (auto client) {
 460          if (verbose) {
 461            logger().info("client {} shutdown" , client->get_myaddrs());
 462          }
 463          client->stop();
 464          return client->shutdown();
 465        });
 466      });
 467    }
 468
 469    void handle_reset(crimson::net::ConnectionRef con) {
 470      available_connections.erase(con);
 471    }
 472
 473    uint64_t servers_count() {
 474      return available_servers.size();
 475    }
 476
 477    uint64_t clients_count() {
 478      return available_clients.size();
 479    }
 480
 481    uint64_t connections_count() {
 482      return available_connections.size();
 483    }
 484 };
 485
 486 void SyntheticDispatcher::ms_handle_reset(crimson::net::ConnectionRef con,
 487                                           bool is_replace) {
 488   workload->handle_reset(con);
 489   clear_pending(con);
 490 }
 491
 492 seastar::future<> reset_conf() {
 493   return seastar::when_all_succeed(
 494     local_conf().set_val("ms_inject_socket_failures", "0"),
 495     local_conf().set_val("ms_inject_internal_delays", "0"),
 496     local_conf().set_val("ms_inject_delay_probability", "0"),
 497     local_conf().set_val("ms_inject_delay_max", "0")
 498   ).then_unpack([] {
 499     return seastar::now();
 500   });
 501 }
 502
 503 // Testing Crimson messenger (with msgr-v2 protocol) robustness against
 504 // network delays and failures. The test includes stress tests and
 505 // socket level delays/failures injection tests, letting time
 506 // and randomness achieve the best test coverage.
 507
 508 // Test Parameters:
 509 // Clients: 8             (stateful)
 510 // Servers: 32            (lossless)
 511 // Connections: 100       (Generated between random clients/server)
 512 // Random Operations: 120 (Generate/Drop Connection, Send Message, Sleep)
 513 seastar::future<> test_stress(thrash_params_t tp)
 514 {
 515
 516   logger().info("test_stress():");
 517
 518   SyntheticWorkload test_msg(tp.servers, tp.clients, 100,
 519                              crimson::net::SocketPolicy::stateful_server(0),
 520                              crimson::net::SocketPolicy::lossless_client(0));
 521
 522   return seastar::do_with(test_msg, [tp]
 523     (SyntheticWorkload& test_msg) {
 524       return seastar::do_until([&test_msg] {
 525         return test_msg.servers_count() == test_msg.servers; },
 526       [&test_msg] {
 527         entity_addr_t bind_addr = get_server_addr();
 528         bind_addr.set_type(entity_addr_t::TYPE_MSGR2);
 529         uint64_t server_num = get_nonce();
 530         return test_msg.init_server(entity_name_t::OSD(server_num),
 531                              "server", server_num , bind_addr);
 532       }).then([&test_msg] {
 533       return seastar::do_until([&test_msg] {
 534         return test_msg.clients_count() == test_msg.clients; },
 535       [&test_msg] {
 536         return test_msg.init_client(entity_name_t::CLIENT(-1),
 537                              "client", get_nonce());
 538         });
 539       }).then([&test_msg, tp] {
 540         return test_msg.generate_connections(tp.connections);
 541       }).then([&test_msg, tp] {
 542         return test_msg.random_op(tp.random_op);
 543       }).then([&test_msg] {
 544         return test_msg.wait_for_done();
 545       }).then([] {
 546         logger().info("test_stress() DONE");
 547       }).handle_exception([] (auto eptr) {
 548         logger().error(
 549           "test_stress() failed: got exception {}",
 550           eptr);
 551         throw;
 552       });
 553     });
 554 }
 555
 556 // Test Parameters:
 557 // Clients: 8              (statefull)
 558 // Servers: 32             (loseless)
 559 // Connections: 100        (Generated between random clients/server)
 560 // Random Operations: 120 (Generate/Drop Connection, Send Message, Sleep)
 561 seastar::future<> test_injection(thrash_params_t tp)
 562 {
 563
 564   logger().info("test_injection():");
 565
 566   SyntheticWorkload test_msg(tp.servers, tp.clients, 100,
 567                              crimson::net::SocketPolicy::stateful_server(0),
 568                              crimson::net::SocketPolicy::lossless_client(0));
 569
 570   return seastar::do_with(test_msg, [tp]
 571     (SyntheticWorkload& test_msg) {
 572       return seastar::do_until([&test_msg] {
 573         return test_msg.servers_count() == test_msg.servers; },
 574       [&test_msg] {
 575         entity_addr_t bind_addr = get_server_addr();
 576         bind_addr.set_type(entity_addr_t::TYPE_MSGR2);
 577         uint64_t server_num = get_nonce();
 578         return test_msg.init_server(entity_name_t::OSD(server_num),
 579                              "server", server_num , bind_addr);
 580       }).then([&test_msg] {
 581       return seastar::do_until([&test_msg] {
 582         return test_msg.clients_count() == test_msg.clients; },
 583       [&test_msg] {
 584         return test_msg.init_client(entity_name_t::CLIENT(-1),
 585                              "client", get_nonce());
 586         });
 587       }).then([] {
 588         return seastar::when_all_succeed(
 589           local_conf().set_val("ms_inject_socket_failures", "30"),
 590           local_conf().set_val("ms_inject_internal_delays", "0.1"),
 591           local_conf().set_val("ms_inject_delay_probability", "1"),
 592           local_conf().set_val("ms_inject_delay_max", "5"));
 593       }).then_unpack([] {
 594         return seastar::now();
 595       }).then([&test_msg, tp] {
 596         return test_msg.generate_connections(tp.connections);
 597       }).then([&test_msg, tp] {
 598         return test_msg.random_op(tp.random_op);
 599       }).then([&test_msg] {
 600           return test_msg.wait_for_done();
 601       }).then([] {
 602         logger().info("test_inejction() DONE");
 603         return seastar::now();
 604       }).then([] {
 605         return reset_conf();
 606       }).handle_exception([] (auto eptr) {
 607         logger().error(
 608           "test_injection() failed: got exception {}",
 609           eptr);
 610         throw;
 611       });
 612     });
 613 }
 614
 615 }
 616
 617 seastar::future<int> do_test(seastar::app_template& app)
 618 {
 619   std::vector<const char*> args;
 620   std::string cluster;
 621   std::string conf_file_list;
 622   auto init_params = ceph_argparse_early_args(args,
 623                                               CEPH_ENTITY_TYPE_CLIENT,
 624                                               &cluster,
 625                                               &conf_file_list);
 626   return crimson::common::sharded_conf().start(init_params.name, cluster)
 627   .then([conf_file_list] {
 628     return local_conf().parse_config_files(conf_file_list);
 629   }).then([&app] {
 630     auto&& config = app.configuration();
 631     verbose = config["verbose"].as<bool>();
 632       return test_stress(thrash_params_t{8, 32, 50, 120})
 633     .then([] {
 634       return test_injection(thrash_params_t{16, 32, 50, 120});
 635     }).then([] {
 636       logger().info("All tests succeeded");
 637       // Seastar has bugs to have events undispatched during shutdown,
 638       // which will result in memory leak and thus fail LeakSanitizer.
 639       return seastar::sleep(100ms);
 640     });
 641   }).then([] {
 642     return crimson::common::sharded_conf().stop();
 643   }).then([] {
 644     return 0;
 645   }).handle_exception([] (auto eptr) {
 646     logger().error("Test failed: got exception {}", eptr);
 647     return 1;
 648   });
 649 }
 650
 651 int main(int argc, char** argv)
 652 {
 653   seastar::app_template app;
 654   app.add_options()
 655     ("verbose,v", bpo::value<bool>()->default_value(false),
 656      "chatty if true");
 657   return app.run(argc, argv, [&app] {
 658     return do_test(app);
 659   });
 660 }