+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * Copyright (C) 2016 XSky <haomai@xsky.com>
- *
- * Author: Haomai Wang <haomaiwang@gmail.com>
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- *
- */
-
-#include <algorithm>
-#include <atomic>
-#include <iostream>
-#include <random>
-#include <string>
-#include <set>
-#include <vector>
-#include <gtest/gtest.h>
-
-#include "acconfig.h"
-#include "include/Context.h"
-
-#include "msg/async/Event.h"
-#include "msg/async/Stack.h"
-
-
-#if GTEST_HAS_PARAM_TEST
-
-class NetworkWorkerTest : public ::testing::TestWithParam<const char*> {
- public:
- std::shared_ptr<NetworkStack> stack;
- string addr, port_addr;
-
- NetworkWorkerTest() {}
- void SetUp() override {
- cerr << __func__ << " start set up " << GetParam() << std::endl;
- if (strncmp(GetParam(), "dpdk", 4)) {
- g_ceph_context->_conf->set_val("ms_type", "async+posix", false);
- addr = "127.0.0.1:15000";
- port_addr = "127.0.0.1:15001";
- } else {
- g_ceph_context->_conf->set_val("ms_type", "async+dpdk", false);
- g_ceph_context->_conf->set_val("ms_dpdk_debug_allow_loopback", "true", false);
- g_ceph_context->_conf->set_val("ms_async_op_threads", "2", false);
- g_ceph_context->_conf->set_val("ms_dpdk_coremask", "0x7", false);
- g_ceph_context->_conf->set_val("ms_dpdk_host_ipv4_addr", "172.16.218.3", false);
- g_ceph_context->_conf->set_val("ms_dpdk_gateway_ipv4_addr", "172.16.218.2", false);
- g_ceph_context->_conf->set_val("ms_dpdk_netmask_ipv4_addr", "255.255.255.0", false);
- addr = "172.16.218.3:15000";
- port_addr = "172.16.218.3:15001";
- }
- stack = NetworkStack::create(g_ceph_context, GetParam());
- stack->start();
- }
- void TearDown() override {
- stack->stop();
- }
- string get_addr() const {
- return addr;
- }
- string get_ip_different_port() const {
- return port_addr;
- }
- string get_different_ip() const {
- return "10.0.123.100:4323";
- }
- EventCenter *get_center(unsigned i) {
- return &stack->get_worker(i)->center;
- }
- Worker *get_worker(unsigned i) {
- return stack->get_worker(i);
- }
- template<typename func>
- class C_dispatch : public EventCallback {
- Worker *worker;
- func f;
- std::atomic_bool done;
- public:
- C_dispatch(Worker *w, func &&_f): worker(w), f(std::move(_f)), done(false) {}
- void do_request(int id) override {
- f(worker);
- done = true;
- }
- void wait() {
- int us = 1000 * 1000 * 1000;
- while (!done) {
- ASSERT_TRUE(us > 0);
- usleep(100);
- us -= 100;
- }
- }
- };
- template<typename func>
- void exec_events(func &&f) {
- std::vector<C_dispatch<func>*> dis;
- for (unsigned i = 0; i < stack->get_num_worker(); ++i) {
- Worker *w = stack->get_worker(i);
- C_dispatch<func> *e = new C_dispatch<func>(w, std::move(f));
- stack->get_worker(i)->center.dispatch_event_external(e);
- dis.push_back(e);
- }
-
- for (auto &&e : dis) {
- e->wait();
- delete e;
- }
- }
-};
-
-class C_poll : public EventCallback {
- EventCenter *center;
- std::atomic<bool> woken;
- static const int sleepus = 500;
-
- public:
- C_poll(EventCenter *c): center(c), woken(false) {}
- void do_request(int r) override {
- woken = true;
- }
- bool poll(int milliseconds) {
- auto start = ceph::coarse_real_clock::now();
- while (!woken) {
- center->process_events(sleepus);
- usleep(sleepus);
- auto r = std::chrono::duration_cast<std::chrono::milliseconds>(
- ceph::coarse_real_clock::now() - start);
- if (r >= std::chrono::milliseconds(milliseconds))
- break;
- }
- return woken;
- }
- void reset() {
- woken = false;
- }
-};
-
-TEST_P(NetworkWorkerTest, SimpleTest) {
- entity_addr_t bind_addr;
- ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
- std::atomic_bool accepted(false);
- std::atomic_bool *accepted_p = &accepted;
-
- exec_events([this, accepted_p, bind_addr](Worker *worker) mutable {
- entity_addr_t cli_addr;
- SocketOptions options;
- ServerSocket bind_socket;
- EventCenter *center = &worker->center;
- ssize_t r = 0;
- if (stack->support_local_listen_table() || worker->id == 0)
- r = worker->listen(bind_addr, options, &bind_socket);
- ASSERT_EQ(0, r);
-
- ConnectedSocket cli_socket, srv_socket;
- if (worker->id == 0) {
- r = worker->connect(bind_addr, options, &cli_socket);
- ASSERT_EQ(0, r);
- }
-
- bool is_my_accept = false;
- if (bind_socket) {
- C_poll cb(center);
- center->create_file_event(bind_socket.fd(), EVENT_READABLE, &cb);
- if (cb.poll(500)) {
- *accepted_p = true;
- is_my_accept = true;
- }
- ASSERT_TRUE(*accepted_p);
- center->delete_file_event(bind_socket.fd(), EVENT_READABLE);
- }
-
- if (is_my_accept) {
- r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
- ASSERT_EQ(0, r);
- ASSERT_TRUE(srv_socket.fd() > 0);
- }
-
- if (worker->id == 0) {
- C_poll cb(center);
- center->create_file_event(cli_socket.fd(), EVENT_READABLE, &cb);
- r = cli_socket.is_connected();
- if (r == 0) {
- ASSERT_EQ(true, cb.poll(500));
- r = cli_socket.is_connected();
- }
- ASSERT_EQ(1, r);
- center->delete_file_event(cli_socket.fd(), EVENT_READABLE);
- }
-
- const char *message = "this is a new message";
- int len = strlen(message);
- bufferlist bl;
- bl.append(message, len);
- if (worker->id == 0) {
- r = cli_socket.send(bl, false);
- ASSERT_EQ(len, r);
- }
-
- char buf[1024];
- C_poll cb(center);
- if (is_my_accept) {
- center->create_file_event(srv_socket.fd(), EVENT_READABLE, &cb);
- {
- r = srv_socket.read(buf, sizeof(buf));
- while (r == -EAGAIN) {
- ASSERT_TRUE(cb.poll(500));
- r = srv_socket.read(buf, sizeof(buf));
- cb.reset();
- }
- ASSERT_EQ(len, r);
- ASSERT_EQ(0, memcmp(buf, message, len));
- }
- bind_socket.abort_accept();
- }
- if (worker->id == 0) {
- cli_socket.shutdown();
- // ack delay is 200 ms
- }
-
- bl.clear();
- bl.append(message, len);
- if (worker->id == 0) {
- r = cli_socket.send(bl, false);
- ASSERT_EQ(-EPIPE, r);
- }
- if (is_my_accept) {
- cb.reset();
- ASSERT_TRUE(cb.poll(500));
- r = srv_socket.read(buf, sizeof(buf));
- if (r == -EAGAIN) {
- cb.reset();
- ASSERT_TRUE(cb.poll(1000*500));
- r = srv_socket.read(buf, sizeof(buf));
- }
- ASSERT_EQ(0, r);
- center->delete_file_event(srv_socket.fd(), EVENT_READABLE);
- srv_socket.close();
- }
- });
-}
-
-TEST_P(NetworkWorkerTest, ConnectFailedTest) {
- entity_addr_t bind_addr;
- ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
-
- exec_events([this, bind_addr](Worker *worker) mutable {
- EventCenter *center = &worker->center;
- entity_addr_t cli_addr;
- SocketOptions options;
- ServerSocket bind_socket;
- int r = 0;
- if (stack->support_local_listen_table() || worker->id == 0)
- r = worker->listen(bind_addr, options, &bind_socket);
- ASSERT_EQ(0, r);
-
- ConnectedSocket cli_socket1, cli_socket2;
- if (worker->id == 0) {
- ASSERT_TRUE(cli_addr.parse(get_ip_different_port().c_str()));
- r = worker->connect(cli_addr, options, &cli_socket1);
- ASSERT_EQ(0, r);
- C_poll cb(center);
- center->create_file_event(cli_socket1.fd(), EVENT_READABLE, &cb);
- r = cli_socket1.is_connected();
- if (r == 0) {
- ASSERT_TRUE(cb.poll(500));
- r = cli_socket1.is_connected();
- }
- ASSERT_TRUE(r == -ECONNREFUSED || r == -ECONNRESET);
- }
-
- if (worker->id == 1) {
- ASSERT_TRUE(cli_addr.parse(get_different_ip().c_str()));
- r = worker->connect(cli_addr, options, &cli_socket2);
- ASSERT_EQ(0, r);
- C_poll cb(center);
- center->create_file_event(cli_socket2.fd(), EVENT_READABLE, &cb);
- r = cli_socket2.is_connected();
- if (r == 0) {
- cb.poll(500);
- r = cli_socket2.is_connected();
- }
- ASSERT_TRUE(r != 1);
- center->delete_file_event(cli_socket2.fd(), EVENT_READABLE);
- }
- });
-}
-
-TEST_P(NetworkWorkerTest, ListenTest) {
- Worker *worker = get_worker(0);
- entity_addr_t bind_addr;
- ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
- SocketOptions options;
- ServerSocket bind_socket1, bind_socket2;
- int r = worker->listen(bind_addr, options, &bind_socket1);
- ASSERT_EQ(0, r);
-
- r = worker->listen(bind_addr, options, &bind_socket2);
- ASSERT_EQ(-EADDRINUSE, r);
-}
-
-TEST_P(NetworkWorkerTest, AcceptAndCloseTest) {
- entity_addr_t bind_addr;
- ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
- std::atomic_bool accepted(false);
- std::atomic_bool *accepted_p = &accepted;
- std::atomic_int unbind_count(stack->get_num_worker());
- std::atomic_int *count_p = &unbind_count;
- exec_events([this, bind_addr, accepted_p, count_p](Worker *worker) mutable {
- SocketOptions options;
- EventCenter *center = &worker->center;
- entity_addr_t cli_addr;
- int r = 0;
- {
- ServerSocket bind_socket;
- if (stack->support_local_listen_table() || worker->id == 0)
- r = worker->listen(bind_addr, options, &bind_socket);
- ASSERT_EQ(0, r);
-
- ConnectedSocket srv_socket, cli_socket;
- if (bind_socket) {
- r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
- ASSERT_EQ(-EAGAIN, r);
- }
-
- C_poll cb(center);
- if (worker->id == 0) {
- center->create_file_event(bind_socket.fd(), EVENT_READABLE, &cb);
- r = worker->connect(bind_addr, options, &cli_socket);
- ASSERT_EQ(0, r);
- ASSERT_TRUE(cb.poll(500));
- }
-
- if (bind_socket) {
- cb.reset();
- cb.poll(500);
- ConnectedSocket srv_socket2;
- do {
- r = bind_socket.accept(&srv_socket2, options, &cli_addr, worker);
- usleep(100);
- } while (r == -EAGAIN && !*accepted_p);
- if (r == 0)
- *accepted_p = true;
- ASSERT_TRUE(*accepted_p);
- // srv_socket2 closed
- center->delete_file_event(bind_socket.fd(), EVENT_READABLE);
- }
-
- if (worker->id == 0) {
- char buf[100];
- cb.reset();
- center->create_file_event(cli_socket.fd(), EVENT_READABLE, &cb);
- int i = 3;
- while (!i--) {
- ASSERT_TRUE(cb.poll(500));
- r = cli_socket.read(buf, sizeof(buf));
- if (r == 0)
- break;
- }
- ASSERT_EQ(0, r);
- center->delete_file_event(cli_socket.fd(), EVENT_READABLE);
- }
-
- if (bind_socket)
- center->create_file_event(bind_socket.fd(), EVENT_READABLE, &cb);
- if (worker->id == 0) {
- *accepted_p = false;
- r = worker->connect(bind_addr, options, &cli_socket);
- ASSERT_EQ(0, r);
- cb.reset();
- ASSERT_TRUE(cb.poll(500));
- cli_socket.close();
- }
-
- if (bind_socket) {
- do {
- r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
- usleep(100);
- } while (r == -EAGAIN && !*accepted_p);
- if (r == 0)
- *accepted_p = true;
- ASSERT_TRUE(*accepted_p);
- center->delete_file_event(bind_socket.fd(), EVENT_READABLE);
- }
- // unbind
- }
-
- --*count_p;
- while (*count_p > 0)
- usleep(100);
-
- ConnectedSocket cli_socket;
- r = worker->connect(bind_addr, options, &cli_socket);
- ASSERT_EQ(0, r);
- {
- C_poll cb(center);
- center->create_file_event(cli_socket.fd(), EVENT_READABLE, &cb);
- r = cli_socket.is_connected();
- if (r == 0) {
- ASSERT_TRUE(cb.poll(500));
- r = cli_socket.is_connected();
- }
- ASSERT_TRUE(r == -ECONNREFUSED || r == -ECONNRESET);
- }
- });
-}
-
-TEST_P(NetworkWorkerTest, ComplexTest) {
- entity_addr_t bind_addr;
- std::atomic_bool listen_done(false);
- std::atomic_bool *listen_p = &listen_done;
- std::atomic_bool accepted(false);
- std::atomic_bool *accepted_p = &accepted;
- std::atomic_bool done(false);
- std::atomic_bool *done_p = &done;
- ASSERT_TRUE(bind_addr.parse(get_addr().c_str()));
- exec_events([this, bind_addr, listen_p, accepted_p, done_p](Worker *worker) mutable {
- entity_addr_t cli_addr;
- EventCenter *center = &worker->center;
- SocketOptions options;
- ServerSocket bind_socket;
- int r = 0;
- if (stack->support_local_listen_table() || worker->id == 0) {
- r = worker->listen(bind_addr, options, &bind_socket);
- ASSERT_EQ(0, r);
- *listen_p = true;
- }
- ConnectedSocket cli_socket, srv_socket;
- if (worker->id == 1) {
- while (!*listen_p) {
- usleep(50);
- r = worker->connect(bind_addr, options, &cli_socket);
- ASSERT_EQ(0, r);
- }
- }
-
- if (bind_socket) {
- C_poll cb(center);
- center->create_file_event(bind_socket.fd(), EVENT_READABLE, &cb);
- int count = 3;
- while (count--) {
- if (cb.poll(500)) {
- r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
- ASSERT_EQ(0, r);
- *accepted_p = true;
- break;
- }
- }
- ASSERT_TRUE(*accepted_p);
- center->delete_file_event(bind_socket.fd(), EVENT_READABLE);
- }
-
- if (worker->id == 1) {
- C_poll cb(center);
- center->create_file_event(cli_socket.fd(), EVENT_WRITABLE, &cb);
- r = cli_socket.is_connected();
- if (r == 0) {
- ASSERT_TRUE(cb.poll(500));
- r = cli_socket.is_connected();
- }
- ASSERT_EQ(1, r);
- center->delete_file_event(cli_socket.fd(), EVENT_WRITABLE);
- }
-
- const size_t message_size = 10240;
- size_t count = 100;
- string message(message_size, '!');
- for (size_t i = 0; i < message_size; i += 100)
- message[i] = ',';
- size_t len = message_size * count;
- C_poll cb(center);
- if (worker->id == 1)
- center->create_file_event(cli_socket.fd(), EVENT_WRITABLE, &cb);
- if (srv_socket)
- center->create_file_event(srv_socket.fd(), EVENT_READABLE, &cb);
- size_t left = len;
- len *= 2;
- string read_string;
- int again_count = 0;
- int c = 2;
- bufferlist bl;
- for (size_t i = 0; i < count; ++i)
- bl.push_back(bufferptr((char*)message.data(), message_size));
- while (!*done_p) {
- again_count = 0;
- if (worker->id == 1) {
- if (c > 0) {
- ssize_t r = 0;
- usleep(100);
- if (left > 0) {
- r = cli_socket.send(bl, false);
- ASSERT_TRUE(r >= 0 || r == -EAGAIN);
- if (r > 0)
- left -= r;
- if (r == -EAGAIN)
- ++again_count;
- }
- if (left == 0) {
- --c;
- left = message_size * count;
- ASSERT_EQ(0U, bl.length());
- for (size_t i = 0; i < count; ++i)
- bl.push_back(bufferptr((char*)message.data(), message_size));
- }
- }
- }
-
- if (srv_socket) {
- char buf[1000];
- if (len > 0) {
- r = srv_socket.read(buf, sizeof(buf));
- ASSERT_TRUE(r > 0 || r == -EAGAIN);
- if (r > 0) {
- read_string.append(buf, r);
- len -= r;
- } else if (r == -EAGAIN) {
- ++again_count;
- }
- }
- if (len == 0) {
- for (size_t i = 0; i < read_string.size(); i += message_size)
- ASSERT_EQ(0, memcmp(read_string.c_str()+i, message.c_str(), message_size));
- *done_p = true;
- }
- }
- if (again_count) {
- cb.reset();
- cb.poll(500);
- }
- }
- if (worker->id == 1)
- center->delete_file_event(cli_socket.fd(), EVENT_WRITABLE);
- if (srv_socket)
- center->delete_file_event(srv_socket.fd(), EVENT_READABLE);
-
- if (bind_socket)
- bind_socket.abort_accept();
- if (srv_socket)
- srv_socket.close();
- if (worker->id == 1)
- cli_socket.close();
- });
-}
-
-class StressFactory {
- struct Client;
- struct Server;
- struct ThreadData {
- Worker *worker;
- std::set<Client*> clients;
- std::set<Server*> servers;
- ~ThreadData() {
- for (auto && i : clients)
- delete i;
- for (auto && i : servers)
- delete i;
- }
- };
-
- struct RandomString {
- size_t slen;
- vector<std::string> strs;
- std::random_device rd;
- std::default_random_engine rng;
-
- RandomString(size_t s): slen(s), rng(rd()) {}
- void prepare(size_t n) {
- static const char alphabet[] =
- "abcdefghijklmnopqrstuvwxyz"
- "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
- "0123456789";
-
- std::uniform_int_distribution<> dist(
- 0, sizeof(alphabet) / sizeof(*alphabet) - 2);
-
- strs.reserve(n);
- std::generate_n(
- std::back_inserter(strs), strs.capacity(), [&] {
- std::string str;
- str.reserve(slen);
- std::generate_n(std::back_inserter(str), slen, [&]() {
- return alphabet[dist(rng)];
- });
- return str;
- }
- );
- }
- std::string &get_random_string() {
- std::uniform_int_distribution<> dist(
- 0, strs.size() - 1);
- return strs[dist(rng)];
- }
- };
- struct Message {
- size_t idx;
- size_t len;
- std::string content;
-
- explicit Message(RandomString &rs, size_t i, size_t l): idx(i) {
- size_t slen = rs.slen;
- len = std::max(slen, l);
-
- std::vector<std::string> strs;
- strs.reserve(len / slen);
- std::generate_n(
- std::back_inserter(strs), strs.capacity(), [&] {
- return rs.get_random_string();
- }
- );
- len = slen * strs.size();
- content.reserve(len);
- for (auto &&s : strs)
- content.append(s);
- }
- bool verify(const char *b, size_t len = 0) const {
- return content.compare(0, len, b, 0, len) == 0;
- }
- };
-
- template <typename T>
- class C_delete : public EventCallback {
- T *ctxt;
- public:
- C_delete(T *c): ctxt(c) {}
- void do_request(int id) override {
- delete ctxt;
- delete this;
- }
- };
-
- class Client {
- StressFactory *factory;
- EventCenter *center;
- ConnectedSocket socket;
- std::deque<StressFactory::Message*> acking;
- std::deque<StressFactory::Message*> writings;
- std::string buffer;
- size_t index = 0;
- size_t left;
- bool write_enabled = false;
- size_t read_offset = 0, write_offset = 0;
- bool first = true;
- bool dead = false;
- StressFactory::Message homeless_message;
-
- class Client_read_handle : public EventCallback {
- Client *c;
- public:
- Client_read_handle(Client *_c): c(_c) {}
- void do_request(int id) override {
- c->do_read_request();
- }
- } read_ctxt;
-
- class Client_write_handle : public EventCallback {
- Client *c;
- public:
- Client_write_handle(Client *_c): c(_c) {}
- void do_request(int id) override {
- c->do_write_request();
- }
- } write_ctxt;
-
- public:
- Client(StressFactory *f, EventCenter *cen, ConnectedSocket s, size_t c)
- : factory(f), center(cen), socket(std::move(s)), left(c), homeless_message(factory->rs, -1, 1024),
- read_ctxt(this), write_ctxt(this) {
- center->create_file_event(
- socket.fd(), EVENT_READABLE, &read_ctxt);
- center->dispatch_event_external(&read_ctxt);
- }
- void close() {
- ASSERT_FALSE(write_enabled);
- dead = true;
- socket.shutdown();
- center->delete_file_event(socket.fd(), EVENT_READABLE);
- center->dispatch_event_external(new C_delete<Client>(this));
- }
-
- void do_read_request() {
- if (dead)
- return ;
- ASSERT_TRUE(socket.is_connected() >= 0);
- if (!socket.is_connected())
- return ;
- ASSERT_TRUE(!acking.empty() || first);
- if (first) {
- first = false;
- center->dispatch_event_external(&write_ctxt);
- if (acking.empty())
- return ;
- }
- StressFactory::Message *m = acking.front();
- int r = 0;
- if (buffer.empty())
- buffer.resize(m->len);
- bool must_no = false;
- while (true) {
- r = socket.read((char*)buffer.data() + read_offset,
- m->len - read_offset);
- ASSERT_TRUE(r == -EAGAIN || r > 0);
- if (r == -EAGAIN)
- break;
- read_offset += r;
-
- std::cerr << " client " << this << " receive " << m->idx << " len " << r << " content: " << std::endl;
- ASSERT_FALSE(must_no);
- if ((m->len - read_offset) == 0) {
- ASSERT_TRUE(m->verify(buffer.data(), 0));
- delete m;
- acking.pop_front();
- read_offset = 0;
- buffer.clear();
- if (acking.empty()) {
- m = &homeless_message;
- must_no = true;
- } else {
- m = acking.front();
- buffer.resize(m->len);
- }
- }
- }
- if (acking.empty()) {
- center->dispatch_event_external(&write_ctxt);
- return ;
- }
- }
-
- void do_write_request() {
- if (dead)
- return ;
- ASSERT_TRUE(socket.is_connected() > 0);
-
- while (left > 0 && factory->queue_depth > writings.size() + acking.size()) {
- StressFactory::Message *m = new StressFactory::Message(
- factory->rs, ++index,
- factory->rd() % factory->max_message_length);
- std::cerr << " client " << this << " generate message " << m->idx << " length " << m->len << std::endl;
- ASSERT_EQ(m->len, m->content.size());
- writings.push_back(m);
- --left;
- --factory->message_left;
- }
-
- while (!writings.empty()) {
- StressFactory::Message *m = writings.front();
- bufferlist bl;
- bl.append(m->content.data() + write_offset, m->content.size() - write_offset);
- ssize_t r = socket.send(bl, false);
- if (r == 0)
- break;
- std::cerr << " client " << this << " send " << m->idx << " len " << r << " content: " << std::endl;
- ASSERT_TRUE(r >= 0);
- write_offset += r;
- if (write_offset == m->content.size()) {
- write_offset = 0;
- writings.pop_front();
- acking.push_back(m);
- }
- }
- if (writings.empty() && write_enabled) {
- center->delete_file_event(socket.fd(), EVENT_WRITABLE);
- write_enabled = false;
- } else if (!writings.empty() && !write_enabled) {
- ASSERT_EQ(0, center->create_file_event(
- socket.fd(), EVENT_WRITABLE, &write_ctxt));
- write_enabled = true;
- }
- }
-
- bool finish() const {
- return left == 0 && acking.empty() && writings.empty();
- }
- };
- friend class Client;
-
- class Server {
- StressFactory *factory;
- EventCenter *center;
- ConnectedSocket socket;
- std::deque<std::string> buffers;
- bool write_enabled = false;
- bool dead = false;
-
- class Server_read_handle : public EventCallback {
- Server *s;
- public:
- Server_read_handle(Server *_s): s(_s) {}
- void do_request(int id) override {
- s->do_read_request();
- }
- } read_ctxt;
-
- class Server_write_handle : public EventCallback {
- Server *s;
- public:
- Server_write_handle(Server *_s): s(_s) {}
- void do_request(int id) override {
- s->do_write_request();
- }
- } write_ctxt;
-
- public:
- Server(StressFactory *f, EventCenter *c, ConnectedSocket s):
- factory(f), center(c), socket(std::move(s)), read_ctxt(this), write_ctxt(this) {
- center->create_file_event(socket.fd(), EVENT_READABLE, &read_ctxt);
- center->dispatch_event_external(&read_ctxt);
- }
- void close() {
- ASSERT_FALSE(write_enabled);
- socket.shutdown();
- center->delete_file_event(socket.fd(), EVENT_READABLE);
- center->dispatch_event_external(new C_delete<Server>(this));
- }
- void do_read_request() {
- if (dead)
- return ;
- int r = 0;
- while (true) {
- char buf[4096];
- bufferptr data;
- if (factory->zero_copy_read) {
- r = socket.zero_copy_read(data);
- } else {
- r = socket.read(buf, sizeof(buf));
- }
- ASSERT_TRUE(r == -EAGAIN || (r >= 0 && (size_t)r <= sizeof(buf)));
- if (r == 0) {
- ASSERT_TRUE(buffers.empty());
- dead = true;
- return ;
- } else if (r == -EAGAIN)
- break;
- if (factory->zero_copy_read) {
- buffers.emplace_back(data.c_str(), 0, data.length());
- } else {
- buffers.emplace_back(buf, 0, r);
- }
- std::cerr << " server " << this << " receive " << r << " content: " << std::endl;
- }
- if (!buffers.empty() && !write_enabled)
- center->dispatch_event_external(&write_ctxt);
- }
-
- void do_write_request() {
- if (dead)
- return ;
-
- while (!buffers.empty()) {
- bufferlist bl;
- auto it = buffers.begin();
- for (size_t i = 0; i < buffers.size(); ++i) {
- bl.push_back(bufferptr((char*)it->data(), it->size()));
- ++it;
- }
-
- ssize_t r = socket.send(bl, false);
- std::cerr << " server " << this << " send " << r << std::endl;
- if (r == 0)
- break;
- ASSERT_TRUE(r >= 0);
- while (r > 0) {
- ASSERT_TRUE(!buffers.empty());
- string &buffer = buffers.front();
- if (r >= (int)buffer.size()) {
- r -= (int)buffer.size();
- buffers.pop_front();
- } else {
- std::cerr << " server " << this << " sent " << r << std::endl;
- buffer = buffer.substr(r, buffer.size());
- break;
- }
- }
- }
- if (buffers.empty()) {
- if (write_enabled) {
- center->delete_file_event(socket.fd(), EVENT_WRITABLE);
- write_enabled = false;
- }
- } else if (!write_enabled) {
- ASSERT_EQ(0, center->create_file_event(
- socket.fd(), EVENT_WRITABLE, &write_ctxt));
- write_enabled = true;
- }
- }
-
- bool finish() {
- return dead;
- }
- };
- friend class Server;
-
- class C_accept : public EventCallback {
- StressFactory *factory;
- ServerSocket bind_socket;
- ThreadData *t_data;
- Worker *worker;
-
- public:
- C_accept(StressFactory *f, ServerSocket s, ThreadData *data, Worker *w)
- : factory(f), bind_socket(std::move(s)), t_data(data), worker(w) {}
- void do_request(int id) override {
- while (true) {
- entity_addr_t cli_addr;
- ConnectedSocket srv_socket;
- SocketOptions options;
- int r = bind_socket.accept(&srv_socket, options, &cli_addr, worker);
- if (r == -EAGAIN) {
- break;
- }
- ASSERT_EQ(0, r);
- ASSERT_TRUE(srv_socket.fd() > 0);
- Server *cb = new Server(factory, &t_data->worker->center, std::move(srv_socket));
- t_data->servers.insert(cb);
- }
- }
- };
- friend class C_accept;
-
- public:
- static const size_t min_client_send_messages = 100;
- static const size_t max_client_send_messages = 1000;
- std::shared_ptr<NetworkStack> stack;
- RandomString rs;
- std::random_device rd;
- const size_t client_num, queue_depth, max_message_length;
- atomic_int message_count, message_left;
- entity_addr_t bind_addr;
- std::atomic_bool already_bind = {false};
- bool zero_copy_read;
- SocketOptions options;
-
- explicit StressFactory(std::shared_ptr<NetworkStack> s, const string &addr,
- size_t cli, size_t qd, size_t mc, size_t l, bool zero_copy)
- : stack(s), rs(128), client_num(cli), queue_depth(qd),
- max_message_length(l), message_count(mc), message_left(mc),
- zero_copy_read(zero_copy) {
- bind_addr.parse(addr.c_str());
- rs.prepare(100);
- }
- ~StressFactory() {
- }
-
- void add_client(ThreadData *t_data) {
- static Mutex lock("add_client_lock");
- Mutex::Locker l(lock);
- ConnectedSocket sock;
- int r = t_data->worker->connect(bind_addr, options, &sock);
- std::default_random_engine rng(rd());
- std::uniform_int_distribution<> dist(
- min_client_send_messages, max_client_send_messages);
- ASSERT_EQ(0, r);
- int c = dist(rng);
- if (c > message_count.load())
- c = message_count.load();
- Client *cb = new Client(this, &t_data->worker->center, std::move(sock), c);
- t_data->clients.insert(cb);
- message_count -= c;
- }
-
- void drop_client(ThreadData *t_data, Client *c) {
- c->close();
- ASSERT_EQ(1U, t_data->clients.erase(c));
- }
-
- void drop_server(ThreadData *t_data, Server *s) {
- s->close();
- ASSERT_EQ(1U, t_data->servers.erase(s));
- }
-
- void start(Worker *worker) {
- int r = 0;
- ThreadData t_data;
- t_data.worker = worker;
- ServerSocket bind_socket;
- if (stack->support_local_listen_table() || worker->id == 0) {
- r = worker->listen(bind_addr, options, &bind_socket);
- ASSERT_EQ(0, r);
- already_bind = true;
- }
- while (!already_bind)
- usleep(50);
- C_accept *accept_handler = nullptr;
- int bind_fd = 0;
- if (bind_socket) {
- bind_fd = bind_socket.fd();
- accept_handler = new C_accept(this, std::move(bind_socket), &t_data, worker);
- ASSERT_EQ(0, worker->center.create_file_event(
- bind_fd, EVENT_READABLE, accept_handler));
- }
-
- int echo_throttle = message_count;
- while (message_count > 0 || !t_data.clients.empty() || !t_data.servers.empty()) {
- if (message_count > 0 && t_data.clients.size() < client_num && t_data.servers.size() < client_num)
- add_client(&t_data);
- for (auto &&c : t_data.clients) {
- if (c->finish()) {
- drop_client(&t_data, c);
- break;
- }
- }
- for (auto &&s : t_data.servers) {
- if (s->finish()) {
- drop_server(&t_data, s);
- break;
- }
- }
-
- worker->center.process_events(1);
- if (echo_throttle > message_left) {
- std::cerr << " clients " << t_data.clients.size() << " servers " << t_data.servers.size()
- << " message count " << message_left << std::endl;
- echo_throttle -= 100;
- }
- }
- if (bind_fd)
- worker->center.delete_file_event(bind_fd, EVENT_READABLE);
- delete accept_handler;
- }
-};
-
-TEST_P(NetworkWorkerTest, StressTest) {
- StressFactory factory(stack, get_addr(), 16, 16, 10000, 1024,
- strncmp(GetParam(), "dpdk", 4) == 0);
- StressFactory *f = &factory;
- exec_events([f](Worker *worker) mutable {
- f->start(worker);
- });
- ASSERT_EQ(0, factory.message_left);
-}
-
-
-INSTANTIATE_TEST_CASE_P(
- NetworkStack,
- NetworkWorkerTest,
- ::testing::Values(
-#ifdef HAVE_DPDK
- "dpdk",
-#endif
- "posix"
- )
-);
-
-#else
-
-// Google Test may not support value-parameterized tests with some
-// compilers. If we use conditional compilation to compile out all
-// code referring to the gtest_main library, MSVC linker will not link
-// that library at all and consequently complain about missing entry
-// point defined in that library (fatal error LNK1561: entry point
-// must be defined). This dummy test keeps gtest_main linked in.
-TEST(DummyTest, ValueParameterizedTestsAreNotSupportedOnThisPlatform) {}
-
-#endif
-
-
-/*
- * Local Variables:
- * compile-command: "cd ../.. ; make ceph_test_async_networkstack &&
- * ./ceph_test_async_networkstack
- *
- * End:
- */
Code Review / stor4nfv.git / blobdiff