c ++ udp recvfrom в конечном итоге зависает через удаленное соединение - PullRequest
Новая
       51

c ++ udp recvfrom в конечном итоге зависает через удаленное соединение

0 голосов
/ 30 июня 2019

Я пишу приложение UDP клиент-сервер на C ++. Приложение работает правильно, когда клиент и сервер работают на одном компьютере, однако, когда я запускаю свой клиент на своем ноутбуке и сервер на экземпляре ec2 в AWS, сервер получает 4 сообщения и затем начинает зависать на recvfrom. Клиент отправляет файл кусками по 500 байт, а сервер записывает их в файл. Сервер отправляет ack после получения каждого чанка. Тайм-аут клиента составляет 2 секунды, если он не получает ack до истечения времени ожидания, он отправляет предыдущий блок. Вот мой код клиента 

void Sender::send_file(const char* host) {
cout << "in send file\n";
int sock = 0;
struct sockaddr_in serv_addr;
size_t length = 8;
unsigned char res_buf[8] = {0};
if((sock = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
    std::cout << "\n Socket creation error \n";
    exit(EXIT_FAILURE);
}
struct timeval tv;
tv.tv_sec = timeout;
tv.tv_usec = 0;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);

// copy 0 into serv_addr members
memset(&serv_addr, '0', sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(port);
serv_addr.sin_addr.s_addr = inet_addr(host);

size_t file_size = buffer.size();
cout << "file size " << to_string(file_size) << "\n";
size_t file_pos = 0;
int next_iteration = 1;
bool establish = true;
while(file_pos < file_size - 1) {
    cout << "waiting for lock\n";
    std::unique_lock<std::mutex> lock(mrun);
    main_ready.wait(lock, [&next_iteration, this]{return next_iteration == current_iteration; });
    lock.unlock();
    cout << "acquired lock\n";
    ++next_iteration;
    Segment segment;
    vector<unsigned char> req_str;
    if(!establish) {
        if(file_pos + mss < file_size) {
            std::vector<unsigned char> file_chunk(buffer.begin() + file_pos, buffer.begin() + file_pos + mss);
            segment.data = file_chunk;
            file_pos += mss;
        }
        else {
            std::vector<unsigned char> file_chunk(buffer.begin() + file_pos, buffer.end());
            segment.data = file_chunk;
            file_pos = file_size;
        }
        segment.seq_num = file_pos - mss;
        req_str = segment.to_bytes(false, mss);
    }
    else {
        segment.seq_num = 0;
        req_str = segment.to_bytes(true, mss);
    }

    bool is_ack = false, timed_out = false;
    std::chrono::high_resolution_clock::time_point start_time, end_time;
    long duration = 0;
    unsigned char* req = req_str.data();
    size_t num_bytes = req_str.size();
    unsigned int len = 0;
    while(!is_ack) {
        cout << "about to send data " << std::to_string(segment.seq_num) << " bytes " << std::to_string(num_bytes) << "\n";
        ssize_t send_res = sendto(sock, req, num_bytes, 0, (const struct sockaddr *) &serv_addr, sizeof(serv_addr));
        printf("Sent to %s:%d\n", inet_ntoa(serv_addr.sin_addr), ntohs(serv_addr.sin_port));
        cout << "send res " << to_string(send_res) << " " << strerror(errno) << "\n";
        start_time = std::chrono::high_resolution_clock::now();
        bzero(res_buf, length);
        cout << "about to read ack\n";
        ssize_t block_sz = recvfrom(sock, res_buf, length, 0, (struct sockaddr *) &serv_addr, &len);
        cout << "read ack\n";
        end_time = std::chrono::high_resolution_clock::now();
        duration = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();
        cout << "block sz " << std::to_string(block_sz) << " duration " << std::to_string(duration) << std::endl;
        if(duration > timeout * 1000)
            timed_out = true;
        if(timed_out || errno == ETIMEDOUT || block_sz < 0) {
            std::cout << "time out/error occurred on read " << to_string(errno) << "\n";
            cout << strerror(errno) << "\n";
            bzero(res_buf, length);
            timed_out = false;
            continue;
        }
        //add_nulls(res_buf);
        is_ack = read_response(segment.seq_num, res_buf);
        cout << "is ack " << is_ack << "\n\n";
        bzero(res_buf, length);
    }
    establish = false;
    lock.lock();
    if(--worker_count == 0) {
        lock.unlock();
        iteration_complete.notify_one();
    }
}
cout << "Out of while loop\n";
close(sock);
files_sent.push_back(1);
}

Он отправляет ровно 508 байтов каждый раз, за ​​исключением 1-го сегмента, который составляет всего 10 байтов. Я печатаю результаты sendto после каждой итерации, которая всегда выглядит следующим образом 

Sent to 52.0.0.0:7735
send res 508 Undefined error: 0

* ** Обновление *** Как только сервер начинает зависать в recvfrom, и время ожидания клиента для ack истекает, клиент начинает непрерывно печатать следующую ошибку, поскольку он продолжает повторную отправку чанка. 

Sent to 52.0.0.0:7735
send res 508 Resource temporarily unavailable

Код сервера выглядит как 

 void Receiver::download_file() {
int server_fd;
struct sockaddr_in serv_addr, cli_addr;
int segment_size = 20;
umap seg_map;
ACK_Segment::init_static();
std::random_device rd;  //Will be used to obtain a seed for the random number engine
std::mt19937 gen(rd()); //Standard mersenne_twister_engine seeded with rd()
std::uniform_real_distribution<> dis(0.0, 1.0);
if((server_fd = socket(AF_INET, SOCK_DGRAM, 0)) == 0) {
    perror("socket failed");
    exit(EXIT_FAILURE);
}
memset(&serv_addr, 0, sizeof(serv_addr));
memset(&cli_addr, 0, sizeof(cli_addr));

serv_addr.sin_family = AF_INET;
serv_addr.sin_addr.s_addr = INADDR_ANY;
serv_addr.sin_port = htons(PORT);

// bind socket to port
//bzero(buffer, segment_size);
ssize_t block_sz = 0;
unsigned int len = sizeof cli_addr;
bool is_set_mss = true;
if(bind(server_fd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
    perror("bind failed");
    exit(EXIT_FAILURE);
}
//unsigned char* buffer = bvec.data();
vector<unsigned char> bvec(segment_size);
unsigned char* buffer = bvec.data();
bool is_resized = false;
while(true) {
    cout << "about to recvfrom\n";
    printf("Receiving from %s:%d\n", inet_ntoa(cli_addr.sin_addr), ntohs(cli_addr.sin_port));
    cout << "segment size " << to_string(segment_size) << " len " << to_string(len) << "\n";
    block_sz = recvfrom(server_fd, buffer, segment_size, 0, ( struct sockaddr *) &cli_addr, &len);
    cout << "block size " << to_string(block_sz) << "errno " << to_string(errno) << " " << strerror(errno) << "\n";
    if(block_sz > 0) {
        cout << "Received data - bytes " << std::to_string(block_sz) << "\n";
        printf("Received from %s:%d\n", inet_ntoa(cli_addr.sin_addr), ntohs(cli_addr.sin_port));
        double rand_val = dis(gen);
        if(is_set_mss) {
            seg_map = read_segment(bvec, block_sz, true);
            if(seg_map["checksum_valid"] == "true") {
                segment_size = mss + HEADER_SIZE;
                is_resized = true;
            }
        }
        else {
            if(block_sz == 0) {
                cout << "received no data\n";
                break;
            }
            seg_map = read_segment(bvec, block_sz, false);
        }
        if(seg_map["checksum_valid"] == "false")
            cout << "invalid checksum\n";
        else if(is_set_mss) {
            vector<unsigned char> ack = get_ack();
            //remove_nulls(ack);
            unsigned char* ack_bytes = ack.data();
            cout << "about to send ack\n";
            sendto(server_fd, ack_bytes, 8, 0, (const struct sockaddr *) &cli_addr, len);
            //send(new_socket, ack, 8, 0);
            is_set_mss = false;
        }
        else {
            if(seg_map["in_order"] == "true") {
                string data = seg_map["data"];
                std::ofstream out;
                out.open(file_name, std::ios_base::app);
                out << data;
                out.close();
        cout << "Done writing file chunk\n";
            }
            vector<unsigned char> ack = get_ack();
            unsigned char* ack_bytes = ack.data();
            sendto(server_fd, ack_bytes, 8, 0, (const struct sockaddr *) &cli_addr, len);
        }
        if(is_resized) {
            bvec.resize(segment_size);
            is_resized = false;
        }
        bvec.clear();
        buffer = bvec.data();
    cout << "Cleared buffer\n\n";
    }
    cout << "out of inner while\n";
}
}

Это печатает 

Receiving from 168.0.0.0:18832
segment size 508 len 16
block size 508 errno 0 Success

Однако он печатает это только 4 раза, затем зависает на recvfrom, в то время как клиент продолжает отправлять 508 байтов, а размер сегмента остается равным 508 в recvfrom. Зачем начинать работать, а потом внезапно останавливаться?

Обновление

Я поливал это на простейшем клиентском сервере udp, который я мог написать, вот клиент: 

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <vector>
#include <iostream>
#include <cstring>
#include <string>

using std::vector;
using std::cout;
using std::to_string;

int main() {
    int sock = 0;
    struct sockaddr_in serv_addr;
    size_t length = 11;
    unsigned char res_buf[11] = {0};
    if((sock = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
        std::cout << "\n Socket creation error \n";
        exit(EXIT_FAILURE);
    }
    struct timeval tv;
    tv.tv_sec = 2;
    tv.tv_usec = 0;
    setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);

    // copy 0 into serv_addr members
    memset(&serv_addr, '0', sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(7735);
    serv_addr.sin_addr.s_addr = inet_addr("52.71.240.175");

    vector<unsigned char> req(20);
    for(int i = 0; i < 20; i++)
        req[i] = 'A';
    unsigned char* req_str = req.data();
    size_t num_bytes = req.size();
    unsigned int len = sizeof serv_addr;
    while(true) {
        cout << "about to send data\n";
        ssize_t send_res = sendto(sock, req_str, num_bytes, 0, (const struct sockaddr *) &serv_addr, sizeof(serv_addr));
        bzero(res_buf, length);
        cout << "about to read ack\n";
        ssize_t block_sz = recvfrom(sock, res_buf, length, 0, (struct sockaddr *) &serv_addr, &len);
        cout << "received " << res_buf <<  "\n";
    }
}

А вот и сервер: 

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <vector>
#include <iostream>
#include <cstring>
#include <string>

using std::vector;
using std::cout;
using std::to_string;

int main() {
    int server_fd;
    struct sockaddr_in serv_addr, cli_addr;
    int segment_size = 20;
    memset(&serv_addr, 0, sizeof(serv_addr));
    memset(&cli_addr, 0, sizeof(cli_addr));

    if((server_fd = socket(AF_INET, SOCK_DGRAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }

    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = INADDR_ANY;
    serv_addr.sin_port = htons(7735);

    unsigned int len = sizeof cli_addr;
    if(bind(server_fd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }

    vector<unsigned char> bvec(segment_size);
    unsigned char* buffer = bvec.data();
    while(true) {
        cout << "about to receive from\n";
        ssize_t block_sz = recvfrom(server_fd, buffer, segment_size, 0, ( struct sockaddr *) &cli_addr, &len);
        if(block_sz > 0) {
            printf("Received from %s:%d\n", inet_ntoa(cli_addr.sin_addr), ntohs(cli_addr.sin_port));
            char* ack_bytes = "ack message";
            cout << "about to send ack\n";
            sendto(server_fd, ack_bytes, 11, 0, (const struct sockaddr *) &cli_addr, len);
        }
    }
}

И такое же поведение происходит, сервер получает 4 сообщения, но затем на пятом сообщении просто висит в recvfrom. Я пытался закрыть и снова открыть сокет в каждой итерации на сервере, но это не имело никакого эффекта.

1 Ответ

1 голос
/ 01 июля 2019

Мне пришлось закрыть и снова открыть сокет на стороне клиента после получения каждого подтверждения от сервера.Мне не нужно было закрывать и открывать на стороне сервера.Я не уверен, что это устраняет основную причину проблемы, с которой я столкнулся, но позволяет моему приложению работать правильно.

...