时间戳

首页 > 解决方案 > 如何使用 C++ 中的 recv 或读取函数从 tcp 套接字读取大请求？

c++ - 如何使用 C++ 中的 recv 或读取函数从 tcp 套接字读取大请求？

问题描述

我正在开发一个将接收 http 请求的服务器。一切正常，但是当请求很大时，例如 30 mb，此请求的一部分会丢失。换句话说，当接收到大请求时，服务器无法读取所有数据。

这是我的 tcp 服务器的代码：

TcpServer.h：

#include <cstring>
#include <string>
#include <vector>
#include <thread>
#include <functional>
#include <fcntl.h>
#include <chrono>
#include <thread>
#include <iostream>
#include <poll.h>

#if defined(_WIN32) || defined(_WIN64)

#include <winsock2.h>
#include <ws2tcpip.h>

typedef SOCKET socket_t;
typedef int msg_size_t;

#define WSA_LAST_ERR WSAGetLastError()
#define SOCKET_SEND SD_SEND
#define SOCKET_RECEIVE SD_RECEIVE

#pragma comment (lib, "Ws2_32.lib")

#elif defined(__unix__) || defined(__linux__)

#include <sys/socket.h>
#include <unistd.h>
#include <arpa/inet.h>

typedef int socket_t;
typedef ssize_t msg_size_t;

#define INVALID_SOCKET (socket_t)(-1)
#define SOCKET_ERROR (-1)
#define SOCKET_SEND SHUT_RDWR
#define SOCKET_RECEIVE SHUT_RDWR
#define WSA_LAST_ERR errno

#else
#error Library is not supported on this platform

#endif

#define DEFAULT_HOST "127.0.0.1"
#define DEFAULT_PORT 8000

#define MAX_BUFF_SIZE 8192 * 8 - 1

typedef std::function<void(const std::string&, const socket_t&)> tcpHandler;

class TcpServer
{
private:
    uint16_t _port;
    const char* _host;
    tcpHandler _handler;
    socket_t _socket;
    sockaddr_in _socketAddr{};
    void startListener();
    void serveConnection(const socket_t& client);
    static int closeSocket(const socket_t& socket);
    static void wsaCleanUp();
    void cleanUp(const socket_t& connection);
    std::string recvAll(const socket_t& connection);
    int init();

    enum ReadResult
    {
        Continue, None
    };

    ReadResult _handleError(char* buffer, int& status);

public:
    struct Context
    {
        const char* host = nullptr;
        uint16_t port = 0;
        tcpHandler handler = nullptr;
    };

    explicit TcpServer(TcpServer::Context ctx);
    void listenAndServe();
    static void send(const char* data, const socket_t& connection);
    static void write(const char* data, size_t bytesToWrite, const socket_t& connection);
    ~TcpServer();
};

TcpServer.cpp：

#include "tcp_server.h"

bool set_socket_blocking(int fd, bool blocking)
{
    if (fd < 0)
    {
        return false;
    }

#if defined(_WIN32) || defined(_WIN64)
    unsigned long mode = blocking ? 0 : 1;
    return ioctlsocket(fd, FIONBIO, &mode) == 0;
#else
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1)
    {
        return false;
    }

    flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
    return fcntl(fd, F_SETFL, flags) == 0;
#endif
}

TcpServer::TcpServer(TcpServer::Context ctx)
{
    if (ctx.host == nullptr)
    {
        ctx.host = DEFAULT_HOST;
    }

    if (ctx.port == 0)
    {
        ctx.port = DEFAULT_PORT;
    }

    this->_host = ctx.host;
    this->_port = ctx.port;

    if (ctx.handler == nullptr)
    {
        throw std::invalid_argument("Context::handler can not be null");
    }

    this->_handler = ctx.handler;

    if (ctx.logger == nullptr)
    {
        ctx.logger = Logger::getInstance();
    }

    this->_socketAddr = {};
    this->_socket = {};
}

TcpServer::~TcpServer()
{
    TcpServer::cleanUp(this->_socket);
}

int TcpServer::init()
{
    this->_socketAddr.sin_family = AF_INET;
    this->_socketAddr.sin_port = htons(this->_port);
    this->_socketAddr.sin_addr.s_addr = inet_addr(this->_host);

    memset(this->_socketAddr.sin_zero, '\0', sizeof this->_socketAddr.sin_zero);

    this->_socket = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);

    if (this->_socket == INVALID_SOCKET)
    {
        std::cout << "Failed to initialize server at port " + std::to_string(this->_port) << '\n';
        TcpServer::wsaCleanUp();
        return INVALID_SOCKET;
    }

    if (bind(this->_socket, (sockaddr*)&this->_socketAddr, sizeof(this->_socketAddr)) == SOCKET_ERROR)
    {
        std::cout << "Failed to bind socket to port " + std::to_string(this->_port) << '\n';
        TcpServer::cleanUp(this->_socket);
        return SOCKET_ERROR;
    }

    if (listen(this->_socket, SOMAXCONN) == SOCKET_ERROR)
    {
        std::cout << "Failed to listen at port " + std::to_string(this->_port) << '\n';
        return SOCKET_ERROR;
    }

    return 0;
}

TcpServer::ReadResult TcpServer::_handleError(
    char* buffer, int& status, int line, const char *function, const char *file
)
{
    switch (errno)
    {
        case EBADF:
        case EFAULT:
        case EINVAL:
        case ENXIO:
            // Fatal error.
            free(buffer);
            throw "Critical error";
        case EIO:
        case ENOBUFS:
        case ENOMEM:
            // Resource acquisition failure or device error.
            free(buffer);
            throw "Resource failure";
        case EINTR:
            // TODO: Check for user interrupt flags.
        case ETIMEDOUT:
        case EAGAIN:
            // Temporary error.
            return ReadResult::Continue;
        case ECONNRESET:
        case ENOTCONN:
            // Connection broken.
            // Return the data we have available and exit
            // as if the connection was closed correctly.
            status = 0;
            break;
        default:
            free(buffer);
            throw "Returned -1";
    }
    return ReadResult::None;
}

void TcpServer::listenAndServe()
{
    if (this->init() != 0)
    {
        return;
    }

#if defined(_WIN32) || defined(_WIN64)
    int status;
    WSADATA wsaData;
    status = WSAStartup(MAKEWORD(2, 2), &wsaData);
    if (status != 0)
    {
        std::cout << "WSAStartup() failed with error #" + std::to_string(status) << '\n';
        return;
    }
#endif

    this->startListener();
    TcpServer::wsaCleanUp();
}

void TcpServer::startListener()
{
    bool listening = true;
    while (listening)
    {
        try
        {
            socklen_t connectionLen = sizeof(this->_socketAddr);
            socket_t connection = accept(this->_socket, (sockaddr*)&this->_socketAddr, &connectionLen);


            if (connection != INVALID_SOCKET)
            {
                    set_socket_blocking(connection, false);

                std::thread newThread(&TcpServer::serveConnection, this, connection);
                newThread.detach();
            }
            else
            {
                std::cout << "Invalid socket connection" << '\n';
            }
        }
        catch (const std::exception& exc)
        {
            std::cout << exc.what() << '\n';
            listening = false;
        }
        catch (const char* exc)
        {
            std::cout << exc << '\n';
            listening = false;
        }
        catch (...)
        {
            std::cout << "Error occurred while listening for socket connection" << '\n';
            listening = false;
        }
    }
}

void TcpServer::serveConnection(const socket_t& connection)
{
    try
    {
        std::string data = TcpServer::recvAll(connection);

        if (!data.empty())
        {
            this->_handler(data, connection);
        }
    }
    catch (const BaseException& exc)
    {
        std::cout << exc.what() << '\n';
    }
    catch (const std::exception& exc)
    {
        std::cout << exc.what() << '\n';
    }
    TcpServer::cleanUp(connection);
}

std::string TcpServer::recvAll(const socket_t& connection)
{
    msg_size_t ret = 0;
    int status = 0;
    unsigned long size = 0;
    std::string data;

    // Poll descriptor structure
    struct pollfd descriptor{};

    // Input stream
    descriptor.fd = connection;
    descriptor.events = POLLIN;

    char* buffer = (char*) calloc(MAX_BUFF_SIZE, sizeof(char));
    do
    {
        // Wait 20 ms
        status = poll(&descriptor, 1, SO_RCVTIMEO);
        if (status == -1)
        {
            this->_handleError(buffer, status, _ERROR_DETAILS_);
        }
        else if (status == 0)
        {
            // Timeout, skip
        }
        else
        {
            // Reset the descriptor.revents to reuse the structure
            if (descriptor.revents == POLLIN)
            {
                descriptor.revents = 0;
            }

            ret = read(connection, buffer, MAX_BUFF_SIZE);
            if (ret > 0)
            {
                data.append(buffer, ret);
                size += ret;
            }
            else if (ret == -1)
            {
                this->_handleError(buffer, status);
            }
        }
    }
    while (status > 0);

    free(buffer);

    if (data.size() != size)
    {
        throw "Invalid request data total size";
    }
    return data;
}

void TcpServer::send(const char* data, const socket_t& connection)
{
    if (::send(connection, data, std::strlen(data), 0) == SOCKET_ERROR)
    {
        throw "Failed to send bytes to socket connection";
    }
}

void TcpServer::write(const char* data, size_t bytesToWrite, const socket_t& connection)
{
    if (::write(connection, data, bytesToWrite) == -1)
    {
        throw "Failed to send bytes to socket connection";
    }
}

int TcpServer::closeSocket(const socket_t& socket)
{
#if defined(_WIN32) || defined(_WIN64)
    return ::closesocket(socket);
#elif defined(__unix__) || defined(__linux__)
    return ::close(socket);
#endif
}

void TcpServer::wsaCleanUp()
{
#if defined(_WIN32) || defined(_WIN64)
    WSACleanup();
#endif
}

void TcpServer::cleanUp(const socket_t& socket)
{
    if (TcpServer::closeSocket(socket) == SOCKET_ERROR)
    {
        std::cout << "Failed to close socket connection" << '\n';
    }
    TcpServer::wsaCleanUp();
}

我认为，问题出在recvAll功能上。我尝试poll在第 19 行将 ' 超时设置为超过 20 毫秒（SO_RCVTIMEO 等于 20），并且服务器开始读取更大的请求。但是增加超时不是解决方案，因为我不知道 1 GB 或更大的请求需要什么超时值。这是一个recvAll函数代码：

std::string TcpServer::recvAll(const socket_t& connection)
{
    msg_size_t ret = 0;
    int status = 0;
    unsigned long size = 0;
    std::string data;

    // Poll descriptor structure
    struct pollfd descriptor{};

    // Input stream
    descriptor.fd = connection;
    descriptor.events = POLLIN;

    char* buffer = (char*) calloc(MAX_BUFF_SIZE, sizeof(char));
    do
    {
        // Wait 20 ms
        status = poll(&descriptor, 1, SO_RCVTIMEO);
        if (status == -1)
        {
            this->_handleError(buffer, status, _ERROR_DETAILS_);
        }
        else if (status == 0)
        {
            // Timeout, skip
        }
        else
        {
            // Reset the descriptor.revents to reuse the structure
            if (descriptor.revents == POLLIN)
            {
                descriptor.revents = 0;
            }

            ret = read(connection, buffer, MAX_BUFF_SIZE);
            if (ret > 0)
            {
                data.append(buffer, ret);
                size += ret;
            }
            else if (ret == -1)
            {
                this->_handleError(buffer, status);
            }
        }
    }
    while (status > 0);

    free(buffer);

    if (data.size() != size)
    {
        throw "Invalid request data total size";
    }
    return data;
}

如何修复此功能，或者服务器如何使其完全读取大型请求？

标签： c++socketstcp

解决方案

您需要继续调用recv，直到请求完成。TCP 无法告诉您请求何时完成 - 这不是它所做的事情（好吧，它可以告诉您连接何时关闭，但这会关闭连接）。因此，您需要另一种方法来知道何时停止阅读。

目前，只要 20 毫秒内没有任何数据，您就会停止阅读。这不是很可靠。网络中断可能会导致 20 毫秒的延迟，然后您会过早停止。

在 HTTP 中，客户端通过使用 Content-Length 标头告诉您请求正文的长度。请求由请求标头和请求正文组成，因此服务器需要读取标头（在空行后停止），然后查看标头以查看请求正文的长度，然后读取正文（停止在那个字节数之后）。

推荐阅读