0

Objective: 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 的消息。

我不确定问题是因为套接字描述符的使用不正确还是因为线程？

4

1 回答 1

1

两个问题：

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

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

请参阅以下 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. */
            ...
于 2012-10-07T16:06:38.413 回答