Decoupled Execution: Working with Multiple Threads Tutorial¶

This tutorial provides a practical guide to building a basic Safe DDS application that uses multiple threads to achieve a decoupled execution model. It demonstrates how to separate the control loop from data publishing, allowing data samples to be prepared and queued for transmission independently of the main execution cycle.

Note

It is recommended to first read the Getting Started section to gain a better understanding of the basic concepts and terminology used in this section.

Understanding the decoupled execution approach¶

Safe DDS is designed to operate in a single-threaded manner by default. However, some applications may benefit from preparing and queuing data samples in separate threads, decoupling data production from the middleware control loop.

This behavior can be safely implemented using the ConcurrentAccessDomainParticipantFactory provided in this example. The factory ensures that internal DDS entities are protected against concurrent access, allowing multiple threads to interact with them safely.

Note

While this approach enables safe concurrent access from multiple threads, it does so by introducing internal blocking mechanisms to synchronize operations. These mechanisms may slightly affect the deterministic behavior of the application’s timing and should be considered in scenarios that require strict real-time determinism.

This design provides a controlled and safe way to work with multiple threads while preserving the familiar single-threaded programming model of Safe DDS.

Safe DDS application¶

The main difference between a typical Safe DDS application and this decoupled execution approach is that the DomainParticipant is created using the ConcurrentAccessDomainParticipantFactory.

ConcurrentAccessDomainParticipantFactory factory;

After instantiating the factory, the rest of the application is built as usual.

DomainId domain_id = 0;
DomainParticipant* participant = factory.create_participant(domain_id, participant_qos, nullptr, NONE_STATUS_MASK);
CHECK_ENTITY_CREATION(participant);

// Type Support
StringTypeSupport type_support;
type_support.register_type(*participant, type_name);

// Entities
Topic* topic = participant->create_topic(topic_name, type_name, TopicQos{}, nullptr, NONE_STATUS_MASK);
CHECK_ENTITY_CREATION(topic);

Subscriber* subscriber = participant->create_subscriber(SubscriberQos{}, nullptr, NONE_STATUS_MASK);
CHECK_ENTITY_CREATION(subscriber);

DataReaderQos datareader_qos;
datareader_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS;
datareader_qos.history().depth = 100;
DataReaderCallback datareader_listener;
DataReader* dds_datareader =
        subscriber->create_datareader(*topic, datareader_qos, &datareader_listener, DATA_AVAILABLE_STATUS);
CHECK_ENTITY_CREATION(dds_datareader);

Publisher* publisher = participant->create_publisher(PublisherQos{}, nullptr, NONE_STATUS_MASK);
CHECK_ENTITY_CREATION(publisher);

DataWriterQos datawriter_qos;
datawriter_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS;
datawriter_qos.history().depth = 100;
DataWriter* dds_datawriter = publisher->create_datawriter(*topic, datawriter_qos, nullptr, NONE_STATUS_MASK);
CHECK_ENTITY_CREATION(dds_datawriter);

Any DataWriter that will be used from outside the control loop must be protected. To do so, request from the ConcurrentAccessDomainParticipantFactory a typed ConcurrentAccessDataWriter given the original DataWriter created by the participant.

ConcurrentAccessDataWriter<StringTypeSupport>* typed_writer =
        factory.get_concurrent_accessible_typed_datawriter<StringTypeSupport>(
    *dds_datawriter);

The resulting ConcurrentAccessDataWriter can then be safely used from a separate execution context to publish data. In this example, several independent publisher routines are created to publish at different rates.

PublisherThread<StringTypeSupport> publisher_thread_1(*typed_writer,
        execution::TimePeriod::from_ms(1000));
PublisherThread<StringTypeSupport> publisher_thread_2(*typed_writer,
        execution::TimePeriod::from_ms(500));
PublisherThread<StringTypeSupport> publisher_thread_3(*typed_writer,
        execution::TimePeriod::from_ms(100));

The publisher class wraps a routine that periodically calls write() on the ConcurrentAccessDataWriter to publish a new sample containing a string with the corresponding identifier.

template<typename T>
class PublisherThread
{
public:

    /**
     * @brief Constructor.
     *
     * @param writer: ConcurrentAccessDataWriter to use for publishing.
     * @param rate: Rate at which to publish data.
     */
    PublisherThread(
            ConcurrentAccessDataWriter<T>& writer,
            execution::TimePeriod rate) noexcept
        : writer_(writer)
        , timer_(rate)
    {
        pthread_create(&thread_, nullptr, &PublisherThread::run, this);
    }

private:

    ConcurrentAccessDataWriter<T>& writer_; //!< Thread-safe TypedDataWriter
    execution::Timer timer_;                //!< Timer to trigger publishing
    pthread_t thread_;                      //!< Thread ID

    /**
     * @brief Thread function.
     *
     * @param arg: Argument passed to the thread.
     *
     * @return nullptr.
     */
    static void* run(
            void* arg)
    {
        PublisherThread<T>* publisher_thread = static_cast<PublisherThread<T>*>(arg);

        const pthread_t tid = pthread_self();
        typename TypedDataWriter<T>::Data data;

        while (true)
        {
            if (publisher_thread->timer_.is_triggered_and_reset())
            {

                data.data = "Hello from publisher " + std::to_string(tid);

                if (dds::ReturnCode::OK != publisher_thread->writer_.write(data, HANDLE_NIL))
                {
                    std::cerr << "Error writing data" << std::endl;
                }
            }

            // Sleep until the next timer trigger
            execution::TimePoint next_timepoint = publisher_thread->timer_.next_trigger();
            execution::TimePoint now = get_platform().get_current_timepoint();

            if (next_timepoint > now)
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(next_timepoint.to_ms() - now.to_ms()));
            }
        }

        return nullptr;
    }

};

Finally, the factory is used to create a default executor, which spins the entities as usual.

execution::ISpinnable* executor = factory.create_default_executor();

while (true)
{
    while (executor->has_pending_work())
    {
        executor->spin(execution::TIME_ZERO);
    }

    execution::TimePoint next_work_timepoint = executor->get_next_work_timepoint();

    executor->spin(next_work_timepoint);
}

Running the application¶

Build and run the application in two terminals like any other Safe DDS application, as explained in the Getting Started section, and observe the output.

There are messages from three publishers on each participant, publishing at the configured rates.

[DW: 16777216] Received: Hello from publisher 131236318897728
[DW: 16777216] Received: Hello from publisher 131236310505024
[DW: 16777216] Received: Hello from publisher 131236302112320
[DW: 0] Received: Hello from publisher 124552277657152
[DW: 16777216] Received: Hello from publisher 131236302112320
[DW: 0] Received: Hello from publisher 124552294442560
[DW: 0] Received: Hello from publisher 124552286049856
[DW: 0] Received: Hello from publisher 124552277657152
[DW: 16777216] Received: Hello from publisher 131236302112320
[DW: 0] Received: Hello from publisher 124552277657152
[DW: 16777216] Received: Hello from publisher 131236302112320
[DW: 0] Received: Hello from publisher 124552277657152