c++ - 通过 TCP 进行 PvP 通信的问题

Question

由于需要通过 Internet 连接实现两个进程之间的通信，我刚刚开始使用 TCP(以及所有相关的库)。我的代码可以工作，但考虑到网络延迟和带宽，与我(可能是由于缺乏经验)所期望的相比，它非常慢。此外，我确信代码还有很多其他问题，它使用的是 UNIX 套接字 API。除非有很好的理由，否则我不希望在我的项目中使用大型库(例如 Boost)。
我包括一个最小的工作示例。尽管我尽最大努力缩短它，但它还是相当长。但是，我认为大多数问题应该在第一个文件(tcp_helpers.h)中，该文件仅由客户端和服务器主程序以相当明显的方式使用。那里的功能没有完全优化，但我很难相信这是问题所在，而不是可能是逻辑中的一些基本缺陷 .
我也想问一些问题 与问题相关:

对于网络性能，我应该担心使用 IPv4 还是 IPv6？会不会是我的网络不喜欢使用 IPv4 并影响性能？

由于 Socket API 模拟了一个流，我认为无论是对较小的数据 block 还是对大块的数据多次调用 send() 都没有关系。但也许它确实很重要，并且使用较小的 block (我每次都为我的自定义协议(protocol)头和数据分别调用发送)会导致问题？

假设两方在发送下一条消息之前通过网络进行通信，对接收到的数据进行处理(如我的示例中所做的那样)。如果这两个进程在 localhost 上花费 x 时间来完成，那么它们在真实网络上花费的时间永远不会超过 (2*x + (网络开销)) ，对吧？如果 x 很小，使计算(即在发送下一条消息之前工作)更快将无济于事，对吧？

我的示例程序在 localhost 上运行时大约需要 4 毫秒，而在我正在使用的本地(大学)网络上运行时需要 >0.7 秒。本地网络的 ping 时间(用 ping 测量)为( min/avg/max/mdev [ms] = 4.36/97.6/405./86.3 )和带宽(用 iperf 测量)约为 70Mbit/s .在网络上运行示例程序时，我得到(使用 wireshark 对相关端口进行过滤测量)190 个数据包，平均吞吐量为 172kB/s，平均数据包大小约为 726 字节。这是现实的吗？对我来说，考虑到这些网络参数，我的程序似乎应该快得多，尽管 ping 时间相当长。

查看示例程序生成的实际网络流量，我开始思考 TCP 的所有“特性”，这些“特性”是在后台完成的。我在某处读到许多程序同时使用多个套接字“以提高速度”。这对这里有帮助吗，例如使用两个套接字，每个套接字仅用于单向通信？特别是，也许以某种方式减少 ack 数据包的数量可以提高性能？

我将消息/标题作为结构编写的方式(至少)有两个我已经知道的大问题。首先，我不强制执行网络字节顺序。如果一个通信方使用 big-endian 而另一个使用 little-endian，则此程序将无法运行。此外，由于结构填充(参见 catb.org/esr/structure-packing/ )，结构的大小可能会因实现或编译器而异，这也会破坏我的程序。我可以添加类似 (for gcc) __attribute__((__packed__))到结构，但这会使它非常特定于编译器，甚至可能导致效率低下。是否有处理这个问题的标准方法(我见过一些关于手动对齐的东西)？ (也许我正在寻找错误的关键字。)

// tcp_helpers.h. // NOTE: Using this code is very ill-advised.
#include <iostream>
#include <string>
#include <sstream>
#include <vector>
#include <unistd.h>  // POSIX specific
#include <sys/socket.h> // POSIX specific
#include <netinet/in.h> // POSIX specific
#include <arpa/inet.h> // POSIX specific
#include <cerrno>  // for checking socket error messages
#include <cstdint> // for fixed length integer types

//////////////////// PROFILING ///////////////////
#include <chrono>
static auto start = std::chrono::high_resolution_clock::now();
void print_now(const std::string &message) {
    auto t2 = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> time_span = t2 - start;
    std::cout << time_span.count() << ": " << message << std::endl;
}
//////////////////// PROFILING ///////////////////

struct TCPMessageHeader {
    uint8_t protocol_name[4];
    uint32_t message_bytes;
};

struct ServerSends {
    uint16_t a;
    uint32_t b;
    uint32_t c;
};

typedef uint8_t ClientSends;

namespace TCP_Helpers {
    template<typename NakedStruct>
    void send_full_message(int fd, TCPMessageHeader header_to_send, const std::vector<NakedStruct> &structs_to_send) {
        print_now("Begin send_full_message");
        if (header_to_send.message_bytes != sizeof(NakedStruct) * structs_to_send.size()) {
            throw std::runtime_error("Struct vector's size does not match the size claimed by message header");
        }
        int bytes_to_send = sizeof(header_to_send);
        int send_retval;
        while (bytes_to_send != 0) {
            send_retval = send(fd, &header_to_send, sizeof(header_to_send), 0);
            if (send_retval == -1) {
                int errsv = errno;  // from errno.h
                std::stringstream s;
                s << "Sending data failed (locally). Errno:" << errsv << " while sending header.";
                throw std::runtime_error("Sending data failed (locally)");
            }
            bytes_to_send -= send_retval;
        }
        bytes_to_send = header_to_send.message_bytes;
        while (bytes_to_send != 0) {
            send_retval = send(fd, &structs_to_send[0], sizeof(NakedStruct) * structs_to_send.size(), 0);
            if (send_retval == -1) {
                int errsv = errno;  // from errno.h
                std::stringstream s;
                s << "Sending data failed (locally). Errno:" << errsv <<
                  " while sending data of size " << header_to_send.message_bytes << ".";
                throw std::runtime_error(s.str());
            }
            bytes_to_send -= send_retval;
        }
        print_now("end send_full_message.");
    }

    template<typename NakedStruct>
    std::vector<NakedStruct> receive_structs(int fd, uint32_t bytes_to_read) {
        print_now("Begin receive_structs");
        unsigned long num_structs_to_read;
        // ensure expected message is non-zero length and a multiple of the SingleBlockParityRequest struct
        if (bytes_to_read > 0 && bytes_to_read % sizeof(NakedStruct) == 0) {
            num_structs_to_read = bytes_to_read / sizeof(NakedStruct);
        } else {
            std::stringstream s;
            s << "Message length (bytes_to_read = " << bytes_to_read <<
              " ) specified in header does not divide into required stuct size (" << sizeof(NakedStruct) << ").";
            throw std::runtime_error(s.str());
        }
        // vector must have size > 0 for the following pointer arithmetic to work 
        // (this method must check this in above code).
        std::vector<NakedStruct> received_data(num_structs_to_read);
        int valread;
        while (bytes_to_read > 0)  // todo need to include some sort of timeout?!
        {
            valread = read(fd,
                           ((uint8_t *) (&received_data[0])) +
                           (num_structs_to_read * sizeof(NakedStruct) - bytes_to_read),
                           bytes_to_read);
            if (valread == -1) {
                throw std::runtime_error("Reading from socket file descriptor failed");
            } else {
                bytes_to_read -= valread;
            }
        }
        print_now("End receive_structs");
        return received_data;
    }

    void send_header(int fd, TCPMessageHeader header_to_send) {
        print_now("Start send_header");
        int bytes_to_send = sizeof(header_to_send);
        int send_retval;
        while (bytes_to_send != 0) {
            send_retval = send(fd, &header_to_send, sizeof(header_to_send), 0);
            if (send_retval == -1) {
                int errsv = errno;  // from errno.h
                std::stringstream s;
                s << "Sending data failed (locally). Errno:" << errsv << " while sending (lone) header.";
                throw std::runtime_error(s.str());
            }
            bytes_to_send -= send_retval;
        }
        print_now("End send_header");
    }

    TCPMessageHeader receive_header(int fd) {
        print_now("Start receive_header (calls receive_structs)");
        TCPMessageHeader retval = receive_structs<TCPMessageHeader>(fd, sizeof(TCPMessageHeader)).at(0);
        print_now("End receive_header (calls receive_structs)");
        return retval;
    }
}

// main_server.cpp
#include "tcp_helpers.h"

int init_server(int port) {
    int server_fd;
    int new_socket;
    struct sockaddr_in address{};
    int opt = 1;
    int addrlen = sizeof(address);
    // Creating socket file descriptor
    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        throw std::runtime_error("socket creation failed\n");
    }

    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
        throw std::runtime_error("failed to set socket options");
    }
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(port);
    // Forcefully attaching socket to the port
    if (bind(server_fd, (struct sockaddr *) &address, sizeof(address)) < 0) {
        throw std::runtime_error("bind failed");
    }
    if (listen(server_fd, 3) < 0) {
        throw std::runtime_error("listen failed");
    }
    if ((new_socket = accept(server_fd, (struct sockaddr *) &address, (socklen_t *) &addrlen)) < 0) {
        throw std::runtime_error("accept failed");
    }
    if (close(server_fd)) // don't need to listen for any more tcp connections (PvP connection).
        throw std::runtime_error("closing server socket failed");
    return new_socket;
}

int main() {
    int port = 20000;
    int socket_fd = init_server(port);
    while (true) {
        TCPMessageHeader rcv_header = TCP_Helpers::receive_header(socket_fd);
        if (rcv_header.protocol_name[0] == 0)   // using first byte of header name as signal to end
            break;
        // receive message
        auto rcv_message = TCP_Helpers::receive_structs<ClientSends>(socket_fd, rcv_header.message_bytes);
        for (ClientSends ex : rcv_message) // example "use" of the received data that takes a bit of time.
            std::cout <<  static_cast<int>(ex) << " ";
        std::cout << std::endl << std::endl;

        // send a "response" containing 1000 structs of zeros
        auto bunch_of_zeros = std::vector<ServerSends>(500);
        TCPMessageHeader send_header{"abc", 500 * sizeof(ServerSends)};
        TCP_Helpers::send_full_message(socket_fd, send_header, bunch_of_zeros);

    }
    exit(EXIT_SUCCESS);
}

// main_client.cpp
#include "tcp_helpers.h"

int init_client(const std::string &ip_address, int port) {
    int sock_fd;
    struct sockaddr_in serv_addr{};

    if ((sock_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        throw std::runtime_error("TCP Socket creation failed\n");
    }
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(port);
    // Convert IPv4 address from text to binary form
    if (inet_pton(AF_INET, ip_address.c_str(), &serv_addr.sin_addr) <= 0) {
        throw std::runtime_error("Invalid address/ Address not supported for TCP connection\n");
    }
    if (connect(sock_fd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
        throw std::runtime_error("Failed to connect to server.\n");
    }
    return sock_fd;
}

int main() {
    // establish connection to server and get socket file descriptor.
    int port = 20000;
    int socket_fd = init_client("127.0.0.1", port);
    for (int i = 0; i < 20; ++i) {  // repeat sending and receiving random data
        // send a message containing 200 structs of zeros
        auto bunch_of_zeros = std::vector<ClientSends>(250);
        TCPMessageHeader send_header{"abc", 250 * sizeof(ClientSends)};
        TCP_Helpers::send_full_message(socket_fd, send_header, bunch_of_zeros);

        // receive response
        TCPMessageHeader rcv_header = TCP_Helpers::receive_header(socket_fd);
        auto rcv_message = TCP_Helpers::receive_structs<ServerSends>(socket_fd, rcv_header.message_bytes);
        for (ServerSends ex : rcv_message) // example "use" of the received data that takes a bit of time.
            std::cout << ex.a << ex.b << ex.c << " ";
        std::cout << std::endl << std::endl;
    }
    auto end_header = TCPMessageHeader{}; // initialized all fields to zero. (First byte of name == 0) is "end" signal.
    TCP_Helpers::send_header(socket_fd, end_header);
    exit(EXIT_SUCCESS);
}

Answer 1

您关心延迟，所以首先要做的是始终确保禁用 Nagle 算法，使用 TCP_NODELAY .另一个答案显示了如何。
当您想要相反时，Nagle 的算法以延迟为代价显式优化吞吐量。

I also want to ask some questions relevant to the problem:

我希望你不会 - 它使这个问题成为一个完全回答的怪物。

1. For network performance, should I worry about using IPv4 vs IPv6? Could it be that my network dislikes the use of IPv4 somehow and penalized performance?

没有明显的理由它应该很重要，如果有的话，v4 堆栈可能会得到更好的优化，因为它(在撰写本文时)仍然被更频繁地使用。
但是，如果你想测试，你已经在使用 iperf - 因此，请自行比较网络上的 v4 和 v6 性能。如果您不理解结果，请提出一个单独的问题。

2. Since the Socket API emulates a stream, I would think it does not matter if you call send() multiple times on smaller chunks of data or once on a big chunk. But perhaps it does matter and doing it with smaller chunks (I call send for my custom protocol header and the data separately each time) leads to issues?

当然，它有所作为。
首先，考虑到网络堆栈需要以某种方式决定如何将该流划分为数据包。使用 Nagle 的算法，这是通过等待计时器(或下一个 ack，这也是它与客户端的延迟 ack 计时器交互的原因)来完成的。与 TCP_NODELAY ，每次调用 send()通常会产生自己的数据包。
由于数据包具有 header ，因此在更多数据包中发送相同数量的用户数据会占用更多网络带宽。默认情况下，延迟和吞吐量效率之间的权衡由 Nagle 算法和延迟确认计时器处理。如果您禁用 Nagle 算法，您可以手动控制权衡，以便您可以做最适合您的程序的事情 - 但这是一种权衡，需要一些思考和努力。
其次，调用 send()本身不是免费的。系统调用比用户空间库调用更昂贵。

3. Suppose that two parties communicate over a network doing work on the received data before sending their next message (as is done in my example). If the two processes take x amount of time on localhost to finish, they should never take longer than (2*x + (network overhead)) on the real network, right? If x is small, making the computations (i.e. work before sending next message) go faster will not help, right?

您的估计看起来合理，但是 - 时间就是时间。
仅仅因为总延迟由网络控制，并不意味着对本地计算的加速没有影响。
如果使计算速度提高 1ns，即使网络延迟为 10ms，总体上仍然快 1ns。您对网络延迟的直接控制也较少，因此可能需要尽可能节省时间。

4. ... To me it seems like my program should be much faster given these network parameters, despite the fairly high ping time.

是的，应该使用 TCP_NODELAY 再试一次和 send() 的正确数量来电。

5. ... Could this help here, for example using two sockets, each for just one-way communication? In particular, maybe somehow reducing the number of ack packets could help performance?

由于延迟的确认计时器，确认对于对称双向通信基本上是免费的。您的 Wireshark 调查应该已经证明了这一点。它们对于单向流不是免费的，因此使用两个半双工套接字要糟糕得多。

6. The way I'm writing messages/headers as structs has (at least) two big problems that I already know. First, I do not enforce network byte order. If one communicating party uses big-endian and the other little-endian, this program will not work. Furthermore, due to struct padding (see [catb.org/esr/structure-packing/][1]), the sizes of the structs may vary between implementations or compilers, which would also break my program. I could add something like (for gcc) __attribute__((__packed__)) to the structs but that would make it very compiler specific and perhaps even lead to inefficiency. Are there standard ways of dealing with this issue (I've seen something about aligning manually)? (Maybe I'm looking for the wrong keywords.)

处理这些问题的标准方法有很多，没有什么类似于单一标准。

Endianness - 最简单的方法是采用当前主机的 native 字节顺序，并使用它。如果您连接具有不同顺序的主机，则需要做额外的工作，但这很可能永远不会发生，您会推迟额外的工作。

填充:
使用 __attribute__((packed))或 #pragma pack当然会导致一些效率低下，但它很方便。请注意，对未对齐字段的指针和引用不需要正常工作，因此这些结构并不是真正通用的。
手动填充是可行的，但很乏味。您只需要弄清楚 native 布局结构中每个字段的实际对齐方式，然后插入填充字节，以便没有其他实现可以以不同方式布局。您也许可以使用 alignas说明符以更好的方式实现相同的目标。
免费获得大部分对齐的一种简单方法是始终从最大到最小排列字段(大小和对齐，但它们通常是相关的)。

通常，序列化是将 native 数据转换为有线格式(反之为反序列化)的名称。这涵盖了从将数据转换为/从 JSON 字符串转换为非常广泛的兼容性到发送精确布局的二进制数据的整个范围。您的延迟限制使您处于最后。