DataReader::read() reads invalid data

[fschoenm]

Is there an already existing issue for this?

• I have searched the existing issues

Expected behavior

Receive the messages exactly as they are sent. (According to https://fast-dds.docs.eprosima.com/en/latest/fastdds/dds_layer/subscriber/dataReader/readingData.html).

Current behavior

DataReader::read() seems to invent data or merges received data from multiple messages.

We have a simple publisher/subscriber scenario with a message type that contains a map/std::map. When publishing multiple messages, subsequent messages don't receive the message that has been sent but message contents of multiple messages are merged into one:

Message sent: {0: "abc"}
Message rcvd: {0: "abc"}
Message sent: {1: "def"}
Message rcvd: {0: "abc", 1: "def"}

This seems to happen when using DataReader::read() (with DataReader::read_next_sample() it seems to work correctly but that is not our use case so I haven't investigated a lot).

Steps to reproduce

Here's a hopefully complete reproduction scenario. Code from IDL is generated with fastddsgen -d idl msg.idl.

main.cpp

#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/SubscriberListener.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>

#include "idl/msg.h"
#include "idl/msgPubSubTypes.h"

#include <fmt/format.h>
#include <fmt/ranges.h>

using namespace std::chrono_literals;

using MsgT = ComplexMessage;
using MsgPubSubT = ComplexMessagePubSubType;

template <class MsgT>
struct SubscriberLoans
{
    eprosima::fastdds::dds::LoanableSequence<MsgT> data;
    eprosima::fastdds::dds::LoanableSequence<eprosima::fastdds::dds::SampleInfo> infos;
};

class Listener : public eprosima::fastdds::dds::SubscriberListener
{
public:
    void on_data_available(eprosima::fastdds::dds::DataReader* reader) override
    {
        while (reader->read(loans.data, loans.infos, eprosima::fastdds::dds::LENGTH_UNLIMITED,
                            eprosima::fastdds::dds::NOT_READ_SAMPLE_STATE) == ReturnCode_t::RETCODE_OK)
        {
            for (eprosima::fastdds::dds::LoanableCollection::size_type i = 0; i < loans.infos.length(); ++i)
            {
                const auto& infos = loans.infos[i];
                const auto& msg = loans.data[i];
                std::cout << "Message rcvd: " << fmt::format("{}\n", msg.dict());
            }

            auto ret = reader->return_loan(loans.data, loans.infos);
            assert(ret == ReturnCode_t::RETCODE_OK);
        }
    }

private:
    SubscriberLoans<MsgT> loans;
};

int
main()
{
    eprosima::fastdds::dds::DomainParticipantQos qos;
    eprosima::fastdds::dds::DomainParticipant* part{
        eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->create_participant(
            1, qos)
    };

    // register types
    eprosima::fastdds::dds::TypeSupport typeSupport(new MsgPubSubT());
    part->register_type(typeSupport);

    // create publisher
    eprosima::fastdds::dds::PublisherQos pubQos;
    auto pub = part->create_publisher(pubQos);
    eprosima::fastdds::dds::TopicQos topicQos;
    auto topic = part->create_topic("/msg", typeSupport.get_type_name(), topicQos);
    eprosima::fastdds::dds::DataWriterQos writerQos;
    auto writer = pub->create_datawriter(topic, writerQos);

    // create subscriber
    Listener listener;
    eprosima::fastdds::dds::SubscriberQos subQos;
    auto sub = part->create_subscriber(subQos);
    eprosima::fastdds::dds::DataReaderQos readerQos;
    auto reader = sub->create_datareader(topic, readerQos, &listener);

    // publish data
    MsgT msg0;
    msg0.dict().insert(std::make_pair(0, "abc"));
    std::cout << "Message sent: " << fmt::format("{}\n", msg0.dict());
    writer->write(&msg0);
    std::this_thread::sleep_for(20ms);

    MsgT msg1;
    msg1.dict().insert(std::make_pair(1, "def"));
    std::cout << "Message sent: " << fmt::format("{}\n", msg1.dict());
    writer->write(&msg1);
    std::this_thread::sleep_for(20ms);

    return 0;
}

msg.idl

struct ComplexMessage {
	map<uint32, string> dict;
};

CMakeLists.txt

cmake_minimum_required(VERSION 3.23)

project(fastdds-bug LANGUAGES CXX)

set(CMAKE_EXPORT_COMPILE_COMMANDS ON)

find_package(fastdds REQUIRED)
find_package(fmt REQUIRED)

add_executable(bug main.cpp)
target_compile_features(bug PUBLIC cxx_std_23)
target_link_libraries(bug PUBLIC fast-dds::fast-dds)
target_link_libraries(bug PUBLIC fmt::fmt)
target_sources(bug PRIVATE idl/msg.cxx idl/msgPubSubTypes.cxx)

Fast DDS version/commit

2.14.2

Platform/Architecture

Windows 10 Visual Studio 2019

Transport layer

Default configuration, UDPv4 & SHM

Additional context

n/a

XML configuration file

n/a

Relevant log output

Message sent: {0: "abc"}
Message rcvd: {0: "abc"}
Message sent: {1: "def"}
Message rcvd: {0: "abc", 1: "def"}

Network traffic capture

n/a

[JesusPoderoso]

Hi @fschoenm, thanks for using Fast DDS.
As you have pointed here, read retrieves a collection of samples, while read_next_sample only retrieves the next, non-previously accessed data value available.
Mind that take methods also remove the samples from the DataReader (they are no longer accessible), meanwhile read methods do not remove them.
That is the reason why you are obtaining all previous samples in each read iteration. Check the read documentation for detailed info

[fschoenm]

That is the reason why you are obtaining all previous samples in each read iteration.

@JesusPoderoso I don't understand what you mean. I'm not receiving previous samples. I'm receiving a sample that I haven't sent, which seems to be merged from previous samples. When not using a map but e.g. a string, everything works as expected and I don't receive previous samples.

It seems like you didn't really understand the description but this is the essence:

Message sent: {0: "abc"}
Message rcvd: {0: "abc"}
Message sent: {1: "def"}
Message rcvd: {0: "abc", 1: "def"}

That is just plain wrong, as I never sent a message with two keys in the map.

Please move back to bugs.

[fschoenm]

@JesusPoderoso Here's a section from your documentation:

The read_next_sample operation is semantically equivalent to the read operation where the input Data sequence has max_length = 1 , the sample_states = NOT_READ_SAMPLE_STATE , the view_states = ANY_VIEW_STATE , and the instance_states = ANY_INSTANCE_STATE .

This is not true as the above code works when using read_next_sample with these parameters instead of read:

This works with maps:

        while (reader->read_next_sample(&msg, &info) == ReturnCode_t::RETCODE_OK) {
            std::cout << "Message rcvd: " << fmt::format("{}\n", msg.dict());
        }

This does not work with maps:

        while (reader->read(loans.data, loans.infos, 1,
                            eprosima::fastdds::dds::NOT_READ_SAMPLE_STATE,
                            eprosima::fastdds::dds::ANY_VIEW_STATE, eprosima::fastdds::dds::ANY_INSTANCE_STATE
        ) == ReturnCode_t::RETCODE_OK)

[JesusPoderoso]

I am moving this issue to the Support discussion forum section according to the Fast DDS CONTRIBUTING guidelines.