当新消息到达时，如何使UDP套接字替换旧消息（尚未recv()'d）？

Question

首先，用一些上下文来解释为什么我采用“UDP 采样”路线：
我想对未知时间段内快速生成的数据进行采样。我想要采样的数据位于另一台机器上，而不是消耗数据的机器上。我在两者之间有专用的以太网连接，因此带宽不是问题。我遇到的问题是消耗数据的机器比生成数据的机器慢得多。另一个限制是，虽然我没有获得所有样本（它们只是样本）也没关系，但我必须获得最后一个样本。

我的第一个解决方案是让数据生产者为每个生成的样本发送一个 UDP 数据报，并让数据消费者尝试获取它可以获取的样本，并在 UDP 套接字已满时让其他样本被套接字层丢弃。此解决方案的问题在于，当新的 UDP 数据报到达且套接字已满时，将丢弃新数据报，而不是旧数据报。所以我不保证能得到最后一张！

我的问题是：有没有办法让 UDP 套接字在新数据报到达时替换旧数据报？

接收器目前是一台 Linux 机器，但将来可能会改用另一种类 UNIX 操作系统（Windows 可能是可能的，因为它实现了 BSD 套接字，但可能性较小）
理想的解决方案是使用广泛的机制（如setsockopt()）来工作。

PS：我想到了其他解决方案，但它们更复杂（涉及对发送者的大量修改），因此我想首先对我所要求的可行性有一个明确的状态！ :)

更新： - 我知道接收机器上的操作系统可以处理网络负载+发送者生成的流量的重组。只是它的默认行为是当套接字缓冲区已满时丢弃新的数据报。由于接收过程中的处理时间，我知道无论我做什么它都会变满（在套接字缓冲区上浪费一半内存不是一个选项:)）。
- 我真的很想避免让辅助进程执行操作系统在数据包分发时可以执行的操作，并浪费资源只是在 SHM 中复制消息。
- 我在修改发件人时看到的问题是，我有权访问的代码只是一个 PleaseSendThisData() 函数，它不知道这可能是很长时间之前最后一次调用它，所以我不知道在那一端看到任何可行的技巧......但我愿意接受建议！ :)

如果真的没有办法改变 BSD 套接字中的 UDP 接收行为，那么好吧......告诉我，我准备接受这个可怕的事实，并且当我去时将开始研究“辅助进程”解决方案回到正题:)

Answer 1

只需将套接字设置为非阻塞，然后循环 recv() 直到它返回 0 与 errno == EAGAIN。然后处理你得到的最后一包，冲洗并重复。

Answer 2

我同意“咖啡馆”。
将套接字设置为非阻塞模式。

每当您在套接字上收到某些内容时 - 循环读取直到没有任何内容为止。然后处理最后读取的数据报。

只有一点需要注意：您应该为套接字设置一个大的系统接收缓冲区

int nRcvBufSize = 5*1024*1024; // or whatever you think is ok
setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char*) &nRcvBufSize, sizeof(nRcvBufSize));

Answer 3

仅在侦听器端很难完全正确，因为它实际上可能会错过网络接口芯片中的最后一个数据包，这将使您的程序没有机会看到它。

操作系统的 UDP 代码将是尝试处理此问题的最佳位置，因为即使它决定丢弃新数据包，它也会收到新数据包，因为它已经有太多排队的数据包。然后它可以决定丢弃旧的或丢弃新的，但我不知道如何告诉它这就是你想要它做的事情。

您可以尝试在接收器上处理此问题，方法是让一个程序或线程始终尝试读取最新的数据包，而另一个程序或线程始终尝试获取最新的数据包。根据您是作为两个单独的程序还是作为两个线程执行此操作，如何执行此操作会有所不同。

作为线程，您需要一个互斥锁（信号量或类似的东西）来保护指向用于保存 1 UDP 有效负载的结构的指针（或引用）以及您想要的任何其他内容（大小、发送者 IP、发送者端口、时间戳等） ）。

实际从套接字读取数据包的线程会将数据包的数据存储在结构中，获取保护该指针的互斥体，将当前指针替换为指向刚刚创建的结构体的指针，释放互斥体，向处理器线程发出信号有事情要做，然后清除它刚刚获得指针的结构，并使用它来保存进入的下一个数据包。

实际处理数据包有效负载的线程应该等待来自其他线程的信号和/或定期等待（500 毫秒左右可能是一个很好的起点，但由您决定）并获取互斥锁，将其指向 UDP 负载结构的指针与现有的结构交换，释放互斥锁，然后如果该结构有任何数据包它应该处理它的数据，然后等待下一个信号。如果它没有任何数据，它应该继续等待下一个信号。

处理器线程可能应该以低于 UDP 侦听器的优先级运行，以便侦听器不太可能错过数据包。当处理最后一个数据包（您真正关心的数据包）时，处理器不会被中断，因为没有新的数据包可供侦听器听到。

您可以通过使用队列而不仅仅是单个指针作为两个线程的交换位置来扩展此功能。单个指针只是一个长度为1的队列，非常容易处理。

您还可以通过尝试让侦听器线程检测是否有多个数据包在等待来扩展此功能，并且仅将最后一个数据包实际放入处理器线程的队列中。执行此操作的方式因平台而异，但如果您使用 *nix，则对于没有任何等待的套接字，这应该返回 0：

while (keep_doing_this()) {
    ssize_t len = read(udp_socket_fd, my_udp_packet->buf, my_udp_packet->buf_len); 
    // this could have been recv or recvfrom
    if (len < 0) {
        error();
    }
    int sz;
    int rc = ioctl(udp_socket_fd, FIONREAD, &sz);
    if (rc < 0) {
        error();
    }
    if (!sz) {
        // There aren't any more packets ready, so queue up the one we got
        my_udp_packet->current_len = len;

        my_udp_packet = swap_udp_packet(my_ucp_packet);
        /* swap_udp_packet is code you would have to write to implement what I talked
           about above. */

        tgkill(this_group, procesor_thread_tid, SIGUSR1);
    } else if (sz > my_udp_packet->buf_len) {
        /* You could resize the buffer for the packet payload here if it is too small.*/
    }
}

必须为每个线程分配 udp_packet，并为交换指针分配 1。如果您使用队列进行交换，那么您必须为队列中的每个位置提供足够的 udp_packets ——因为指针只是一个长度为 1 的队列，因此它只需要 1。

如果您使用的是 POSIX 系统，则考虑不使用实时数据包信号用于信令，因为它们排队。使用常规信号将允许您将多次收到信号视为与仅收到一次信号相同，直到信号被处理，而实时信号会排队。定期醒来检查队列还可以让您处理在检查是否有任何新数据包之后、调用 pause 等待信号之前最后一个信号到达的可能性。

Answer 4

另一个想法是有一个专用的读取器进程，它除了在套接字上循环并将传入数据包读取到共享内存中的循环缓冲区之外什么都不做（您必须担心正确的写入顺序）。类似 kfifo 的东西。非阻塞在这里也很好。新数据覆盖旧数据。然后其他进程将始终可以访问队列头部的最新块以及所有尚未覆盖的先前块。

对于一个简单的单向读者来说可能太复杂了，只是一种选择。

Answer 5

我很确定这是一个无法解决的问题，与两军问题密切相关。

我可以想到一个肮脏的解决方案：建立一个 TCP“控制”边带连接，该连接携带最后一个数据包，这也是“结束传输”指示。否则，您需要使用工程方法中提到的更通用的实用方法之一。

Answer 6

这是一个老问题，但您基本上想将套接字队列（先进先出）转换为堆栈（后进先出）。这是不可能的，除非你想摆弄内核。

您需要将数据报从内核空间移动到用户空间，然后进行处理。最简单的方法是像这样的循环...

1. 阻塞，直到套接字上有数据（参见 select、poll、epoll）
2. 排空套接字，根据您自己的选择策略存储数据报
3. 处理存储的数据报
4. 重复