DDS: Data Distribution Service

 What is DDS?

    DDS means Data Distribution Service and its a standard for machine-to-machine real-time systems communication that uses a publish-subscribe pattern to allow complex high dependable systems to handle their internal communication in a scalable and interoperable way.

    It works as a middleware that simplifies complex network programming. As we've discoursed on the topic about ROS 2 fundamentals,  a pub/sub architecture is a sort of re-reading of the classical producer-consumer problem, where we have n producers and m consumers.

    Each node can be consuming or producing a message, and we could have a lot of messages dealing with a lot of things running through the system in parallel.

    In order to keep things organized, every "parallel" messaging queue is called a topic.

    So a pub/sub system works as follows: 

• A publisher is a node that intends to produce a message. This publisher will publish the message in a topic.
• A consumer is a node whose job is to listen to any message of a specific topic and react to it appropriately. So a consumer will subscribe to a topic in order to be alerted by the pub/sub system that a message is available to be read.

    This architecture has a concept of a QoS (quality of service) which are some parameters to alter the expected behavior of this middleware up-from. For example, you could make it more reliable (but slower), because you can't afford to lose messages. 

 Eclipse Cyclone DDS

    The Eclipse Cyclone DDS is an implementation of the OMG specification that is a part of the Eclipse Foundation's IoT project and that is currently (Galactic version) the default middleware for ROS 2.

    The Cyclone DDS aims at full coverage of the specs and today already covers most of this. Some optional parts are still missing. With references to the individual OMG specifications:

• DCPS the base specification
• zero configuration discovery (if multicast works)
• publish/subscribe messaging
• configurable storage of data in subscribers
• many QoS settings - liveliness monitoring, deadlines, historical data, 
• ...
• coverage includes the Minimum, Ownership and (partially) Content profiles
• DDS Security - providing authentication, access control and encryption
• DDS C++ API
• DDS XTypes - the structural type system (some caveats here)
• DDSI-RTPS - the interoperable network protocol

Let's dig into it

    The tutorial bellow will allow us to better understand DDS. It is not required to run ROS 2, but it's good to understand how it works.

    First and foremost, we need to install the Cyclone DDS and then we can take a look at a code as an example of using this lib.

    1. Cloning the latest version

    

        To test this DDS implementation, lets first clone it and build it from source

git clone https://github.com/eclipse-cyclonedds/cyclonedds.git
cd cyclonedds

    2. Building and installing iceoryx.io

    

        Iceoryx.io is a lib to allow message communication using shared memory, instead of using UDP. First, we install the dependencies by issuing

sudo apt install gcc g++ cmake libacl1-dev libncurses5-dev pkg-config

        Then, we'll download the source code and compile / install it

git clone https://github.com/eclipse-iceoryx/iceoryx.git
cd iceoryx
cmake -Bbuild -Hiceoryx_meta -DCMAKE_PREFIX_PATH=$(PWD)/build/dependencies/

        Next, compile the source code and install it to the system

cmake --build build
sudo cmake --build build --target install

    3. Building Cyclone DDS from the source

    

        Finally, let's build the Cyclone DDS from the source

mkdir build
cd build
cmake -DCMAKE_INSTALL_PREFIX=/opt/cyclonedds -S..
cmake --build .
cmake --build . --target install

 Analyzing the Hello World example

    Let's first analyze a code that comes with the source code that implements a "hello world" for this DDS.  For this implementation, we'll have a publisher that will send a Hello World message once a subscriber shows up.

    We'll also have the subscriber to subscribe, consume and show this message.

    The first thing we have is to define a message format using the IDL language. This language is responsible for defining an interface. The "source code" for this definition should be compiled to C to be integrated into the final binary file

So, go to the folder examples/helloworld and take a look at the file HelloWorldData.idl

module HelloWorldData

{

  struct Msg

  {

    @key

    long userID;

    string message;

  };

};

To compile it, we issue the following command

/opt/cyclonedds/bin/idlc ./HelloWorldData.idl

The above command will generate two files: HelloWorldData.c, HelloWorldData.h implementing this structure.

Then, we can join those two files with the publisher.c code and create the hello world publisher

gcc -o publisher publisher.c HelloWorldData.c -I/opt/cyclonedds/include/ -I. -L/opt/cyclonedds/lib -lddsc

After that, we build the subscriber

gcc -o subscriber subscriber.c HelloWorldData.c -I/opt/cyclonedds/include/ -I. -L/opt/cyclonedds/lib -lddsc

Finally, we need to link the libs on /opt/cyclone/lib to /usr/lib and link those libraries.

sudo ln -s /opt/cyclonedds/lib/libddsc.so.0 /usr/lib/libddsc.so.0
ldconfig -v

Now, we can run those examples and see the results

./publisher 
=== [Publisher]  Waiting for a reader to be discovered ...

./subscriber 
=== [Subscriber] Waiting for a sample ...
=== [Subscriber] Received : Message (1, Hello World)

The Publisher Code

The publisher code is very straightforward. First, we create a participant, which is an actor in the DDS world

/* Create a Participant. */

participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);

if (participant < 0)

DDS_FATAL("dds_create_participant: %s\n", dds_strretcode(-participant));

Then, we create a topic to publish to

topic = dds_create_topic (

participant, &HelloWorldData_Msg_desc, "HelloWorldData_Msg", NULL, NULL);

if (topic < 0)

DDS_FATAL("dds_create_topic: %s\n", dds_strretcode(-topic));

Next, we create a writer, which is the method to create a publisher implicitly.

writer = dds_create_writer (participant, topic, NULL, NULL);

if (writer < 0)

DDS_FATAL("dds_create_writer: %s\n", dds_strretcode(-writer));

After that, we call the dds_set_status_mask( ) function which is responsible for enabling actions based on status mask. In this example, the chosen mask is DDS_PUBLICATION_MATCHED _STATUS which means that the writer has found a reader that matches the topic and has a compatible QoS

rc = dds_set_status_mask(writer, DDS_PUBLICATION_MATCHED_STATUS);

if (rc != DDS_RETCODE_OK)

DDS_FATAL("dds_set_status_mask: %s\n", dds_strretcode(-rc));

Finally we put the publisher to sleep, waiting for a subscriber to subscribe to a topic. 

while(!(status & DDS_PUBLICATION_MATCHED_STATUS))

{

rc = dds_get_status_changes (writer, &status);

if (rc != DDS_RETCODE_OK)

DDS_FATAL("dds_get_status_changes: %s\n", dds_strretcode(-rc));

/* Polling sleep. */

dds_sleepfor (DDS_MSECS (20));

}

When the subscriber comes to live, we execute dds_write ( ) to send the message.

  msg.userID = 1;

msg.message = "Hello World";

printf ("=== [Publisher] Writing : ");

printf ("Message (%"PRId32", %s)\n", msg.userID, msg.message);

fflush (stdout);

rc = dds_write (writer, &msg);

if (rc != DDS_RETCODE_OK)

DDS_FATAL("dds_write: %s\n", dds_strretcode(-rc));

The Subscriber Code

    For the subscriber code, the code is also straightforward and very similar to the publisher.

    First, we also need to create a participant, which in this case is the subscriber.

participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);

if (participant < 0)

DDS_FATAL("dds_create_participant: %s\n", dds_strretcode(-participant));

  

   Then, we create a topic to subscribe to

topic = dds_create_topic (

participant, &HelloWorldData_Msg_desc, "HelloWorldData_Msg", NULL, NULL);

if (topic < 0)

DDS_FATAL("dds_create_topic: %s\n", dds_strretcode(-topic));

    Next, we set QoS and config it to be a reliable reader. A reliable reader will check data for transmission error, assuring that it is well transmitted (much like a TCP connection would do).

  

qos = dds_create_qos ();

dds_qset_reliability (qos, DDS_RELIABILITY_RELIABLE, DDS_SECS (10));

  

  Finally, we create a reader which implicitly creates a subscriber to a topic.

reader = dds_create_reader (participant, topic, qos, NULL);

if (reader < 0)

DDS_FATAL("dds_create_reader: %s\n", dds_strretcode(-reader));

dds_delete_qos(qos);

    Then, we read the incoming message using the dds_reader ( ) function.

  rc = dds_read (reader, samples, infos, MAX_SAMPLES, MAX_SAMPLES);

if (rc < 0)

DDS_FATAL("dds_read: %s\n", dds_strretcode(-rc));

Final Thoughts

    DDS is a very nice pub/sub specification to allow decoupling modules in a complex system. 

    We chose the current official implementation of DDS on ROS  2 as an example, but you can find other examples to study. Most of them have a pretty similar structure.

    I hope you've learned something on this post. Let's move on.... there's yet a lot to learn!