subscriber can't receive a message

[llddang]

Is there an already existing issue for this?

• I have searched the existing issues

Expected behavior

My domain configuration is like this.

Participant  ------------------------------------------------------
      |--------Publisher                                          |
      |           |-----------DataWriter1 ------------------------|
      |           |-----------DataWriter2 ------------------------|
      |---------Subscriber                                      Topic
                  |-----------DataReader -------------------------|

One Participant, Topic, Publisher, Subscriber, DataReader, and two DataWriters.

In this state, the reader expects to receive two messages if communication is attempted.
So this is what I though log output.

 [ WRITER : on_publication_matched ] Writer1 matched.
 [ WRITER : on_publication_matched ] Writer2 matched.
 < READER : on_subscription_matched > Reader matched.
 < READER : on_subscription_matched > Reader matched.
Message: HelloWorld1 with index: 1 SENT
< READER : on_data_available > reader get message HelloWorld1 from 1 RECEIVED
Message: HelloWorld2 with index: 2 SENT
< READER : on_data_available > reader get message HelloWorld2 from 2 RECEIVED
 [ WRITER : on_publication_matched ] Writer1 unmatched.
 [ WRITER : on_publication_matched ] Writer2 unmatched.

Current behavior

 [ WRITER : on_publication_matched ] Writer2 matched.
 < READER : on_subscription_matched > Reader matched.
 [ WRITER : on_publication_matched ] Writer1 matched.
 < READER : on_subscription_matched > Reader matched.
Message: HelloWorld1 with index: 1 SENT
< READER : on_data_available > reader get message HelloWorld1 from 1 RECEIVED
Message: HelloWorld2 with index: 2 SENT
 [ WRITER : on_publication_matched ] Writer1 unmatched.
 [ WRITER : on_publication_matched ] Writer2 unmatched.

From log we can see that index 1 and 2 had been send, but reader receive only index 1.

My qos as bellow:
TopicQos tqos = TOPIC_QOS_DEFAULT;
tqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
tqos.durability_service().service_cleanup_delay = Duration_t(317782863 * 1e-9);
tqos.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
tqos.durability_service().history_depth = 1004623649;
tqos.durability_service().max_samples = 1155043293;
tqos.durability_service().max_instances = 4039424;
tqos.durability_service().max_samples_per_instance = 1051145751;
tqos.latency_budget().duration = Duration_t(853141231 * 1e-9);
tqos.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
tqos.reliability().max_blocking_time = Duration_t(991706505 * 1e-9);
tqos.history().kind = KEEP_LAST_HISTORY_QOS;
tqos.history().depth = 1843769184;
tqos.resource_limits().max_samples = 2134960815;
tqos.resource_limits().max_instances = 1533973938;
tqos.resource_limits().max_samples_per_instance = 1980921081;
tqos.transport_priority().value = 1487440300;
tqos.lifespan().duration = Duration_t(164455410 * 1e-9);
tqos.ownership().kind = SHARED_OWNERSHIP_QOS;

PublisherQos pubqos = PUBLISHER_QOS_DEFAULT;
pubqos.presentation().access_scope = TOPIC_PRESENTATION_QOS;
pubqos.presentation().coherent_access = true;
pubqos.presentation().ordered_access = true;

DataWriterQos wqos1 = DATAWRITER_QOS_DEFAULT;
wqos1.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
wqos1.durability_service().service_cleanup_delay = Duration_t(567311302 * 1e-9);
wqos1.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
wqos1.durability_service().history_depth = 972947330;
wqos1.durability_service().max_samples = 1744157312;
wqos1.durability_service().max_instances = 301739589;
wqos1.durability_service().max_samples_per_instance = 1011714986;
wqos1.latency_budget().duration = Duration_t(921824426 * 1e-9);
wqos1.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
wqos1.reliability().max_blocking_time = Duration_t(852261378 * 1e-9);
wqos1.history().kind = KEEP_ALL_HISTORY_QOS;
wqos1.history().depth = 157146035;
wqos1.resource_limits().max_samples = 1933789796;
wqos1.resource_limits().max_instances = 598012395;
wqos1.resource_limits().max_samples_per_instance = 1322252895;
wqos1.transport_priority().value = 1485709160;
wqos1.lifespan().duration = Duration_t(823257944 * 1e-9);
wqos1.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
wqos1.ownership_strength().value = 502680922;

DataWriterQos wqos2 = DATAWRITER_QOS_DEFAULT;
wqos2.durability().kind = VOLATILE_DURABILITY_QOS;
wqos2.durability_service().service_cleanup_delay = Duration_t(532849506 * 1e-9);
wqos2.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
wqos2.durability_service().history_depth = 312845482;
wqos2.durability_service().max_samples = 762842861;
wqos2.durability_service().max_instances = 320602300;
wqos2.durability_service().max_samples_per_instance = 321976501;
wqos2.latency_budget().duration = Duration_t(487440300 * 1e-9);
wqos2.reliability().kind = RELIABLE_RELIABILITY_QOS;
wqos2.reliability().max_blocking_time = Duration_t(510011137 * 1e-9);
wqos2.history().kind = KEEP_ALL_HISTORY_QOS;
wqos2.history().depth = 1193831150;
wqos2.resource_limits().max_samples = 1633341369;
wqos2.resource_limits().max_instances = 414659463;
wqos2.resource_limits().max_samples_per_instance = 1627120252;
wqos2.transport_priority().value = 972947330;
wqos2.lifespan().duration = Duration_t(213303973 * 1e-9);
wqos2.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
wqos2.ownership_strength().value = 301739589;

SubscriberQos subqos = SUBSCRIBER_QOS_DEFAULT;
subqos.presentation().access_scope = INSTANCE_PRESENTATION_QOS;
subqos.presentation().coherent_access = true;
subqos.presentation().ordered_access = true;

DataReaderQos rqos = DATAREADER_QOS_DEFAULT;
rqos.durability().kind = VOLATILE_DURABILITY_QOS;
rqos.latency_budget().duration = Duration_t(604916512 * 1e-9);
rqos.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
rqos.reliability().max_blocking_time = Duration_t(637368419 * 1e-9);
rqos.history().kind = KEEP_LAST_HISTORY_QOS;
rqos.history().depth = 640808567;
rqos.resource_limits().max_samples = 1191646692;
rqos.resource_limits().max_instances = 715291965;
rqos.resource_limits().max_samples_per_instance = 1052517764;
rqos.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
rqos.time_based_filter().minimum_separation = Duration_t(441727422 * 1e-9);
rqos.reader_data_lifecycle().autopurge_no_writer_samples_delay = Duration_t(28215143 * 1e-9);
rqos.reader_data_lifecycle().autopurge_disposed_samples_delay = Duration_t(883375490 * 1e-9);

Steps to reproduce

Here is the main.cpp file with qos settings as mentioned above.

#include "HelloWorldPubSubTypes.h"

#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastrtps/attributes/PublisherAttributes.h>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/qos/PublisherQos.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
#include <fastrtps/attributes/SubscriberAttributes.h>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
#include <fastdds/dds/subscriber/DataReaderListener.hpp>
#include <fastdds/dds/core/status/SubscriptionMatchedStatus.hpp>

using namespace eprosima::fastdds::dds;

class PubListener : public DataWriterListener{
public:
    PubListener()
        : matched_(0)
        , firstConnected_(false)
    {
    }

    void on_offered_deadline_missed(DataWriter* writer, const OfferedDeadlineMissedStatus& status){
        std::cout << " [ WRITER : on_offered_deadline_missed ] " << id_ << std::endl;
    }
    void on_offered_incompatible_qos(DataWriter* writer, const OfferedIncompatibleQosStatus& status){
        std::cout << " [ WRITER : on_offered_incompatible_qos ] " << id_ << std::endl;
    }
    void on_liveliness_lost(DataWriter* writer, const LivelinessLostStatus& status){
        std::cout << " [ WRITER : on_liveliness_lost ] " << id_ << std::endl;
    }
    void on_publication_matched(DataWriter* writer, const PublicationMatchedStatus& info){
        std::cout << " [ WRITER : on_publication_matched ] Writer" << id_;
        if (info.current_count_change == 1){
            matched_ = info.total_count;
            std::cout << " matched." << std::endl;
        } else if (info.current_count_change == -1){
            matched_ = info.total_count;
            std::cout << " unmatched." << std::endl;
        } else {
            std::cout << info.current_count_change
                    << " is not a valid value for PublicationMatchedStatus current count change" << std::endl;
        }
    }
    void on_unacknowledged_sample_removed(DataWriter* writer, const InstanceHandle_t& instance){
        std::cout << " [ WRITER : on_unacknowledged_sample_removed ] " << id_ << std::endl;
    }

    int matched_;
    bool firstConnected_;
    int id_;

} writer_listener1_, writer_listener2_;

class CustomDataReaderListener : public DataReaderListener{
public:
    CustomDataReaderListener()
        : matched_(0)
        , samples_(0)
    {
    }
    
    void on_data_available(DataReader* reader){
        SampleInfo info;
        if (reader->take_next_sample(&hello_, &info) == ReturnCode_t::RETCODE_OK){
            if (info.instance_state == ALIVE_INSTANCE_STATE){
                samples_++;
                std::cout << "< READER : on_data_available > reader get message " << hello_.message() << " from " << hello_.index() << " RECEIVED" << std::endl;
            } 
        } 
    }
    void on_subscription_matched(DataReader* reader, const SubscriptionMatchedStatus& info){
        std::cout << " < READER : on_subscription_matched > ";
        if (info.current_count_change == 1){
            matched_ = info.total_count;
            std::cout << "Reader matched." << std::endl;
        } else if (info.current_count_change == -1){
            matched_ = info.total_count;
            std::cout << "Reader unmatched." << std::endl;
        } else {
            std::cout << info.current_count_change
                    << " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
        }
    }
    void on_sample_lost(DataReader* reader, const SampleLostStatus& status){
        std::cout << " < RADER : on_sample_lost > " << std::endl;
    }
    void on_requested_incompatible_qos(DataReader* reader, const RequestedIncompatibleQosStatus& status){
        std::cout << " < READER : on_requested_incompatible_qos > " << status.last_policy_id << std::endl;
    }
    void on_requested_deadline_missed(DataReader* reader, const RequestedDeadlineMissedStatus& status){
        std::cout << " < READER : on_requested_deadline_missed > " << std::endl;
    }
    void on_sample_rejected(DataReader* reader, const SampleRejectedStatus& status){
        std::cout << " < READER : on_sample_rejected > " << std::endl;
    }
    void on_liveliness_changed(DataReader* reader, const LivelinessChangedStatus& status){
        std::cout << " < READER : on_liveliness_changed > " << std::endl;
    }

    HelloWorld hello_;
    int matched_;
    uint32_t samples_;

} reader_listener_;

int main(){
    /* setting the participant qos */
    DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT;

    /* create participant */
    DomainParticipantFactory* factory = DomainParticipantFactory::get_instance();
    DomainParticipant* participant_ = factory->create_participant(0, pqos);

    /* register the type */
    TypeSupport type_(new HelloWorldPubSubType());
    type_.register_type(participant_);

    /* setting the topic qos */
    TopicQos tqos = TOPIC_QOS_DEFAULT;
    tqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
    tqos.durability_service().service_cleanup_delay = Duration_t(317782863 * 1e-9);
    tqos.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
    tqos.durability_service().history_depth = 1004623649;
    tqos.durability_service().max_samples = 1155043293;
    tqos.durability_service().max_instances = 4039424;
    tqos.durability_service().max_samples_per_instance = 1051145751;
    tqos.latency_budget().duration = Duration_t(853141231 * 1e-9);
    tqos.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
    tqos.reliability().max_blocking_time = Duration_t(991706505 * 1e-9);
    tqos.history().kind = KEEP_LAST_HISTORY_QOS;
    tqos.history().depth = 1843769184;
    tqos.resource_limits().max_samples = 2134960815;
    tqos.resource_limits().max_instances = 1533973938;
    tqos.resource_limits().max_samples_per_instance = 1980921081;
    tqos.transport_priority().value = 1487440300;
    tqos.lifespan().duration = Duration_t(164455410 * 1e-9);
    tqos.ownership().kind = SHARED_OWNERSHIP_QOS;

    /* create topic */
    Topic* topic_ = participant_->create_topic("Topic1", "HelloWorld", tqos); 

    /* setting the publisiher qos */
    PublisherQos pubqos = PUBLISHER_QOS_DEFAULT;
    pubqos.presentation().access_scope = TOPIC_PRESENTATION_QOS;
    pubqos.presentation().coherent_access = true;
    pubqos.presentation().ordered_access = true;

    /* create publisher */
    Publisher* publisher_ = participant_->create_publisher(pubqos, nullptr);

    /* setting the writer1 qos */
    DataWriterQos wqos1 = DATAWRITER_QOS_DEFAULT;
    wqos1.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
    wqos1.durability_service().service_cleanup_delay = Duration_t(567311302 * 1e-9);
    wqos1.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
    wqos1.durability_service().history_depth = 972947330;
    wqos1.durability_service().max_samples = 1744157312;
    wqos1.durability_service().max_instances = 301739589;
    wqos1.durability_service().max_samples_per_instance = 1011714986;
    wqos1.latency_budget().duration = Duration_t(921824426 * 1e-9);
    wqos1.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
    wqos1.reliability().max_blocking_time = Duration_t(852261378 * 1e-9);
    wqos1.history().kind = KEEP_ALL_HISTORY_QOS;
    wqos1.history().depth = 157146035;
    wqos1.resource_limits().max_samples = 1933789796;
    wqos1.resource_limits().max_instances = 598012395;
    wqos1.resource_limits().max_samples_per_instance = 1322252895;
    wqos1.transport_priority().value = 1485709160;
    wqos1.lifespan().duration = Duration_t(823257944 * 1e-9);
    wqos1.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
    wqos1.ownership_strength().value = 502680922;

    writer_listener1_.id_ = 1;

    /* create writer1 */
    DataWriter* writer1_ = publisher_->create_datawriter(topic_, wqos1, &writer_listener1_);

    /* setting the writer2 qos */
    DataWriterQos wqos2 = DATAWRITER_QOS_DEFAULT;
    wqos2.durability().kind = VOLATILE_DURABILITY_QOS;
    wqos2.durability_service().service_cleanup_delay = Duration_t(532849506 * 1e-9);
    wqos2.durability_service().history_kind = KEEP_ALL_HISTORY_QOS;
    wqos2.durability_service().history_depth = 312845482;
    wqos2.durability_service().max_samples = 762842861;
    wqos2.durability_service().max_instances = 320602300;
    wqos2.durability_service().max_samples_per_instance = 321976501;
    wqos2.latency_budget().duration = Duration_t(487440300 * 1e-9);
    wqos2.reliability().kind = RELIABLE_RELIABILITY_QOS;
    wqos2.reliability().max_blocking_time = Duration_t(510011137 * 1e-9);
    wqos2.history().kind = KEEP_ALL_HISTORY_QOS;
    wqos2.history().depth = 1193831150;
    wqos2.resource_limits().max_samples = 1633341369;
    wqos2.resource_limits().max_instances = 414659463;
    wqos2.resource_limits().max_samples_per_instance = 1627120252;
    wqos2.transport_priority().value = 972947330;
    wqos2.lifespan().duration = Duration_t(213303973 * 1e-9);
    wqos2.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
    wqos2.ownership_strength().value = 301739589;

    writer_listener2_.id_ = 2;

    /* create writer2 */
    DataWriter* writer2_ = publisher_->create_datawriter(topic_, wqos2, &writer_listener2_);

    /* setting the subscriber qos */
    SubscriberQos subqos = SUBSCRIBER_QOS_DEFAULT;
    subqos.presentation().access_scope = INSTANCE_PRESENTATION_QOS;
    subqos.presentation().coherent_access = true;
    subqos.presentation().ordered_access = true;

    /* create the subscriber qos */
    Subscriber* subscriber_ = participant_->create_subscriber(subqos, nullptr);

    /* setting the reader qos */
    DataReaderQos rqos = DATAREADER_QOS_DEFAULT;
    rqos.durability().kind = VOLATILE_DURABILITY_QOS;
    rqos.latency_budget().duration = Duration_t(604916512 * 1e-9);
    rqos.reliability().kind = BEST_EFFORT_RELIABILITY_QOS;
    rqos.reliability().max_blocking_time = Duration_t(637368419 * 1e-9);
    rqos.history().kind = KEEP_LAST_HISTORY_QOS;
    rqos.history().depth = 640808567;
    rqos.resource_limits().max_samples = 1191646692;
    rqos.resource_limits().max_instances = 715291965;
    rqos.resource_limits().max_samples_per_instance = 1052517764;
    rqos.ownership().kind = EXCLUSIVE_OWNERSHIP_QOS;
    rqos.time_based_filter().minimum_separation = Duration_t(441727422 * 1e-9);
    rqos.reader_data_lifecycle().autopurge_no_writer_samples_delay = Duration_t(28215143 * 1e-9);
    rqos.reader_data_lifecycle().autopurge_disposed_samples_delay = Duration_t(883375490 * 1e-9);

    /* create reader */
    DataReader* reader_ = subscriber_->create_datareader(topic_, rqos, &reader_listener_);

    /* wirter write the message */
    HelloWorld hello_1;
    hello_1.index(1);
    hello_1.message("HelloWorld1");
    std::cout << "Message: " << hello_1.message() << " with index: " << hello_1.index() << " SENT" << std::endl;
    writer1_->write(&hello_1);

    HelloWorld hello_2;
    hello_2.index(2);
    hello_2.message("HelloWorld2");
    std::cout << "Message: " << hello_2.message() << " with index: " << hello_2.index() << " SENT" << std::endl;
    writer2_->write(&hello_2);

    /* sleep */
    sleep(10);

    /* delete everything */
    if (reader_ != nullptr) subscriber_->delete_datareader(reader_);
    if (writer1_ != nullptr) publisher_->delete_datawriter(writer1_);
    if (writer2_ != nullptr) publisher_->delete_datawriter(writer2_);
    if (subscriber_ != nullptr) participant_->delete_subscriber(subscriber_);
    if (publisher_ != nullptr) participant_->delete_publisher(publisher_);
    if (topic_ != nullptr) participant_->delete_topic(topic_);
    DomainParticipantFactory::get_instance()->delete_participant(participant_);

    return 0;
}

Fast DDS version/commit

v2.11.1

Platform/Architecture

Other. Please specify in Additional context section.

Transport layer

Intra-process

Additional context

ubuntu 22.04 amd64

XML configuration file

No response

Relevant log output

[ WRITER : on_publication_matched ] Writer2 matched.
 < READER : on_subscription_matched > Reader matched.
 [ WRITER : on_publication_matched ] Writer1 matched.
 < READER : on_subscription_matched > Reader matched.
Message: HelloWorld1 with index: 1 SENT
< READER : on_data_available > reader get message HelloWorld1 from 1 RECEIVED
Message: HelloWorld2 with index: 2 SENT
 [ WRITER : on_publication_matched ] Writer1 unmatched.
 [ WRITER : on_publication_matched ] Writer2 unmatched.

Network traffic capture

No response

[JesusPoderoso]

Hi @llddang, this behavior is caused by the ownership QoS. Setting all the endpoints with EXCLUSIVE_OWNERSHIP_QOS, and the first writer ownership strength higher than the second writer makes the DataReader unable to receive any message from the second DataWriter.
If you set the second DataWriter's ownership strength higher than the first one, DataReader receives both messages but it will not receive any more messages from the first DataWriter.

Setting all of their ownership QoS as SHARED_OWNERSHIP_QOS would make your example work as expected.

I am moving this question to the support discussion forum as per Fast DDS CONTRIBUTING guidelines.

[llddang]

Hi, @JesusPoderoso,
Thank you for your kind explanation. However, I have some questions about the following statement:
"If you set the second DataWriter's ownership strength higher than the first one, DataReader receives both messages but it will not receive any more messages from the first DataWriter."

I conducted an experiment with the second DataWriter having a higher strength than the first one. As you explained, the DataReader received both messages, but after that, DataReader didn't receive any more messages from the first DataWriter. I'm wondering if this behavior is implemented correctly. Based on my understanding of your explanation, if the second DataWriter has a higher strength than the first one, the DataReader should only receive messages from the second DataWriter. I'm uncertain if the behavior of the DataReader, which receives both messages, is implemented correctly.

I would appreciate it if you could provide me with a document or explanation to address my concerns.
Best regards.

[JLBuenoLopez]

@llddang

Ownership QoS policy affects instances. Thus, the same DataWriter might publish in some instance and being received whereas in other instances is prevented by stronger DataWriters if the ownership is set to EXCLUSIVE. The DataReader is receiving two samples because if the first received sample comes from a new instance, the DataReader does not know yet that a stronger DataWriter might claim ownership of that instance. When the second sample is received in the same instance and with a higher strength value, then the first DataWriter is ignored whenever he publish in that same instance (publishing in any other instance where the other DataWriter has not published yet, will be also received). More information can be found in the DDS specification section 2.2.3.9 dedicated to the Ownership QoS.

[llddang]

Thank you for your quick reply. From the explanation, it seems that my understanding of specification is still unclear. I will clearly understand the contents of the specification and check it through the experiment.

Best regard