RTPS协议之Behavior Module

交互要求
- 基本要求
- RTPS Writer 行为
- RTPS Reader行为
RTPS协议的实现
与Reader匹配的Writer的行为
涉及到的类型
RTPS Writer实现
- RTPS Writer
- RTPS StatelessWriter
- RTPS ReaderLocator
- RTPS StatefulWriter
- RTPS ReaderProxy
- RTPS ChangeForReader
RTPS StatelessWriter Behavior
- Best-Effort StatelessWriter Behavior
- - Reliable StatelessWriter Behavior
RTPS StatefulWriter Behavior

主要描述rtps实体的动态行为，主要记录rtps writer和rtps reader之间的有效的序列消息交换，和这些消息交换的时间限制。RTPS Writer和RTPS Reader的时序交互如下：
RTPS Endpoints交互
说明：

DDS DataWriter使用writer发送数据
DataWriter调用RTPS Writer的new_change创建一个新的CacheChange，每个CacheChange使用SequenceNumber唯一标识；
new_change返回；
DDS的DataWriter操作使用add_change操作将CacheChange添加到RTPS的Writer的HistoryCache中；
add_change操作返回
writer操作返回，用户完成写数据
RTPS writer使用Data Submessage发送CacheChange的内容给RTPS reader，并通过同时发送一个Heartbeat的Submessage的ack请求；
RTPS reader接收到Data消息，通过add_change操作将CacheChange添加到RTPS reader的HistoryCache中；
add_change返回，此时CacheChange对于DDS DataReader和DDS user是有效且可见的，这依赖于RTPS reader的reliabilityLevel属性；
a. 对于RELIABLE 的DDS DataReader，只有当之前的RTPS Reader’s HistoryCache的changes可见时，当前的changes可见；
b. 对于BEST_EFFORT DDS DataReader，只有没有后面的changes是可见的时候即可（i.e., if there are no changes in the RTPS Receiver’s HistoryCache with a higher sequence number）
DDS user收到通知（通过listener或者WaitSet）并通过DataReader的take函数读取数据
DDS 的DataReader通过HistoryCache的get_change函数读取change；
get_change操作返回CacheChange给DataReader；
take操作返回数据给DDS user；
RTPS Reader发送AckNack消息表示CacheChange已存放到Reader的HistoryCache中，AckNack消息中包含RTPS reader的GUID，和change的SequenceNumber。这些操作和通知DDS user或take操作是独立的（即并行的）；
StatefulWriter记录RTPS Reader已经收到CacheChange并使用acked_changes_set操作将其加入到由ReaderProxy维护的acked_changes的set中。
DDS user在DataReader上调用return_loan操作表明不再使用前面通过take操作获取到的数据；
DataReader使用remove_change操作从HistoryCache中删除数据；
remove_change操作返回；
return_loan操作返回；
DataWriter通过is_acked_by_all查看和该StatefulWriter匹配的哪些RTPS reader的CacheChanges接收到了；
DataWriter通过remove_change从RTPS Writer的HistoryCache中删除指定seq_num的change；这一步操作也需要考虑到DDS Qos如DURABILITY
remove_change返回

交互要求

基本要求

所有的通信必须使用RTPS Messages
必须实现RTPS Message Receiver

RTPS Writer 行为

writers发送数据不能out-of-order，即必须要按添加到HistoryCache中的数据发送
如果reader要求，Writers必须包含 in-line Qos
Writers只有在reliable下必须周期性发送HEARTBEAT Messages。
- Writer必须通过周期性的发送HEARTBEAT message通知每个匹配的reliable readers存在有效的data sample，这个HEARTBEAT message中包含有效sample的Sequence number；如果没有有效的samples，就没有HEARTBEAT Message需要发送；
- 对于严格使用reliable通信的情况，Writer必须持续性给Readers发送HEARTBEAT Messages，直到Reader要么确认收到了所有有效的samples，要么消失
- 其他情况下，HEARTBEAT消息的发送可以是特定实现的，也可以是有限的
在reliable模式下（且只在该模式下），Writer必须对negative acknowledgment作出回应。当收到ACKNACK Message，表明Reader丢失了一些data sample，Writer必须要么发送丢失的data sample，发送一个GAP message当sample是无关的时候；要么当sample不再有效时发送一个HEARTBEAT message；Writer可以立即回应，或者选择未来的某个时间点发送；可以合并多个相关的response一起发送；
属于一个Group的Writer会发送HEARTBEAT or GAP Submessages给和他匹配的Readers，即使是Reader接收到了Writer的所有的samples。This is necessary for the Subscriber to detect the
group sequence numbers that are not available in that Writer. The exception to this rule is when the Writer has
sent DATA or DATA_FRAG Submessages that contain the same information.

RTPS Reader行为

best-effort Reader只接收数据，不会发送任何数据；所以下面描述的要求仅限于reliable Reader。

Readers在收到一个没有设置最终flag的HEARTBEAT消息时必须作出响应。一旦收到一个未设置最终flag的HEARTBEAT消息时，Reader必须回应一个ACKNACK Message。这个ACKNACK Message消息可以用于回应收到了所有的data sample，或者一些data sample丢失了。响应可能会延迟，以避免消息风暴（在RTPS协议中，当一个节点收到一个HEARTBEAT消息时，它并不一定立即作出响应。相反，它可能会延迟一段时间后才做出响应，以防止在短时间内收到大量的HEARTBEAT消息，导致网络拥塞或过载。通过延迟响应，系统可以更有效地管理通信流量，确保通信的稳定性和可靠性）。
Readers在收到一个HEARTBEAT消息后发现data sample丢失了的时候必须响应一个ACKNACK Message。这一要求只有在Reader的缓存可以容纳这些丢失的data sample时才适用，并且与HEARTBEAT消息中的最终标志的设置无关。换句话说，在RTPS协议中，如果接收者的缓存能够处理缺失的数据样本，并且即使在HEARTBEAT消息中未设置最终标志，接收者仍然需要按照先前提到的规则做出响应。这强调了接收者在处理数据丢失时的一致性和可靠性。
- 这个消息可以延迟，以避免消息风暴
- 当一个liveness HEARTBEAT同时设置了liveness标志和final标志时，表明这是一个仅包含liveness信息的消息，此时不需要进行响应。换句话说，在RTPS协议中，如果一个HEARTBEAT消息被设置了liveness标志和flag标志，那么接收者不需要做出响应，因为这个消息只是用来表明发送者的存活状态，而不需要接收者采取任何措施。
一旦一个Reader使用ACKNACK消息积极地确认接收了一个数据样本，那么它就不能在之后的时间点再次使用负面确认（NAK）消息来否定地确认相同的数据样本。
- 当一个Writer从所有Readers那里都收到了积极的确认消息（positive acknowledgement），表示所有Readers都成功接收了相应的数据样本后，Writer可以收回与该数据样本相关的任何资源。这意味着Readers可以释放内存或其他资源，因为数据已经成功传输给了所有的Readers。
- 如果一个Writer之前收到了积极的确认消息，表示某个数据样本已经成功接收，然后在之后又收到了一个负面的确认消息（negative acknowledgement），表示该样本未成功接收，但Readers仍能够处理这个请求，那么Readers应该重新发送这个数据样本。这样做是为了确保数据的完整性和可靠性，即使在之前已经收到积极确认的情况下，如果有必要，仍然应该重新发送未成功接收的数据样本。——？TODO 是否矛盾？
Readers只能发送ACKNACK message用于回应HEARTBEAT Message。在稳定状态下，ACKNACK消息只能作为对来自Writer的HEARTBEAT消息的响应而发送。当Reader首次发现写入者时，可以发送ACKNACK消息作为一种优化。Reader不需要对这些预先发送的ACKNACK消息做出响应。

RTPS协议的实现

Stateless实现：被优化用于可伸缩性。它在远端entities上几乎不保存任何状态，因此在大型系统中具有非常好的可伸缩性。这涉及到一种权衡，因为改进的可伸缩性和减少的内存使用可能需要额外的带宽使用。Stateless的实现非常适合通过组播进行best-effort通信。
Stateful实现：在远端entities上维护完整状态。这种方法最小化了带宽使用，但需要更多的内存，并可能导致可伸缩性降低。与Stateless实现相比，它可以保证严格可靠的通信，并能够在写入者端应用基于QoS或基于内容的过滤。
实际的实现不需要完全遵守上面的规则，可以根据需要结合以上特点。Stateless Reference Implementation在远程实体上保持了最少的信息和状态。因此，它无法在写入端执行基于时间的过滤，因为这需要跟踪每个远程读取器及其属性。它也无法在读取器端丢弃无序样本，因为这需要跟踪从每个远程写入器接收到的最大序列号。一些实现可能模仿 Stateless Reference Implementation，但选择存储足够的额外状态以避免上述一些限制。所需的额外信息可以永久性地存储，这种情况下，实现接近于状态参考实现，或者可以慢慢老化并在需要时保留，以尽可能地模拟保持状态时的行为。

与Reader匹配的Writer的行为

与Reader匹配的RTPS Writer的行为，取决于RTPS Writer和RTPS Reader的reliabilityLevel属性（reliable和best-effort）的设置。
Writer和Reader允许的匹配：

RTPS Writer的reliabilityLevel属性设置为RELIABLE
RTPS Writer和RTPS Reader的reliabilityLevel属性都设置为BEST_EFFORT

这是因为DDS标准要求，BEST_EFFORT DDS DataWriter只能BEST_EFFORT DDS DataReader，RELIABLE DDS DataWriter可以匹配RELIABLE 和 BEST_EFFORT DDS DataReader
以上说明只限定于匹配，对于Stateful和Stateless来说，它们是可以通信的。

涉及到的类型

Attribute type	Purpose
Duration_t	时间
ChangeForReaderStatusKind	ChangeForReader.的状态，有以下值：UNSENT, UNACKNOWLEDGED, REQUESTED, ACKNOWLEDGED, UNDERWAY
ChangeFromWriterStatusKind	ChangeFromWriter.的状态值，有以下：LOST, MISSING, RECEIVED, UNKNOWN
InstanceHandle_t
ParticipantMessageData

RTPS Writer实现

RTPS Writer和RTPS Reader是对RTPS Endpoint的特化实现，StatelessWriter和StatefulWriter是RTPS Reader的特化，它们的区别在于处理匹配的Reader端点的信息时，它们的方法有所不同
TODO P70

RTPS Writer

RTPS Writer属性

attribute	type	meaning	relation to DDS
pushMode	bool
heartbeatPeriod	Duration_t
nackResponseDelay	Duration_t
nackSuppression Duration	Duration_t
lastChangeSequenceNumber	SequenceNumber_t
writer_cache	HistoryCache
dataMaxSize Serialized

默认的时间相关的值
new：new一个RTPS Writer
new_change：new一个CacheChange并添加到RTPS Writer的HistoryCache，sequenceNumber相对上次自动+1；

++this.lastChangeSequenceNumber;
a_change := new CacheChange(kind, this.guid, this.lastChangeSequenceNumber,data, inlineQos, handle);
RETURN a_change;

RTPS StatelessWriter

StatelessWriter不需要知道匹配的readers的数量，也不需要维护匹配的Reader的任何状态；只维护RTPS ReaderLocator，用于向已经匹配的readers发送信息
StatelessWriter属性：

attribute	type	meaning	relation to DDS
reader_locators	ReaderLocator[*]	StatelessWriter维护一个locators列表，用于发送CacheChanges.	N/A

StatelessWriter用于以下场景：

writer的HistoryCache交小
通信方式是best-effort
writer通过multicast和很多readers通信

StatelessWriter的方法：

new：创建一个StatelessWriter，并进行初始化：this.readerlocators := <empty>;
reader_locator_add：将ReaderLocator对象添加到StatelessWriter::reader_locators中，伪代码：ADD a_locator TO {this.reader_locators};；
reader_locator_remove：将ReaderLocator对象从StatelessWriter::reader_locators中移除，伪代码：REMOVE a_locator FROM {this.reader_locators};
unsent_changes_reset：This operation modifies the set of ‘unsent_changes’ for all the ReaderLocators in the StatelessWriter::reader_locators. The list of unsent changes is reset to match the complete list of changes available in the writer’s HistoryCache. This operation is useful when called periodically to cause the StatelessWriter to keep re-sending all available changes in its HistoryCache.伪代码：FOREACH readerLocator in {this.reader_locators} DO readerLocator.unsent_changes := {this.writer_cache.changes}

RTPS ReaderLocator

用于跟踪所有匹配的Readers的locators，用于跟踪和管理与特定Writer匹配的所有远程Reader的位置信息。这使得Writer能够有效地定位和发送数据到正确的Reader。具体来说，ReaderLocator的主要职责是存储和管理远程Reader的网络地址（通常是IP地址和端口号）
RTPS ReaderLocator属性：

attribute	type	meaning
requested_changes	CacheChange[*]	A list of changes in the writer’s HistoryCache that were requested by remote Readers at this ReaderLocator.
unsent_changes	CacheChange[*]	A list of changes in the writer’s HistoryCache that have not been sent yet to this ReaderLocator.
locator	Locator_t	Unicast or multicast locator through which the readers represented by this ReaderLocator can be reached.
expectsInlineQos	bool	Specifies whether the readers represented by this ReaderLocator expect inline QoS to be sent with every Data Message.

RTPS ReaderLocator operations：

new
next_requested_change
next_unsent_change
requested_changes
requested_changes_set
unsent_changes

RTPS StatefulWriter

RTPS StatefulWriter用于配置所有匹配的RTPS Reader，维护每个RTPS Reader的状态，这包括每个匹配的Reader的状态，包括序列号，确认的变化，缓冲的更改，过去的交互等等。这种状态使得StatefulWriter能够提供丰富的增强功能，如确保可靠的信息交付，处理不同的发送速率和网络故障，以及跟踪未确认的更改。
通过维护每个匹配的RTPS Reader端点的状态，RTPS StatefulWriter可以确定所有匹配的RTPS Reader端点是否已收到特定的CacheChange，并且可以通过避免向已接收到Writer的HistoryCache中所有changes的reader发送announcements，从而最优化地使用网络带宽。它维护的信息也简化了写入端的基于QoS的过滤。
RTPS StatefulWriter Attributes

attribute	type	meaning	relation to DDS
matched_readers	ReaderProxy[*]	StatefulWriter使用该字段跟踪与其匹配的所有RTPS readers。每个匹配的reader由ReaderProxy类的一个实例表示。

StatefulWriter Operations

new：new一个StatefulWriter，并将matched_readers初始化为空：this.matched_readers := <empty>;
is_acked_by_all：
matched_reader_add
matched_reader_remove
matched_reader_lookup

RTPS ReaderProxy

用于维护和StatefulWriter匹配的每一个Reader的信息；
RTPS ReaderProxy Attributes：

attribute	type	meaning	relation to DDS
remoteReaderGuid	GUID_t	远端匹配的RTPS Reader的guid
remoteGroupEntityId	EntityId_t	group中匹配的reader的EntityId	DataReader所属的Subscriber的EntityId
unicastLocatorList	Locator_t[*]	用于向匹配的RTPS reader发送消息的unicast locators (transport, address, port combinations)，该列表可能是空的
multicastLocatorList	Locator_t[*]	用于向匹配的RTPS reader发送的multicast locators (transport, address, port combinations)，可能为空
changes_for_reader	CacheChange[*]	和RTPS Reader相关的CacheChange的列表——什么作用？
expectsInlineQos	bool	Specifies whether the remote matched RTPS Reader expects in-line QoS to be sent along with any data.
isActive	bool	远端Reader是否需要向writer回应

StatefulWriter将把Writer的HistoryCache中的CacheChange changes发送到由ReaderProxy表示的匹配的RTPS reader中。

ReaderProxy Operations

new：创建一个ReaderProxy对象
acked_changes_set
next_requested_change
next_unsent_change
requested_changes
requested_changes_set
unsent_changes
unacked_changes

RTPS ChangeForReader

RTPS的ChangeForReader是一个关联类，它保存了RTPS Writer的HistoryCache中的CacheChange信息，这些信息与ReaderProxy所代表的RTPS Reader有关。换句话说，这个类是用来记录和管理RTPS Writer中的缓存改变（CacheChange）对于特定的RTPS Reader（由ReaderProxy代表）的相关信息。

RTPS StatelessWriter Behavior

Best-Effort StatelessWriter Behavior

TODO p81， Best-Effort StatelessWriter和每个ReaderLocator的交互

Transition	state	event	next state
T1	initial	给RTPS Best-Effort StatelessWriter配置RTPS ReaderLocator	idle
T2	idle	`GuardCondition: RL::unsent_changes() != <empty>`	pushing
T3	pushing	`GuardCondition: RL::unsent_changes() == <empty>`	idle
T4	pushing	`GuardCondition: RL::can_send() == true`	pushing
T5	any state	RTPS Writer is configured to no longer have the ReaderLocator	final

T1：这一步是服务发现阶段完成，给RTPS Best-Effort StatelessWriter配置DataReader到RTPS ReaderLocator。

a_locator := new ReaderLocator( locator, expectsInlineQos ); 
the_rtps_writer.reader_locator_add( a_locator );

表示RTPS Writer的HistoryCache中有一些changes还没有被发送给RTPS ReaderLocator
[RL::unsent_changes() == <empty>] 表明所有RTPS Writer的HistoryCache中的changes被发送到RTPS ReaderLocator中。但这并不意味着这些changes被接收到。
[RL::can_send() == true]表示RTPS Writer有一些resources需要发送change到RTPS ReaderLocator 对象中，伪代码如下：

a_change := the_reader_locator.next_unsent_change();
IF a_change IN the_writer.writer_cache.changes {
  DATA = new DATA(a_change);
  IF (the_reader_locator.expectsInlineQos) {
    DATA.inlineQos := the_writer.related_dds_writer.qos;
    DATA.inlineQos += a_change.inlineQos;
  }
  DATA.readerId := ENTITYID_UNKNOWN;
  sendto the_reader_locator.locator, DATA;
}
ELSE {
  GAP = new GAP(a_change.sequenceNumber);
  GAP.readerId := ENTITYID_UNKNOWN;
  sendto the_reader_locator.locator, GAP;
}

RTPS Writer不再发送消息给RTPS ReaderLocator。这一步也是由服务发现完成。

the_rtps_writer.reader_locator_remove(the_reader_locator);
delete the_reader_locator;

Reliable StatelessWriter Behavior

TODO P83 Reliable StatelessWriter with respect to each ReaderLocator

Transition	state	event	next state
T1	initial	RTPS Writer is configured with a ReaderLocator	announcing
T2	announcing	`GuardCondition: RL::unsent_changes() != <empty>`	pushing
T3	pushing	`GuardCondition: RL::unsent_changes() == <empty>`	announcing
T4	pushing	GuardCondition: RL::can_send() == true	pushing
T5	announcing	after(W::heartbeatPeriod)	announcing
T6	waiting	ACKNACK message is received	waiting
T7	waiting	GuardCondition: RL::requested_changes() !=	must_repair
T8	must_repair	ACKNACK message is received	must_repair
T9	must_repair	after(W::nackResponseDelay)	repairing
T10	repairing	GuardCondition: RL::can_send() == true	repairing
T11	repairing	GuardCondition: RL::requested_changes() ==	waiting
T12	any state	RTPS Writer is configured to no longer have the ReaderLocator	final

通过服务发现，配置Reliable StatelessWriter的ReaderLocator，将所有匹配的DataReader添加到ReaderLocator.

a_locator := new ReaderLocator( locator, expectsInlineQos ); 
the_rtps_writer.reader_locator_add( a_locator );

[RL::unsent_changes() != ] 表明，RTPS Writer HistoryCache中还有changes没有发送给ReaderLocator；
[RL::unsent_changes == ] 表明，所有RTPS Writer HistoryCache中的消息都已发送到ReaderLocator，但这并不保证这些changes已经被接收；
[RL::can_send() == true]表明RTPS Writer有一些resources需要发送给ReaderLocator对象。执行流程如下：

a_change := the_reader_locator.next_unsent_change(); 
DATA = new DATA(a_change);
IF (the_reader_locator.expectsInlineQos) { 
    DATA.inlineQos := the_writer.related_dds_writer.qos;
}
DATA.readerId := ENTITYID_UNKNOWN;
sendto the_reader_locator.locator, DATA;

After the transition the following post-conditions hold:

( a_change BELONGS-TO the_reader_locator.unsent_changes() ) == FALSE

这个状态转变由W::heartbeatPeriod.的timer触发，执行逻辑如下：

seq_num_min := the_rtps_writer.writer_cache.get_seq_num_min();
seq_num_max := the_rtps_writer.writer_cache.get_seq_num_max();
HEARTBEAT := new HEARTBEAT(the_rtps_writer.writerGuid, seq_num_min,
seq_num_max);
HEARTBEAT.FinalFlag := SET;
HEARTBEAT.readerId := ENTITYID_UNKNOWN;
sendto the_reader_locator, HEARTBEAT;

这个状态转换是由于RTPS StatelessWriter接收到一个由某个RTPS Reader发送的ACKNACK message而触发的，执行以下逻辑：

FOREACH reply_locator_t IN { Receiver.unicastReplyLocatorList,
Receiver.multicastReplyLocatorList }
reader_locator := the_rtps_writer.reader_locator_lookup(reply_locator_t); reader_locator.requested_changes_set(ACKNACK.readerSNState.set);

需要注意的是处理这类消息使用RTPS Receiver中的reply locators。这是StatelessWriter确定发送回复的位置的唯一信息源。协议的正常运行需要RTPS Reader在AckNack前插入一个InfoReply Submessage，以便正确设置这些字段。

这个状态转换由guard condition[RL::requested_changes() != <empty>]触发，表明RTPS ReaderLocator中的有一些RTPS Reader的request
这个状态转换是由于RTPS StatelessWriter收到由RTPS Reader发送的ACKNACK消息。
这个状态转换是由一个定时器触发的，表示自从进入must_repair状态以来，W::nackResponseDelay的持续时间已经过去。不执行任何逻辑操作。
todo
todo
todo