summaryrefslogtreecommitdiffstats
path: root/src/test/crimson/test_alien_echo.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/test/crimson/test_alien_echo.cc')
-rw-r--r--src/test/crimson/test_alien_echo.cc281
1 files changed, 281 insertions, 0 deletions
diff --git a/src/test/crimson/test_alien_echo.cc b/src/test/crimson/test_alien_echo.cc
new file mode 100644
index 00000000..1dbe8133
--- /dev/null
+++ b/src/test/crimson/test_alien_echo.cc
@@ -0,0 +1,281 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+
+#include "auth/Auth.h"
+#include "messages/MPing.h"
+#include "crimson/net/Connection.h"
+#include "crimson/net/Dispatcher.h"
+#include "crimson/net/Messenger.h"
+#include "crimson/net/Config.h"
+#include "crimson/thread/Condition.h"
+#include "crimson/thread/Throttle.h"
+
+#include <seastar/core/alien.hh>
+#include <seastar/core/app-template.hh>
+#include <seastar/core/future-util.hh>
+#include <seastar/core/reactor.hh>
+
+
+enum class echo_role {
+ as_server,
+ as_client,
+};
+
+namespace seastar_pingpong {
+struct DummyAuthAuthorizer : public AuthAuthorizer {
+ DummyAuthAuthorizer()
+ : AuthAuthorizer(CEPH_AUTH_CEPHX)
+ {}
+ bool verify_reply(bufferlist::const_iterator&,
+ std::string *connection_secret) override {
+ return true;
+ }
+ bool add_challenge(CephContext*, const bufferlist&) override {
+ return true;
+ }
+};
+
+struct Server {
+ ceph::thread::Throttle byte_throttler;
+ ceph::net::Messenger& msgr;
+ struct ServerDispatcher : ceph::net::Dispatcher {
+ unsigned count = 0;
+ seastar::condition_variable on_reply;
+ seastar::future<> ms_dispatch(ceph::net::ConnectionRef c,
+ MessageRef m) override {
+ std::cout << "server got ping " << *m << std::endl;
+ // reply with a pong
+ return c->send(MessageRef{new MPing(), false}).then([this] {
+ ++count;
+ on_reply.signal();
+ });
+ }
+ seastar::future<ceph::net::msgr_tag_t, bufferlist>
+ ms_verify_authorizer(peer_type_t peer_type,
+ auth_proto_t protocol,
+ bufferlist& auth) override {
+ return seastar::make_ready_future<ceph::net::msgr_tag_t, bufferlist>(
+ 0, bufferlist{});
+ }
+ seastar::future<std::unique_ptr<AuthAuthorizer>>
+ ms_get_authorizer(peer_type_t) override {
+ return seastar::make_ready_future<std::unique_ptr<AuthAuthorizer>>(
+ new DummyAuthAuthorizer{});
+ }
+ } dispatcher;
+ Server(ceph::net::Messenger& msgr)
+ : byte_throttler(ceph::net::conf.osd_client_message_size_cap),
+ msgr{msgr}
+ {
+ msgr.set_crc_header();
+ msgr.set_crc_data();
+ }
+};
+
+struct Client {
+ ceph::thread::Throttle byte_throttler;
+ ceph::net::Messenger& msgr;
+ struct ClientDispatcher : ceph::net::Dispatcher {
+ unsigned count = 0;
+ seastar::condition_variable on_reply;
+ seastar::future<> ms_dispatch(ceph::net::ConnectionRef c,
+ MessageRef m) override {
+ std::cout << "client got pong " << *m << std::endl;
+ ++count;
+ on_reply.signal();
+ return seastar::now();
+ }
+ } dispatcher;
+ Client(ceph::net::Messenger& msgr)
+ : byte_throttler(ceph::net::conf.osd_client_message_size_cap),
+ msgr{msgr}
+ {
+ msgr.set_crc_header();
+ msgr.set_crc_data();
+ }
+};
+} // namespace seastar_pingpong
+
+class SeastarContext {
+ seastar::file_desc begin_fd;
+ ceph::thread::Condition on_end;
+
+public:
+ SeastarContext()
+ : begin_fd{seastar::file_desc::eventfd(0, 0)}
+ {}
+
+ template<class Func>
+ std::thread with_seastar(Func&& func) {
+ return std::thread{[this, func = std::forward<Func>(func)] {
+ // alien: are you ready?
+ wait_for_seastar();
+ // alien: could you help me apply(func)?
+ func();
+ // alien: i've sent my request. have you replied it?
+ // wait_for_seastar();
+ // alien: you are free to go!
+ on_end.notify();
+ }};
+ }
+
+ void run(seastar::app_template& app, int argc, char** argv) {
+ app.run(argc, argv, [this] {
+ return seastar::now().then([this] {
+ return set_seastar_ready();
+ }).then([this] {
+ // seastar: let me know once i am free to leave.
+ return on_end.wait();
+ }).handle_exception([](auto ep) {
+ std::cerr << "Error: " << ep << std::endl;
+ }).finally([] {
+ seastar::engine().exit(0);
+ });
+ });
+ }
+
+ seastar::future<> set_seastar_ready() {
+ // seastar: i am ready to serve!
+ ::eventfd_write(begin_fd.get(), 1);
+ return seastar::now();
+ }
+
+private:
+ void wait_for_seastar() {
+ eventfd_t result = 0;
+ if (int r = ::eventfd_read(begin_fd.get(), &result); r < 0) {
+ std::cerr << "unable to eventfd_read():" << errno << std::endl;
+ }
+ }
+};
+
+static seastar::future<>
+seastar_echo(const entity_addr_t addr, echo_role role, unsigned count)
+{
+ std::cout << "seastar/";
+ if (role == echo_role::as_server) {
+ return ceph::net::Messenger::create(entity_name_t::OSD(0), "server",
+ addr.get_nonce(), 0)
+ .then([addr, count] (auto msgr) {
+ return seastar::do_with(seastar_pingpong::Server{*msgr},
+ [addr, count](auto& server) mutable {
+ std::cout << "server listening at " << addr << std::endl;
+ // bind the server
+ server.msgr.set_policy_throttler(entity_name_t::TYPE_OSD,
+ &server.byte_throttler);
+ return server.msgr.bind(entity_addrvec_t{addr})
+ .then([&server] {
+ return server.msgr.start(&server.dispatcher);
+ }).then([&dispatcher=server.dispatcher, count] {
+ return dispatcher.on_reply.wait([&dispatcher, count] {
+ return dispatcher.count >= count;
+ });
+ }).finally([&server] {
+ std::cout << "server shutting down" << std::endl;
+ return server.msgr.shutdown();
+ });
+ });
+ });
+ } else {
+ return ceph::net::Messenger::create(entity_name_t::OSD(1), "client",
+ addr.get_nonce(), 0)
+ .then([addr, count] (auto msgr) {
+ return seastar::do_with(seastar_pingpong::Client{*msgr},
+ [addr, count](auto& client) {
+ std::cout << "client sending to " << addr << std::endl;
+ client.msgr.set_policy_throttler(entity_name_t::TYPE_OSD,
+ &client.byte_throttler);
+ return client.msgr.start(&client.dispatcher)
+ .then([addr, &client] {
+ return client.msgr.connect(addr, entity_name_t::TYPE_OSD);
+ }).then([&disp=client.dispatcher, count](ceph::net::ConnectionXRef conn) {
+ return seastar::do_until(
+ [&disp,count] { return disp.count >= count; },
+ [&disp,conn] { return (*conn)->send(MessageRef{new MPing(), false})
+ .then([&] { return disp.on_reply.wait(); });
+ });
+ }).finally([&client] {
+ std::cout << "client shutting down" << std::endl;
+ return client.msgr.shutdown();
+ });
+ });
+ });
+ }
+}
+
+int main(int argc, char** argv)
+{
+ namespace po = boost::program_options;
+ po::options_description desc{"Allowed options"};
+ desc.add_options()
+ ("help,h", "show help message")
+ ("role", po::value<std::string>()->default_value("pong"),
+ "role to play (ping | pong)")
+ ("port", po::value<uint16_t>()->default_value(9010),
+ "port #")
+ ("nonce", po::value<uint32_t>()->default_value(42),
+ "a unique number to identify the pong server")
+ ("count", po::value<unsigned>()->default_value(10),
+ "stop after sending/echoing <count> MPing messages")
+ ("v2", po::value<bool>()->default_value(false),
+ "using msgr v2 protocol");
+ po::variables_map vm;
+ std::vector<std::string> unrecognized_options;
+ try {
+ auto parsed = po::command_line_parser(argc, argv)
+ .options(desc)
+ .allow_unregistered()
+ .run();
+ po::store(parsed, vm);
+ if (vm.count("help")) {
+ std::cout << desc << std::endl;
+ return 0;
+ }
+ po::notify(vm);
+ unrecognized_options = po::collect_unrecognized(parsed.options, po::include_positional);
+ } catch(const po::error& e) {
+ std::cerr << "error: " << e.what() << std::endl;
+ return 1;
+ }
+
+ entity_addr_t addr;
+ if (vm["v2"].as<bool>()) {
+ addr.set_type(entity_addr_t::TYPE_MSGR2);
+ } else {
+ addr.set_type(entity_addr_t::TYPE_LEGACY);
+ }
+ addr.set_family(AF_INET);
+ addr.set_port(vm["port"].as<std::uint16_t>());
+ addr.set_nonce(vm["nonce"].as<std::uint32_t>());
+
+ echo_role role = echo_role::as_server;
+ if (vm["role"].as<std::string>() == "ping") {
+ role = echo_role::as_client;
+ }
+
+ auto count = vm["count"].as<unsigned>();
+ seastar::app_template app;
+ SeastarContext sc;
+ auto job = sc.with_seastar([&] {
+ auto fut = seastar::alien::submit_to(0, [addr, role, count] {
+ return seastar_echo(addr, role, count);
+ });
+ fut.wait();
+ });
+ std::vector<char*> av{argv[0]};
+ std::transform(begin(unrecognized_options),
+ end(unrecognized_options),
+ std::back_inserter(av),
+ [](auto& s) {
+ return const_cast<char*>(s.c_str());
+ });
+ sc.run(app, av.size(), av.data());
+ job.join();
+}
+
+/*
+ * Local Variables:
+ * compile-command: "make -j4 \
+ * -C ../../../build \
+ * unittest_seastar_echo"
+ * End:
+ */