Memory leak in generation DDS network

[ehgus607]

I am working on a project to create a DDS network automatically by creating a wrapping code in API for each DDS vendor and reading files containing the information of the DDS network. I made a wrapping code using API in Open dds, but I uploaded a question because I think I am doing the wrong allocation and free of memory. Please let me know if there is any information that can help me with debugging. Here are a network file defined by me and main file.

And I attached my project PoC.zip. Please make a build folder and do cmake .. and cmake -- build . in it.

domain.txt

<domain>
	<participant_num>1</participant_num>
	<participant>
		<id>1</id>
		<qos>
			<EntityFactoryQosPolicy_>
				<autoenable_created_entities>true</autoenable_created_entities>
			</EntityFactoryQosPolicy_>
		</qos>
		<publisher_num>0</publisher_num>
		<subscriber_num>1</subscriber_num>
		<subscriber>
			<id>1_1</id>
			<qos>
				<PresentationQosPolicy_>
					<access_scope>GROUP_PRESENTATION_QOS</access_scope>
					<coherent_access>true</coherent_access>
					<ordered_access>true</ordered_access>
				</PresentationQosPolicy_>
				<EntityFactoryQosPolicy_>
					<autoenable_created_entities>true</autoenable_created_entities>
				</EntityFactoryQosPolicy_>
			</qos>
			<datareader_num>1</datareader_num>
			<datareader>
				<id>1_1_1</id>
				<qos>
					<DurabilityQosPolicy_>
						<kind>VOLATILE_DURABILITY_QOS</kind>
					</DurabilityQosPolicy_>
					<DeadlineQosPolicy_>
						<period>4294967295</period>
					</DeadlineQosPolicy_>
					<LatencyBudgetQosPolicy_>
						<duration>233665123</duration>
					</LatencyBudgetQosPolicy_>
					<LivelinessQosPolicy_>
						<kind>AUTOMATIC_LIVELINESS_QOS</kind>
						<lease_duration>4294967295</lease_duration>
					</LivelinessQosPolicy_>
					<ReliabilityQosPolicy_>
						<kind>RELIABLE_RELIABILITY_QOS</kind>
						<max_blocking_time>468703135</max_blocking_time>
					</ReliabilityQosPolicy_>
					<DestinationOrderQosPolicy_>
						<kind>BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS</kind>
					</DestinationOrderQosPolicy_>
					<HistoryQosPolicy_>
						<kind>KEEP_ALL_HISTORY_QOS</kind>
						<depth>9802</depth>
					</HistoryQosPolicy_>
					<ResourceLimitsQosPolicy_>
						<max_samples>9922</max_samples>
						<max_instances>1369133069</max_instances>
						<max_samples_per_instance>9856</max_samples_per_instance>
					</ResourceLimitsQosPolicy_>
					<OwnershipQosPolicy_>
						<kind>EXCLUSIVE_OWNERSHIP_QOS</kind>
					</OwnershipQosPolicy_>
					<TimeBasedFilterQosPolicy_>
						<minimum_separation>0</minimum_separation>
					</TimeBasedFilterQosPolicy_>
					<ReaderDataLifecycleQosPolicy_>
						<autopurge_no_writer_samples_delay>89018456</autopurge_no_writer_samples_delay>
						<autopurge_disposed_samples_delay>628175011</autopurge_disposed_samples_delay>
					</ReaderDataLifecycleQosPolicy_>
				</qos>
				<topic_id>1</topic_id>
			</datareader>
		</subscriber>
	</participant>
	<topic_num>1</topic_num>
	<topic>
		<id>1</id>
		<qos>
			<DurabilityQosPolicy_>
				<kind>TRANSIENT_LOCAL_DURABILITY_QOS</kind>
			</DurabilityQosPolicy_>
			<DurabilityServiceQosPolicy_>
				<service_cleanup_delay>846930886</service_cleanup_delay>
				<history_kind>KEEP_ALL_HISTORY_QOS</history_kind>
				<history_depth>1714636915</history_depth>
				<max_samples>1952264381</max_samples>
				<max_instances>719885386</max_instances>
				<max_samples_per_instance>1940997780</max_samples_per_instance>
			</DurabilityServiceQosPolicy_>
			<DeadlineQosPolicy_>
				<period>4294967295</period>
			</DeadlineQosPolicy_>
			<LatencyBudgetQosPolicy_>
				<duration>649760492</duration>
			</LatencyBudgetQosPolicy_>
			<LivelinessQosPolicy_>
				<kind>AUTOMATIC_LIVELINESS_QOS</kind>
				<lease_duration>4294967295</lease_duration>
			</LivelinessQosPolicy_>
			<ReliabilityQosPolicy_>
				<kind>RELIABLE_RELIABILITY_QOS</kind>
				<max_blocking_time>189641421</max_blocking_time>
			</ReliabilityQosPolicy_>
			<DestinationOrderQosPolicy_>
				<kind>BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS</kind>
			</DestinationOrderQosPolicy_>
			<HistoryQosPolicy_>
				<kind>KEEP_LAST_HISTORY_QOS</kind>
				<depth>27</depth>
			</HistoryQosPolicy_>
			<ResourceLimitsQosPolicy_>
				<max_samples>9785</max_samples>
				<max_instances>2044897763</max_instances>
				<max_samples_per_instance>9513</max_samples_per_instance>
			</ResourceLimitsQosPolicy_>
			<TransportPriorityQosPolicy_>
				<value>1967513926</value>
			</TransportPriorityQosPolicy_>
			<LifespanQosPolicy_>
				<duration>540383426</duration>
			</LifespanQosPolicy_>
			<OwnershipQosPolicy_>
				<kind>SHARED_OWNERSHIP_QOS</kind>
			</OwnershipQosPolicy_>
		</qos>
		<participant_id>1</participant_id>
		<datawriter_num>0</datawriter_num>
		<datawriter>
		</datawriter>
		<datareader_num>1</datareader_num>
		<datareader>
			<id>1_1_1</id>
		</datareader>
	</topic>
</domain>

PoC.cpp

int main(int argc, char** argv){
    std::string domain_file = read_from_file("domain.txt");
    NodeDomain* domain = new NodeDomain(domain_file);
    domain->init();

    delete domain;
    return 0;
}

NodeDomain constructs a dds network by receiving and parsing strings. The example above is just an example of creating a subscriber, but running it will result in more than 100 memory leaks.

The following is the disconstructor for each wrapping code:

NodeDomain.cpp

NodeDomain::NodeDomain(std::string shadow){
    participant_count_ = 0;
    topic_count_ = 0;
    shadow_ = shadow;

    int argc=3; 
    char argv1[] = "./Open-dds";
    char argv2[] = "-DCPSConfigFile";
    char argv3[] = "Open-dds_rtps.ini";
    char* argv[] = {argv1, argv2, argv3};
    factory_ = TheParticipantFactoryWithArgs(argc, argv);
}

NodeDomain::~NodeDomain(){
    std::vector<DomainParticipant*> participants;
    
    while(!participants_.empty()){
        NodeParticipant *p  = participants_.back();
        participants_.pop_back();
        participants.push_back(p->get_participant());   
        delete p;  
    }

    while(!topics_.empty()){
        Topic* topic = topics_.back();
        topics_.pop_back();
        DomainParticipant* participant = topic->get_participant();

        if(participant->delete_topic(topic) != DDS::RETCODE_OK){
            std::cout << "[ERROR] There is something worng with remove topic (" << topic->get_name() << ")" << std::endl;
        }
    }
    while(!participants.empty()){
        DomainParticipant* p = participants.back();
        participants.pop_back();
        p->delete_contained_entities();
        factory_->delete_participant(p);
    }
}

NodeParticipant.cpp

NodeParticipant::NodeParticipant(std::string id, DomainParticipantQos dp_qos, std::string shadow, DomainParticipantFactory* factory)
    :type_(new Messenger::MessageTypeSupportImpl())
{
    id_ = id;
    publisher_count_ = 0;
    subscriber_count_ = 0;
    shadow_ = shadow;

    factory_ = factory;

    participant_ = factory_->create_participant(42, dp_qos, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    if( !participant_ ){
        std::cout << "[ERROR] There is something worng with create participant (" << id << ")" << std::endl;
        raise(SIGSEGV);
    }
    type_->register_type(participant_, "");
}

NodeParticipant::~NodeParticipant(){
    while(!publishers_.empty()){
        NodePublisher* pub = publishers_.back();
        publishers_.pop_back();
        delete pub;
    }
    while(!subscribers_.empty()){
        NodeSubscriber* sub = subscribers_.back();
        subscribers_.pop_back();
        delete sub;
    }
}

NodePublisher.cpp

NodePublisher::NodePublisher(DomainParticipant* participant, PublisherQos pub_qos, std::string id, std::string shadow){
    id_ = id;
    participant_ = participant;
    datawriter_count_ = 0;
    shadow_ = shadow;

    publisher_ = participant_->create_publisher(pub_qos, 0, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    if( !publisher_ ){
        std::cout << "[ERROR] There is something worng with create publisher (" << id << ")" << std::endl;
        raise(SIGSEGV);
    }
}

NodePublisher::~NodePublisher(){
    while(!datawriters_.empty()){
        delete datawriters_.back();
        datawriters_.pop_back();
    }  
    publisher_->delete_contained_entities();
    if( participant_->delete_publisher(publisher_) != DDS::RETCODE_OK){
        std::cout << "[ERROR] There is something worng with remove publisher (" << id_ << ")" << std::endl;
        raise(SIGSEGV);
    }
}

NodeSubscriber.cpp

NodeSubscriber::NodeSubscriber(DomainParticipant* participant, SubscriberQos sub_qos, std::string id, std::string shadow){
    id_ = id;
    participant_ = participant;
    datareader_count_ = 0;
    listener_.id_ = id;

    shadow_ = shadow;
    
    subscriber_ = participant_->create_subscriber(sub_qos, &listener_, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    if(subscriber_ == nullptr){
        std::cout << "[ERROR] There is something worng with create subscriber (" << id << ")" << std::endl;
        raise(SIGSEGV);
    }
}

NodeSubscriber::~NodeSubscriber(){
    while(!datareaders_.empty()){
        delete datareaders_.back();
        datareaders_.pop_back();
    }
    subscriber_->delete_contained_entities();
    if( participant_->delete_subscriber(subscriber_) != DDS::RETCODE_OK){
        std::cout << "[ERROR] There is something worng with remove subscriber (" << id_ << ")" << std::endl;
        raise(SIGSEGV);
    };
}

NodeDatareader.cpp

NodeDatareader::NodeDatareader(Subscriber* sub, Topic* topic, DataReaderQos dr_qos, std::string id){
    id_ = id;
    subscriber_ = sub;
    listener_.id_ = id;

    datareader_ = subscriber_->create_datareader(topic, dr_qos, &listener_, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    if( !datareader_ ){
        std::cout << "[ERROR] There is something worng with create datareader (" << id << ")" << std::endl;
        raise(SIGSEGV);
    }
}

NodeDatareader::~NodeDatareader(){
    datareader_->delete_contained_entities();
    if( subscriber_->delete_datareader(datareader_) != DDS::RETCODE_OK){
        std::cout << "[ERROR] There is something worng with remove datareader (" << id_ << ")" << std::endl;
        raise(SIGSEGV);
    } 
}

NodeDatawriter.cpp

NodeDatawriter::NodeDatawriter(Publisher* pub, Topic* topic, DataWriterQos dw_qos, std::string id){
    id_ = id;
    publisher_ = pub;
    listener_.id_ = id;

    datawriter_ = publisher_->create_datawriter(topic, dw_qos, &listener_, OpenDDS::DCPS::DEFAULT_STATUS_MASK);
    if( !datawriter_ ){
        std::cout << "[ERROR] There is something worng with create datawriter (" << id << ")" << std::endl;
        raise(SIGSEGV);
    }
}

NodeDatawriter::~NodeDatawriter(){
    if( publisher_->delete_datawriter(datawriter_) != DDS::RETCODE_OK){
        std::cout << "[ERROR] There is something worng with remove datawriter (" << id_ << ")" << std::endl;
        raise(SIGSEGV);
    }
}

Disconstructors free memory one after another, calling from domain to datareader/datawriter in succession. I have to add the functions to add/remove entities later, so I'm trying not to use endelete_contained_entities() as much as possible.

I think it's because of the difference and characteristics between _var and _ptr.

As a side note, can't CORBA::string_var be used with std::string? There were a lot of memory leaks because the memory generated during the copy process couldn't be free.

Thank you for reading the long text.

[iguessthislldo]

I am working on a project to create a DDS network automatically by creating a wrapping code in API for each DDS vendor and reading files containing the information of the DDS network.

That sounds very similar to bench, specifically the builder library that's part of bench, though bench only currently works with OpenDDS. @simpsont-oci may be able to speak on how easy it would be to have it use other DDS vendors. Regardless of that, it would probably serve as a good reference.

I think it's because of the difference and characteristics between _var and _ptr.

This is part of the CORBA memory management. It's reference counted like std::shared_ptr, but unfortunately it requires much more care to be taken. The OpenDDS interfaces are allocated and returned as raw pointers back (_ptr), but these usually should go into their respective _var objects so they can be tracked. Otherwise it will never free the memory.

The members that are currently raw pointers should definitely be _var types:

class NodeDatawriter{
public:
    Publisher_var publisher_;
    DataWriter_var datawriter_;

For the constructor arguments, I would either make them raw pointers that require a _duplicate:

NodeDatawriter::NodeDatawriter(Publisher* pub, Topic* topic, DataWriterQos dw_qos, std::string id){
   id_ = id;
   publisher_ = Publisher::_duplicate(pub);
   listener_.id_ = id;

or a _var& that doesn't:

NodeDatawriter::NodeDatawriter(Publisher_var& pub, Topic* topic, DataWriterQos dw_qos, std::string id){
   id_ = id;
   publisher_ = pub;
   listener_.id_ = id;

(This is what bench builder does)

If you want a details on how CORBA memory management actually works, this page looks like it provides a decent explaination. There's also the IDL-to-C++ section on interface mapping.

As a side note, can't CORBA::string_var be used with std::string? There were a lot of memory leaks because the memory generated during the copy process couldn't be free.

This is for a similar reason as the DataReaders, etc. The pointer needs the _var object to delete the allocated memory. If you want it as a std::string, you'll need to copy it. For example:

std::string type_name;
{
  CORBA::String_var type_name_sv = ts->get_type_name();
  type_name = type_name_sv.in();
}

If you have C++17, you could avoid the copy and use std::string_view, but you'd need to keep the CORBA::String_var around.

At some point in the future all of this will modernized as part of "Remove dependency on TAO" in the roadmap, but we have other tasks to do before that.

[ehgus607]

Hi @iguessthislldo! Thank you so much for your response.

The references you mentioned seem to helpful to me. In particular, the reference on CORBA memory management will be very helpful.
Also, I'll refer to the bench code as well.

Thank you so much.