linux - 套接字传入连接不能将元素 push_back 并发到全局定义的 std::vector

Question

我是套接字编程的新手，此刻我遇到了一个我无法解决的问题。我从多个来源了解到，C++ 标准模板 (STL) 容器不是线程安全的，因此作为程序员必须强加一种机制，确保多个线程不会修改数据一个容器。

例如，Thread safety std::vector push_back and reserve

我用过std::mutex类以确保在编程时没有人同时在同一个容器中写入数据threads .但是，当我使用 sockets 时，这对我不起作用。 .

假设我有 4 个客户端，每个客户端按以下顺序向服务器发送数据 ( int ):

client_0: 4
client_1: 8
client_2: 5
client_4: 7

观察以下简单服务器的代码:

#define PORT 60000

#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <vector>
#include <string>
#include <iostream>
#include <mutex>

using namespace std;

vector<int> inputQueue; //<--------!
mutex mtx; //<---------------------!

void printVector(vector<int> input) {
    cout << "inputQueue: [";
    for (unsigned int i = 0; i < input.size(); i++ ) {
        if (i != input.size() - 1)
            cout << input[i] << ", ";
        else
            cout << input[i];
    }
    cout << "]." << endl;
}

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

    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
        perror("setsockopt");
        exit(EXIT_FAILURE);
    }
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons( PORT );
    if (bind(server_fd, (struct sockaddr *)&address, sizeof(address))<0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }
    if (listen(server_fd, 10) < 0) {
        perror("listen");
        exit(EXIT_FAILURE);
    }
    while(1) {
        char buffer[4];
        if ((client_fd = accept(server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen))<0) {
            perror("accept");
            exit(EXIT_FAILURE);
        }
        if (!fork()) {
            recv(client_fd, buffer, 4, MSG_WAITALL);
            int receivedInt = int(
                    (unsigned char)(buffer[0]) << 24 |
                    (unsigned char)(buffer[1]) << 16 |
                    (unsigned char)(buffer[2]) << 8 |
                    (unsigned char)(buffer[3])
            );
            mtx.lock(); //<-------------------------------------!
            inputQueue.push_back(receivedInt); //<--------------!
            cout << "Client context. Integer registered: " << receivedInt << ": inputQueue length is " << inputQueue.size() << endl;
            printVector(inputQueue); //<------------------------!
            mtx.unlock(); //<-----------------------------------!
            close(server_fd); close(client_fd);
        }
        cout << "Server context: inputQueue length is " << inputQueue.size() << endl;
        printVector(inputQueue);
    }
    return 0;
}

服务器必须确保它们以相同的顺序接收数据，并将它们各自的数据注册到一个整数向量中，即 std::vector<int> inputQueue , 使用 push_back()方法，所以 inputQueue = {4, 8, 5, 7}在客户端接收到所有数据结束时。

我必须澄清 inputQueue是一个全局变量，当开始执行服务器时，它不包含元素，但它们是在客户端注册时添加的。

问题是没有客户端在inputQueue中注册元素。请注意，在以下代码中，取决于您放置 cout << 的位置指令，你可以看到inputQueue尺寸不同。这表明在客户端的上下文中，每个客户端都会覆盖 inputQueue 的第一个元素。 ，但在它之外，没有任何客户端能够在 inputQueue 中注册单个元素。 .

显然，每个套接字都有自己的 inputQueue 副本，所以当它被销毁时，inputQueue 的修改副本也被摧毁。

输出如下:

Server context: inputQueue length is 0
inputQueue: [].
Client context. Integer registered: 4: inputQueue length is 1
inputQueue: [4].
Server context: inputQueue length is 1
inputQueue: [4].
Server context: inputQueue length is 0
inputQueue: [].
Client context. Integer registered: 8: inputQueue length is 1
inputQueue: [8].
Server context: inputQueue length is 0
inputQueue: [].
Server context: inputQueue length is 1
inputQueue: [8].
Client context. Integer registered: 5: inputQueue length is 1
inputQueue: [5].
Server context: inputQueue length is 1
inputQueue: [5].
Server context: inputQueue length is 0
inputQueue: [].
Client context. Integer registered: 7: inputQueue length is 1
inputQueue: [7].
Server context: inputQueue length is 1
inputQueue: [7].

有没有人知道为什么会发生这种情况以及他们如何解决？我希望你可以帮助我。谢谢

Answer 1

if (!fork()) {

fork() 使用自己的虚拟内存地址空间创建一个全新的独立进程。显然，所示代码期望子进程和原始进程通过同一个对象(即由互斥锁锁定的向量)进行交互。

事实并非如此。您现在有两个完全独立的进程。这与同时或快速连续运行程序两次没有什么不同。您是否希望程序的两个运行副本以某种方式共享相同的向量和互斥体？当然不是。

相反，您要做的是使用 std::thread 在同一进程中创建一个新的执行线程。您的 C++ 书籍应该包含有关如何使用 std::thread 创建新执行线程的更多信息。

此外，即使您将 fork() 替换为类似的执行线程:仍然不能解决这里的所有问题。您还需要正确处理多个执行线程之间的同步。具体来说:在其他执行线程尝试 printVector 其内容之前，无法保证新的执行线程会将某些内容插入向量中。在原始执行线程进入 printVector 之前，新的执行线程可以设法做到这一点。或者它可能没有，并且 printVector 找到了一个完全空的向量，因为另一个执行线程没有足够快地设法将某些东西插入其中。您现在有两个完全独立的执行线程同时运行，并且您无法保证哪个线程先执行什么操作。

您甚至可以在每次运行所示程序的多线程版本时得到不同的结果(您可能会这样做)。

当您准备好开始解决这个新问题时，您的 C++ 书籍将解释如何使用条件变量和互斥体来正确实现多线程同步。不幸的是，这不是一个可以在 stackoverflow.com 上的简短回答中完全涵盖的主题，但在您的 C++ 书中应该有几个专门的章节，您可以在其中找到更多信息。

附言您的输出在输入队列中显示任何内容的唯一原因是因为当子进程退出其 if 语句并最终自行调用 printVector 时，没有什么可以阻止子进程继续执行程序。它不是来自父进程。每个子进程最终打印它自己插入到自己的向量中的值。