Custom Executor Tutorial¶

This tutorial provides a practical guide to implementing a custom executor in Safe DDS. It demonstrates how to create a custom execution model by implementing the ISpinnable interface and managing the execution of DDS entities and transport manually.

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 custom executor approach¶

Safe DDS provides a default executor that handles the execution of all entities and the transport. However, some applications may require specific execution models, such as prioritizing certain entities, handling execution in a specific order, or integrating with external event loops.

This behavior can be achieved by implementing a custom executor. The custom executor is responsible for “spinning” the entities and the transport, which means giving them CPU time to process their pending work.

All Safe DDS entities and the transport implement the ISpinnable interface, which provides the methods required to be managed by an executor:

spin(time): Process pending work until the specified time. Note that it might return earlier.
has_pending_work(): Check if there is any work to be done.
get_next_work_timepoint(): Get the time point when the next work is scheduled.

By collecting the ISpinnable interfaces of the entities and the transport, a custom executor can orchestrate their execution according to the application’s needs.

Note

The ISpinnable interface of an entity can be retrieved using the get_spinnable() method, while the ISpinnable interface of the transport is obtained from the DomainParticipantFactory via get_transport_spinnable().

Safe DDS Application¶

The application consists of a CustomExecutor class and a main function that sets up the DDS entities as described in the Getting Started tutorial and uses the custom executor.

Custom Executor Implementation¶

The CustomExecutor class implements the ISpinnable interface. It holds references to the ISpinnable interfaces of the entities and the transport.

The core of the executor is the spin method. In this example, the executor prioritizes the transport by spinning it first, followed by the entities in a specific order. Unlike a round-robin approach, if an entity has pending work after a spin, it is spun repeatedly until all work is completed. This strategy ensures that the transport processes incoming data before entities react to it. It is important to clarify that outgoing data is sent when the associated writer is spun, not when the transport is spun. To conclude the cycle, the transport is spun once more using any remaining time to await new events.

Note

The transport is handled as a separate ISpinnable, distinct from the list of entities. This separation allows the executor to block on the transport waiting for new messages or events, which is crucial for efficient resource usage.

/**
 * @brief Spin method custom implementation
 *
 * @param tm TimePoint until which to spin
 */
void spin(
        const execution::TimePoint& tm) noexcept override
{
    // Spin the transport first
    while (transport_spinnable_->has_pending_work())
    {
        transport_spinnable_->spin(execution::TIME_ZERO);
    }

    // Spin the entities in the stablished order
    for (auto spinnable : entities_spinnables_)
    {
        // Spin the entity
        while (spinnable->has_pending_work())
        {
            spinnable->spin(execution::TIME_ZERO);
        }
    }

    // Determine the maximum timepoint to block
    execution::TimePoint next_timepoint = tm;

    if (execution::TIME_ZERO != tm)
    {
        next_timepoint = execution::TimePoint::min(tm, get_next_work_timepoint());
    }

    // Block on transport
    transport_spinnable_->spin(next_timepoint);
}

The executor also needs to implement the remaining virtual methods of the ISpinnable interface to report pending work and the next work time point. These methods aggregate the status of the managed ISpinnable objects.

/**
 * @brief Check if there is pending work
 */
bool has_pending_work() const noexcept override
{
    bool has_pending_work = transport_spinnable_->has_pending_work();

    for (auto spinnable : entities_spinnables_)
    {
        has_pending_work = has_pending_work || spinnable->has_pending_work();
    }

    return has_pending_work;
}

/**
 * @brief Get the next work timepoint
 */
execution::TimePoint get_next_work_timepoint() const noexcept override
{
    execution::TimePoint next_work_timepoint = transport_spinnable_->get_next_work_timepoint();

    for (auto spinnable : entities_spinnables_)
    {
        next_work_timepoint = execution::TimePoint::min(
            next_work_timepoint, spinnable->get_next_work_timepoint());
    }

    return next_work_timepoint;
}

Main Application¶

The main application creates the DDS entities (Participant, Publisher, Subscriber, DataWriter, DataReader, Topic) as usual, following the steps in the Getting Started tutorial. Once the entities are created, their ISpinnable interfaces are collected into a vector. The order in which they are added to the vector determines the order in which they will be spun by the CustomExecutor.

// Create the list of entities to be managed by the CustomExecutor in custom priority order
std::vector<execution::ISpinnable*> entities_spinnables;
entities_spinnables.push_back(subscriber->get_spinnable());
entities_spinnables.push_back(datareader->get_spinnable());
entities_spinnables.push_back(publisher->get_spinnable());
entities_spinnables.push_back(datawriter->get_spinnable());
entities_spinnables.push_back(topic->get_spinnable());
entities_spinnables.push_back(participant->get_spinnable());

The transport spinnable is retrieved directly from the DomainParticipantFactory.

// Get the transport as spinnable
execution::ISpinnable* transport_spinnable = factory.get_transport_spinnable();

Then, the CustomExecutor is instantiated with the list of the ISpinnable entities and the ISpinnable transport.

// Create the CustomExecutor
CustomExecutor custom_executor{entities_spinnables, transport_spinnable};

Finally, the application enters the main loop, where it periodically publishes data and spins the custom executor. The spin call drives the execution of the entire middleware stack according to the custom policy defined in the executor. In this case, we simply spin the executor until the next publishing timepoint, although it might return earlier.

while (true)
{
    if (publish_timer.is_triggered_and_reset())
    {
        publish();
    }

    custom_executor.spin(publish_timer.next_trigger());
}

Running the application¶

Build and run the application in a terminal as explained in the Getting Started section.

./custom_executor

The application will start publishing and receiving messages using the custom executor. You should see output indicating that messages are being sent and received.

[DW: 0] Message: Hello DDS Custom Executor! 0
[DW: 0] Message: Hello DDS Custom Executor! 1
[DW: 0] Message: Hello DDS Custom Executor! 2
...