--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include "global/global_init.h"
+#include "common/ceph_argparse.h"
+
+#include "DirectMessenger.h"
+#include "msg/FastStrategy.h"
+#include "msg/QueueStrategy.h"
+#include "messages/MPing.h"
+
+
+/// mock dispatcher that calls the given callback
+class MockDispatcher : public Dispatcher {
+ std::function<void(Message*)> callback;
+ public:
+ MockDispatcher(CephContext *cct, std::function<void(Message*)> callback)
+ : Dispatcher(cct), callback(std::move(callback)) {}
+ bool ms_handle_reset(Connection *con) override { return false; }
+ void ms_handle_remote_reset(Connection *con) override {}
+ bool ms_handle_refused(Connection *con) override { return false; }
+ bool ms_dispatch(Message *m) override {
+ callback(m);
+ m->put();
+ return true;
+ }
+};
+
+/// test synchronous dispatch of messenger and connection interfaces
+TEST(DirectMessenger, SyncDispatch)
+{
+ auto cct = g_ceph_context;
+
+ // use FastStrategy for synchronous dispatch
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new FastStrategy());
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(&client));
+
+ bool got_request = false;
+ bool got_reply = false;
+
+ MockDispatcher client_dispatcher(cct, [&] (Message *m) {
+ got_reply = true;
+ });
+ client.add_dispatcher_head(&client_dispatcher);
+
+ MockDispatcher server_dispatcher(cct, [&] (Message *m) {
+ got_request = true;
+ ASSERT_EQ(0, m->get_connection()->send_message(new MPing()));
+ });
+ server.add_dispatcher_head(&server_dispatcher);
+
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(0, server.start());
+
+ // test DirectMessenger::send_message()
+ ASSERT_EQ(0, client.send_message(new MPing(), server.get_myinst()));
+ ASSERT_TRUE(got_request);
+ ASSERT_TRUE(got_reply);
+
+ // test DirectConnection::send_message()
+ {
+ got_request = false;
+ got_reply = false;
+ auto conn = client.get_connection(server.get_myinst());
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ ASSERT_TRUE(got_request);
+ ASSERT_TRUE(got_reply);
+ }
+
+ // test DirectMessenger::send_message() with loopback address
+ got_request = false;
+ got_reply = false;
+ ASSERT_EQ(0, client.send_message(new MPing(), client.get_myinst()));
+ ASSERT_FALSE(got_request); // server should never see this
+ ASSERT_TRUE(got_reply);
+
+ // test DirectConnection::send_message() with loopback address
+ {
+ got_request = false;
+ got_reply = false;
+ auto conn = client.get_connection(client.get_myinst());
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ ASSERT_FALSE(got_request); // server should never see this
+ ASSERT_TRUE(got_reply);
+ }
+
+ // test DirectConnection::send_message() with loopback connection
+ {
+ got_request = false;
+ got_reply = false;
+ auto conn = client.get_loopback_connection();
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ ASSERT_FALSE(got_request); // server should never see this
+ ASSERT_TRUE(got_reply);
+ }
+
+ ASSERT_EQ(0, client.shutdown());
+ client.wait();
+
+ ASSERT_EQ(0, server.shutdown());
+ server.wait();
+}
+
+/// test asynchronous dispatch of messenger and connection interfaces
+TEST(DirectMessenger, AsyncDispatch)
+{
+ auto cct = g_ceph_context;
+
+ // use QueueStrategy for async replies
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new QueueStrategy(1));
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(&client));
+
+ // condition variable to wait on ping reply
+ std::mutex mutex;
+ std::condition_variable cond;
+ bool done = false;
+
+ auto wait_for_reply = [&] {
+ std::unique_lock<std::mutex> lock(mutex);
+ while (!done) {
+ cond.wait(lock);
+ }
+ done = false; // clear for reuse
+ };
+
+ // client dispatcher signals the condition variable on reply
+ MockDispatcher client_dispatcher(cct, [&] (Message *m) {
+ std::lock_guard<std::mutex> lock(mutex);
+ done = true;
+ cond.notify_one();
+ });
+ client.add_dispatcher_head(&client_dispatcher);
+
+ MockDispatcher server_dispatcher(cct, [&] (Message *m) {
+ // hold the lock over the call to send_message() to prove that the client's
+ // dispatch is asynchronous. if it isn't, it will deadlock
+ std::lock_guard<std::mutex> lock(mutex);
+ ASSERT_EQ(0, m->get_connection()->send_message(new MPing()));
+ });
+ server.add_dispatcher_head(&server_dispatcher);
+
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(0, server.start());
+
+ // test DirectMessenger::send_message()
+ ASSERT_EQ(0, client.send_message(new MPing(), server.get_myinst()));
+ wait_for_reply();
+
+ // test DirectConnection::send_message()
+ {
+ auto conn = client.get_connection(server.get_myinst());
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ }
+ wait_for_reply();
+
+ // test DirectMessenger::send_message() with loopback address
+ {
+ // hold the lock to test that loopback dispatch is asynchronous
+ std::lock_guard<std::mutex> lock(mutex);
+ ASSERT_EQ(0, client.send_message(new MPing(), client.get_myinst()));
+ }
+ wait_for_reply();
+
+ // test DirectConnection::send_message() with loopback address
+ {
+ auto conn = client.get_connection(client.get_myinst());
+ // hold the lock to test that loopback dispatch is asynchronous
+ std::lock_guard<std::mutex> lock(mutex);
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ }
+ wait_for_reply();
+
+ // test DirectConnection::send_message() with loopback connection
+ {
+ auto conn = client.get_loopback_connection();
+ // hold the lock to test that loopback dispatch is asynchronous
+ std::lock_guard<std::mutex> lock(mutex);
+ ASSERT_EQ(0, conn->send_message(new MPing()));
+ }
+ wait_for_reply();
+
+ ASSERT_EQ(0, client.shutdown());
+ client.wait();
+
+ ASSERT_EQ(0, server.shutdown());
+ server.wait();
+}
+
+/// test that wait() blocks until shutdown()
+TEST(DirectMessenger, WaitShutdown)
+{
+ auto cct = g_ceph_context;
+
+ // test wait() with both Queue- and FastStrategy
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new QueueStrategy(1));
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(&client));
+
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(0, server.start());
+
+ std::atomic<bool> client_waiting{false};
+ std::atomic<bool> server_waiting{false};
+
+ // spawn threads to wait() on each of the messengers
+ std::thread client_thread([&] {
+ client_waiting = true;
+ client.wait();
+ client_waiting = false;
+ });
+ std::thread server_thread([&] {
+ server_waiting = true;
+ server.wait();
+ server_waiting = false;
+ });
+
+ // give them time to start
+ std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+ ASSERT_TRUE(client_waiting);
+ ASSERT_TRUE(server_waiting);
+
+ // call shutdown to unblock the waiting threads
+ ASSERT_EQ(0, client.shutdown());
+ ASSERT_EQ(0, server.shutdown());
+
+ client_thread.join();
+ server_thread.join();
+
+ ASSERT_FALSE(client_waiting);
+ ASSERT_FALSE(server_waiting);
+}
+
+/// test connection and messenger interfaces after mark_down()
+TEST(DirectMessenger, MarkDown)
+{
+ auto cct = g_ceph_context;
+
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new FastStrategy());
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(&client));
+
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(0, server.start());
+
+ auto client_to_server = client.get_connection(server.get_myinst());
+ auto server_to_client = server.get_connection(client.get_myinst());
+
+ ASSERT_TRUE(client_to_server->is_connected());
+ ASSERT_TRUE(server_to_client->is_connected());
+
+ // mark_down() breaks the connection on both sides
+ client_to_server->mark_down();
+
+ ASSERT_FALSE(client_to_server->is_connected());
+ ASSERT_EQ(-ENOTCONN, client_to_server->send_message(new MPing()));
+ ASSERT_EQ(-ENOTCONN, client.send_message(new MPing(), server.get_myinst()));
+
+ ASSERT_FALSE(server_to_client->is_connected());
+ ASSERT_EQ(-ENOTCONN, server_to_client->send_message(new MPing()));
+ ASSERT_EQ(-ENOTCONN, server.send_message(new MPing(), client.get_myinst()));
+
+ ASSERT_EQ(0, client.shutdown());
+ client.wait();
+
+ ASSERT_EQ(0, server.shutdown());
+ server.wait();
+}
+
+/// test connection and messenger interfaces after shutdown()
+TEST(DirectMessenger, SendShutdown)
+{
+ auto cct = g_ceph_context;
+
+ // put client on the heap so we can free it early
+ std::unique_ptr<DirectMessenger> client{
+ new DirectMessenger(cct, entity_name_t::CLIENT(1),
+ "client", 0, new FastStrategy())};
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ ASSERT_EQ(0, client->set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(client.get()));
+
+ ASSERT_EQ(0, client->start());
+ ASSERT_EQ(0, server.start());
+
+ const auto client_inst = client->get_myinst();
+ const auto server_inst = server.get_myinst();
+
+ auto client_to_server = client->get_connection(server_inst);
+ auto server_to_client = server.get_connection(client_inst);
+
+ ASSERT_TRUE(client_to_server->is_connected());
+ ASSERT_TRUE(server_to_client->is_connected());
+
+ // shut down the client to break connections
+ ASSERT_EQ(0, client->shutdown());
+ client->wait();
+
+ ASSERT_FALSE(client_to_server->is_connected());
+ ASSERT_EQ(-ENOTCONN, client_to_server->send_message(new MPing()));
+ ASSERT_EQ(-ENOTCONN, client->send_message(new MPing(), server_inst));
+
+ // free the client connection/messenger to test that calls to the server no
+ // longer try to dereference them
+ client_to_server.reset();
+ client.reset();
+
+ ASSERT_FALSE(server_to_client->is_connected());
+ ASSERT_EQ(-ENOTCONN, server_to_client->send_message(new MPing()));
+ ASSERT_EQ(-ENOTCONN, server.send_message(new MPing(), client_inst));
+
+ ASSERT_EQ(0, server.shutdown());
+ server.wait();
+}
+
+/// test connection and messenger interfaces after bind()
+TEST(DirectMessenger, Bind)
+{
+ auto cct = g_ceph_context;
+
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new FastStrategy());
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ entity_addr_t client_addr;
+ client_addr.set_family(AF_INET);
+ client_addr.set_port(1);
+
+ // client bind succeeds before set_direct_peer()
+ ASSERT_EQ(0, client.bind(client_addr));
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+ ASSERT_EQ(0, server.set_direct_peer(&client));
+
+ // server bind fails after set_direct_peer()
+ entity_addr_t empty_addr;
+ ASSERT_EQ(-EINVAL, server.bind(empty_addr));
+
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(0, server.start());
+
+ auto client_to_server = client.get_connection(server.get_myinst());
+ auto server_to_client = server.get_connection(client.get_myinst());
+
+ ASSERT_TRUE(client_to_server->is_connected());
+ ASSERT_TRUE(server_to_client->is_connected());
+
+ // no address in connection to server
+ ASSERT_EQ(empty_addr, client_to_server->get_peer_addr());
+ // bind address is reflected in connection to client
+ ASSERT_EQ(client_addr, server_to_client->get_peer_addr());
+
+ // mark_down() with bind address breaks the connection
+ server.mark_down(client_addr);
+
+ ASSERT_FALSE(client_to_server->is_connected());
+ ASSERT_FALSE(server_to_client->is_connected());
+
+ ASSERT_EQ(0, client.shutdown());
+ client.wait();
+
+ ASSERT_EQ(0, server.shutdown());
+ server.wait();
+}
+
+/// test connection and messenger interfaces before calls to set_direct_peer()
+TEST(DirectMessenger, StartWithoutPeer)
+{
+ auto cct = g_ceph_context;
+
+ DirectMessenger client(cct, entity_name_t::CLIENT(1),
+ "client", 0, new FastStrategy());
+ DirectMessenger server(cct, entity_name_t::CLIENT(2),
+ "server", 0, new FastStrategy());
+
+ // can't start until set_direct_peer()
+ ASSERT_EQ(-EINVAL, client.start());
+ ASSERT_EQ(-EINVAL, server.start());
+
+ ASSERT_EQ(0, client.set_direct_peer(&server));
+
+ // only client can start
+ ASSERT_EQ(0, client.start());
+ ASSERT_EQ(-EINVAL, server.start());
+
+ // client has a connection but can't send
+ auto conn = client.get_connection(server.get_myinst());
+ ASSERT_NE(nullptr, conn);
+ ASSERT_FALSE(conn->is_connected());
+ ASSERT_EQ(-ENOTCONN, conn->send_message(new MPing()));
+ ASSERT_EQ(-ENOTCONN, client.send_message(new MPing(), server.get_myinst()));
+
+ ASSERT_EQ(0, client.shutdown());
+ client.wait();
+}
+
+int main(int argc, char **argv)
+{
+ // command-line arguments
+ vector<const char*> args;
+ argv_to_vec(argc, (const char **)argv, args);
+ env_to_vec(args);
+
+ auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_ANY,
+ CODE_ENVIRONMENT_DAEMON, 0);
+ common_init_finish(cct.get());
+
+ ::testing::InitGoogleTest(&argc, argv);
+ return RUN_ALL_TESTS();
+}
Code Review / stor4nfv.git / blobdiff