带有异步 I/O 的消息排序 (epoll)

Question

假设我已经实现了一个基于 epoll 的 TCP 服务器，其中每个线程都在运行与下面非常相似的东西（取自 epoll 手册页，其中 kdpfd 是 epoll 文件描述符，侦听器是一个正在侦听端口的套接字):

struct epoll_event ev, *events;
for(;;) {
    nfds = epoll_wait(kdpfd, events, maxevents, -1);
    for(n = 0; n < nfds; ++n) {
        if(events[n].data.fd == listener) {
            client = accept(listener, (struct sockaddr *) &local,
                            &addrlen);
            if(client < 0){
                perror("accept");
                continue;
            }
            setnonblocking(client);
            ev.events = EPOLLIN | EPOLLET;
            ev.data.fd = client;
            if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, client, &ev) < 0) {
                fprintf(stderr, "epoll set insertion error: fd=%d0,
                        client);
                return -1;
            }
        }
        else
            do_use_fd(events[n].data.fd);
    }
}

对于上面的do_use_fd(events[n].data.fd)，假设我们要将收到的所有内容写入标准输出：

int do_use_fd(int fd) {
    int err;
    char buf[512];
    while ((err = read(fd, buf, 512)) > 0) {
        write(1, buf, err);
    }

    if (err == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
       // do some error handling and return -1

    return 0;
}

现在，假设我有 10k+ 连接，所有这些连接都在很长一段时间内向我发送了大量消息。假设我的客户每隔几秒就向我发送消息hello, my name is {client's name}。假设（不知何故）这条消息足够大，以至于它必须作为多个数据包传输。

因此，read(fd, buf, 512) 可能偶尔会返回 -1 并带有 errno 指示它会阻塞。因此，我认为上述解决方案最终可能会得到以下输出：

hello, my nam
hello, my name is Pau
e is John Le
hello, my name is Geo
nnon
l McCartney
rge
hello, my name is Ringo
Starr
 Harrison

因为一旦一个连接上的读取阻塞，另一个读取就可以在另一个连接上开始。相反，我希望打印以下内容：

hello, my name is John Lennon
hello, my name is Paul McCartney
hello, my name is George Harrison
hello, my name is Ringo Starr

是否有处理此问题的推荐方法？一种选择是为每个连接保留一个缓冲区，并检查消息是否已完成，并且仅在发生这种情况时打印。但是有 10k+ 个连接，这是个好主意吗？一方面，有些事情告诉我这个解决方案不能很好地扩展。另一方面，如果消息只有 500 字节，有 10k 连接，则此解决方案仅占用 5MB。

提前致谢。

Answer 1

我认为在您的情况下，每个连接都使用一个缓冲区是可以的。然而，为每个不完整的消息创建一个缓冲区可能更优雅。这意味着您必须以某种方式知道您的消息何时完成，因此您需要一个小协议，例如使用长度字段或终止符（并且可能在一定时间后超时以终止不完整的消息）。这也将保证没有分配未使用的内存，因为缓冲区可以在消息完成并传递后立即释放。例如，您可以使用连接 5 元组作为键通过 hashmap 访问这些缓冲区。如果您决定使用消息绑定标识符，这当然会产生额外的开销，您甚至可以从用于一次传输多条消息的单个 tcp 连接中解复用消息。

如果您需要在这些消息中强制执行排序，则必须详细说明您的情况，因为在许多情况下排序是一个棘手的问题。

编辑：对不起，我现在有很多事情要做，所以我不能尽快回答。你是对的，使用基于连接的方法更容易。使用的连接越稀疏，基于消息的优势就越大。如果您可以期望所有连接始终接收消息，那只是开销。如果连接有时空闲一段时间，它可能会大大减少内存使用量。另请注意，您的应用程序内存使用量不再随客户端数量而是随消息数量而变化，这通常很好，因为消息速率通常会有所不同。您对 TCP 流的排序也是正确的。只要您一次只通过连接发送一条完整的消息，TCP 就会确保排序。一些应用程序，例如 HTTP2 重用相同的 TCP 连接来同时发送多个消息。在这种情况下，TCP 将无济于事，因为消息片段以未指定的顺序到达，您需要将它们解复用（例如通过 HTTP2 中的流 ID）。