dds

What?

DDS（Data Distribution Service）数据分发服务是用于分布式软件应用程序通信的以数据为中心的通信协议。 它描述了通信应用程序编程接口（API）和通信语义，使数据提供者和数据使用者之间能够进行通信。

由于它是一个以数据为中心的发布订阅（DCPS: Data-Centric Publish Subscribe）模型，因此在其实现中定义了三个关键的应用程序实体：

发布实体(publication entities)，用于定义信息生成对象及其属性；
	
订阅实体(subscription entities)，用于定义消耗信息的对象及其属性；
	
配置实体(configuration entities)，这些实体定义作为主题传输的信息的类型，并利用其服务质量（QoS: Quality of Service）属性创建发布者和订阅者，从而确保上述实体的正确性能。
Why?

如何构建一个可靠、健壮、实时性的通信系统是很麻烦的一件事。

如：在自动驾驶系统开发过程中，需要实时且可靠的进行各种传感器数据的传输。

如果基于传统的Linux/Unix IPC（Inter-Process Communication）机制进行封装，实现起来往往非常复杂，不便于进行产品开发和大型商用项目应用。

已有解决方案：

D-Bus系列（底层基于Socket）：FDBus、KDBus、Qt DBus、gDBus ...
	
Binder（底层基于Binder driver）：Android中应用非常广泛，单拷贝，性能极高
个人觉得D-Bus系列性能欠佳（对自动驾驶来说实时性相当重要），Binder性能很强，但是依赖内核驱动支持，需要移植Google的Binder驱动。

DDS是一个标准，可以有不同的实现，如RTI的Connext，eProsima的Fast DDS。

本文只讨论eProsima Fast DDS，其特点：

高性能（High performance）: performance benchmarks.
	
跨平台（Multi-Platform）: Windows, Linux, Mac OS, QNX, VxWorks, iOS, Android, Raspbian.
	
易用（Easy to use）: See a hello world video tutorial here.
	
开源（Free and Open Source）: Apache License 2.0
因此选择DDS作为数据分发的中间件。

How？

Fast DDS上手起来比较容易，墙裂建议直接阅读官方文档。

在此先上一个Fast DDS的架构图：




	
Application layer. The user application that makes use of the Fast DDS API for the implementation of communications in distributed systems.
	
	

	
Fast DDS layer. Robust implementation of the DDS communications middleware. It allows the deployment of one or more DDS domains in which DomainParticipants within the same domain exchange messages by publishing/subscribing under a domain topic.
	
	

	
RTPS layer. Implementation of the Real-Time Publish-Subscribe (RTPS) protocol for interoperability with DDS applications. This layer acts an abstraction layer of the transport layer.
	
	

	
Transport Layer. Fast DDS can be used over various transport protocols such as unreliable transport protocols (UDP), reliable transport protocols (TCP), or shared memory transport protocols (SHM).
	
DDS layer


	
QoS. The mechanism by which the behavior of each of the entities is defined.
	
	

	
Listener. The mechanism by which the entities are notified of the possible events that arise during the application’s execution.
	
Below are listed the DDS Entities together with their description and functionality. For a more detailed explanation of each entity, their QoS, and their listeners, please refer to DDS Layer section.


	
Domain. A positive integer which identifies the DDS domain. Each DomainParticipant will have an assigned DDS domain, so that DomainParticipants in the same domain can communicate, as well as isolate communications between DDS domains. This value must be given by the application developer when creating the DomainParticipants.
	
	

	
DomainParticipant. Object containing other DDS entities such as publishers, subscribers, topics and multitopics. It is the entity that allows the creation of the previous entities it contains, as well as the configuration of their behavior.
	
	

	
Publisher. The Publisher publishes data under a topic using a DataWriter, which reads the data from the transport. It is the entity that creates and configures the DataWriter entities it contains, and may contain one or more of them.
	
	

	
DataWriter. It is the entity in charge of publishing messages. The user must provide a Topic when creating this entity which will be the Topic under which the data will be published. Publication is done by writing the data-objects as a change in the DataWriterHistory.
	
	

	
DataWriterHistory. This is a list of changes to the data-objects. When the DataWriter proceeds to publish data under a specific Topic, it actually creates a change in this data. It is this change that is registered in the History. These changes are then sent to the DataReader that subscribes to that specific topic.
	
	

	
Subscriber. The Subscriber subscribes to a topic using a DataReader, which reads the data from the transport. It is the entity that creates and configures the DataReader entities it contains, and may contain one or more DataReader entities.
	
	

	
DataReader. It is the entity that subscribes to the topics for the reception of publications. The user must provide a subscription Topic when creating this entity. A DataReader receives the messages as changes in its HistoryDataReader.
	
	

	
DataReaderHistory. It contains the changes in the data-objects that the DataReader receives as a result of subscribing to a certain Topic.
	
	

	
Topic. Entity that binds Publishers’ DataWriters with Subscribers’ DataReaders.
	
RTPS layer

the RTPS protocol in Fast DDS allows the abstraction of DDS application entities from the transport layer. According to the graph shown above, the RTPS layer has four main Entities.


	
RTPSDomain. It is the extension of the DDS domain to the RTPS protocol.
	
	

	
RTPSParticipant. Entity containing other RTPS entities. It allows the configuration and creation of the entities it contains.
	
	

	
RTPSWriter. The source of the messages. It reads the changes written in the DataWriterHistory and transmits them to all the RTPSReaders to which it has previously matched.
	
	

	
RTPSReader. Receiving entity of the messages. It writes the changes reported by the RTPSWriter into the DataReaderHistory.
	
Transport layer

Fast DDS supports the implementation of applications over various transport protocols. Those are UDPv4, UDPv6, TCPv4, TCPv6 and Shared Memory Transport (SHM). By default, a DomainParticipant implements a UDPv4 and a SHM transport protocol. The configuration of all supported transport protocols is detailed in the Transport Layer section.

 

Reference：

eProsima Fast DDS Documentation (https://fast-dds.docs.eprosima.com/en/latest/fastdds/library_overview/library_overview.html)

DDS与openDDS
DDS

什么是DDS？

Data Distribution Service(DDS)，根据字面理解就是数据分发服务。这套服务，在分布式应用环境下，可以高效率地分发参与者（应用）间的数据信息。
订阅/发布体系结构

考虑到应用实现，肯定要定一套体系结构。DDS采用订阅/发布体系结构，以数据为中心，也就是通过订阅/发布这个结构来实现消息(数据)的交换。
P/S与RTPS

上面说的订阅/发布，从通信角度考虑。其实，就是P/S通信模式。P(Publish)是发布者，S(Subscribe)是订阅者，订阅者订阅发布者提供的某些服务后，P与S间就存在了通信关系。P发布了相应的消息后，订阅者便可以收到这些消息（数据）。

RTPS又是什么？

RTPS指Real Time Publish/Subscribe，它继承自P/S模型。简单来说，相比P/S，RTPS多出了QoS属性(Quality of Service)、将发布/订阅模块化等等优点。

因此，OMG将RTPS标准化为：DDS的实施互操作协议。
OMG

一个组织，全名为Object Management Group。DDS就是OMG这个组织规定的规范。
全局数据空间

DDS以数据为中心，也就是基于数据去考虑如何分发消息。因此在DDS中存在全局数据空间的概念。这个全局数据空间是虚拟（逻辑）上的，所有数据都在这个空间中，每个分布式结点可以向这个空间写数据，也可以从这个空间读数据。

这个空间可以总结为三点：All Data、Read and Write、Cache。



DCPS
OMG将RPTS重定义为DCPS，DDS采用DCPS通信机制。DCPS主要组成有：

Domain : 域。参与者只有在同一个域内，才可以通信。不同域间，不可通信。
DomainParticipant: 域参与者。一个域的入口点。
Topic：主题。只有使用相同主题(主题名)的P和S，才能够通信。
DataWriter：数据写入者，负责向Publisher写入消息。
Publisher：发布者，负责发布DataWriter写入的消息。
Subscriber：订阅者，负责接收已订阅的消息。
DataReader数据读取写，负责处理从Subscriber接收的消息。



DLRL与DCPS
DDS规范定义了两类Interfaces结构


DCPS:DDS核心，数据分布基础架构。




DLRL(A Data Local Reconstruction Layer):更高的抽象接口层，隐藏细节实现。其实，Data Local Reconstruction可以理解为将App与DDS直接的数据，根据应用环境重构为需要的格式。




OpenDDS

根据OpenDDS(目前最新版本是3.14)官网开发者手册中的介绍


OpenDDS is an open source implementation of the OMG Data Distribution Service (DDS) for

Real-Time Systems Specification v1.4 (OMG Document formal/2015-04-10) and the Realtime Publish-Subscribe Wire Protocol DDS Interoperability Wire Protocol Specification

(DDSI-RTPS) v2.3 (OMG Document formal/2019-04-03). OpenDDS also implements the

DDS Security Specification v1.1 (OMG Document formal/2018-04-01).


因此OpenDDS，是DDS规范的实现。当然还有其他一些DDS规范的实现。例如：RTI DDS、OpenSplice DDS。
OpenDDS框架：下图为OpenDDS官网给出的Framework图。可以看出，主要部分为Topic、Transport、Discovery。也就是，数据如何定义？如何传输信息？如何发现两端（P/S)？


OpenDDS Topic：根据IDL，定义Topic结构，利用OpenDDS提供的工具，根据IDL文件自动生成主题文件。例：https://github.com/adver1991/DDS-Example/blob/master/Messenger.idl
OpenDDS Transport: OpenDDS提供了多种传输协议，TCP、UDP、Multicast、Shared-Memory、RTPS_UDP。使用者可以通过配置文件，指定使用的传输协议。

OpenDDS Discovery
DDS应用在分布式环境中，订阅者和发布者两端，如何发现对方？OpenDDS提供了两种节点寻找方式:


DCPSInfoRepo（默认）：一个集中式的仓库，相当于发现中心，它需要运行在独立的进程中（例：P端一个进程，S端一个进程，还需要一个进程运行InfoRepo）。




RTPS：使用RTPS协议，点对点，不需要额外的进程运行其他服务。




使用OpenDDS

平台（Ubuntu 16.4)
下载解压安装报，根据官网提示操作即可。
https://opendds.org/quickstart/GettingStartedLinux.html
例子(参考自OpenDDS官网开发手册2.1.3节），使用RTPS发现方式，使用rtps_udp传输协议。这里只给出代码，代码中具体的含义请参考OpenDDS开发者手册。代码见：

https://github.com/adver1991/DDS-Example
关于例子的简单说明：例子中，包括订阅者、发布者、数据写入者、数据读取者（包含Listener）、Topic（IDL定义）部分。根据IDL文件，使用OpenDDS提供的工具，生成Topic相关的代码。订阅者与发布者，在同一个域内，使用相同的主题(主题名）实现数据的分发。

参考

OpenDDS官网
OpenDDS开发者手册

DDS系列
【DDS】DDS与OpenDDS

【DDS】DDS-RPC通信机制

【DDS】基于OpenDDS的DDS-RPC实现

【DDS】DDSI-RTPS规范

基于OpenDDS的DDS-RPC实现
DDS-RPC：

DDS指Data Distribution Service，是OMG(Object Management Group)提出的规范。DDS类似于总线的概念，用于数据交互。
RPC指Remote Procedure Calls，远过程调用。比方说，远程服务端提供一个了方法，客户端可以调用这个方法。

既然，DDS可以进行数据交互，那么理论上基于DDS便可以实现RPC机制。

关于DDS-RPC的详细概念原理参考下文：

DDS-RPC通信机制

DDS-RPC实现
OpenDDS没有实现RPC，但根据RPC over DDS给出的规范。可以利用OpenDDS实现 DDS-RPC。在实现上，主要包括以下几个部分：


IDL定义

根据DDS-RPC规范，定义OpenDDS可以识别的IDL文件(主题)。


主题文件生成

根据上一步，定义的IDL文件，利用OpenDDS提供的工具生成主体文件(.h/.cpp)


Publisher/Subscribe

服务端和客户端，相当于两个Participation。每个Participation需要包含一个P端，一个S端。


DataWriter/DataReader。

DateWriter负责写入请求数据，DataReader负责读取订阅的数据。


本地实现

当服务器端的DataReader接收到请求端的数据时，根据调用的方法，调用具体的本地实现。方法调用后，将返回值通过DataWriter告知客户端。

当客户端的DataReader接收到数据后，调用相应的本地实现，告知其返回结果。


代码源码GitHub地址如下：

https://github.com/adver1991/DDS-Example/tree/dds-rpc


IDL文件示例


#ifdef INTERFACE
module robot {

@DDSService
interface RobotControl 
{
  float setSpeed(float speed);
  float getSpeed();
};

}; //module robot
#endif // INTERFACE

#ifndef MY_BASIC
#define MY_BASIC

#include <rpc_types.idl>

module robot {

/***********************************************/
/*                   Request Types             */
/***********************************************/
@nested
struct RobotControl_setSpeed_In 
{
    float speed;
};

@nested
struct RobotControl_getSpeed_In 
{ 
  DDS::RPC::UnusedMember dummy; 
};

const long RobotControl_setSpeed_Hash  = 1;
const long RobotControl_getSpeed_Hash  = 2;

@nested
union RobotControl_Call switch(long) 
{
    default:
       DDS::RPC::UnknownOperation unknownOp;

    case RobotControl_setSpeed_Hash:
       RobotControl_setSpeed_In setSpeed;

    case RobotControl_getSpeed_Hash:
       RobotControl_getSpeed_In getSpeed;
};

@topic
struct RobotControl_Request 
{
  DDS::RPC::RequestHeader header;
  RobotControl_Call       data;
};

/***********************************************/
/*                   Reply Types               */
/***********************************************/
@nested
struct RobotControl_setSpeed_Out 
{
  float return_;
};

@nested
struct RobotControl_getSpeed_Out 
{ 
  float return_; 
};

@nested
union RobotControl_setSpeed_Result switch(DDS::RPC::RemoteExceptionCode_t)
{
  default:
    DDS::RPC::UnknownException unknownEx;

  case DDS::RPC::REMOTE_EX_OK:
    RobotControl_setSpeed_Out result;
};

@nested
union RobotControl_getSpeed_Result switch(DDS::RPC::RemoteExceptionCode_t)
{
  default:
    DDS::RPC::UnknownException unknownEx;
  
  case DDS::RPC::REMOTE_EX_OK:
    RobotControl_getSpeed_Out result;
};

@nested
union RobotControl_Return switch(long)
{
  default: 
    DDS::RPC::UnknownOperation unknownOp;

  case RobotControl_setSpeed_Hash:
    RobotControl_setSpeed_Result setSpeed;

  case RobotControl_getSpeed_Hash:
    RobotControl_getSpeed_Result getSpeed;
};

@topic
struct RobotControl_Reply
{
  DDS::RPC::ReplyHeader  header;
  RobotControl_Return    data;
};

}; // module robot
#endif /* MY_BASIC */

DDS系列
【DDS】DDS与OpenDDS

【DDS】DDS-RPC通信机制

【DDS】基于OpenDDS的DDS-RPC实现

【DDS】DDSI-RTPS规范