c - TCP server 可以同时处理两个不同的client写请求而不会互相阻塞

Question

我正在尝试编写一个可以处理两个不同客户端的 TCP 服务器。我有一个请求者和提供者客户。提供程序是多线程的，可以向服务器添加和删除新服务。每次添加或删除新服务时，它都应将其发送到服务器，服务器将打印更新。请求者客户端允许用户输入服务，然后检查服务器以查看该服务是否存在。

我遇到的问题是 recv() 函数。我在我的程序中调用了两次，一次是从生产者客户端读取，另一次是从请求者读取。问题是服务器只收到一条消息然后卡住。它应该在每次线程运行时更新。问题似乎正在发生，因为第二个 recv() 调用阻止了它，因为它正在等待请求者。我试图通过使用非阻塞事件标志 (MSG_DONTWAIT) 使第二个 recv() 调用成为非阻塞，但这并没有解决我的问题。

如何编写一个 TCP 服务器来处理两个不同的客户端写入请求并防止它们相互阻塞？我的代码如下。

客户 1- 提供者

#include <stdio.h> 
#include <string.h> 
#include <stdlib.h>  
#include <unistd.h>
#include <time.h> 
#include <sys/socket.h> 
#include <arpa/inet.h> 
#include <stdbool.h>
#include <pthread.h>

#define PORT 8080

int SIZE = 100;
int counter = 0;
int semaphore = 1; //set to false

struct values 
       {
       int serviceArray[100]; 
       int portArray[100];  
       }input;

void callServer()
{
struct sockaddr_in address; 
    int sock = 0, valread; 
    struct sockaddr_in serv_addr; 

    if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) 
    { 
        printf("\n Socket creation error \n"); 

    } 

    memset(&serv_addr, '0', sizeof(serv_addr)); 

    serv_addr.sin_family = AF_INET; 
    serv_addr.sin_port = htons(PORT); 

    // Convert IPv4 and IPv6 addresses from text to binary form 
    if(inet_pton(AF_INET, "127.0.0.1", &serv_addr.sin_addr)<=0)  
    { 
        printf("\nInvalid address/ Address not supported \n"); 

    } 

    if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) 
    { 
        printf("\nConnection Failed \n"); 

    } 
    send(sock , &input , sizeof(input) , 0 ); 


} 






int arraySearch(int number){
    int i=0; 
    for(i=0; i< 100; i++){
        if(input.serviceArray[i] == number)
        {
            return 0; //is found
        }

    }

    return 1;
}


void *createService()
{

    while(counter < SIZE)
    {
        sleep(2);
        if(semaphore ==1)
        {

            int randomValue = rand() % SIZE;

            if(arraySearch(randomValue) == 1)
            {
                input.serviceArray[counter] = randomValue;
                input.portArray[counter] = randomValue + PORT;
                printf("Thread 1 is adding service number: %d and port number: %d\n", input.serviceArray[counter], input.portArray[counter]);
                semaphore = 0;//unlock
                counter = counter + 1;
                callServer();

            }
        }
    }
}
void *removeService()
{

    while(counter < SIZE)
    {
        sleep(4);
        if(semaphore ==0)
        {
            printf("Thread 2 is removing service number: %d and port number: %d\n", input.serviceArray[counter - 1], input.portArray[counter - 1]);
            input.serviceArray[counter -1] = 0;
            input.portArray[counter - 1] = 0;
            semaphore = 1; //lock
            callServer();

        }
    }
}



int main(void)
{

    //create threads
        pthread_t thread_id1, thread_id2;  


        pthread_create(&thread_id1, NULL, createService, NULL);
        pthread_create(&thread_id2, NULL, removeService, NULL); 


        pthread_join(thread_id1, NULL);  
        pthread_join(thread_id2, NULL); 



}

客户 2 - 请求者

#include <stdio.h> 
#include <string.h> 
#include <stdlib.h>  
#include <unistd.h>
#include <time.h> 
#include <sys/socket.h> 
#include <arpa/inet.h> 
#include <stdbool.h>
#include <pthread.h>
#define PORT 8080

void callServer(int serviceNum)
{
struct sockaddr_in address; 
    int sock = 0, valread; 
    struct sockaddr_in serv_addr; 

    if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) 
    { 
        printf("\n Socket creation error \n"); 

    } 

    memset(&serv_addr, '0', sizeof(serv_addr)); 

    serv_addr.sin_family = AF_INET; 
    serv_addr.sin_port = htons(PORT); 

    // Convert IPv4 and IPv6 addresses from text to binary form 
    if(inet_pton(AF_INET, "127.0.0.1", &serv_addr.sin_addr)<=0)  
    { 
        printf("\nInvalid address/ Address not supported \n"); 

    } 

    if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) 
    { 
        printf("\nConnection Failed \n"); 

    } 
    send(sock , &serviceNum , sizeof(serviceNum) , 0 ); 


} 


void main(){
int serviceNum;

printf("Which service would you like to run?\n");
scanf("%d",&serviceNum);
printf("You entered: %d", serviceNum);
callServer(serviceNum);

}

服务器

#include <stdio.h> 
#include <string.h> 
#include <stdlib.h>  
#include <unistd.h>
#include <time.h> 
#include <sys/socket.h> 
#include <arpa/inet.h> 
#include <stdbool.h>
#include <pthread.h>

#define PORT 8080 

 int sockfd; 
 int serviceNum;

    struct sockaddr_in servaddr, cliaddr; 

struct values {
       int serviceArray[100]; 
       int portArray[100];
       }input;






int main()
{

    int server_fd, server_fd2, new_socket, valread, valread2; 

    struct sockaddr_in address; 

    int opt = 1; 

    int addrlen = sizeof(address); 





    // Creating socket file descriptor 

    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 



    // Forcefully attaching socket to the port 8080 

    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | 0, 

                                                  &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 ); 



    // Forcefully attaching socket to the port 8080 

    if (bind(server_fd, (struct sockaddr *)&address,  

                                 sizeof(address))<0) 

    { 

        perror("bind failed"); 

        exit(EXIT_FAILURE); 

    } 

    if (listen(server_fd, 3) < 0) 

    { 

        perror("listen"); 

        exit(EXIT_FAILURE); 

    } 

    if ((new_socket = accept(server_fd, (struct sockaddr *)&address,  

                       (socklen_t*)&addrlen))<0) 

    { 

        perror("accept"); 

        exit(EXIT_FAILURE); 

    } 



        if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 


while(1){
   valread = recv( new_socket , &input, sizeof(input), 0); 




   int j;
        for(j = 0; j < 5; j++)
        {
            printf("service: %d and port: %d\n", input.serviceArray[j], input.portArray[j]);

        }

recv( new_socket , &serviceNum, sizeof(serviceNum), MSG_DONTWAIT)

    printf("The service number you passed is %d", serviceNum);

    }


} 

编辑 - 这是对服务器的更新并且是多线程的。我仍然遇到阻塞问题。

#include <stdio.h> 
#include <string.h> 
#include <stdlib.h>  
#include <unistd.h>
#include <time.h> 
#include <sys/socket.h> 
#include <arpa/inet.h> 
#include <stdbool.h>
#include <pthread.h>

#define PORT 8080 

 int sockfd; 
 int serviceNum;

    struct sockaddr_in servaddr, cliaddr; 

struct values {
       int serviceArray[100]; 
       int portArray[100];
       }input;

int server_fd, server_fd2, new_socket, valread, valread2; 

    struct sockaddr_in address; 

    int opt = 1; 

    int addrlen = sizeof(address);     



void *producer()
{
     // Creating socket file descriptor 

    if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 



    // Forcefully attaching socket to the port 8080 

    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | 0, 

                                                  &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 ); 



    // Forcefully attaching socket to the port 8080 

    if (bind(server_fd, (struct sockaddr *)&address,  

                                 sizeof(address))<0) 

    { 

        perror("bind failed"); 

        exit(EXIT_FAILURE); 

    } 

    if (listen(server_fd, 3) < 0) 

    { 

        perror("listen"); 

        exit(EXIT_FAILURE); 

    } 

    if ((new_socket = accept(server_fd, (struct sockaddr *)&address,  

                       (socklen_t*)&addrlen))<0) 

    { 

        perror("accept"); 

        exit(EXIT_FAILURE); 

    } 



        if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 
    while(1)
    {
   valread = read( new_socket , &input, sizeof(input));   
   int j;
        for(j = 0; j < 5; j++)
        {
            printf("service: %d and port: %d\n", input.serviceArray[j], input.portArray[j]);

        }
}
}

void *requestor()
{
     // Creating socket file descriptor 

    if ((server_fd2 = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 



    // Forcefully attaching socket to the port 8080 

    if (setsockopt(server_fd2, SOL_SOCKET, SO_REUSEADDR | 0, 

                                                  &opt, sizeof(opt))) 

    { 

        perror("setsockopt"); 

        exit(EXIT_FAILURE); 

    } 

    address.sin_family = AF_INET; 

    address.sin_addr.s_addr = INADDR_ANY; 

    address.sin_port = htons(8090); 





    if (bind(server_fd2, (struct sockaddr *)&address,  

                                 sizeof(address))<0) 

    { 

        perror("bind failed"); 

        exit(EXIT_FAILURE); 

    } 

    if (listen(server_fd2, 3) < 0) 

    { 

        perror("listen"); 

        exit(EXIT_FAILURE); 

    } 

    if ((new_socket = accept(server_fd2, (struct sockaddr *)&address,  

                       (socklen_t*)&addrlen))<0) 

    { 

        perror("accept"); 

        exit(EXIT_FAILURE); 

    } 



        if ((server_fd2 = socket(AF_INET, SOCK_STREAM, 0)) == 0) 

    { 

        perror("socket failed"); 

        exit(EXIT_FAILURE); 

    } 
    while(1)
    {
    valread2 = read( new_socket , &serviceNum, sizeof(serviceNum)); 
    printf("The service number you passed is %d", serviceNum);
    }
}


int main()
{

//create threads
        pthread_t thread_id1, thread_id2;  


        pthread_create(&thread_id1, NULL, producer, NULL);
        pthread_create(&thread_id2, NULL, requestor, NULL); 


        pthread_join(thread_id1, NULL);  
        pthread_join(thread_id2, NULL); 







} 

Answer 1

默认情况下，套接字以阻塞 模式运行。因此，当您在一个套接字上调用 recv() 时，它会在等待数据到达时阻塞同一线程上的其他套接字，这是有道理的。

对于您尝试执行的操作，您需要将服务器代码更改为:

• 让每个已接受的套接字处于阻塞模式，并在它们各自的工作线程或分支进程中对它们进行操作。

• 将每个接受的套接字切换到非阻塞模式(fctrl(FIONBIO)等)，然后使用select()、(e)poll() 或其他类似机制在一个线程中一起监视套接字，然后在它告诉您各个套接字何时有您需要处理的事件时作出 react (即，不要' 从套接字读取数据，直到确实有可读取的数据，等等)。

• (仅限 Windows)通过重叠 I/O 操作异步使用每个接受的套接字。让操作系统在每个套接字上有事件时通知您。