application的heap一直增長(zhǎng)，不下降

linuxfellow

用了一個(gè)開(kāi)源的application:  http://zeromq.org/
[1516]     0  1516    22300     7806   0       0             0 zeromg

cat /proc/1516/smap
00011000-03544000 rw-p 00000000 00:00 0          [heap]
Size:              54476 kB
Rss:               30380 kB
Pss:               30380 kB
Shared_Clean:          0 kB
Shared_Dirty:          0 kB
Private_Clean:         0 kB
Private_Dirty:     30380 kB
Referenced:        30380 kB
Anonymous:         30380 kB
AnonHugePages:         0 kB
Swap:                  0 kB
KernelPageSize:        4 kB
MMUPageSize:           4 kB
Locked:                0 kB

該進(jìn)程有29MB heap。開(kāi)源代碼不至于犯這樣低級(jí)錯(cuò)誤吧。 是不是系統(tǒng)里有些參數(shù)沒(méi)有調(diào)好？
用的是uclibc,  #define DEFAULT_TRIM_THRESHOLD (256 * 1024)
uclibc里有沒(méi)有辦法讓heap早點(diǎn)回歸系統(tǒng)？

鎮(zhèn)水鐵牛

zeromp中有很多函數(shù)有陷阱，有的函數(shù)調(diào)用后，需要你主動(dòng)去釋放內(nèi)存。

鎮(zhèn)水鐵牛

你把代碼貼上來(lái)，我給你瞅瞅端倪？

linuxfellow

太感謝樓上。這些都是應(yīng)用層的代碼。他們有問(wèn)題查不出來(lái)，讓我們做內(nèi)核的來(lái)分析。明天向他們要代碼去。

linuxfellow

回復(fù) 3# 鎮(zhèn)水鐵牛
zmq_msg_t zmq_msg;
        void* context, * rsock,* ssock;
        int len,status;
        unsigned char *msg;

       context = zmq_ctx_new();
       rsock=zmq_socket(context,ZMQ_SUB);

       status =zmq_bind (rsock, argv[1] );


       zmq_setsockopt(rsock, ZMQ_SUBSCRIBE, NULL, 0);

       ssock =zmq_socket(context,ZMQ_PUB);
       status =zmq_bind (ssock, argv[2] );

       
       while(1)
       {
             zmq_msg_init (&zmq_msg);

             len = zmq_msg_recv (&zmq_msg, rsock, 0);
             if (len >= 0)
             {
                         zmq_msg_send (&zmq_msg, ssock, 0);
             }

             zmq_msg_close (&zmq_msg);
       }




   

linuxfellow

回復(fù) 3# 鎮(zhèn)水鐵牛


    代碼超簡(jiǎn)單，難道zmq_msg_t zmq_msg 里面有隱含指針要釋放？

鎮(zhèn)水鐵牛

看起來(lái)好像沒(méi)什么問(wèn)題，建議你開(kāi)發(fā)的時(shí)候可以基于czmq這個(gè)高級(jí)zmq庫(kù).
給你個(gè)xpub-xsub的demo.
int main (void)
{
    setup_test_environment();
    void *ctx = zmq_ctx_new ();
    assert (ctx);

    //  First, create an intermediate device
    void *xpub = zmq_socket (ctx, ZMQ_XPUB);
    assert (xpub);
    int rc = zmq_bind (xpub, "tcp://127.0.0.1:5560");
    assert (rc == 0);
    void *xsub = zmq_socket (ctx, ZMQ_XSUB);
    assert (xsub);
    rc = zmq_bind (xsub, "tcp://127.0.0.1:5561");
    assert (rc == 0);

    //  Create a publisher
    void *pub = zmq_socket (ctx, ZMQ_PUB);
    assert (pub);
    rc = zmq_connect (pub, "tcp://127.0.0.1:5561");
    assert (rc == 0);

    //  Create a subscriber
    void *sub = zmq_socket (ctx, ZMQ_SUB);
    assert (sub);
    rc = zmq_connect (sub, "tcp://127.0.0.1:5560");
    assert (rc == 0);

    //  Subscribe for all messages.
    rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);
    assert (rc == 0);

    //  Pass the subscription upstream through the device
    char buff [32];
    rc = zmq_recv (xpub, buff, sizeof (buff), 0);
    assert (rc >= 0);
    rc = zmq_send (xsub, buff, rc, 0);
    assert (rc >= 0);

    //  Wait a bit till the subscription gets to the publisher
    msleep (SETTLE_TIME);

    //  Send an empty message
    rc = zmq_send (pub, NULL, 0, 0);
    assert (rc == 0);

    //  Pass the message downstream through the device
    rc = zmq_recv (xsub, buff, sizeof (buff), 0);
    assert (rc >= 0);
    rc = zmq_send (xpub, buff, rc, 0);
    assert (rc >= 0);

    //  Receive the message in the subscriber
    rc = zmq_recv (sub, buff, sizeof (buff), 0);
    assert (rc == 0);

    //  Clean up.
    rc = zmq_close (xpub);
    assert (rc == 0);
    rc = zmq_close (xsub);
    assert (rc == 0);
    rc = zmq_close (pub);
    assert (rc == 0);
    rc = zmq_close (sub);
    assert (rc == 0);
    rc = zmq_ctx_term (ctx);
    assert (rc == 0);

    return 0 ;
}

鎮(zhèn)水鐵牛

zmsg的demo，這個(gè)更容易掌握。
//  Selftest

int
zmsg_test (bool verbose)
{
    printf (" * zmsg: ");

    int rc = 0;
    //  @selftest
    zctx_t *ctx = zctx_new ();
    assert (ctx);

    void *output = zsocket_new (ctx, ZMQ_PAIR);
    assert (output);
    zsocket_bind (output, "inproc://zmsg.test");
    void *input = zsocket_new (ctx, ZMQ_PAIR);
    assert (input);
    zsocket_connect (input, "inproc://zmsg.test");

    //  Test send and receive of single-frame message
    zmsg_t *msg = zmsg_new ();
    assert (msg);
    zframe_t *frame = zframe_new ("Hello", 5);
    assert (frame);
    zmsg_push (msg, frame);
    assert (zmsg_size (msg) == 1);
    assert (zmsg_content_size (msg) == 5);
    rc = zmsg_send (&msg, output);
    assert (msg == NULL);
    assert (rc == 0);

    msg = zmsg_recv (input);
    assert (msg);
    assert (zmsg_size (msg) == 1);
    assert (zmsg_content_size (msg) == 5);
    zmsg_destroy (&msg);

    //  Test send and receive of multi-frame message
    msg = zmsg_new ();
    assert (msg);
    rc = zmsg_addmem (msg, "Frame0", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame1", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame2", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame3", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame4", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame5", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame6", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame7", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame8", 6);
    assert (rc == 0);
    rc = zmsg_addmem (msg, "Frame9", 6);
    assert (rc == 0);
    zmsg_t *copy = zmsg_dup (msg);
    assert (copy);
    rc = zmsg_send (&copy, output);
    assert (rc == 0);
    rc = zmsg_send (&msg, output);
    assert (rc == 0);

    copy = zmsg_recv (input);
    assert (copy);
    assert (zmsg_size (copy) == 10);
    assert (zmsg_content_size (copy) == 60);
    zmsg_destroy (&copy);

    msg = zmsg_recv (input);
    assert (msg);
    assert (zmsg_size (msg) == 10);
    assert (zmsg_content_size (msg) == 60);

    // create empty file for null test
    FILE *file = fopen ("zmsg.test", "w");
    assert (file);
    fclose (file);

    file = fopen ("zmsg.test", "r");
    zmsg_t *null_msg = zmsg_load (NULL, file);
    assert (null_msg == NULL);
    fclose (file);
    remove ("zmsg.test");

    //  Save to a file, read back
    file = fopen ("zmsg.test", "w");
    assert (file);
    rc = zmsg_save (msg, file);
    assert (rc == 0);
    fclose (file);

    file = fopen ("zmsg.test", "r");
    rc = zmsg_save (msg, file);
    assert (rc == -1);
    fclose (file);
    zmsg_destroy (&msg);

    file = fopen ("zmsg.test", "r");
    msg = zmsg_load (NULL, file);
    assert (msg);
    fclose (file);
    remove ("zmsg.test");
    assert (zmsg_size (msg) == 10);
    assert (zmsg_content_size (msg) == 60);

    //  Remove all frames except first and last
    int frame_nbr;
    for (frame_nbr = 0; frame_nbr < 8; frame_nbr++) {
        zmsg_first (msg);
        frame = zmsg_next (msg);
        zmsg_remove (msg, frame);
        zframe_destroy (&frame);
    }
    //  Test message frame manipulation
    assert (zmsg_size (msg) == 2);
    frame = zmsg_last (msg);
    assert (zframe_streq (frame, "Frame9"));
    assert (zmsg_content_size (msg) == 12);
    frame = zframe_new ("Address", 7);
    assert (frame);
    zmsg_wrap (msg, frame);
    assert (zmsg_size (msg) == 4);
    rc = zmsg_addstr (msg, "Body");
    assert (rc == 0);
    assert (zmsg_size (msg) == 5);
    frame = zmsg_unwrap (msg);
    zframe_destroy (&frame);
    assert (zmsg_size (msg) == 3);
    char *body = zmsg_popstr (msg);
    assert (streq (body, "Frame0"));
    free (body);
    zmsg_destroy (&msg);

    //  Test encoding/decoding
    msg = zmsg_new ();
    assert (msg);
    byte *blank = (byte *) zmalloc (100000);
    assert (blank);
    rc = zmsg_addmem (msg, blank, 0);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 1);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 253);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 254);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 255);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 256);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 65535);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 65536);
    assert (rc == 0);
    rc = zmsg_addmem (msg, blank, 65537);
    assert (rc == 0);
    free (blank);
    assert (zmsg_size (msg) == 9);
    byte *buffer;
    size_t buffer_size = zmsg_encode (msg, &buffer);
    zmsg_destroy (&msg);
    msg = zmsg_decode (buffer, buffer_size);
    assert (msg);
    free (buffer);
    zmsg_destroy (&msg);

    //  Now try methods on an empty message
    msg = zmsg_new ();
    assert (msg);
    assert (zmsg_size (msg) == 0);
    assert (zmsg_unwrap (msg) == NULL);
    assert (zmsg_first (msg) == NULL);
    assert (zmsg_last (msg) == NULL);
    assert (zmsg_next (msg) == NULL);
    assert (zmsg_pop (msg) == NULL);
    assert (zmsg_send (&msg, output) == -1);
    assert (msg != NULL);
    zmsg_destroy (&msg);

    zctx_destroy (&ctx);
    //  @end
    printf ("OK\n");
    return 0;
}

linuxfellow

回復(fù) 7# 鎮(zhèn)水鐵牛

幫我再想想， 這么超級(jí)簡(jiǎn)單的調(diào)用為啥會(huì)有l(wèi)eak? zmq內(nèi)部自己有l(wèi)eak?


   

鎮(zhèn)水鐵牛

    while(1)
       {
             zmq_msg_init (&zmq_msg);

             len = zmq_msg_recv (&zmq_msg, rsock, 0); //對(duì)端發(fā)送的消息是幾部分？
             if (len >= 0)
             {
                         zmq_msg_send (&zmq_msg, ssock, 0);
             }

             zmq_msg_close (&zmq_msg);
       }

你把收發(fā)端的代碼完整貼一下，我自己測(cè)試下看看。

你也可以Valgrind看看是哪里的異常。