Errors sending large data

[spiderkeys]

Context: Sending video data with each sample being up to 150Kb in size. The first several frames go through, followed by failures on almost every other frame. data_msg_length has a peculiar value of ~ULONG_MAX. It was my understanding that the one step required for large data is to set the publisher mode to asynchronous to allow for fragmentation. Please let me know if I have set up anything incorrectly here.

IDL definition:

module Video
{
    const long VIDEO_FRAME_MAX_SIZE = 200000;

    struct TAccessUnit
    {
        sequence<octet, VIDEO_FRAME_MAX_SIZE> payload;
    };
};

Following is the setup:

// Set participant attributes
    ParticipantAttributes partAttr;
    partAttr.rtps.sendSocketBufferSize = 201000;
    partAttr.rtps.builtin.domainId = 0;
    partAttr.rtps.builtin.leaseDuration = c_TimeInfinite;
    partAttr.rtps.setName("Participant_pub");

    // Create participant
    LOG( INFO ) << "Creating participant...";
    mp_participant = Domain::createParticipant( partAttr );
    if( mp_participant == nullptr )
    {
        throw std::runtime_error( "Failed to create participant" );
    }

    // Register types
    LOG(INFO ) << "Registering TAccessUnit type...";
    if( !Domain::registerType( mp_participant, &m_type ) )
    {
        throw std::runtime_error( "Failed to register TAccessUnit type" );
    }

    // Set publisher attributes
    PublisherAttributes pubAttr;

    // Topic
    pubAttr.topic.topicKind         = NO_KEY;
    pubAttr.topic.topicDataType     = "Video::TAccessUnit";
    pubAttr.topic.topicName         = "rovVideo";

    // History
    pubAttr.topic.historyQos.kind   = KEEP_LAST_HISTORY_QOS;
    pubAttr.topic.historyQos.depth  = 1;

    // Reliability
    pubAttr.qos.m_reliability.kind  = BEST_EFFORT_RELIABILITY_QOS;

    // Durability
    pubAttr.qos.m_durability.kind   = VOLATILE_DURABILITY_QOS;

    // Asynchronous
    pubAttr.qos.m_publishMode.kind = ASYNCHRONOUS_PUBLISH_MODE;

    // Create publisher
    LOG(INFO ) << "Creating publisher...";
    mp_publisher = Domain::createPublisher( mp_participant, pubAttr, nullptr );
    if( mp_publisher == nullptr )
    {
        throw std::runtime_error( "Failed to create publisher" );
    }

Sending data:

        // Sets the frame information within the data to the correct size
        // based on the buffer size
        channel->videoFrame.payload().resize( frameLength );

        // Copy the data from the buffer into the object to be sent using
        // the middleware.
        std::copy( sourceBuffer, sourceBuffer + frameLength, channel->videoFrame.payload().begin() );

        mp_publisher->write((void*)&channel->videoFrame);

Added logging at this location:
https://github.com/eProsima/Fast-RTPS/blob/master/src/cpp/rtps/messages/RTPSMessageGroup.cpp#L240

std::cout << "cdrmsg_fullmsg->max_size = " << cdrmsg_fullmsg->max_size << std::endl;
std::cout << "cdrmsg_fullmsg->length = " << cdrmsg_fullmsg->length << std::endl;
std::cout << "data_msg_length = " << data_msg_length << std::endl;

The resulting output:

2017-03-13 23:19:51,068546 Writing payload size: 29596
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 29600
2017-03-13 23:19:51,101922 Writing payload size: 30232
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 4294932568
[RTPS_WRITER Error] A problem occurred when adding a message -> Function send_Changes_AsData
2017-03-13 23:19:51,135171 Writing payload size: 29284
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 29288
2017-03-13 23:19:51,168482 Writing payload size: 28836
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 4294931172
[RTPS_WRITER Error] A problem occurred when adding a message -> Function send_Changes_AsData
2017-03-13 23:19:51,201824 Writing payload size: 29244
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 29248
2017-03-13 23:19:51,234751 Writing payload size: 96262
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 31302
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 31302
2017-03-13 23:19:51,268308 Writing payload size: 23552
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 23556
2017-03-13 23:19:51,301845 Writing payload size: 33692
cdrmsg_fullmsg->max_size = 65500
cdrmsg_fullmsg->length = 32
data_msg_length = 4294936028
[RTPS_WRITER Error] A problem occurred when adding a message -> Function send_Changes_AsData

[jespersmith]

Probably the UDP buffers overflow. Use a throughputController for your publisher, see documentation (I'm using a wrapper so I don't know the code by heart). Keep the period small and set a useful amount of bytes.

[richiware]

Do you still notice the problem with last version of Fast RTPS? This version fixes several bugs and introduces piggyback heartbeats to improve sending of large data.

[richiware]

Sorry, I close the issue accidentally

[cvilas]

Is this issue resolved? I am trying to do something similar but using the RTPS layer. An example program is attached, following guidelines for sending large files in the user manual. Can you spot what is it that I am doing wrong? Thanks.

large_data_example.zip

[cvilas]

Just to add, for payload sizes just above 65kB, I do get the following error messages on the writer side:

RTPS_WRITER Error] Error sending change 1 -> Function send_any_unsent_changes [RTPS_WRITER Error] Cannot add DATA submsg to the CDRMessage. Buffer too small -> Function add_data

These messages do not appear if the payload is much larger (for example: 640x480 - as in sending a VGA grayscale image - the above test program was derived from such a use case, with reliability set to BEST_EFFORT)

Also, when the writer is pushing messages out in a loop (as you would do with a camera driver), RTPS consumption of system memory explodes within a few seconds and consumes all RAM)

Vilas

[richiware]

The error message is shown because the CacheChange cannot be inserted entirely in a RTPS message. In low level RTPS API when a user works with a CacheChange, he should specify if the CacheChange has to be fragmented. In Pub/Sub API this is done automatically by the library.

You can introduce this code line when using a CacheChange:

pChange->setFragmentSize(64000);

About the memory increasing, the history attribute maximumReservedCaches is set to 0, that is equivalent to infinite. It means each time you request a new CacheChange the library will allocate a new one. Logic never will enter in this block of code:

  while (pChange == nullptr)
  {
    pWriter->remove_older_changes(nMaxChangesToRemove);
    ++nRemoved;
    pChange = pWriter->new_change(dataSizeFunc, ChangeKind_t::ALIVE);
  }

If you set the history attribute to 1, for example, the internal CacheChange pool only mateins one CacheChange.

[cvilas]

Sorry, I am still not able to see any received messages on the reader end. Is there anything special I have to do other than setFragmentSize(). Reader's onNewCacheChangeAdded() never gets called for any data size after setFragmentSize.

[spiderkeys]

@cvilas You may want to try enabling asynchronous publishing mode on your publisher. This is required for sending data larger than a single UDP packet (or perhaps in this case, the user specified fragment size). When I originally tried to stream video way back when, I didn't know about the setFragmentSize call, so maybe these two things combined are the magic words to get this working?

// Asynchronous Publish Mode
pubAttr.qos.m_publishMode.kind = ASYNCHRONOUS_PUBLISH_MODE;
mp_publisher = Domain::createPublisher( mp_participant, pubAttr, nullptr );

With RTI's implementation, I usually set my fragment size to:

<element>
   <name>dds.transport.UDPv4.builtin.parent.message_size_max</name>
   <!-- 64 KB minus header sizes -->
   <value>65507</value>
</element>

[cvilas]

@spiderkeys Thanks. I am working with the RTPS layer rather than pub/sub api (which apparently doesn't require setFragmentSize to be called anyway). The asynchronous mode is set on the writer and when registering with the participant.

The pub/sub API would be simpler to use I suppose, but I am now committed to using the RTPS layer due to various design choices we made on our end.

[richiware]

@cvilas Are you using the same example you attached? I've added the lines I told you and it's working:

83c83
<   historyAttr.maximumReservedCaches = 0;
---
>   historyAttr.maximumReservedCaches = 1;
194c194
<   historyAttr.maximumReservedCaches = 0;
---
>   historyAttr.maximumReservedCaches = 1;
279a280
>   pChange->setFragmentSize(64000);

@spiderkeys He is not using the Pub/Sub API. He is using a low level API that is slightly different. But he enables the asynchronous writing. Thanks for your answer.

[cvilas]

Indeed, it does work! Thanks. I am not sure what I was looking at just a moment ago. Perhaps time for a coffee break. Thanks very much @richiware. I appreciate your time on the issue.

[cvilas]

I can confirm that I can reliably transmit 10MB data between RTPS readers and writers, and also transmit 640*480 VGA image data at 50 fps in best-effort mode - the two scenarios described in the user manual.

Thanks again @richiware @spiderkeys

[MiguelCompany]

Since all the users involved seem happy with the solution provided, I close this (very) old issue