Как подготовить пакет TCP заранее, чтобы уменьшить задержку записи? - PullRequest
0 голосов
/ 13 сентября 2018

Это продолжение Почему задержка записи TCP хуже при чередовании работы?

В этом вопросе мы обнаружили, что при интенсивной работе ЦП между write вызовами к сокету TCP задержка write увеличивается в 5 раз. Это связано с тем, что в отсутствие интенсивной работы ЦП исходящие байты упаковываются перед передачей в виде пакетов TCP на устройство. Интенсивная загрузка процессора позволяет очищать буфер отправки, так что каждый новый write запускает полную конструкцию пакета, которая включает служебные данные. (В качестве дополнительного вопроса, что именно влечет за собой эта конструкция пакета? Заголовок TCP составляет <20 байтов, поэтому я не уверен, откуда на самом деле происходит большая часть служебных данных.) </p>

В свете этого я ищу способ «подготовить» следующий пакет. Это было бы полезно в среде, чувствительной к задержке, когда вы знаете, что в какой-то момент в будущем вам понадобится отправить пакет, поэтому вы хотите покончить с издержками на создание пакета раньше.

Моя первая идея состояла в том, чтобы установить нижнюю отметку SO_SNDLOWAT на 2, а затем подготовить пакет, не отправляя его с вызовом write только одним байтом. Теоретически, SO_SNDLOWAT должен предотвратить попадание этого пакета на устройство, поэтому, когда я измеряю задержку последующего write, передающего фактические данные, он должен быть быстрым. Но это вовсе не уменьшает задержки (я несколько скептически отношусь к тому, что SO_SNDLOWAT делает то, что я ожидаю).

Вот мой код сервера:

// Server side C/C++ program to demonstrate Socket programming
// #include <iostream>
#include <boost/timer.hpp>
#include <ctime>
#include <sched.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <string.h>
#include <unistd.h>
#include <time.h>

// Function to count clock cycles
__inline__ uint64_t rdtsc(void)
{
    uint32_t lo, hi;
    __asm__ __volatile__ (
            "xorl %%eax,%%eax \n        cpuid"
            ::: "%rax", "%rbx", "%rcx", "%rdx");
    __asm__ __volatile__ ("rdtsc" : "=a" (lo), "=d" (hi));
    return (uint64_t)hi << 32 | lo;
}

// Set up some blocking work.
bool isPrime(int n) {
    if (n < 2) {
        return false;
    }

    for (int i = 2; i < n; i++) {
        if (n % i == 0) {
            return false;
        }
    }

    return true;
}

// Compute the nth largest prime. Takes ~1 sec for n = 10,000
int getPrime(int n) {
    int numPrimes = 0;
    int i = 0;
    while (true) {
        if (isPrime(i)) {
            numPrimes++;
            if (numPrimes >= n) {
                return i;
            }
        }
        i++;
    }
}

int main(int argc, char const *argv[])
{
    int server_fd, new_socket, valread;
    struct sockaddr_in address;
    int opt = 1;

    // Low water mark for socket
    int lowat = 2;

    int lowat2;
    socklen_t optlen;
    int addrlen = sizeof(address);
    int result;

    // Create socket for TCP server
    server_fd = socket(AF_INET, SOCK_STREAM, 0);

    setsockopt(server_fd, SOL_SOCKET, SO_SNDLOWAT, &lowat, sizeof(lowat));

    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(8080);

    bind(server_fd, (struct sockaddr *)&address, sizeof(address));

    listen(server_fd, 3);

    // Accept one client connection
    new_socket = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen);
    setsockopt(new_socket, SOL_SOCKET, SO_SNDLOWAT, &lowat, sizeof(lowat));

    // Check that SO_SNDLOWAT was updated
    getsockopt(new_socket, SOL_SOCKET, SO_SNDLOWAT, &lowat2, &optlen);
    printf("New lowat value: %d\n", lowat2);

    char sendBuffer[1] = {0};
    int primes[20] = {0};

    int N = 10;
    for (int i = 0; i < N; i++) {
        sendBuffer[0] = 97 + i;
        boost::timer t;

        auto start = rdtsc();
        write(new_socket, sendBuffer, 1);
        auto end = rdtsc();
        printf("%d mics (%llu cycles) to write\n", int(1e6 * t.elapsed()), end-start);

        // Inserting blocking work here slows down the `write` calls by a
        // factor of 5.
        primes[i] = getPrime(10000 + i);

        // Attempt to prep the next packet without sending it, by writing 'X'.
        sendBuffer[0] = 88;
        write(new_socket, sendBuffer, 1);
        primes[i] = getPrime(1000 + i);
    }

    // Prevent the compiler from optimizing away the prime computation.
    printf("prime: %d\n", primes[8]);
}

И код клиента:

// Server side C/C++ program to demonstrate Socket programming
// #include <iostream>
#include <unistd.h>
#include <stdio.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <string.h>
#include <unistd.h>

int main(int argc, char const *argv[])
{
    int sock, valread;
    struct sockaddr_in address;
    int opt = 1;
    int addrlen = sizeof(address);

    // We'll be passing uint32's back and forth
    unsigned char recv_buffer[1024] = {0};

    // Create socket for TCP server
    sock = socket(AF_INET, SOCK_STREAM, 0);

    // Set TCP_NODELAY so that writes won't be batched
    setsockopt(sock, SOL_SOCKET, TCP_NODELAY, &opt, sizeof(opt));

    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(8080);

    // Accept one client connection
    if (connect(sock, (struct sockaddr *)&address, (socklen_t)addrlen) != 0) {
        throw("connect failed");
    }

    int N = 10;
    int loc[N+1];
    int nloc, curloc;
    for (nloc = curloc = 0; curloc < N; nloc++) {
        int n = read(sock, recv_buffer + curloc, sizeof recv_buffer-curloc);
        if (n <= 0) {
                break;
        }
        curloc += n;
        loc[nloc] = curloc;
        // usleep(100000);
    }

    int last = 0;
    for (int i = 0; i < nloc; i++) {
        printf("%*.*s ", loc[i] - last, loc[i] - last, recv_buffer + last);
        last = loc[i];
    }
    printf("\n");
}

Вывод:

New lowat value: 2
14 mics (31252 cycles) to write
25 mics (49088 cycles) to write
26 mics (55558 cycles) to write
26 mics (53618 cycles) to write
26 mics (54468 cycles) to write
28 mics (58382 cycles) to write

Удаление простых вычислений в целом уменьшает задержку write до ~ 5000 циклов (в 10 раз или более).

Мне интересно, если у меня что-то не так с моей реализацией SO_SNDLOWAT, или, наоборот, есть ли более чистый способ подготовки пакета.

Вывод клиента (где пробелы обозначают отдельные вызовы read) предполагает, что SO_SNDLOWAT не работает: a X b X c X d X e X.

Обновление: согласно предложению Джила, я пытался использовать флаг MSG_MORE, когда отправляю пакеты X в качестве сигнала для удержания на фактической записи устройства. Похоже, это работает (после того, как вторая блокировка работает <200 мс), клиент выводит <code>a Xb Xc Xd Xe Xf. Но нелогично, полезная нагрузка write с фактически становится медленнее (100 000 циклов против 50 000 циклов без MSG_MORE против 5000 циклов без блокировки работы). MSG_MORE код:

// Attempt to prep the next packet without sending it, by writing 'X'.
sendBuffer[0] = 88;
send(new_socket, sendBuffer, 1, MSG_MORE);
primes[i] = getPrime(1000 + i + 1);
...