c - select() 未检测到传入数据

Question

目标: N 个节点(运行在不同的机器上)应该通过相互建立 TCP 连接来相互通信。发送和接收消息是由进程创建的 2 个线程完成的。最初，主进程将所有节点相互连接，创建 2 个线程，并为其提供一个文件描述符列表，线程可以使用该列表来发送和接收数据。下面的结构由主进程填充并传递给线程。

typedef struct
{
    char hostName[MAXIMUM_CHARACTERS_IN_HOSTNAME];  /* Host name of the node */
    char portNumber[MAXIMUM_PORT_LENGTH];           /* Port number of the node */
    char nodeId[MAXIMUM_NODE_ID_LENGTH];            /* Node ID of the node */
    int socketFd;                                   /* Socket file descriptor */
    int socketReady;        /* Flag to indicate if socket information is filled */
}SNodeInformation;

PS:socketFd是由accept()或socket()接收的套接字描述符，具体取决于连接的建立方式(监听来自节点的连接或连接到节点)。

使用大小为 MAX_NUM_OF_NODES 的 SNodeInformation 数组。

发送线程遍历nodeInformation并向所有节点发送消息“Hello”，如下所示。

void *sendMessageThread(void *pNodeInformation) {

    int i;
    int ownNodeId;
    int bytesSent = 0;
    char ownHostName[MAXIMUM_CHARACTERS_IN_HOSTNAME];

    SNodeInformation *nodeInformation = (SNodeInformation *) pNodeInformation;
    SNodeInformation *iterNodeInformation;

    printf("SendMessageThread: Send thread created\n");

    if(gethostname(ownHostName, MAXIMUM_CHARACTERS_IN_HOSTNAME) != 0) {
        perror("Error: sendMessageThread, gethostname failed\n");
        exit(1);
    }

    for(i=0, iterNodeInformation=nodeInformation ; i<MAXIMUM_NUMBER_OF_NODES ; i++, iterNodeInformation++) {
        if(strcmp((const char*) iterNodeInformation->hostName, (const char*) ownHostName) != 0)  {
            /* Send message to all nodes except yourself */
            bytesSent = send(iterNodeInformation->socketFd, "Hello", 6, 0);

            if(bytesSent == -1) {
                printf("Error: sendMessageThread, sending failed, code: %s FD %d\n",     strerror(errno), iterNodeInformation->socketFd);
            }
        }
    }

    pthread_exit(NULL);
}

接收线程遍历nodeInformation，设置文件描述符集并使用select等待传入数据，如下所示。

void *receiveMessageThread(void *pNodeInformation)
{
    int i;
    int fileDescriptorMax = -1;
    int doneReceiving = 0;
    int numberOfBytesReceived = 0;
    int receiveCount = 0;
    fd_set readFileDescriptorList;
    char inMessage[6];

    SNodeInformation *nodeInformation = (SNodeInformation *) pNodeInformation;
    SNodeInformation *iterNodeInformation;

    printf("ReceiveMessageThread: Receive thread created\n");

    /* Initialize the read file descriptor */
    FD_ZERO(&readFileDescriptorList);

    for(i=0, iterNodeInformation=nodeInformation ; i<MAXIMUM_NUMBER_OF_NODES ; i++, iterNodeInformation++) {
        FD_SET(iterNodeInformation->socketFd, &readFileDescriptorList);

        if(iterNodeInformation->socketFd > fileDescriptorMax) {
            fileDescriptorMax = iterNodeInformation->socketFd;
        }
    }

    printf("ReceiveMessageThread: fileDescriptorMax:%d\n", fileDescriptorMax);

    while(!doneReceiving) {
        if (select(fileDescriptorMax+1, &readFileDescriptorList, NULL, NULL, NULL) == -1) {
            perror("Error receiveMessageThread, select failed \n");
            return -1;
        }

        for(i=0 ; i<fileDescriptorMax ; i++) {
            if (FD_ISSET(i, &readFileDescriptorList)) {
                /* Check if any FD was set */
                printf("ReceiveThread: FD set %d\n", i);

                /* Receive data from one of the nodes */
                if ((numberOfBytesReceived = recv(i, &inMessage, 6, 0)) <= 0) {
                    /* Got error or connection closed by client */
                    if (numberOfBytesReceived == 0) {
                        /* Connection closed */
                        printf("Info: receiveMessageThread, node %d hung up\n", i);
                    }
                    else {
                        perror("Error: receiveMessageThread, recv FAILED\n");
                    }

                    close(i);

                    /* Remove from Master file descriptor set */
                    FD_CLR(i, &readFileDescriptorList); 
                    doneReceiving = 1;
                }
                else {
                    /* Valid data from a node */
                    inMessage[6] = '\0';

                    if(++receiveCount == MAXIMUM_NUMBER_OF_NODES-1) {
                        doneReceiving = 1;
                    }

                    printf("ReceiveThread: %s received, count: %d\n", inMessage, rece    iveCount);
                }
            }
        }
    }
    pthread_exit(NULL);
}

预期输出:我尝试只使用 2 个进程，P1(首先启动)和 P2 在 machine1 上运行，另一个在 machine2 上运行。机器中的两个进程都应该首先连接，然后线程应该发送和接收消息“Hello”并退出。

观察到的输出:P1 能够发送消息，P2(接收线程)能够接收消息“Hello”。但是P1(接收线程)无法从P2(发送线程)获取消息。两台机器中的应用程序代码相同，但每次，首先启动的进程都不会从其他进程获取消息。我添加了一个打印来检查是否设置了某个文件描述符，但我没有在 P1 中看到它，而只在 P2 中看到它。接收过程中发送并没有失败，返回了6。我检查了文件描述符的最大值，它是正确的。

如果我先启动 P2，然后启动 P1，那么我可以看到 P1 接收到来自 P2 的消息并存在，而 P2 则无限等待来自 P1 的消息。

我不确定问题是由于套接字描述符使用不正确还是由于线程所致？

Answer 1

两个问题:

1 对正在设置的文件描述符进行循环测试，不包括放入该集合中的所有文件描述符。 (此编程错误预计是OP中描述的故障的原因。)

2 传递给 select() 的文件描述符集被 select() 修改，因此需要在 select() 之前重新初始化该集合再次。 (只有从多个套接字接收数据时，编程错误才会值得注意。)

请参阅 OP 代码的以下 mod/s:

void *receiveMessageThread(void *pNodeInformation)
{
    ...

    printf("ReceiveMessageThread: Receive thread created\n");

    while(!doneReceiving) {
        /* Initialize the read-set of file descriptors */

        /* Issue 2 fixed from here ... */
        FD_ZERO(&readFileDescriptorList);

        for(i=0, iterNodeInformation=nodeInformation ; i<MAXIMUM_NUMBER_OF_NODES ; i++, iterNodeInformation++) {
            FD_SET(iterNodeInformation->socketFd, &readFileDescriptorList);

            if (iterNodeInformation->socketFd > fileDescriptorMax) {
                fileDescriptorMax = iterNodeInformation->socketFd;
            }
        }
        /* ... up to here. */

        printf("ReceiveMessageThread: fileDescriptorMax:%d\n", fileDescriptorMax);

        if (select(fileDescriptorMax+1, &readFileDescriptorList, NULL, NULL, NULL) == -1) {
            perror("Error receiveMessageThread, select failed \n");
            return -1;
        }

        for(i=0 ; i <= fileDescriptorMax ; i++) { /* Issue 1 fixed here. */
            ...