Многопоточный TCP-сервер

Решение: Многопоточный TCP-сервер на C++

Архитектура

Сервер состоит из двух основных частей:

1. accept loop — главный поток слушает входящие подключения
2. client handler threads — отдельный поток для каждого клиента выполняет echo

Полная реализация

#include <iostream>
#include <thread>
#include <vector>
#include <mutex>
#include <atomic>
#include <cstring>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>

class TcpEchoServer {
private:
    uint16_t port;
    int serverSocket;
    std::atomic<bool> running;
    std::vector<std::thread> clientThreads;
    std::mutex threadsMutex;
    static TcpEchoServer* instance;  // для обработки сигналов
    
    void handleClient(int clientSocket) {
        char buffer[4096];
        const char* addr = inet_ntoa(((struct sockaddr_in*)&serverAddr)->sin_addr);
        
        std::cout << "[INFO] Client connected from " << addr << std::endl;
        
        while (running) {
            int bytesRead = recv(clientSocket, buffer, sizeof(buffer) - 1, 0);
            
            if (bytesRead < 0) {
                std::cerr << "[ERROR] recv() failed" << std::endl;
                break;
            }
            
            if (bytesRead == 0) {
                std::cout << "[INFO] Client disconnected" << std::endl;
                break;  // клиент закрыл соединение
            }
            
            buffer[bytesRead] = '\0';
            std::cout << "[ECHO] Received: " << buffer;
            
            // Отправляем обратно
            int bytesSent = send(clientSocket, buffer, bytesRead, 0);
            if (bytesSent < 0) {
                std::cerr << "[ERROR] send() failed" << std::endl;
                break;
            }
        }
        
        close(clientSocket);
    }
    
    void acceptLoop() {
        std::cout << "[INFO] Server listening on port " << port << std::endl;
        
        while (running) {
            struct sockaddr_in clientAddr;
            socklen_t addrLen = sizeof(clientAddr);
            
            // accept с timeout для проверки флага running
            int clientSocket = accept(serverSocket, 
                                     (struct sockaddr*)&clientAddr, 
                                     &addrLen);
            
            if (clientSocket < 0) {
                if (running) {  // не из-за shutdown
                    std::cerr << "[ERROR] accept() failed" << std::endl;
                }
                continue;
            }
            
            // Создаём поток для обработки клиента
            std::lock_guard<std::mutex> lock(threadsMutex);
            clientThreads.emplace_back(&TcpEchoServer::handleClient, this, clientSocket);
            
            // Чистим завершённые потоки
            for (auto it = clientThreads.begin(); it != clientThreads.end(); ) {
                if (it->joinable()) {
                    ++it;
                } else {
                    // Этот код не будет вызван, так как потоки всегда joinable
                    // Но покажем корректный паттерн
                    ++it;
                }
            }
        }
    }
    
    static void signalHandler(int signal) {
        if (instance) {
            std::cout << "\n[INFO] Received SIGINT, shutting down..." << std::endl;
            instance->stop();
        }
    }
    
public:
    struct sockaddr_in serverAddr;
    
    explicit TcpEchoServer(uint16_t port) 
        : port(port), serverSocket(-1), running(false) {
        instance = this;
        signal(SIGINT, TcpEchoServer::signalHandler);
    }
    
    ~TcpEchoServer() {
        stop();
    }
    
    void start() {
        // Создаём сокет
        serverSocket = socket(AF_INET, SOCK_STREAM, 0);
        if (serverSocket < 0) {
            std::cerr << "[ERROR] socket() failed" << std::endl;
            return;
        }
        
        // SO_REUSEADDR разрешает переиспользование адреса сразу после перезапуска
        int opt = 1;
        if (setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, 
                      &opt, sizeof(opt)) < 0) {
            std::cerr << "[ERROR] setsockopt() failed" << std::endl;
            close(serverSocket);
            return;
        }
        
        // Биндим сокет к адресу
        serverAddr.sin_family = AF_INET;
        serverAddr.sin_addr.s_addr = INADDR_ANY;
        serverAddr.sin_port = htons(port);
        
        if (bind(serverSocket, (struct sockaddr*)&serverAddr, 
                sizeof(serverAddr)) < 0) {
            std::cerr << "[ERROR] bind() failed" << std::endl;
            close(serverSocket);
            return;
        }
        
        // Переводим в режим прослушивания
        if (listen(serverSocket, SOMAXCONN) < 0) {
            std::cerr << "[ERROR] listen() failed" << std::endl;
            close(serverSocket);
            return;
        }
        
        running = true;
        acceptLoop();  // блокирует до stop()
    }
    
    void stop() {
        if (!running) return;
        
        running = false;
        
        // Закрываем серверный сокет для прерывания accept()
        if (serverSocket >= 0) {
            shutdown(serverSocket, SHUT_RDWR);
            close(serverSocket);
            serverSocket = -1;
        }
        
        // Ждём завершения всех потоков
        {
            std::lock_guard<std::mutex> lock(threadsMutex);
            for (auto& thread : clientThreads) {
                if (thread.joinable()) {
                    thread.join();
                }
            }
            clientThreads.clear();
        }
        
        std::cout << "[INFO] Server stopped" << std::endl;
    }
};

TcpEchoServer* TcpEchoServer::instance = nullptr;

Использование

int main() {
    TcpEchoServer server(8888);
    server.start();  // блокирует; выход по Ctrl+C
    return 0;
}

Тестирование из терминала:

$ nc localhost 8888
Hello Server
Hello Server

Ключевые детали

1. SO_REUSEADDR

int opt = 1;
setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

Без этого может быть ошибка "Address already in use" при перезапуске.

2. Обработка сигналов

signal(SIGINT, TcpEchoServer::signalHandler);

Позволяет корректно завершить работу по Ctrl+C.

3. Мьютекс для управления потоками

std::lock_guard<std::mutex> lock(threadsMutex);

Гарантирует потокобезопасность при добавлении новых потоков.

Graceful Shutdown

Наша реализация уже поддерживает graceful shutdown:

1. Получен SIGINT → вызываем stop()
2. Устанавливаем running = false
3. Закрываем serverSocket → accept() вернёт ошибку
4. acceptLoop() выходит
5. Ждём join() всех clientThreads
6. Каждый поток заканчивает обработку текущих данных
7. Все сокеты закрыты, программа завершается

Улучшения для production

1. Пул потоков вместо создания нового потока на клиента:

std::queue<std::function<void()>> taskQueue;
std::vector<std::thread> workers;  // фиксированное количество

2. Асинхронный I/O (epoll/kqueue) вместо одного потока на клиента:

• Масштабируется до 10k+ клиентов
• Используется в nginx, Redis

3. Обработка ошибок socket при shutdown:

if (clientSocket < 0) {
    if (errno == EINTR) continue;  // сигнал перервал accept
    if (!running) break;  // нормальное завершение
    // остальные ошибки — логируем
}

4. Таймауты на операции socket:

struct timeval tv;
tv.tv_sec = 30;  // 30 секунд
setsockopt(clientSocket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));

Анализ безопасности

✅ Нет утечек файловых дескрипторов (используем close) ✅ Нет утечек потоков (используем join) ✅ Корректная обработка отключения клиента ✅ Защита от переполнения буфера (4KB max) ✅ Graceful shutdown при сигналах