Libzmq: more dealer send message to a router,heap leak!

[boostiewj]

Hi，60,000 dealers connect a router,then dealers send message to the router.
The router‘s RES memory is 1.3G~1.5G~1.9G~2.6G. Then disconnect dealers,but the memory also is 2.6G...

start router server command line: ./test_zmq s 192.168.10.110:1993
start dealer client command line: ./test_zmq c 1 60001 192.168.10.110:1993
Press enter stop dealer client;
Also,press enter stop router server;

OS:centos 6.7 ulimit -n 300000
zeromq version:4.1.4 build epoll mode

 PID USER      PR  NI  VIRT  RES  SHR S %CPU %MEM    TIME+  COMMAND                                                          
 7133 root      20   0 3252m 2.5g 2156 S  6.3  8.1   3:28.09 test_zmq 

stop and restart dealer client again.the router server:

 PID USER      PR  NI  VIRT  RES  SHR S %CPU %MEM    TIME+  COMMAND                                                          
 7133 root      20   0 3252m 2.8g 2156 S 19.6  9.0   5:03.71  test_zmq 

code is:

#include <list>
#include <string>
#include <thread>
#include <zmq.h>
using namespace std;
const std::string zmq_error()
{
    int no = zmq_errno();
    std::string r(" zmq-error:");
    r += std::to_string(no);
    if (const char * str = zmq_strerror(no))
    {
        r += " "; r += str;
    }
    return r;
}
template< class T>
bool setzmq_sockopt(void *s, int option_, T v)
{
    const int rc = zmq_setsockopt(s, option_, (void*)&v, sizeof(T));
    if (rc != 0)
        puts(zmq_error().c_str());
    assert(0 == rc);
    return 0 == rc;
}
void send_msgx(void* s, string str, bool more = false)
{
    zmq_send(s, str.c_str(), str.length(), (more ? ZMQ_SNDMORE : 0)|ZMQ_DONTWAIT);
}
int recv_msgx(void* s,int timeout_msc = 500)
{
    zmq_pollitem_t items[1] ={ { s, 0, ZMQ_POLLIN, 0 } };
    int rc = zmq_poll(items,1, timeout_msc);
    if (rc < 0)
        return -1;
    if (items[0].revents & ZMQ_POLLIN)
    {
        while (true)
        {
            int more(0);
            size_t more_size = sizeof(more);
            zmq_msg_t msg;
            zmq_msg_init(&msg);
            rc = zmq_msg_recv(&msg, s, 0);
            zmq_msg_close(&msg);
            if (-1 == rc)
            {
                return 0;
            }
            rc = zmq_getsockopt(s, ZMQ_RCVMORE, &more, &more_size);
            if(-1 == rc || !more)
                break;
        }
    }
    return 0;
}
void router_server(string ip,volatile bool &exit_loop)
{
    shared_ptr<void> sp_ctx(zmq_ctx_new(), zmq_ctx_destroy);
    void* ctx = sp_ctx.get();
    zmq_ctx_set(ctx, ZMQ_IO_THREADS,1);
    zmq_ctx_set(ctx, ZMQ_MAX_SOCKETS,25*10000);
    shared_ptr<void> sp_s(zmq_socket(ctx, ZMQ_ROUTER),zmq_close);
    void* s = sp_s.get();
    setzmq_sockopt(s, ZMQ_SNDHWM, int(10));
    setzmq_sockopt(s, ZMQ_RCVHWM, int(10));
    setzmq_sockopt(s, ZMQ_LINGER, int(1));

    setzmq_sockopt(s, ZMQ_TCP_KEEPALIVE, int(1));
    setzmq_sockopt(s, ZMQ_TCP_KEEPALIVE_IDLE, int(30));
    setzmq_sockopt(s, ZMQ_TCP_KEEPALIVE_INTVL, int(20));
    setzmq_sockopt(s, ZMQ_TCP_KEEPALIVE_CNT, int(2));
    string addr = "tcp://" + ip;
    printf("server:%s\n", addr.c_str());
    int rc = zmq_bind(s, addr.c_str());
    if (0 != rc)
    {
        printf("bind error %s\n", zmq_error().c_str());
        return;
    }
    long n(0);
    while (!exit_loop)
    {
        rc = recv_msgx(s);
        if (-1 == rc)
        {
            printf("recv error\n");
            break;
        }
        if (0 == (++n % 5000))
            printf("recv:%ld\n", n);
    }
    printf("exit loop\n");
}
void delear_client(string ip, long start_id, long end_id,volatile bool &exit_loop)
{
    shared_ptr<void> sp_ctx(zmq_ctx_new(), zmq_ctx_destroy);
    void* ctx = sp_ctx.get();
    zmq_ctx_set(ctx, ZMQ_IO_THREADS, 1);
    zmq_ctx_set(ctx, ZMQ_MAX_SOCKETS, 25 * 10000);
    string addr = "tcp://" + ip;
    printf("client %s\n", addr.c_str());
    list<shared_ptr<void>> v;
    int rc(0);
    for (long i(start_id); i <= end_id;++i)
    {
        shared_ptr<void> sp(zmq_socket(ctx, ZMQ_DEALER), zmq_close);
        void*s = sp.get();
        v.push_back(sp);
        string id = to_string(i);
        zmq_setsockopt(s, ZMQ_IDENTITY,id.c_str(),id.length());
        setzmq_sockopt(s, ZMQ_SNDHWM, int(10));
        setzmq_sockopt(s, ZMQ_RCVHWM, int(10));
        setzmq_sockopt(s, ZMQ_LINGER, int(1));
        rc = zmq_connect(s, addr.c_str());
        if (0 != rc)
        {
            printf("%ld connect error %s\n",i, zmq_error().c_str());
            return;
        }
        printf("connect %ld\n", i);
    }
    string str("012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789");
    long n(0);
    while (!exit_loop)
    {
        for (auto &sp : v)
        {
            if (exit_loop)
                break;
            void* s = sp.get();
            send_msgx(s, str);
            if (0 == (++n % 5000))
                printf("send:%ld\n", n);
            std::this_thread::sleep_for(std::chrono::microseconds(20));
        }
    }
    printf("exit loop\n");
}
#define error_return do{printf("arg error\n"); return -1;}while (0)
/*
./test_zmq s 192.168.10.110:1993
./test_zmq c 1 60001 192.168.10.110:1993
*/
int main(int argc, char** argv)
{
    std::thread t;
    volatile bool exit_loop(false);
    if (argc < 2)error_return;
    char tag = argv[1][0];
    if ('s' == tag)
    {
        if(argc != 3)error_return;
        string ip = argv[2];
        t = std::thread(router_server,ip, std::ref(exit_loop));
    }
    else if ('c' == tag)
    {
        if (argc != 5) error_return;
        long start_id = atol(argv[2]);
        long end_id = atol(argv[3]);
        string ip = argv[4];
        t = std::thread(delear_client,ip,start_id,end_id, std::ref(exit_loop));
    }
    else
        error_return;
    getchar();
    exit_loop = true;
    if (t.joinable())
        t.join();
    return 0;
}

[boostiewj]

Also, if 180,000 dealers connect a router, then kill 120,000 dealers,the router go to the dead cycle for 12 minutes .

PID   USER  PR  NI  VIRT   RES  SHR  S %CPU    %MEM       TIME+      COMMAND                                                          
53134 root  20  0   12.7g  9.4g 2772 R 112.5   30.0       444:08.62  test_zmq         

[boostiewj]

@hitstergtd
How are things going?
And this problem also has existed at the older version of 3.2.5. I am troubled that I can't work now.

I find the code of the following problem is not exists at newer version.I think,Heartbeat connection causes a memory leak，the problems have been not solved.
Heartbeat connection causes a memory leak
Allow session to flush commands on an engine error

thanks for your help

[sunkin351]

Idk about you, but I cant follow those links. Code 404: Page not found

[boostiewj]

@sunkin351 those links now are ok.

[sunkin351]

I agree, this is one of the biggest issues I have seen in libzmq. Which sockets does it affect exactly?

[boostiewj]

this bug feel depressed

[bluca]

Does it still happen on 4.2.0?

[boostiewj]

@bluca yes.

[santupari]

@bluca Is there any update on this issue? We are also observing continuous memory growth of our process which opens several dealer sockets. RSS growth never reduces until we stop the process (graceful exit). We would like to know if there is any update on memory leak. Is there any setting on dealer socket that cleans the pipepair periodically?

Thanks,
Santosh

[sleepycoder72]

Does anyone know the status of this? I am working with thousands of sockets and haven't tested for this error. I would like not to duplicate effort if necessary.

[honglei]

I found in dealer-router model, dealer will create many tcp connections and resend tcp block again and again, while no any useful zmq message can be sended to Router port. Version:4.3.2

[heliarmk]

Any update for this？