L5RDMA/many2OneBench.cpp at master · pfent/L5RDMA · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#include <iostream>
#include <thread>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <include/MulticlientRDMATransport.h>
#include <include/MulticlientTCPTransport.h>
#include <util/ycsb.h>
#include "rdma/Network.hpp"
#include "rdma/QueuePair.hpp"
#include "util/bench.h"
#include "rdma/RcQueuePair.h"
#include "rdma/UcQueuePair.h"
#include "rdma/UdQueuePair.h"
#include <unistd.h>

using namespace std;
using namespace l5::transport;

static constexpr uint16_t port = 1234;
static const char *ip = "127.0.0.1";
static constexpr auto MESSAGES = 1024 * 1024;

template<class Client, class Server>
void doRun(size_t clients, bool isClient) {
    if (isClient) {
        RandomString rand;
        char testdata[64];
        rand.fill(64, testdata);

        sleep(1);
        std::vector<std::thread> clientThreads;
        for (size_t c = 0; c < clients; ++c) {
            clientThreads.emplace_back([&] {
                auto client = Client();
                for (int i = 0;; ++i) {
                    try {
                        client.connect(ip, port);
                        break;
                    } catch (...) {
                        std::this_thread::sleep_for(20ms);
                        if (i > 10) throw;
                    }
                }

                std::vector<char> buf(64);

                for (size_t m = 0; m < MESSAGES; ++m) {
                    client.send(reinterpret_cast<const uint8_t *>(testdata), 64);
                    client.receive(buf.data(), 64);

                    for (size_t i = 0; i < 64; ++i) {
                        if (testdata[i] != buf[i]) throw runtime_error("NEQ");
                    }
                }
            });
        }
        for (auto &t : clientThreads) {
            t.join();
        }
    } else {
        auto server = Server(to_string(port));
        for (size_t i = 0; i < clients; ++i) {
            server.accept();
        }

        std::vector<uint8_t> buf(64);
        bench(MESSAGES * clients, [&] {
            for (size_t m = 0; m < MESSAGES * clients; ++m) {
                auto client = server.receive(buf.data(), 64);
                server.send(client, buf.data(), 64);
            }
        });
    }
}

int main(int argc, char **argv) {
    if (argc < 3) {
        cout << "Usage: " << argv[0] << " <client / server> <#clients> <(optional) 127.0.0.1>" << endl;
        return -1;
    }
    const auto isClient = argv[1][0] == 'c';
    const auto clients = atoi(argv[2]);
    if (argc > 3) {
        ip = argv[3];
    }

    cout << "clients, messages, seconds, msgps, user, kernel, total\n";
    if (!isClient) {
        cout << clients << ", ";
    }
    doRun<MulticlientTCPTransportClient, MulticlientTCPTransportServer>(clients, isClient);
}