客户端与服务器的独立和并行工作

Posted 2023-02-16

【中文标题】客户端与服务器的独立和并行工作

【英文标题】：Independent and parallel work of clients with the server

【发布时间】：2021-04-10 19:24:50

【问题描述】：

程序要点：服务器创建 n 个客户端进程。在客户端上，用户输入一个发送到服务器的字符串。在服务器上，字符串的处理方式如下：计算输入字符串中元音和数字出现的频率。此外，此信息会发送给客户端，客户端会打印答案。

备注：

客户端是一次性的 服务器的应答输出不应该存在，它的存在仅用于调试

错误要点：要在服务器上进行即时处理并将响应从服务器即时返回到客户端，您需要按照客户端的创建顺序处理它们。但是，如果我们在除第一个客户端之外的所有客户端中输入一行，那么我们将不会看到来自服务器端的任何操作（它不会打印任何内容）。但是一旦你在第一个客户端输入一行，所有客户端的信息都会立即处理，并按照创建客户端的顺序进行处理（根据服务器的调试输出）。

问题：如何实现每个客户端与其创建队列的独立性？

服务器：

#include <windows.h> 
#include <iostream>
#include <tchar.h>
#include <strsafe.h>
#include <limits>
#include <string.h>
#include <sstream>

#pragma warning(disable : 4996)

using namespace std;

#ifdef max
#undef max
#endif


#define PIPE_TIMEOUT 5000
#define BUFSIZE 4096


int main(int argc, char* argv[])

    // ***ROUTINE STAFF***

    cout << "Server is lauched\n"
        "It will be terminated when all clients exits\n";


    if (argc > 2 || (argc == 2 && argv[argc - 1] != "--help")) 
        cout << "Program should be lauched this way:\n"
            "name_of_program --help\n"
            "Note: --help is optional\n";
        return EXIT_FAILURE;
    
    if (argc == 2 && argv[argc - 1] == "--help") 
        cout << "The server creates n client processes. On the client user\n"
            "enters a character string that is sent to the server.On server\n"
            "the string is processed as follows : the frequency is counted\n"
            "the occurence of vowels and numbers in the entered string. Further,\n"
            "this information is sent to the client.\n";
        return EXIT_SUCCESS;
    

    int n = 0;
    cout << "Enter number of clients: ";
    if (!(cin >> n) || (n < 0))
    
        cout << "Invalid input\n";
        return EXIT_FAILURE;
    
    cin.ignore(numeric_limits<streamsize>::max(), '\n'); // cleaning buffer



    // ***CREATING PROCESSES***

    cout << "Generating " << n << " clients...\n";
    // struct _STARTUPINFO: Specifies the window station, desktop, standard handles, and appearance of the main window for a process at creation time.
    STARTUPINFO* si_arr = new STARTUPINFO[n];
    // struct _PROCESS_INFORMATION: Contains information about a newly created process and its primary thread. It is used with the CreateProcess() (and other).
    PROCESS_INFORMATION* pi_arr = new PROCESS_INFORMATION[n];

    for (int i = 0; i < n; ++i) 
        // ZeroMemory macro: Fills a block of memory with zeros.
        /*
        void ZeroMemory(
        [in] PVOID  Destination,
        [in] SIZE_T Length
        );
        */
        ZeroMemory(&si_arr[i], sizeof(si_arr[i]));
        // DWORD STARTUPINFO.cb: The size of the structure, in bytes.
        si_arr[i].cb = sizeof(si_arr[i]);
        ZeroMemory(&pi_arr[i], sizeof(pi_arr[i]));

        if (!CreateProcess(
        
            TEXT("C:\\Users\\andrewshnitko\\source\\repos\\pipe_client\\Debug\\pipe_client.exe"),   // name of program (like in cmd)
            NULL,        // arguments for program (like in cmd after name of program)
            NULL,           // Process handle not inheritable
            NULL,           // Thread handle not inheritable
            FALSE,          // Set handle inheritance to FALSE
            CREATE_NEW_CONSOLE, // dwCreationFlags - The new process gets a new console instead of inheriting the parent one 
            NULL,           // Use parent's environment block
            NULL,           // Use parent's starting directory 
            &si_arr[i],            // Pointer to STARTUPINFO structure
            &pi_arr[i])           // Pointer to PROCESS_INFORMATION structure
            )
        
            printf("CreateProcess failed (%d).\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "All processes (pipe clients) created\n";


    // ***CREATING PIPE INSTANCES***

    HANDLE* pipe_instances = new HANDLE[n];

    for (int i = 0; i < n; i++)
    
        pipe_instances[i] = CreateNamedPipe(
            TEXT("\\\\.\\pipe\\os_lab4_pipe"),            // pipe name 
            PIPE_ACCESS_DUPLEX,      // read/write access 
            PIPE_TYPE_MESSAGE |      // message-type pipe 
            PIPE_READMODE_MESSAGE |  // message-read mode 
            PIPE_WAIT,               // blocking mode 
            n,               // number of instances 
            1024,   // output buffer size 
            1024,   // input buffer size 
            PIPE_TIMEOUT,            // client time-out 
            NULL);                   // default security attributes 

        if (pipe_instances[i] == INVALID_HANDLE_VALUE)
        
            printf("CreateNamedPipe failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "All pipe instances created\n";

    // ***CONNECTING PIPE INSTANCES***

    for (int i = 0; i < n; i++)
    
        if (!ConnectNamedPipe(pipe_instances[i], NULL))
        
            printf("ConnectNamedPipe failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "All pipe instances connected to clients\n";

    // ***PROCESSING***

    char buf[1024];
    DWORD read_bytes;
    DWORD written_bytes;

    for (int i = 0; i < n; i++)
    
        if (ReadFile(pipe_instances[i], buf, 1024, &read_bytes, NULL))
        
            char* str = new char[read_bytes];
            strncpy(str, buf, read_bytes);
            //char answer[1024]; // ready c-str to WriteFile

            int total_amount = 0;

            int vowel_amount = 0;
            int digit_amount = 0;
            double vowel_frequency = 0.0;
            double digit_frequency = 0.0;


            char vowels[] = "aeiouy";
            char digits[] = "0123456789";


            total_amount = strlen(str);
            printf("Total: %i\n", total_amount);

            // vowel
            char* vowel_search;
            vowel_search = strpbrk(str, vowels); // check for first occurence of vovel in str
            while (vowel_search != NULL)  // while vovels not end up in str
                vowel_amount++;
                vowel_search = strpbrk(vowel_search + 1, vowels);
            
        
            vowel_frequency = (double)vowel_amount / (double)total_amount * 100.0;
        

            // digit
            char* digit_search;
            digit_search = strpbrk(str, digits); // check for first occurence of digit in str
            while (digit_search != NULL)  // while digits not end up in str
                digit_amount++;
                digit_search = strpbrk(digit_search + 1, digits);
            
        
            digit_frequency = (double)digit_amount / (double)total_amount * 100.0;
        
            string pre_str;
            pre_str = "Total: " + to_string(total_amount) + "\n"
                "Vowels: " + to_string(vowel_amount) + "\n"
                "Frequency: " + to_string(vowel_frequency) + "\n"
                "Digits: " + to_string(digit_amount) + "\n"
                "Frequency:" + to_string(digit_frequency) + "\n";
        

            cout << pre_str;

            const char* answer = pre_str.c_str();
   
            if (!WriteFile(pipe_instances[i], answer, 1024, &written_bytes, NULL)) 
                printf("WriteFile failed with %d.\n", GetLastError());
                return EXIT_FAILURE;
            
        
        else 
            printf("ReadFile failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "Reading, processing and writting was successful\n";

    // ***CLOSING PIPE INSTANCES***

    for (int i = 0; i < n; i++)
    
        CloseHandle(pipe_instances[i]);
    

    // !? - WaitMultipleObjects()


    delete[] pipe_instances;

    cout << "All pipe instances closed\n";

    // ***CLOSING PROCESSES***
    // The code written below is needed in order for the server to shutdown not earlier than the clients

    HANDLE* ev_hndl_arr = new HANDLE[n];
    for (int i = 0; i < n; i++) 
        ev_hndl_arr[i] = pi_arr[i].hProcess;
    

    // Wait until EACH child process exits.
    WaitForMultipleObjects(n, ev_hndl_arr, TRUE, INFINITE);

    // Close process and thread handles.
    for (int i = 0; i < n; i++) 
        CloseHandle(pi_arr[i].hProcess);
        CloseHandle(pi_arr[i].hThread);
    

    delete[] si_arr;
    delete[] pi_arr;
    delete[] ev_hndl_arr;

    cout << "All processes (pipe clients) closed\n";

    cout << "This is the end of server execution\n";
    system("pause");
    return EXIT_SUCCESS;

客户：

#include <iostream>
#include <stdio.h> // fgets()
#include <string.h> // strpbrk()

#include <Windows.h>

using namespace std;

int main() 
    cout << "Client is launched\n";

    HANDLE hndlNP = CreateFile(
        TEXT("\\\\.\\pipe\\os_lab4_pipe"),
        GENERIC_READ | GENERIC_WRITE,
        NULL, // Prevents other processes from opening a file or device
        NULL, // cannot be inherited by any child processes
        OPEN_EXISTING,
        NULL, // no attributes
        NULL // no template
        );
    if (hndlNP == INVALID_HANDLE_VALUE) 
        cout << "CreateFile error\n";
        return EXIT_FAILURE;
    
    cout << "Pipe connection established\n";

    char text[1024];


    printf("Enter string (max 1023 symbols): ");
    gets_s(text, 1024); // replacement \n for \0


    char answer[1024];
    DWORD read_bytes;
    DWORD written_bytes;

    if (WriteFile(hndlNP, text, 1024, &written_bytes, NULL)) 
   
        if (!ReadFile(hndlNP, answer, 1024, &read_bytes, NULL)) 
            printf("ReadFile failed with %d.\n", GetLastError());
            system("pause"); // TEMPORARY
            return EXIT_FAILURE;
        
    
    else 
        printf("WriteFile failed with %d.\n", GetLastError());
        system("pause"); // TEMPORARY
        return EXIT_FAILURE;
    
    cout << "Writting and reading was successful\n";

    cout << answer;

    // ***CLOSING PIPE CONNECTION***
    CloseHandle(hndlNP);

    system("pause");

    return 0;

服务器：

Server is launched
It will be terminated when all clients exits
Enter number of clients: 2
Generating 2 clients...
All processes (pipe clients) created
All pipe instances created
All pipe instances connected to clients
***BREAKPOINT 1: no answer from server, no debug output in server side (waiting state)***
(waiting)
***BREAKPOINT 2: immediate response from server and debug output from server(firstly about
first client(wrote for the second time), and then about second client(wrote for the first 
time)). All clients get answer from server***
Total: 16
Total: 16
Vowels: 4
Frequency: 25.000000
Digits: 8
Frequency:50.000000
Total: 11
Total: 11
Vowels: 2
Frequency: 18.181818
Digits: 5
Frequency:45.454545
Reading, processing and writting was successful
All pipe instances closed
(waiting for closure of clients(press any key...))
(termination)

Client (2nd): // 第一个输入到第二个客户端

Client is launched
Pipe connection established
Enter string (max 1023 symbols): 12345qwerty
***BREAKPOINT 1: no answer from server, no debug output in server side (waiting state)***
(waiting)
***BREAKPOINT 2: immediate response from server and debug output from server(firstly about
first client(wrote for the second time), and then about second client(wrote for the first 
time)). All clients get answer from server***
Writting and reading was successful
Total: 11
Vowels: 2
Frequency: 18.181818
Digits: 5
Frequency:45.454545
Press any key to continue . . .
(press)
(termination)

客户（第一名）：

Client is launched
Pipe connection established
Enter string (max 1023 symbols): qwiouryq18927468
***BREAKPOINT 2: immediate response from server and debug output from server(firstly about 
first client(wrote for the second time), and then about second client(wrote for the first 
time)). All clients get answer from server***
Writting and reading was successful
Total: 16
Vowels: 4
Frequency: 25.000000
Digits: 8
Frequency:50.000000
Press any key to continue . . .
(press)
(termination)

注意：首先创建客户端（第一个），然后创建客户端（第二个）

【问题讨论】：

minimal reproducible example 必须是最小的。输入的处理似乎不是问题的一部分，只需将其删除即可。

当您为每个客户循环执行if (ReadFile(pipe_instances[i], buf, 1024, &read_bytes, NULL)) 时......已经可以说所有设计都是错误的。每个连接都必须独立处理

@RbMm，绝对没有办法只更改部分代码？必须重写所有内容？

我确定需要重写所有内容。但这只是我的意见。

【参考方案1】：

可以通过多个线程检测不同的管道，可以参考以下代码：

服务器.cpp

#include <windows.h> 
#include <iostream>
#include <tchar.h>
#include <strsafe.h>
#include <limits>
#include <string.h>
#include <sstream>
#include <thread>

#pragma warning(disable : 4996)

using namespace std;

#ifdef max
#undef max
#endif


#define PIPE_TIMEOUT 5000
#define BUFSIZE 4096
HANDLE* pipe_instances = NULL;
int n = 0;


int f(int i)

    char buf[1024];
    DWORD read_bytes;
    DWORD written_bytes;
    if (ReadFile(pipe_instances[i], buf, 1024, &read_bytes, NULL))
    
        char* str = new char[read_bytes];
        strncpy(str, buf, read_bytes);
        //char answer[1024]; // ready c-str to WriteFile

        int total_amount = 0;

        int vowel_amount = 0;
        int digit_amount = 0;
        double vowel_frequency = 0.0;
        double digit_frequency = 0.0;


        char vowels[] = "aeiouy";
        char digits[] = "0123456789";


        total_amount = strlen(str);
        printf("Total: %i\n", total_amount);

        // vowel
        char* vowel_search;
        vowel_search = strpbrk(str, vowels); // check for first occurence of vovel in str
        while (vowel_search != NULL)  // while vovels not end up in str
            vowel_amount++;
            vowel_search = strpbrk(vowel_search + 1, vowels);
        

        vowel_frequency = (double)vowel_amount / (double)total_amount * 100.0;


        // digit
        char* digit_search;
        digit_search = strpbrk(str, digits); // check for first occurence of digit in str
        while (digit_search != NULL)  // while digits not end up in str
            digit_amount++;
            digit_search = strpbrk(digit_search + 1, digits);
        

        digit_frequency = (double)digit_amount / (double)total_amount * 100.0;

        string pre_str;
        pre_str = "Total: " + to_string(total_amount) + "\n"
            "Vowels: " + to_string(vowel_amount) + "\n"
            "Frequency: " + to_string(vowel_frequency) + "\n"
            "Digits: " + to_string(digit_amount) + "\n"
            "Frequency:" + to_string(digit_frequency) + "\n";


        cout << pre_str;

        const char* answer = pre_str.c_str();

        if (!WriteFile(pipe_instances[i], answer, 1024, &written_bytes, NULL)) 
            printf("WriteFile failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    else 
        printf("ReadFile failed with %d.\n", GetLastError());
        return EXIT_FAILURE;
    



int main(int argc, char* argv[])

    // ***ROUTINE STAFF***

    cout << "Server is lauched\n"
        "It will be terminated when all clients exits\n";


    //if (argc > 2 || (argc == 2 && argv[argc - 1] != "--help")) 
    //    cout << "Program should be lauched this way:\n"
    //        "name_of_program --help\n"
    //        "Note: --help is optional\n";
    //    return EXIT_FAILURE;
    //
    //if (argc == 2 && argv[argc - 1] == "--help") 
    //    cout << "The server creates n client processes. On the client user\n"
    //        "enters a character string that is sent to the server.On server\n"
    //        "the string is processed as follows : the frequency is counted\n"
    //        "the occurence of vowels and numbers in the entered string. Further,\n"
    //        "this information is sent to the client.\n";
    //    return EXIT_SUCCESS;
    //

    cout << "Enter number of clients: ";
    if (!(cin >> n) || (n < 0))
    
        cout << "Invalid input\n";
        return EXIT_FAILURE;
    
    cin.ignore(numeric_limits<streamsize>::max(), '\n'); // cleaning buffer



    // ***CREATING PROCESSES***

    cout << "Generating " << n << " clients...\n";
    // struct _STARTUPINFO: Specifies the window station, desktop, standard handles, and appearance of the main window for a process at creation time.
    STARTUPINFO* si_arr = new STARTUPINFO[n];
    // struct _PROCESS_INFORMATION: Contains information about a newly created process and its primary thread. It is used with the CreateProcess() (and other).
    PROCESS_INFORMATION* pi_arr = new PROCESS_INFORMATION[n];

    for (int i = 0; i < n; ++i) 
        // ZeroMemory macro: Fills a block of memory with zeros.
        /*
        void ZeroMemory(
        [in] PVOID  Destination,
        [in] SIZE_T Length
        );
        */
        ZeroMemory(&si_arr[i], sizeof(si_arr[i]));
        // DWORD STARTUPINFO.cb: The size of the structure, in bytes.
        si_arr[i].cb = sizeof(si_arr[i]);
        ZeroMemory(&pi_arr[i], sizeof(pi_arr[i]));

        if (!CreateProcess(

            TEXT("D:\\VS_test_projects\\Project2\\Debug\\c.exe"),   // name of program (like in cmd)
            NULL,        // arguments for program (like in cmd after name of program)
            NULL,           // Process handle not inheritable
            NULL,           // Thread handle not inheritable
            FALSE,          // Set handle inheritance to FALSE
            CREATE_NEW_CONSOLE, // dwCreationFlags - The new process gets a new console instead of inheriting the parent one 
            NULL,           // Use parent's environment block
            NULL,           // Use parent's starting directory 
            &si_arr[i],            // Pointer to STARTUPINFO structure
            &pi_arr[i])           // Pointer to PROCESS_INFORMATION structure
            )
        
            printf("CreateProcess failed (%d).\n", GetLastError());
            return EXIT_FAILURE;
        
    





    cout << "All processes (pipe clients) created\n";


    // ***CREATING PIPE INSTANCES***

    pipe_instances = new HANDLE[n];

    for (int i = 0; i < n; i++)
    
        pipe_instances[i] = CreateNamedPipe(
            TEXT("\\\\.\\pipe\\os_lab4_pipe"),            // pipe name 
            PIPE_ACCESS_DUPLEX,      // read/write access 
            PIPE_TYPE_MESSAGE |      // message-type pipe 
            PIPE_READMODE_MESSAGE |  // message-read mode 
            PIPE_WAIT,               // blocking mode 
            n,               // number of instances 
            1024,   // output buffer size 
            1024,   // input buffer size 
            PIPE_TIMEOUT,            // client time-out 
            NULL);                   // default security attributes 

        if (pipe_instances[i] == INVALID_HANDLE_VALUE)
        
            printf("CreateNamedPipe failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "All pipe instances created\n";

    // ***CONNECTING PIPE INSTANCES***

    for (int i = 0; i < n; i++)
    
        if (!ConnectNamedPipe(pipe_instances[i], NULL))
        
            printf("ConnectNamedPipe failed with %d.\n", GetLastError());
            return EXIT_FAILURE;
        
    
    cout << "All pipe instances connected to clients\n";

    // ***PROCESSING***

    thread* t = new thread[n];
    for (int i = 0; i < n; i++)
    
        t[i] = thread(f, i);
    


    cout << "Reading, processing and writting was successful\n";

    // ***CLOSING PIPE INSTANCES***
    for (int i = 0; i < n; i++)
    
        t[i].join();
    
    for (int i = 0; i < n; i++)
    
        CloseHandle(pipe_instances[i]);
    

    // !? - WaitMultipleObjects()


    delete[] pipe_instances;

    cout << "All pipe instances closed\n";

    // ***CLOSING PROCESSES***
    // The code written below is needed in order for the server to shutdown not earlier than the clients

    HANDLE* ev_hndl_arr = new HANDLE[n];
    for (int i = 0; i < n; i++) 
        ev_hndl_arr[i] = pi_arr[i].hProcess;
    

    // Wait until EACH child process exits.
    WaitForMultipleObjects(n, ev_hndl_arr, TRUE, INFINITE);

    // Close process and thread handles.
    for (int i = 0; i < n; i++) 
        CloseHandle(pi_arr[i].hProcess);
        CloseHandle(pi_arr[i].hThread);
    

    delete[] si_arr;
    delete[] pi_arr;
    delete[] ev_hndl_arr;

    cout << "All processes (pipe clients) closed\n";

    cout << "This is the end of server execution\n";
    system("pause");
    return EXIT_SUCCESS;

【讨论】：

您好，这个答案能解决您的问题吗？如果您有任何问题，请随时告诉我，如果有帮助，请接受。