Memory leak on listener

[AlexSch31]

Hello,

I have two applications, the first one is a streamer which capture frames from a camera and send it to DDS, the second one is a subscriber application to the topic of the streamer. They both are executed on the same computer which is on Ubuntu 20.04.

The streamer application send a frame of 4.608.000 bytes (1920*1200 and 2 bytes for 1 pixel, this is a raw image) every 50ms to DDS.

I notice that I have some memory leaks on the subscriber application. It's as if the data received by the listener are not released after the function DataReaderListenerImple::on_data_available is finished. I modified the Messenger_zeroCopy example to test if I reproduce the memory leak on a example provided because I didn't find any clue on my part and I reproduce the memory leak.

Maybe this leak is from a incorrect implementation from me. Below is the code of my streamer and the DataReaderListenerImpl class from the Messenger_zeroCopy that I have modified. The subcriber.cpp file is also modified to follow the structure of my topic, but don't change drastically.

IDL :

RAW_IMAGE_SIZE = 4608000;
struct RawImage {
  @key string deviceName;
  unsigned long frame_id;
  sequence<octet, RAW_IMAGE_SIZE> data;
};

Streamer

void Capture(int domainId, string topicName) {
    
    int frameId = 0;
    RawImage rawImage;
    rawImage.deviceName ="Camera";
    while(RUNNING) {
        clock_t start = clock()
        char *byteArray = captureCamera();
        if(strlen(byteArray) == 0) {
            std::cerr << "Unable to capture a frame" << std::endl;
            sleep(start, clock()); // (clock() - start)/CLOCKS_PER_SEC = sleeping time
            continue;
        }
        rawImage.frame_id = frameId;
        rawImage.data = RawImage::_tao_seq_CORBA_Octet__4608000(4608000, (uchar *)byteArray);
        postToDDS(rawImage); // DataWriter->write(rawImage)
        sleep(start, clock()); // (clock() - start)/CLOCKS_PER_SEC = sleeping time
        frameId++;
    }
}

DataReaderListenerImpl.cpp

DataReaderListenerImpl::on_data_available(DDS::DataReader_ptr reader)
{
    RawImageDataReader_var reader_i = 
        RawImageDataReader::_narrow(reader);
    if(CORBA::IS_NIL(reader_i.in())) {
        ACE_ERROR((LM_ERROR,
                   ACE_TEXT("ERROR: %N:%l: on_data_available() -"),
                   ACE_TEXT(" _narrow failed!\n")));
        ACE_OS::exit(-1);
    }

    RawImageSeq rawImages;
    DDS::SampleInfoSeq infos;

    DDS::ReturnCode_t error = reader_i->take(rawImages,
                                             infos,
                                             DDS::LENGTH_UNLIMITED,
                                             DDS::ANY_SAMPLE_STATE,
                                             DDS::ANY_VIEW_STATE,
                                             DDS::ANY_INSTANCE_STATE);
    if(error == DDS::RETCODE_OK) {
        std::cout << "SampleInfo.sample_rank = " << infos[0].sample_rank << std::endl;
        std::cout << "SampleInfo.instance_state = " << infos[0];instance_state << std::endl;

        if(info[0].valid_data) {
            std::cout << "RawImage: deviceName = " << rawImages[0].deviceName << std::endl
                      << "          frame_id = " << rawImages[0].frame_id << std::endl
                      << "          data.length = " << rawImages[0].data.length() << std::endl;
        } else {
            ACE_ERROR((LM_ERROR,
                       ACE_TEXT("ERROR: %N:%l: on_data_available() -"),
                       ACE_TEXT(" take failed!\n")));
        }
    }

    reader_i->return_loan(rawImages, infos);
}

[simpsont-oci]

How are you determining that there's a memory leak? Are you watching an application like top or looking at the output of a memory checking tool like valgrind or a compiler's LeakSanitizer? If the latter, can you share the stack trace?

What does your configuration file look like? Which discovery / transport are you using?

[AlexSch31]

I utilize htop to monitor the memory consumption and the memory grow continuously until the swap, then, if I don't do anything, the computer froze and I need to reboot it. When I run valgrind, it didn't detect anything.

My configuration file look like this :

[common]
DCPSInfoRepo=corbaloc::172.20.1.52:12345/DCPSInfoRepo
DCPSGlobalTransportConfig=config1

[config/config1]
transports=the_rtps_transport

[transport/the_rtps_transport]
transport_type=rtps_udp
use_multicast=0

[AlexSch31]

FYI when I modify the configuration file to use shared memory I don't have memory leak

[common]
DCPSInfoRepo=corbaloc::172.20.1.52:12345/DCPSInfoRepo
DCPSGlobalTransportConfig=config1

[config/config1]
transports=shared_memory

[transport/shared_memory]
transport_type=shmem

[KelsOnFire]

Hi @simpsont-oci

Here is a project based on Messenger example with the same memory leak issue.
https://github.com/KelsOnFire/opendds-memory-leak

[simpsont-oci]

It sounds like this isn't exactly a true memory leak, but possibly some non-optimal storage for partially collected samples in the rtps_udp transport, specifically for larger sample sizes. I'm guessing that if you reduced your sample size or reduced the write frequency, this would be less of a problem? Unless you need RTPS for a particular reason (e.g. DDS interoperability or DDS security), you might be better off using the tcp (or shmem) transport for larger sample sizes for now.

[KelsOnFire]

@simpsont-oci Actually, we cannot really reduce the sample size, it's a RAW image.
We did some test by reducing the write frequency (writing each second), but the memory consumption kept growing until swapping.
TCP and SHMEM might not be a suitable approach because we would like "real time" data delivery accross several computers.

Maybe using the Messenger example to test this use case was not a good choice, did you check the project https://github.com/KelsOnFire/opendds-memory-leak?
By the way, if you have an example with big samples to send, it would be interesting to compare.

Thanks you help!

[AlexSch31]

@simpsont-oci Unfortunately we can't reduce the sample size, as I said in my first post it's a RAW image. Is DDS not supposed to allow large data sample to be send in real-time ? If it is a problem with my code what can I do to fix the problem ?

Have you tried the example of @KelsOnFire. I also have a memory leak on this project too.

[AlexSch31]

@simpsont-oci I activate the option DCPSDebugLevel on my subscriber app and I notice that I have more malloc call than free (from the Cached_Allocator_With_Overflow).

Can it be the issue ? If yes, how can I fix it ?

The trace :
consumer.log

The lines like below are from my code, they show me if I received a frame and the frame's number (frame_id). If the number don't follow one or more frames are lost.

message.deviceName: LWIR
message.data.length(): 4608000
message.frame_id: 18
message.timestamp: 0
message.capturedTime: 0

[AlexSch31]

FYI, even with a empty on_data_available the memory continue to grow.

[heshiwen1989]

FYI, even with a empty on_data_available the memory continue to grow.

i meet this problem too, i use opendds to send pointcloud(about 3MByte) and 10Hz, and the publisher memory is grow, it seem that the pointcloud message's memory not release even if the message was received by subscriber.

[heshiwen1989]

i use the opendds's example to send my message, almost no any modified.

[heshiwen1989]

try not update the sample keyword, because sample with new keyword which send that will create new instance, and the new instance will load to memory grow up!