如何在C/C++中进行TCP连接池

Question

我正在设计一个带有C++的分布式服务器/客户端系统,其中许多客户端通过TCP向服务器发送请求,服务器抛出一个线程来处理请求并发回它的响应.在我的用例中,只有有限数量的客户端将访问服务器,我需要非常高的性能.从客户端和服务器发送的数据都很小,但非常频繁.因此,在使用后创建连接并将其拆除是很昂贵的.所以我想使用连接缓存来解决这个问题:一旦创建连接,它将被存储在缓存中供将来使用.(假设客户端的数量不会超出缓存的大小).

我的问题是:

1. 我看到有人说连接池是一种客户端技术.如果此连接池仅用于客户端,则第一次连接到服务器并发送数据.此连接操作会触发服务器端的accept()函数,该函数返回用于从客户端接收的套接字套接字.因此,当客户端想要使用现有连接(在缓存中)时,它不会建立新连接,而只是发送数据.问题是,如果没有建立连接,谁会在服务器端触发accept()并抛出一个线程？
2. 如果还需要在服务器端实现连接池,我怎么知道请求的来源？由于只能从accept()获取客户端地址,但同时accept()已经为该请求创建了一个新套接字,因此没有必要使用缓存连接.

任何答案和建议将不胜感激.或者任何人都可以给我一个连接池或连接缓存的例子？

Answer 1

我看到有人说连接池是一种客户端技术....如果没有建立连接,谁将在服务器端触发accept()并抛出一个线程？

首先,连接池不仅仅是一种客户端技术; 这是一种连接模式技术.它适用于两种类型的对等体("服务器"和"客户端").

其次,accept不需要调用启动线程.程序可以出于他们喜欢的任何原因启动线程......他们可以在大规模并行化的线程创建循环中启动线程以启动更多线程.(编辑:我们称之为"叉炸弹")

最后,高效的线程池实现不会为每个客户端启动一个线程.每个线程通常占用512KB-4MB(计算堆栈空间和其他上下文信息),所以如果你有10000个客户端,每个占用那么多,那就浪费了很多内存.

我想这样做,但只是不知道如何在多线程情况下这样做.

你不应该在这里使用多线程...至少,直到你有一个使用单个线程的解决方案,并且你认为它不够快.目前你没有那些信息; 你只是在猜测,而猜测并不能保证优化.

在世纪之交,有FTP服务器解决了C10K问题 ; 他们能够在任何给定时间处理10000个客户端,浏览,下载或空闲,就像用户在FTP服务器上所做的那样.他们不是通过使用线程解决了这个问题,而是使用非阻塞和/或异步套接字和/或调用.

为了澄清,这些Web服务器在单个线程上处理了数千个连接!一种典型的方法是使用select,但我并不特别喜欢这种方法,因为它需要一系列相当丑陋的循环.我更喜欢ioctlsocket用于Windows和fcntl其他POSIX操作系统将文件描述符设置为非阻塞模式,例如:

#ifdef WIN32
ioctlsocket(fd, FIONBIO, (u_long[]){1});
#else
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
#endif

在这一点上,recv而read不会在运行时阻止fd; 如果没有可用的数据,它们将立即返回错误值,而不是等待数据到达.这意味着你可以循环多个套接字.

如果还需要在服务器端实现连接池,我怎么知道请求的来源？

存储客户端fd一起端了struct sockaddr_storage,你需要存储有关客户的任何其他状态信息,在struct但你申报你的感觉.如果这最终是4KB(这是一个相当大的struct,通常大约他们需要得到的那么大)那么其中10000个将只占用大约40000KB(~40MB).即便是今天的手机也应该没有问题处理.考虑根据您的需求完成以下代码:

struct client {
    struct sockaddr_storage addr;
    socklen_t addr_len;
    int fd;
    /* Other stateful information */
};

#define BUFFER_SIZE 4096
#define CLIENT_COUNT 10000

int main(void) {
    int server;
    struct client client[CLIENT_COUNT] = { 0 };
    size_t client_count = 0;
    /* XXX: Perform usual bind/listen */
    #ifdef WIN32
    ioctlsocket(server, FIONBIO, (u_long[]){1});
    #else
    fcntl(server, F_SETFL, fcntl(server, F_GETFL, 0) | O_NONBLOCK);
    #endif

    for (;;) {
        /* Accept connection if possible */
        if (client_count < sizeof client / sizeof *client) {
            struct sockaddr_storage addr = { 0 };
            socklen_t addr_len = sizeof addr;
            int fd = accept(server, &addr, &addr_len);
            if (fd != -1) {
#               ifdef WIN32
                ioctlsocket(fd, FIONBIO, (u_long[]){1});
#               else
                fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
#               endif
                client[client_count++] = (struct client) { .addr = addr
                                                         , .addr_len = addr_len
                                                         , .fd = fd };
            }
        }
        /* Loop through clients */
        char buffer[BUFFER_SIZE];
        for (size_t index = 0; index < client_count; index++) {
            ssize_t bytes_recvd = recv(client[index].fd, buffer, sizeof buffer, 0);
#           ifdef WIN32
            int closed = bytes_recvd == 0
                      || (bytes_recvd < 0 && WSAGetLastError() == WSAEWOULDBLOCK);
#           else
            int closed = bytes_recvd == 0
                      || (bytes_recvd < 0 && errno == EAGAIN) || errno == EWOULDBLOCK;
#           endif
            if (closed) {
                close(client[index].fd);
                client_count--;
                memmove(client + index, client + index + 1, (client_count - index) * sizeof client);
                continue;
            }
            /* XXX: Process buffer[0..bytes_recvd-1] */
        }

        sleep(0); /* This is necessary to pass control back to the kernel,
                   * so it can queue more data for us to process
                   */
    }
}

假设您想在客户端池连接,代码看起来非常相似,除了显然不需要accept相关的代码.假设您有一个client您想要的数组connect,您可以使用非阻塞连接调用一次执行所有连接,如下所示:

size_t index = 0, in_progress = 0;
for (;;) {
    if (client[index].fd == 0) {
        client[index].fd = socket(/* TODO */);
#       ifdef WIN32
        ioctlsocket(client[index].fd, FIONBIO, (u_long[]){1});
#       else
        fcntl(client[index].fd, F_SETFL, fcntl(client[index].fd, F_GETFL, 0) | O_NONBLOCK);
#       endif
    }
#   ifdef WIN32
    in_progress += connect(client[index].fd, (struct sockaddr *) &client[index].addr, client[index].addr_len) < 0
                && (WSAGetLastError() == WSAEALREADY
                ||  WSAGetLastError() == WSAEWOULDBLOCK
                ||  WSAGetLastError() == WSAEINVAL);
#   else
    in_progress += connect(client[index].fd, (struct sockaddr *) &client[index].addr, client[index].addr_len) < 0
                && (errno == EALREADY
                ||  errno == EINPROGRESS);
#   endif
    if (++index < sizeof client / sizeof *client) {
        continue;
    }
    index = 0;
    if (in_progress == 0) {
        break;
    }
    in_progress = 0;
}

至于优化,假设这应该能够处理10000个客户端,可能有一些小的调整,你不应该需要多个线程.

尽管如此,通过将来自mutex集合的项与clients 关联并且在非阻塞套接字操作之前与非阻塞相关联pthread_mutex_trylock,上述循环可以适于在处理同一组套接字的同时在多个线程中同时运行.这为所有符合POSIX标准的平台提供了工作模型,无论是Windows,BSD还是Linux,但它并不是一个完美的最佳平台.为了实现最优化,我们必须进入异步世界,因系统而异:

• Windows使用WSA*具有回调功能.
• BSD和Linux使用的有些相似kqueue,并epoll分别.

编写前面提到的"非阻塞套接字操作"抽象可能是值得的,因为两个异步机制在它们的接口方面差异很大.与其他所有内容一样,遗憾的是我们必须编写抽象,以便我们的Windows相关代码在POSIX兼容系统上保持清晰.作为奖励,这将允许我们将服务器处理(即accept随后的任何事情)与客户端处理(即随后的任何事情)混合在一起connect,因此我们的服务器循环可以成为客户端循环(反之亦然).