如何使用 recv 或 C++ 中的读取功能从 tcp 套接字读取大请求

Question

我正在开发一个将接收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, '', 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功能上。我尝试在第 20 行将poll的超时设置为 19 毫秒以上(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;
}

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

Answer 1

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

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

在 HTTP 中，客户端通过使用内容长度标头告诉您请求正文的长度。请求由请求标头和请求正文组成，因此服务器需要读取标头(在空行后停止)，然后查看标头以查看请求正文的长度，然后读取正文(在该字节数后停止)。