C++ Socket多客户端通信异常：服务器接收空消息问题排查

问题场景 #

• 搭建1个Socket服务器与3个Socket客户端
• 每个客户端每次发送消息时新建连接，收到响应后关闭连接
• 服务器无需并发处理连接，只需串行处理消息

编译命令 #

/usr/bin/g++ mycode.cpp -g -lpthread -lrt -Wall -o mycode

可运行代码 #

#include <iostream>
#include <arpa/inet.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <unordered_map>
#include <thread>

using namespace std;

void Warning(string msg) { std::cout<< msg << std::endl; }
     
namespace mySocket {
    class Memcached {
    public:
        // start a server
        static void controller(int port=7111) { std::thread (server, port).detach(); }

        // open a new connection to send a message: 
        //  1. open a connection
        //  2. send the message
        //  3. read the message
        //  4. close the connection
        std::string sendMessage(string msg, string host, int port=7111) {
            int sock = 0, client_fd;
            struct sockaddr_in serv_addr;
            char buffer[1024] = { 0 };
            if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                std::cout << "Socket creation error, msg: " << msg << ", host: " << host << ", port: " << port << std::endl;
                exit(1);
            }
            serv_addr.sin_family = AF_INET;
            serv_addr.sin_port = htons(port);
            if (inet_pton(AF_INET, host.c_str(), &serv_addr.sin_addr) <= 0) {
                std::cout << "\nInvalid address/ Address not supported, kmsgey: " << msg << ", host: " << host << ", port: " << port << std::endl;
                exit(1);
            }
            while ((client_fd = connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr))) < 0) { sleep(10*1000); }
            std::cout << "client sends a message:"<<msg<<", msg size:"<<msg.size()<<std::endl;
            send(sock, msg.c_str(), msg.size(), 0);
            read(sock, buffer, 1024);
            close(client_fd);
            return std::string(buffer, strlen(buffer));
        }

     private:
        // start a server
        //  1. open a file descriptor
        //  2. listen the fd with queue size 10
        //  3. accept one connection at a time
        //  4. deal with message in the connection
        //  5. accept the next connection
        //  6. repeat step 3
        static void server(int port) {
            int server_fd, new_socket;
            struct sockaddr_in address;
            int opt = 1;
            int addrlen = sizeof(address);
            char buffer[1024] = { 0 };
            unordered_map<string,string> data;
            if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
                Warning("socket failed"); exit(1);
            }
            if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
                Warning("setsockopt failed"); exit(1);
            }
            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) {
                Warning("bind failed"); exit(1);
            }
            // the queue size is 10 > 3
            if (listen(server_fd, 10) < 0) {
                Warning("listen failed"); exit(1);
            }
            while(1)
            {
                if ((new_socket = accept(server_fd, (struct sockaddr*)&address, (socklen_t*)&addrlen)) < 0) {
                    std::cout << "accept failed"; exit(1);
                }
                memset(&buffer, 0, sizeof(buffer)); //clear the buffer
                read(new_socket, buffer, 1024);
                std::string msg = std::string(buffer, strlen(buffer));
                if (msg.size()==0) {
                    std::cout<<"I can't believe it"<<std::endl;
                }
                std::cout<<"received msg from the client:"<<msg<<",msg size:"<<msg.size()<<std::endl;
                std::string results="response from the server:["+msg+"]";
                send(new_socket, results.c_str(), results.length(), 0);
                //usleep(10*1000);
            }
            if (close(new_socket)<0){
                std::cout <<"close error"<<std::endl; 
            }
            shutdown(server_fd, SHUT_RDWR);
        }
    } ;
}

void operation(int client_id) {
    auto obj = new mySocket::Memcached();
    for (int i=0; i<10;i++){
        int id=client_id*100+i;
        std::cout<<obj->sendMessage(std::to_string(id), "127.0.0.1", 7111)<<std::endl<<std::endl;
    }
}

int main(int argc, char const* argv[]) {
    // start a socket server
    mySocket::Memcached::controller();
    
    // start 3 socket clients
    std::thread t1(operation, 1);
    std::thread t2(operation, 2);
    std::thread t3(operation, 3);
    t1.join();
    t2.join();
    t3.join();
}

问题现象 #

客户端始终发送长度为3的消息，但服务器会接收到长度为0的消息，进而引发后续错误。临时缓解方式：

• 在服务器的while循环内添加短时间usleep(10*1000)
• 仅使用单个客户端

根本原因分析 #

1. 未关闭已处理的客户端连接：服务器在处理完单个客户端的消息并发送响应后，没有调用close(new_socket)关闭当前连接描述符。当多个客户端并发发起连接时，服务器会不断通过accept获取新的连接描述符，但旧的连接描述符一直占用系统资源，且客户端关闭连接后，服务器端的连接处于半关闭状态，后续read调用会读到EOF（返回0），导致缓冲区无有效数据，最终构造出空字符串。

2. 连接关闭的时序冲突：多个客户端并发创建连接、发送消息后会立即关闭连接。当服务器处理速度跟不上客户端关闭速度时，服务器的read操作可能在客户端已经关闭连接后才执行，此时read返回0，strlen(buffer)结果为0，就会出现收到空消息的情况。

添加usleep能缓解问题，是因为给了服务器足够时间接收客户端的完整消息，同时延迟了下一次accept的时间，减少了时序冲突的概率；单个客户端时，连接是串行创建和关闭的，服务器能及时处理消息，不会出现连接关闭后才执行read的情况。

修复方案 #

1. 及时关闭已处理的连接：在服务器发送响应后立即关闭当前连接描述符：

std::string results="response from the server:["+msg+"]";
send(new_socket, results.c_str(), results.length(), 0);
close(new_socket); // 添加此行，关闭当前连接

2. 检查read返回值：避免处理空消息或错误情况：

memset(&buffer, 0, sizeof(buffer));
ssize_t bytes_read = read(new_socket, buffer, 1024);
if (bytes_read <= 0) {
    std::cout<<"Connection closed or read error"<<std::endl;
    close(new_socket);
    continue;
}
std::string msg = std::string(buffer, bytes_read);

内容的提问来源于stack exchange，提问作者stickers