Multiple data writers and one data reader

[MMarcus95]

Hello,

in my application I have an arbitrary number of data writer (e.g. 100) publishing on the same topic and a data reader listening on that topic. I would like to be 100% sure that the data reader receive all the messages. How can I do?

I tried with several configurations, e.g using KEEP_ALL_HISTORY_QOS, RELIABLE_RELIABILITY_QOS and topic instances (adjusting the resource limits accordingly) but still I continue to lose messages when the publishing rate is high. I don't care if the write operation blocks, I just want to make sure that I can reconstruct all the messages sent by each data writer.

Thanks!

[JesusPoderoso]

Hi @MMarcus95, thanks for using Fast DDS.
I've successfully tested launching 100 publishers with a single subscriber on the same topic, and checking that all the samples are received. The parameters I've set up to do so are:

• Launch 100 publishers on hello world topic
• Launch 1 subscriber on hello world topic
• Publishers write operation is performed every 10 milliseconds
• Publishers send 200 hello world messages and then exit.
• Publishers' data writer runs a wait for acknowledgment method to make sure that the last sample sent is received (before exit process).
• Using UDPv4 transport
• Both endpoints with the following QoS:
  • TRANSIENT_LOCAL Durability
  • RELIABLE Reliability, with max blocking time at 1 second
  • KEEP_ALL History
  • Unlimited Resource limits

Let us know if it suits your deployment case. I am sure that some of those QoS could be less restrictive.

To test it, I've used a refactored Hello World example that will be released in the next Fast DDS v3.0.0 version, and added a couple of modifications (see the header of the attached script). Hope it helps!

XML profile file: issue_4618_profile.xml

<?xml version="1.0" encoding="UTF-8" ?>
<profiles xmlns="http://www.eprosima.com" >
    <participant profile_name="hello_world_participant_profile" is_default_profile="true">
        <domainId>113</domainId>
        <rtps>
            <name>hello_world_participant</name>
            <builtinTransports>UDPv4</builtinTransports>
        </rtps>
    </participant>
    <data_writer profile_name="hello_world_datawriter_profile" is_default_profile="true">
        <qos>
            <durability>
                <kind>TRANSIENT_LOCAL</kind>
            </durability>
            <reliability>
                <kind>RELIABLE</kind>
                <max_blocking_time>
                    <sec>1</sec>
                </max_blocking_time>
            </reliability>
        </qos>
        <topic>
            <historyQos>
                <kind>KEEP_ALL</kind>
            </historyQos>
            <resourceLimitsQos>
                <max_samples>0</max_samples>
                <max_instances>0</max_instances>
                <max_samples_per_instance>0</max_samples_per_instance>
            </resourceLimitsQos>
        </topic>
    </data_writer>

    <data_reader profile_name="hello_world_datareader_profile" is_default_profile="true">
        <qos>
            <durability>
                <kind>TRANSIENT_LOCAL</kind>
            </durability>
            <reliability>
                <kind>RELIABLE</kind>
                <max_blocking_time>
                    <sec>1</sec>
                </max_blocking_time>
            </reliability>
        </qos>
        <topic>
            <historyQos>
                <kind>KEEP_ALL</kind>
            </historyQos>
            <resourceLimitsQos>
                <max_samples>0</max_samples>
                <max_instances>0</max_instances>
                <max_samples_per_instance>0</max_samples_per_instance>
            </resourceLimitsQos>
        </topic>
    </data_reader>
</profiles>

Bash script: generator.sh Run bash generator.sh --help for more information

#!/bin/bash

### !!! IMPORTANT: Modify the following line to print writer GUID while subscriber receives messages
###
### diff --git a/examples/cpp/hello_world/ListenerSubscriberApp.cpp b/examples/cpp/hello_world/ListenerSubscriberApp.cpp
### --- a/examples/cpp/hello_world/ListenerSubscriberApp.cpp
### +++ b/examples/cpp/hello_world/ListenerSubscriberApp.cpp
### @@ -134,7 +134,7 @@ void ListenerSubscriberApp::on_data_available(
###              received_samples_++;
###              // Print Hello world message data
###              std::cout << "Message: '" << hello_.message() << "' with index: '" << hello_.index()
### -                      << "' RECEIVED" << std::endl;
### +                      << "' RECEIVED. ID: " << info.sample_identity.writer_guid() << std::endl;
###              if (samples_ > 0 && (received_samples_ >= samples_))
###              {
###
### !!! IMPORTANT: Modify the following line to force publishers to wait until last sample is acknowledged (1 min suggested, as the timeout of the applications)
###
### diff --git a/examples/cpp/hello_world/PublisherApp.cpp b/examples/cpp/hello_world/PublisherApp.cpp
### --- a/examples/cpp/hello_world/PublisherApp.cpp
### +++ b/examples/cpp/hello_world/PublisherApp.cpp
### @@ -140,6 +140,15 @@ void PublisherApp::run()
###                      return is_stopped();
###                  });
###      }
### +    ReturnCode_t ret = writer_->wait_for_acknowledgments(Duration_t { 60, 0 });
### +    if (ReturnCode_t::RETCODE_OK == ret)
### +    {
### +        std::cout << "All messages have been confirmed" << std::endl;
### +    }
### +    else
### +    {
### +        std::cout << "Some messages have not been confirmed. Closing Publisher." << std::endl;
### +    }
###  }
###
###
### !!! OPTIONAL: Modify this period with the desired publication interval (10 ms suggested)
###
### diff --git a/examples/cpp/hello_world/PublisherApp.hpp b/examples/cpp/hello_world/PublisherApp.hpp
### --- a/examples/cpp/hello_world/PublisherApp.hpp
### +++ b/examples/cpp/hello_world/PublisherApp.hpp
### @@ -88,7 +88,7 @@ private:
###
###      std::atomic<bool> stop_;
###
### -    const uint32_t period_ms_ = 100; // in ms
### +    const uint32_t period_ms_ = 10; // in ms
###  };

verbose=false
output_file=""

# script usage print method
print_usage()
{
    echo "Usage: $0 [-h|--help] [-v|--verbose] [-f|--file <name>]"
    echo "  -h, --help          Print usage"
    echo "  -v, --verbose       Enable verbose flag"
    echo "  -f, --file  <name>  Record output to file <name>"
}

run_example()
{
    # Mind changing following variables
    project_path=~/Documents/test_4618/
    xml_profile_path=~/Documents/test_4618/issue_4618_profile.xml
    n_publishers=100
    n_message_per_publisher=200
    timeout_seconds=60  # 1 minute

    # dependant required variables
    sourcing_path=$project_path/install/setup.bash
    example_path=$project_path/install/fastdds/share/fastdds/examples/cpp/hello_world/bin/hello_world
    n_messages_expected=$(( $n_publishers * $n_message_per_publisher ))

    # Download devel hello world branch and build project if not already built
    if [[ ! -f $example_path ]]; then
        prev_dir=$(pwd)
        cd $project_path
        git clone --branch feature/example_refactor/hello_world https://github.com/eProsima/Fast-DDS.git src/fastdds > /dev/null &
        git clone https://github.com/eProsima/Fast-CDR.git src/fastcdr > /dev/null &
        git clone https://github.com/eProsima/foonathan_memory_vendor.git src/foonathan_memory_vendor > /dev/null &
        wait
        colcon build --packages-up-to fastdds --cmake-args -DCOMPILE_EXAMPLES=ON -DINSTALL_EXAMPLES=ON
        wait
        cd $prev_dir
    fi

    # source Fast DDS environment
    source $sourcing_path

    # source XML profile
    export FASTDDS_DEFAULT_PROFILES_FILE=$xml_profile_path

    # print configuration
    if [[ $verbose == true ]]; then
        echo "QoS configuration:"
        cat $xml_profile_path
    fi
    echo "Running hello world example with $n_publishers publishers, $n_message_per_publisher messages per publisher, expecting $n_messages_expected messages"

    # Launch publishers in background, given samples per publisher
    for ((i=1; i<=$n_publishers; i++)); do
        if [[ $verbose == true ]]; then
            timeout $timeout_seconds $example_path publisher -s $n_message_per_publisher &
        else
            timeout $timeout_seconds $example_path publisher -s $n_message_per_publisher > /dev/null &
        fi
    done

    echo "Running subscriber. Timeout set at $timeout_seconds seconds"

    # Run a subscriber in foreground, waiting samples
    sub_output=$(timeout $timeout_seconds $example_path subscriber -s $n_messages_expected &)

    wait

    echo "Publishers and subscriber finished. Processing output..."

    declare -A hello_world_message_counter
    declare -A hello_world_message_last
    declare -A hello_world_message_missed

    # Extract the last word of each line and count the occurrences and misses
    timed_out=false
    while IFS= read -r line; do
        last_word=$(echo "$line" | awk '{print $NF}')
        first_word=$(echo "$line" | awk '{print $1}')
        if [[ $first_word == "Message:" ]]; then
            # check the amount of messages received
            if [[ -z ${hello_world_message_counter[$last_word]} ]]; then
                hello_world_message_counter[$last_word]=1
            else
                ((hello_world_message_counter[$last_word]++))
            fi

            # check if sample gap and register missed messages
            index=(${line//\'/ })
            index=${index[5]}
            if [[ -z ${hello_world_message_last[$last_word]} ]]; then
                hello_world_message_last[$last_word]=$index
            else
                if [[ ${hello_world_message_last[$last_word]}+1 -eq $index ]]; then
                    hello_world_message_last[$last_word]=$index
                else
                    if [[ -z ${hello_world_message_missed[$last_word]} ]]; then
                        hello_world_message_missed[$last_word]=$index
                    else
                        hello_world_message_missed[$last_word]="${hello_world_message_missed[$last_word]}, $index"
                    fi
                fi

            fi
        fi
        # print if required
        if [[ $verbose == true ]]; then
            echo "$line"
        fi
        # register if subscriber finshed by timeout
        if [[ $line == "SIGTERM received, stopping Subscriber execution." ]]; then
            timed_out=true
        fi
    done <<< "$sub_output"

    # print the hello world message counters and missed samples
    i=0
    total_messages=0
    for id in "${!hello_world_message_counter[@]}"; do
        ((i++))
        total_messages=$((total_messages + hello_world_message_counter[$id]))
        if [[ ${hello_world_message_counter[$id]} -ne $n_message_per_publisher ]]; then
            if [[ -z ${hello_world_message_missed[$id]} ]]; then
                hello_world_message_missed[$id]="last $(( $n_message_per_publisher-${hello_world_message_counter[$id]} ))"
            fi
            echo "[$i] Publisher GUID: $id, Received ${hello_world_message_counter[$id]}, Lost $(( $n_message_per_publisher-${hello_world_message_counter[$id]} )), Missed: ${hello_world_message_missed[$id]}"
        fi
    done
    # print if subscriber timed out
    if [[ $timed_out == true ]]; then
        echo "Subscriber timed out"
    fi

    # print final status
    if [[ $total_messages -eq $n_messages_expected ]]; then
        echo "All messages received"
    else
        echo "Total messages received [$( printf "%.0f" $(( $total_messages * 100 / $n_messages_expected )) )%]: $total_messages out of $n_messages_expected. $(( $n_messages_expected - $total_messages )) messages lost"
    fi
}

# Check arguments
while [[ $# -gt 0 ]]; do
    key="$1"
    case $key in
        -h|--help)
        print_usage
        exit 0
        ;;
        -v|--verbose)
        verbose=true
        shift
        ;;
        -f|--file)
        output_file="$2"
        shift
        shift
        ;;
        *)
        echo "Invalid argument: $1"
        print_usage
        exit 1
        ;;
    esac
done

if [[ -n $output_file ]]; then
    run_example > $output_file &
else
    run_example
fi