[ZMQ] ZMQ_PUB和ZMQ_SUB 关于缓存区的问题

问题描述

环境

1. ubuntu18.04

2. zmq version:4.3.3

设置了一个发送端和一个接收端，发送端是ZMQ_PUB，接收端ZMQ_SUB，并且通过zmq_setsockopt设置了ZMQ_SNDHWM,ZMQ_RCVHWM,ZMQ_SNDBUF,ZMQ_RCVBUF。

设置了代码内的缓存区大小

设置完成后，通过实际代码测试，并没有起到作用，接收端依旧会收到很久的消息，具体可以跑下面的代码来查看。
接收端一直持续接收到3757帧之后刷新。

设置了内核套接字缓存区

$ sudo sysctl -a | grep mem      
net.core.optmem_max = 20480
net.core.rmem_default = 212992
net.core.rmem_max = 212992
net.core.wmem_default = 212992
net.core.wmem_max = 212992
net.ipv4.fib_sync_mem = 524288
net.ipv4.igmp_max_memberships = 20
net.ipv4.tcp_mem = 475113	633486	950226
net.ipv4.tcp_rmem = 4096	131072	6291456
net.ipv4.tcp_wmem = 4096	16384	4194304
net.ipv4.udp_mem = 950229	1266972	1900458
net.ipv4.udp_rmem_min = 4096
net.ipv4.udp_wmem_min = 4096
...

执行sudo sysctl -a | grep mem可以看到默认的写缓存区大小为212992，我这里修改为5000后重新测试，发现接收端接收到数据比只设置代码内的缓存区接收到的数据更新一些。

接收端一直持续接收到22帧之后刷新。

总结原因

目前确定是因为内核缓存区导致的旧数据一直存在，这个很合理，但是不清楚为什么zmq的zmq_setsockopt设置内核缓存区无效。
有了解的大神请解答一下。

代码示例

发送端

//
// Created by kongshui on 22-8-23.
//
#include <unistd.h>
#include <iostream>
#include <zmq.hpp>

void qqq()
{
    const int first_count = 10000000;
    int hwm = 5;
    int time = 10;

    auto con = zmq_ctx_new();

    // Set up bind socket
    void *pub_socket = zmq_socket(con, ZMQ_PUB);
    zmq_setsockopt(pub_socket, ZMQ_SNDTIMEO, &time, sizeof(time));
    zmq_setsockopt(pub_socket, ZMQ_RCVTIMEO, &time, sizeof(time));
    zmq_setsockopt(pub_socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));
    zmq_setsockopt(pub_socket, ZMQ_RCVHWM, &hwm, sizeof(hwm));
    zmq_bind(pub_socket, "ipc:///tmp/qos_out");

    int buf_size = 5000;
    auto rrr = zmq_setsockopt(pub_socket, ZMQ_SNDBUF, &buf_size, sizeof(buf_size));
    printf("rrr: %d", rrr);
    hwm = 10;
    size_t hwm_size = sizeof(hwm);
    rrr = zmq_getsockopt(pub_socket, ZMQ_SNDBUF, &hwm, &hwm_size);
    printf("rrr: %d, size: %d\n", rrr, hwm);

    sleep(3);

    // Send messages
    int send_count = 0;
    int recv_count = 0;

    std::string msg;
    while (send_count < first_count)
    {
        msg = std::to_string(send_count);
        int res = zmq_send(pub_socket, msg.c_str(), msg.length(), ZMQ_DONTWAIT);

        printf("res:%d, msg:%s", res, msg.c_str());
        if (msg.length() == res)
            ++send_count;
        else
            break;

        usleep(1000);
    }

    // Clean up
    zmq_close(pub_socket);
}

void ttt()
{
    zmq::context_t context(1);

    const int first_count = 15;
    int hwm = 10;
    int buf_size = 10000;
    zmq::socket_t pub_socket(context, ZMQ_PUB);
    zmq_setsockopt(pub_socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));
    zmq_setsockopt(pub_socket, ZMQ_SNDBUF, &buf_size, sizeof(buf_size));

    pub_socket.bind("ipc:///tmp/qos_in");

    std::string msg = "111";

    while (true)
    {
        auto res = zmq_send(pub_socket, msg.c_str(), msg.length(), ZMQ_DONTWAIT);
        printf("res:%d\n", res);
    }
}

int main(int, char **)
{
    qqq();

    return 0;
}

接收端

//
// Created by kongshui on 22-8-23.
//
#include <unistd.h>
#include <iostream>
#include <zmq.hpp>

void qqq()
{
    const int first_count = 15;
    int hwm = 5;
    int time = 10;

    auto con = zmq_ctx_new();

    hwm = 10;
    // Set up connect socket
    void *sub_socket = zmq_socket(con, ZMQ_SUB);
    zmq_setsockopt(sub_socket, ZMQ_SNDTIMEO, &time, sizeof(time));
    zmq_setsockopt(sub_socket, ZMQ_RCVTIMEO, &time, sizeof(time));
    zmq_setsockopt(sub_socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));
    zmq_setsockopt(sub_socket, ZMQ_RCVHWM, &hwm, sizeof(hwm));

    int buf_size = 5000;
    std::string topic = "www";

    auto rrr = zmq_setsockopt(sub_socket, ZMQ_RCVBUF, &buf_size, sizeof(buf_size));
    printf("rrr: %d", rrr);
    hwm = 10;
    size_t hwm_size = sizeof(hwm);
    rrr = zmq_getsockopt(sub_socket, ZMQ_RCVBUF, &hwm, &hwm_size);
    printf("rrr: %d, size: %d\n", rrr, hwm);


    zmq_connect(sub_socket, "ipc:///tmp/qos_out");
//    zmq_setsockopt(sub_socket, ZMQ_SUBSCRIBE, topic.c_str(), topic.size());
    zmq_setsockopt(sub_socket, ZMQ_SUBSCRIBE, nullptr, 0);

    sleep(3);

    // Send messages
    int send_count = 0;
    int recv_count = 0;

    std::string msg;

    char res[1024] = {0};
    while (zmq_recv(sub_socket, res, 1021, ZMQ_DONTWAIT) > 0)
    {
        ++recv_count;
        printf("res:%s", res);
        printf("first_count: %d, recv_count: %d\n", first_count, recv_count);
        memset(res, 0, sizeof(res));
    }


    while (1)
    {
        if (zmq_recv(sub_socket, res, 1021, ZMQ_DONTWAIT) > 0)
        {
            ++recv_count;
            printf("res:%s", res);
            memset(res, 0, sizeof(res));
            printf("first_count: %d, recv_count: %d\n", first_count, recv_count);

            sleep(1);
        }
    }

    // Clean up
    zmq_close(sub_socket);
}

void ttt()
{
    zmq::context_t context(1);

    const int first_count = 15;
    int hwm = 10;
    int buf_size = 10000;
    zmq::socket_t pub_socket(context, ZMQ_PUB);
    zmq_setsockopt(pub_socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));
    zmq_setsockopt(pub_socket, ZMQ_SNDBUF, &buf_size, sizeof(buf_size));

    pub_socket.bind("ipc:///tmp/qos_in");

    std::string msg = "111";

    while (true)
    {
        auto res = zmq_send(pub_socket, msg.c_str(), msg.length(), ZMQ_DONTWAIT);
        printf("res:%d\n", res);
    }
}

int main(int, char **)
{
    qqq();

    return 0;
}