为什么特定的 UDP 消息总是低于特定的缓冲区大小？

Question

3 条不同的消息以不同的速率发送到同一个端口：

消息  大小（字节）  每发送一次 kbd>传输速度
高           232                10 ms          100Hz
中等     148                                                                          50Hz                    
低20                   60 ms         16.6Hz                

我只能每 ~ 6 毫秒处理一条消息。
单线程。阻止读取。

---

发生了一种奇怪的情况，我对此没有任何解释。
当我将接收缓冲区设置为 4,799 字节时，我的所有低速消息都会被丢弃。
我看到可能有一两个得到处理，然后什么都没有。

当我将接收缓冲区设置为4,800（或更高！）时，似乎所有低速消息都开始得到处理。我看到大约 16/17 秒。

---

这一直被观察到。发送数据包的应用程序总是在接收应用程序之前启动。接收应用程序在创建套接字之后和开始处理之前总是有很长的延迟。所以处理开始时缓冲区总是满的，而且每次发生测试时它都不是同一个开始缓冲区。这是因为套接字是在发送者已经发送消息之后创建的，因此接收者可能会在发送周期的中间开始监听。

为什么将接收缓冲区大小增加一个字节，会导致低速消息处理发生巨大变化？

我建了一个表格来更好地可视化预期的处理：

随着其中一些消息得到处理，更多消息可能会被放入队列而不是被丢弃。

尽管如此，我希望4,799 字节缓冲区的行为方式与4,800 字节相同。

但这不是我观察到的。

---

我认为这个问题与低速消息与其他两条消息同时发送的事实有关。它总是在高速/中速消息之后接收。 （这已经通过wireshark得到证实）。

例如，假设缓冲区一开始是空的，很明显低速消息需要比其他消息更长的排队时间。
*每 6 毫秒 1 条消息大约是每 30 毫秒 5 条消息。

这仍然不能解释缓冲区大小。

我们正在运行 VxWorks，并使用他们的 sockLib，它是 Berkeley 套接字的一个实现。这是我们创建套接字的样子的 sn-p：
SOCKET_BUFFER_SIZE 是我要更改的内容。

struct sockaddr_in tSocketAddress;                          // Socket address
int     nSocketAddressSize = sizeof(struct sockaddr_in);    // Size of socket address structure
int     nSocketOption = 0;

// Already created
if (*ptParameters->m_pnIDReference != 0)
    return FALSE;

// Create UDP socket
if ((*ptParameters->m_pnIDReference = socket(AF_INET, SOCK_DGRAM, 0)) == ERROR)
{
    // Error
    CreateSocketMessage(ptParameters, "CreateSocket: Socket create failed with error.");

    // Not successful
    return FALSE;
}

// Valid local address
if (ptParameters->m_szLocalIPAddress != SOCKET_ADDRESS_NONE_STRING && ptParameters->m_usLocalPort != 0)
{
    // Set up the local parameters/port
    bzero((char*)&tSocketAddress, nSocketAddressSize);
    tSocketAddress.sin_len = (u_char)nSocketAddressSize;
    tSocketAddress.sin_family = AF_INET;
    tSocketAddress.sin_port = htons(ptParameters->m_usLocalPort);

    // Check for any address
    if (strcmp(ptParameters->m_szLocalIPAddress, SOCKET_ADDRESS_ANY_STRING) == 0)
        tSocketAddress.sin_addr.s_addr = htonl(INADDR_ANY);
    else
    {
        // Convert IP address for binding
        if ((tSocketAddress.sin_addr.s_addr = inet_addr(ptParameters->m_szLocalIPAddress)) == ERROR)
        {
            // Error
            CreateSocketMessage(ptParameters, "Unknown IP address.");

            // Cleanup socket
            close(*ptParameters->m_pnIDReference);
            *ptParameters->m_pnIDReference = ERROR;

            // Not successful
            return FALSE;
        }
    }

    // Bind the socket to the local address
    if (bind(*ptParameters->m_pnIDReference, (struct sockaddr *)&tSocketAddress, nSocketAddressSize) == ERROR)
    {
        // Error
        CreateSocketMessage(ptParameters, "Socket bind failed.");

        // Cleanup socket
        close(*ptParameters->m_pnIDReference);
        *ptParameters->m_pnIDReference = ERROR;

        // Not successful
        return FALSE;
    }
}

// Receive socket
if (ptParameters->m_eType == SOCKTYPE_RECEIVE || ptParameters->m_eType == SOCKTYPE_RECEIVE_AND_TRANSMIT)
{
    // Set the receive buffer size
    nSocketOption = SOCKET_BUFFER_SIZE;
    if (setsockopt(*ptParameters->m_pnIDReference, SOL_SOCKET, SO_RCVBUF, (char *)&nSocketOption, sizeof(nSocketOption)) == ERROR)
    {
        // Error
        CreateSocketMessage(ptParameters, "Socket buffer size set failed.");

        // Cleanup socket
        close(*ptParameters->m_pnIDReference);
        *ptParameters->m_pnIDReference = ERROR;

        // Not successful
        return FALSE;
    }
}

并且在无限循环中调用的套接字接收：
*缓冲区大小肯定足够大

int SocketReceive(int nSocketIndex, char *pBuffer, int nBufferLength)
{
    int nBytesReceived = 0;
    char szError[256];

    // Invalid index or socket
    if (nSocketIndex < 0 || nSocketIndex >= SOCKET_COUNT || g_pnSocketIDs[nSocketIndex] == 0)
    {
        sprintf(szError,"SocketReceive: Invalid socket (%d) or ID (%d)", nSocketIndex, g_pnSocketIDs[nSocketIndex]);
        perror(szError);
        return -1;
    }

    // Invalid buffer length
    if (nBufferLength == 0)
    {
        perror("SocketReceive: zero buffer length");
        return 0;
    }

    // Send data
    nBytesReceived = recv(g_pnSocketIDs[nSocketIndex], pBuffer, nBufferLength, 0);

    // Error in receiving
    if (nBytesReceived == ERROR)
    {
        // Create error string
        sprintf(szError, "SocketReceive: Data Receive Failure: <%d> ", errno);

        // Set error message
        perror(szError);

        // Return error
        return ERROR;
    }

    // Bytes received
    return nBytesReceived;
}

任何关于为什么将缓冲区大小增加到 4,800 会导致成功且一致地读取低速消息的线索？

Answer 1

对于为什么SO_RCVBUF 大小为 4799 会导致丢失低速消息而大小为 4800 的问题的基本答案是，随着 UDP 数据包的混合进入，它们的到来速率在，您处理传入数据包的速率，以及 vxWorks 内核中 mbuff 的大小和集群编号允许足够的网络堆栈吞吐量，从而不会丢弃较大大小的低速消息。

上面评论中提到的setsockopt() 手册页中的http://www.vxdev.com/docs/vx55man/vxworks/ref/sockLib.html#setsockopt 中的SO_SNDBUF 选项描述说明了指定的大小和对mbuff 使用的影响：

设置缓冲区最大大小的效果（对于SO_SNDBUF 和 SO_RCVBUF，如下所述）实际上并不是分配 mbufs 来自 mbuf 池。相反，效果是设置高水位线 在协议数据结构中，稍后用于限制 mbuf 分配的数量。

UDP 数据包是离散的单元。如果您发送 10 个大小为 232 的数据包，则在连续内存中不被视为 2320 字节的数据。相反，网络堆栈中有 10 个内存缓冲区，因为 UDP 是离散的数据包，而 TCP 是连续的字节流。

请参阅 DDS 社区网站中的 How do I tune the network buffering in VxWorks 5.4?，其中讨论了 mbuff 大小和网络集群混合的相互依赖关系。

请参阅 DDS 社区网站中的 How do I resolve a problem with VxWorks buffers?。

请参阅此 PDF 幻灯片演示文稿，A New Tool to study Network Stack Exhaustion in VxWorks 从 2004 年开始讨论使用各种工具（例如 mBufShow 和 inetStatShow）来了解网络堆栈中发生的情况。

Answer 2

如果不详细分析您的 UDP 消息发送路径上的每个网络堆栈实现，几乎不可能说明结果行为。

UDP 实现可以自行决定丢弃任何数据包。通常，当堆栈得出结论认为需要丢弃数据包才能接收新数据包时，就会发生这种情况。没有正式要求丢弃的数据包是最旧的或最新接收的。也可能是由于内部内存管理策略，某个大小类受到的影响更大。

从涉及的 IP 堆栈中，最有趣的是接收机器上的那个。

如果您将接收端更改为有一个接收缓冲区，该缓冲区将在几秒钟内充满预期的消息，您肯定会获得更好的接收体验。我会从至少 10k 开始。

从 4,799 到 4,800 时观察到的行为“变化”可能是由于后者只允许在需要再次删除其中一条小消息之前接收它，而较小的大小只是导致它被稍微删除早些时候。如果接收应用程序足够快地读取待处理消息，您将在一种情况下收到小消息，而在另一种情况下不会收到小消息。