c++ - 使用 pthreads 重新排列数据包

Question

所以这是我第一次真正在这里提问，尽管我已经使用这个网站很多年了!

我的问题有点棘手。我正在尝试开发一个用于发送大文件的客户端服务器应用程序，使用 UDP 和我自己的错误检查和流量控制。现在，我开发了一个功能齐全的服务器和客户端。客户端请求特定文件，服务器开始发送。文件被部分读入缓冲区，以避免每次发送数据包时都必须读取文件的一小部分，从而节省处理时间。数据包由 1400 字节的实际数据 + 28 字节的 header (序列号、确认号、校验和等)组成。

所以我掌握了基础知识，一个简单的停止等待协议(protocol)。在发送下一个数据包之前发送数据包并接收确认。

为了能够实现更智能的流量控制算法，对于只有一些窗口的初学者，我必须在两个不同的线程中运行发送部分和接收确认部分。现在这就是我遇到问题的地方。这是我第一次使用线程，所以请多多包涵。

我的问题是从客户端的数据包写入的文件已损坏。好吧，当用一个小的 jpg 文件进行测试时，该文件只有 50% 的时间是损坏的，而当用一个 MP4 文件进行测试时，它总是损坏的!所以我想也许线程以某种方式重新排列了数据包发送的顺序？我使用的是序列号，所以问题一定发生在给数据包分配序列号之前...

我确信我拆分文件的部分是正确的，而且我在客户端重新组装它的部分也是正确的，因为我在尝试实现线程之前已经测试过它。还应该注意的是，我将代码的确切发送部分复制到发送线程中，这在将其放入线程之前也能正常工作。这也是为什么我只发布我的线程部分代码，因为这显然是造成问题的原因(并且因为项目的整个代码会占用大量空间)

我的发送线程代码:

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t condition_var = PTHREAD_COND_INITIALIZER;

static void *send_thread(void *){
if (file.is_open()) {
    while(!file.reachedEnd()){
        pthread_mutex_lock(& mutex);

        if(seq <= upperwindow) {

            int blocksize = file.getNextBlocksize();
            senddata = new unsigned char[blocksize + 28];
            Packet to_send;
            to_send.data = new char[blocksize];
            to_send.sequenceNumber = seq;
            to_send.ackNumber = 0;
            to_send.type = 55; // DATA

            file.readBlock(*to_send.data);
            createPacket(senddata, to_send, blocksize + 28);
            if (server.sendToClient(reinterpret_cast<char*>(senddata), blocksize + 28) == -1)
                perror("sending failed");

            incrementSequenceNumber(seq);

            /* free memory */
            delete [] to_send.data;
            delete [] senddata;

        }
        pthread_mutex_unlock(& mutex);
    }
    pthread_exit(NULL);
} else {
    perror("file opening failed!");
    pthread_exit(NULL);
}
}

我的接收ack线程代码:

static void *wait_for_ack_thread(void *){
while(!file.reachedEnd()){
    Packet ack;

    if (server.receiveFromClient(reinterpret_cast<char*>(receivedata), 28) == -1) {
        perror("error receiving ack");
    } else {
        getPacket(receivedata, ack, 28);
        pthread_mutex_lock(& mutex);
        incrementSequenceNumber(upperwindow);

        pthread_mutex_unlock(& mutex)

    }
}
pthread_exit(NULL);
}

非常感谢所有评论! :)

编辑:添加readBlock函数的代码:

void readBlock(char & in){

memcpy(& in, buffer + block_position, blocksize);
block_position = block_position + blocksize;
if(block_position == buffersize){
    buf_position ++;
    if(buf_position == buf_reads){
        buffersize = filesize % buffersize;
    }
    fillBuffer();
    block_position = 0;
}
if(blocksize < MAX_DATA_SIZE){
    reached_end = true;
    return;
}
if((buffersize - block_position) < MAX_DATA_SIZE){
    blocksize = buffersize % blocksize;

}

}

Answer 1

创建一个表示通信状态的数组。

0 表示未发送，或已发送但接收方报错。 1 表示发送。 2 表示已发送，已确认。

分配这个数组，并用互斥锁保护对它的访问。

发送线程在数组中保留两个指针——“已发送至”和“下一个应该发送”。这些由发送线程拥有。

ack 线程简单地获取 ack 数据包，锁定数组，并在状态上进行转换。

发送线程锁定数组，检查它是否可以推进“已发送到”指针(或者它是否应该重新发送旧的东西)。如果它注意到一个错误，它会减少指向它的“下一个应该发送”指针。

然后它会查看接下来是否应该发送内容。如果应该，它将节点标记为“正在发送”，解锁数组并发送它。

如果发送线程没有工作，并且发现无事可做，它会在超时时进入休眠状态，并且可能会被 ack 线程“唤醒”。

现在，请注意客户端可能会以错误的顺序接收由此发送的数据包，除非您将其限制为只有 1 个数据包在传输中。

连接状态数组不必是文字数组，但如果从它开始并在以后进行优化会更容易。

在接收端，您必须注意序列号，因为数据包可能会乱序到达那里。要对此进行测试，请编写一个故意以错误顺序发送数据包的服务器，并确保客户端设法将其正确拼接在一起。