Deadlock when calling RTPSDomain::createRTPSWriter in a ReaderListener::on_new_cache_change_added call.

[owillebo]

Is there an already existing issue for this?

• I have searched the existing issues

Expected behavior

No dead lock.

Current behavior

Dead lock.

Steps to reproduce

Call RTPSDomain::createRTPSWriter in a ReaderListener::on_new_cache_change_added call.

Fast DDS version/commit

repositories:
    foonathan_memory_vendor:
        type: git
        url: https://github.com/eProsima/foonathan_memory_vendor.git
        version: v1.3.1
    fastcdr:
        type: git
        url: https://github.com/eProsima/Fast-CDR.git
        version: v2.2.5
    fastdds:
        type: git
        url: https://github.com/eProsima/Fast-DDS.git
        version: v3.1.0
    fastddsgen:
        type: git
        url: https://github.com/eProsima/Fast-DDS-Gen.git
        version: v4.0.2

Platform/Architecture

Other. Please specify in Additional context section.

Transport layer

UDPv4

Additional context

Windows 10 Visual Studio 2022

Not Flagged		12732	0	Worker Thread	dds.udp.7413	msvcp140d.dll!00007ffd2ef42107
>	fastddsd-3.1.dll!eprosima::fastdds::rtps::RTPSParticipantImpl::create_writer<eprosima::fastdds::rtps::BaseWriter * <lambda>(const eprosima::fastdds::rtps::GUID_t &, eprosima::fastdds::rtps::WriterAttributes &, eprosima::fastdds::rtps::FlowController *, eprosima::fastdds::rtps::IPersistenceService *, bool)>(eprosima::fastdds::rtps::RTPSWriter * * writer_out, eprosima::fastdds::rtps::WriterAttributes & param, const eprosima::fastdds::rtps::EntityId_t & entity_id, bool is_builtin, const eprosima::fastdds::rtps::RTPSParticipantImpl::create_writer::__l2::eprosima::fastdds::rtps::BaseWriter * <lambda>(const eprosima::fastdds::rtps::GUID_t &, eprosima::fastdds::rtps::WriterAttributes &, eprosima::fastdds::rtps::FlowController *, eprosima::fastdds::rtps::IPersistenceService *, bool) & callback) Line 1002	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::RTPSParticipantImpl::create_writer(eprosima::fastdds::rtps::RTPSWriter * * WriterOut, eprosima::fastdds::rtps::WriterAttributes & watt, eprosima::fastdds::rtps::WriterHistory * hist, eprosima::fastdds::rtps::WriterListener * listen, const eprosima::fastdds::rtps::EntityId_t & entityId, bool isBuiltin) Line 1265	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::RTPSParticipantImpl::createWriter(eprosima::fastdds::rtps::RTPSWriter * * WriterOut, eprosima::fastdds::rtps::WriterAttributes & param, eprosima::fastdds::rtps::WriterHistory * hist, eprosima::fastdds::rtps::WriterListener * listen, const eprosima::fastdds::rtps::EntityId_t & entityId, bool isBuiltin) Line 1201	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::RTPSDomain::createRTPSWriter(eprosima::fastdds::rtps::RTPSParticipant * p, eprosima::fastdds::rtps::WriterAttributes & watt, eprosima::fastdds::rtps::WriterHistory * hist, eprosima::fastdds::rtps::WriterListener * listen) Line 344	C++
 	XXXX.dll!rtps::capplication::addPublication(const std::string & topic, const std::string & type, rtps::reliability_t reliability) Line 261	C++
 	XXXX.dll!XXXX::Bus::onData(const std::string & contents, const rtps::csubscription_handle & sub) Line 746	C++
 	XXXX.dll!rtps::Listener::on_new_cache_change_added(eprosima::fastdds::rtps::RTPSReader * reader, const eprosima::fastdds::rtps::CacheChange_t * const change) Line 325	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::StatelessReader::change_received(eprosima::fastdds::rtps::CacheChange_t * change) Line 402	C++
>	fastddsd-3.1.dll!eprosima::fastdds::rtps::StatelessReader::process_data_msg(eprosima::fastdds::rtps::CacheChange_t * change) Line 660	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::MessageReceiver::process_data_message_without_security::__l2::<lambda>(eprosima::fastdds::rtps::BaseReader * reader) Line 229	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::MessageReceiver::findAllReaders<void <lambda>(eprosima::fastdds::rtps::BaseReader *)>(const eprosima::fastdds::rtps::EntityId_t & readerID, const eprosima::fastdds::rtps::MessageReceiver::process_data_message_without_security::__l2::void <lambda>(eprosima::fastdds::rtps::BaseReader *) & callback) Line 727	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::MessageReceiver::process_data_message_without_security(const eprosima::fastdds::rtps::EntityId_t & reader_id, eprosima::fastdds::rtps::CacheChange_t & change, bool __formal) Line 232	C++


RTPSParticipantImpl::create_writer
1001    {
1002        std::lock_guard<shared_mutex> _(endpoints_list_mutex);     <=========
1003        m_allWriterList.push_back(SWriter);
1004
1005        if (!is_builtin)
1006        {
1007            m_userWriterList.push_back(SWriter);
1008        }
1009    }

StatelessReader::process_data_msg
567     {
568         assert(change);
569     
570         std::unique_lock<RecursiveTimedMutex> lock(mp_mutex);
571     
572         if (acceptMsgFrom(change->writerGUID, change->kind))
573         {
     
...     
...     
659                 // Perform reception of cache change
660                 if (!change_received(change_to_add))     <=========
661                 {
662                     EPROSIMA_LOG_INFO(RTPS_MSG_IN,
663                             IDSTRING "MessageReceiver not add change " << change_to_add->sequenceNumber);
664                     change_to_add->serializedPayload.payload_owner->release_payload(change_to_add->serializedPayload);
665                     change_pool_->release_cache(change_to_add);
666                     return false;
667                 }
668             }
669         }


Not Flagged	>	10756	0	Worker Thread	dds.udp.7410	msvcp140d.dll!00007ffd2ef42107
>	fastddsd-3.1.dll!eprosima::fastdds::rtps::StatelessReader::matched_writer_is_matched(const eprosima::fastdds::rtps::GUID_t & writer_guid) Line 320	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::EDP::pairing_writer_proxy_with_any_local_reader::__l2::<lambda>(eprosima::fastdds::rtps::BaseReader & r) Line 1236	C++
>	fastddsd-3.1.dll!eprosima::fastdds::rtps::RTPSParticipantImpl::forEachUserReader<bool <lambda>(eprosima::fastdds::rtps::BaseReader &)>(eprosima::fastdds::rtps::EDP::pairing_writer_proxy_with_any_local_reader::__l2::bool <lambda>(eprosima::fastdds::rtps::BaseReader &) f) Line 1018	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::EDP::pairing_writer_proxy_with_any_local_reader(const eprosima::fastdds::rtps::GUID_t & participant_guid, eprosima::fastdds::rtps::WriterProxyData * wdata) Line 1189	C++
 	fastddsd-3.1.dll!eprosima::fastdds::rtps::EDPBasePUBListener::add_writer_from_change::__l5::<lambda>(int request_ret_status, eprosima::fastdds::rtps::WriterProxyData * temp_writer_data) Line 125	C++

StatelessReader::matched_writer_is_matched
319    {
320        std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);     <=========
321        if (std::any_of(matched_writers_.begin(), matched_writers_.end(),
322                [writer_guid](const RemoteWriterInfo_t& item)
323                {
324                    return item.guid == writer_guid;
325                }))
326        {
327            return true;
328        }

RTPSParticipantImpl::forEachUserReader
989    {
990        // check if we are reentrying
991        shared_lock<shared_mutex> _(endpoints_list_mutex);     <=========
992
993        for (BaseReader* pr : m_userReaderList)
994        {
995            if (!f(*pr))
996            {
997                break;
998            }
999        }

XML configuration file

NA

Relevant log output

NA

Network traffic capture

NA

[cferreiragonz]

Hi @owillebo, thank you for using Fast DDS.

The approach you are using is something that should never be done. The on_new_cache_change_added method is intended solely for processing new CacheChanges received in incoming messages, not for creating new entities. Using it for entity creation can lead to unintended behavior.

We recommend reviewing our RTPS example for proper usage of this methods.
If you have further questions, feel free to ask!

[owillebo]

Thanks @cferreiragonz,

I reviewed the examples extensively. You are right that in these examples no entities are created in the on_new_cache_change_added.
I'm replacing a homebrew RTPS implementation with Fast-DDS (RTPS). The homebrew RTPS implementation allowed for creating entities upon incoming changes and this is used for a command-reply mechanism. With this command-reply mechanism participants have a reply and command reader. A reply writer is created upon receiving a command, hence the creation of a writer in on_new_cache_change_added.

Thanks again for making clear that this is not possible in Fast-DDS. I already have implemented a workaround.