Does Fast-DDS-python support communication through shared memory?

[layemiao]

Does Fast-DDS-python support communication through shared memory? What I want to achieve is to send data using the C++ version of FastDDS through shared memory and receive it using the shared memory of Fast-DDS-python. Both use the same version of FastDDS, which is 2.13.1, and the versions of fastddsgen and Fast-DDS-python are 3.2.1 and v1.4.2 respectively. Can Fast-DDS-python support the above function implementation? If so, are there any conditions or restrictions? Could you provide a shared memory example of Fast-DDS-python?

[Javgilavi]

Hello @layemiao,

We moved your issue to discussion because it's a question and not an issue itself.

Regarding your question, Fast-DDS Python does fully support shared memory communication. The scenario you described should work without any problems as long as you’re using matching versions. The only restriction is that shared memory isn’t supported if you mix Fast-DDS v2 with Fast-DDS v3, but that doesn’t apply to your current setup.

[layemiao]

Thank you very much for your reply. Could you please provide an example of using Fast-DDS Python version 1.4.2（fastdd-v2.13.1） for communication via shared memory?

[Javgilavi]

Hi again @layemiao,

No example is needed because shared memory is used as default, if possible, without needing to do anything. In this picture you can see how the transport layer of Fast-DDS, and so Fast-DDS-python, works:

If you are using everything in the same host, it will use shared memory communication as default.

If what you are meaning is using data loan (Zero-Copy communication), that it's not support with Fast-DDS-python.

[layemiao]

Thank you for your reply.
However, it seems that direct use is not possible. Here is my code: I am sending data through shared memory using the C++ version of Fast DDS and receiving data using Fast DDS Python. But from the running results, there seems to be a problem with the communication between the two sides. Does Fast DDS Python need to use an XML file for some settings and cannot be used directly?
####Both use the same IDL.####
struct HelloWorld
{
unsigned long index;
string message;
};

C++ publisher

#include "HelloWorldPublisher.h"

#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.h>

#include

using namespace eprosima::fastdds::dds;
using namespace eprosima::fastdds::rtps;
using namespace eprosima::fastrtps::rtps;

HelloWorldPublisher::HelloWorldPublisher()
: participant_(nullptr)
, publisher_(nullptr)
, topic_(nullptr)
, writer_(nullptr)
, type_(new HelloWorldPubSubType())
{
}

bool HelloWorldPublisher::init()
{
hello_ = std::make_shared();
hello_->index(0);
hello_->message("HelloWorld");
Log::SetVerbosity(Log::Kind::Info);
//CREATE THE PARTICIPANT
DomainParticipantQos pqos;
pqos.wire_protocol().builtin.discovery_config.discoveryProtocol = DiscoveryProtocol_t::SIMPLE;
pqos.wire_protocol().builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true;
pqos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true;
pqos.wire_protocol().builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true;
pqos.wire_protocol().builtin.discovery_config.leaseDuration = eprosima::fastrtps::c_TimeInfinite;
pqos.name("Participant_pub");

// Explicit configuration of SharedMem transport
pqos.transport().use_builtin_transports = false;

auto shm_transport = std::make_shared<SharedMemTransportDescriptor>();
shm_transport->segment_size(2 * 1024 * 1024);
pqos.transport().user_transports.push_back(shm_transport);

participant_ = DomainParticipantFactory::get_instance()->create_participant(0, pqos);

if (participant_ == nullptr)
{
    return false;
}

//REGISTER THE TYPE
type_.register_type(participant_);

//CREATE THE PUBLISHER
publisher_ = participant_->create_publisher(PUBLISHER_QOS_DEFAULT);

if (publisher_ == nullptr)
{
    return false;
}

//CREATE THE TOPIC
topic_ = participant_->create_topic("HelloWorldSharedMemTopic", "HelloWorld", TOPIC_QOS_DEFAULT);

if (topic_ == nullptr)
{
    return false;
}

//CREATE THE DATAWRITER
DataWriterQos wqos;
wqos.history().kind = KEEP_LAST_HISTORY_QOS;
wqos.history().depth = 30;
wqos.resource_limits().max_samples = 50;
wqos.resource_limits().allocated_samples = 20;
wqos.reliable_writer_qos().times.heartbeatPeriod.seconds = 2;
wqos.reliable_writer_qos().times.heartbeatPeriod.nanosec = 200 * 1000 * 1000;
wqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
wqos.publish_mode().kind = ASYNCHRONOUS_PUBLISH_MODE;

writer_ = publisher_->create_datawriter(topic_, wqos, &listener_);

if (writer_ == nullptr)
{
    return false;
}

return true;

}

HelloWorldPublisher::~HelloWorldPublisher()
{
if (writer_ != nullptr)
{
publisher_->delete_datawriter(writer_);
}
if (publisher_ != nullptr)
{
participant_->delete_publisher(publisher_);
}
if (topic_ != nullptr)
{
participant_->delete_topic(topic_);
}
DomainParticipantFactory::get_instance()->delete_participant(participant_);
}

void HelloWorldPublisher::PubListener::on_publication_matched(
eprosima::fastdds::dds::DataWriter*,
const eprosima::fastdds::dds::PublicationMatchedStatus& info)
{
if (info.current_count_change == 1)
{
matched_ = info.total_count;
first_connected_ = true;
std::cout << "Publisher matched." << std::endl;
}
else if (info.current_count_change == -1)
{
matched_ = info.total_count;
std::cout << "Publisher unmatched." << std::endl;
}
else
{
std::cout << info.current_count_change
<< " is not a valid value for PublicationMatchedStatus current count change" << std::endl;
}
}

void HelloWorldPublisher::runThread(
uint32_t samples,
uint32_t sleep)
{
if (samples == 0)
{
while (!stop)
{
if (publish(false))
{
std::cout << "Message: " << hello_->message() << " with index: " << hello_->index() << " SENT" <<
std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(sleep));
}
}
else
{
for (uint32_t i = 0; i < samples; ++i)
{
if (!publish())
{
--i;
}
else
{
std::cout << "Message: " << hello_->message() << " with index: " << hello_->index() << " SENT" <<
std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(sleep));
}
}
}

void HelloWorldPublisher::run(
uint32_t samples,
uint32_t sleep)
{
stop = false;
std::thread thread(&HelloWorldPublisher::runThread, this, samples, sleep);
if (samples == 0)
{
std::cout << "Publisher running. Please press enter to stop the Publisher at any time." << std::endl;
std::cin.ignore();
stop = true;
}
else
{
std::cout << "Publisher running " << samples << " samples." << std::endl;
}
thread.join();
}

bool HelloWorldPublisher::publish(
bool waitForListener)
{
if (listener_.first_connected_ || !waitForListener || listener_.matched_ > 0)
{
hello_->index(hello_->index() + 1);
size_t data_size = hello_->data().size();
std::string s = "BigData" + std::to_string(hello_->index() % 10);
strcpy(&hello_->data()[data_size - s.length() - 1], s.c_str());

    writer_->write(hello_.get());

    return true;
}
return false;

}

########fastdds-python Reader#########
class ReaderListener(fastdds.DataReaderListener):
def init(self):
super().init()

def on_data_available(self, reader):
    info = fastdds.SampleInfo()
    data = HelloWorld.HelloWorld()
    reader.take_next_sample(data, info)

    print("Received {message} : {index}".format(message=data.message(), index=data.index()))

def on_subscription_matched(self, datareader, info) :
    if (0 < info.current_count_change) :
        print ("Subscriber matched publisher {}".format(info.last_publication_handle))
    else :
        print ("Subscriber unmatched publisher {}".format(info.last_publication_handle))

class Reader():
def init(self, domain):
factory = fastdds.DomainParticipantFactory.get_instance()
self.participant_qos = fastdds.DomainParticipantQos()
factory.get_default_participant_qos(self.participant_qos)
self.participant = factory.create_participant(domain, self.participant_qos)

self.topic_data_type = HelloWorld.HelloWorldPubSubType()
self.topic_data_type.setName("HelloWorld")
self.type_support = fastdds.TypeSupport(self.topic_data_type)
self.participant.register_type(self.type_support)

self.topic_qos = fastdds.TopicQos()
self.participant.get_default_topic_qos(self.topic_qos)
self.topic = self.participant.create_topic("HelloWorldSharedMemTopic", self.topic_data_type.getName(), self.topic_qos)

self.subscriber_qos = fastdds.SubscriberQos()
self.participant.get_default_subscriber_qos(self.subscriber_qos)
self.subscriber = self.participant.create_subscriber(self.subscriber_qos)

self.listener = ReaderListener()
self.reader_qos = fastdds.DataReaderQos()
self.subscriber.get_default_datareader_qos(self.reader_qos)
self.reader = self.subscriber.create_datareader(self.topic, self.reader_qos, self.listener)

def del(self):
factory = fastdds.DomainParticipantFactory.get_instance()
self.participant.delete_contained_entities()
factory.delete_participant(self.participant)

def run(self):
try:
input('Press any key to stop')
except:
pass

Using the C++ version of FastDDS alone can achieve sending and receiving based on shared memory, and using FastDDS-Python alone can also achieve sending and subscribing. However, when sending with the C++ version of FastDDS and subscribing with FastDDS-Python, communication cannot be established. The running result is as follows.
Looking forward to receiving your reply

[Javgilavi]

Hi again @layemiao,

It seems the issue comes from forcing shared memory in your C++ publisher but not applying the same configuration in your Python subscriber. You’re doing in C++:

// Explicit configuration of SharedMem transport
pqos.transport().use_builtin_transports = false;

auto shm_transport = std::make_shared<SharedMemTransportDescriptor>();
shm_transport->segment_size(2 * 1024 * 1024);
pqos.transport().user_transports.push_back(shm_transport);

In your Python subscriber, you must configure the QoS similarly, disabling built-in transports and pushing the shared-memory transport. Please check TransportConfigQos for Fast-DDS Python API and configure use_builtin_transports = false and user_transports.push_back(shm_transport); as you did in C++.

Additionally, I highly recommend configuring the Participant QoS via an XML file and applying it to both the C++ Publisher and Python Subscriber. This ensures they use an identical transport configuration.

Hope that helps! Let us know if it resolves your issue.

[layemiao]

Hello, I made the following modifications to the Python code as per your suggestion, but the program reported an error when running. Does FastDDS not support setting shared memory communication directly in the program, or did I set it up incorrectly?

the python code

class Reader():
def init(self, domain):
factory = fastdds.DomainParticipantFactory.get_instance()

#配置domainparticipantQOS
self.participant_qos = fastdds.DomainParticipantQos()
factory.get_default_participant_qos(self.participant_qos)

#配置TransportConfigQos
transport_qos = fastdds.TransportConfigQos()
transport_qos.use_builtin_transports = False     #禁用内置传输
shm_transport = fastdds.SharedMemTransportDescriptor()
shm_transport.segment_size = 2*1024*1024
transport_qos.user_transports.push_back(shm_transport)  #添加shm传输

#将TransportConfigQos应用到dominparticipantQOS
self.participant_qos.transport().set_qos(transport_qos)

#创建participant
self.participant = factory.create_participant(domain, self.participant_qos)

#注册数据类型
self.topic_data_type = HelloWorld.HelloWorldPubSubType()
#self.topic_data_type.set_name("HelloWorldDataType")
self.topic_data_type.setName("HelloWorldDataType")
self.type_support = fastdds.TypeSupport(self.topic_data_type)
self.participant.register_type(self.type_support)

#创建topic
self.topic_qos = fastdds.TopicQos()
self.participant.get_default_topic_qos(self.topic_qos)
# self.topic = self.participant.create_topic("myTopic", self.topic_data_type.get_name(), self.topic_qos)
self.topic = self.participant.create_topic("myTopic", self.topic_data_type.getName(), self.topic_qos)

#创建subscriber
self.subscriber_qos = fastdds.SubscriberQos()
self.participant.get_default_subscriber_qos(self.subscriber_qos)
self.subscriber = self.participant.create_subscriber(self.subscriber_qos)

self.listener = ReaderListener()
self.reader_qos = fastdds.DataReaderQos()
self.subscriber.get_default_datareader_qos(self.reader_qos)
self.reader = self.subscriber.create_datareader(self.topic, self.reader_qos, self.listener)

def del(self):
factory = fastdds.DomainParticipantFactory.get_instance()
self.participant.delete_contained_entities()
factory.delete_participant(self.participant)

def run(self):
try:
input('Press any key to stop')
except:
pass

#####the error ####
ImportError: cannot import name 'SharedMemTransportDescriptor' from 'fastdds'
Thank you very much for your reply.

[Javgilavi]

Hi @layemiao,

Thanks for reaching out!

The issue you're encountering stems from a limitation in the fastdds Python API, which doesn't allow full configuration of the transport QoS settings directly in code. Instead, you'll need to modify these settings via an XML configuration file.

For more detailed information and examples on creating and configuring your XML file, please refer to the FastDDS XML Profiles documentation.

If you have any further questions or need additional assistance, please don't hesitate to reply.

[layemiao]

Thank you very much for your kind help. I tried using the XML file for configuration, and now the C++ version of Fast DDS and Fast DDS-Python can communicate normally. Thank you.

[Javgilavi]

Glad to hear that, @layemiao!

Since the issue has been resolved, I’ll be closing this discussion.

Feel free to reopen it if needed.