Este documento profundiza en las implementaciones de referencia de lectores RTPS (Real-Time Publish-Subscribe), detallando sus estructuras, configuraciones y comportamientos operativos, con un enfoque particular en las distinciones entre lectores stateless (sin estado) y stateful (con estado). También se explican los protocolos subyacentes para el mantenimiento de la vitalidad de los escritores y el manejo de datos de gran tamaño, junto con pautas de implementación para optimizar el rendimiento.
8.4.10 Implementación de Referencia del Lector RTPS
La implementación de referencia del lector RTPS se basa en especializaciones de la clase RTPS Reader, introducida previamente. Este apartado describe las clases adicionales necesarias para modelar estas implementaciones.
8.4.10.1 RTPS Reader
Un RTPS Reader es una especialización de RTPS Endpoint que recibe CacheChange de uno o más puntos de acceso de escritura RTPS. Las implementaciones de referencia StatelessReader y StatefulReader son especializaciones de RTPS Reader que difieren en el conocimiento que retienen sobre los puntos de acceso de escritura coincidentes.
Las propiedades de configuración para RTPS Reader se detallan en la Tabla 8.62, permitiendo un ajuste fino del comportamiento del protocolo. Las operaciones disponibles se listan en la Tabla 8.63.
Tabla 8.62 - Propiedades de Configuración de RTPS Reader
| Propiedad | Tipo | Significado |
|---|---|---|
heartbeatResponseDelay |
Duration_t | Parámetro de ajuste del protocolo que permite al lector RTPS retrasar el envío de acuses de recibo positivos o negativos. |
heartbeatSuppressionDuration |
Duration_t | Parámetro de ajuste del protocolo que permite al lector RTPS ignorar los HEARTBEAT que llegan "demasiado pronto" después de recibir uno anterior. |
reader_cache |
HistoryCache | Contiene el historial de cambios de CacheChange para este lector RTPS. |
expectsInlineQos |
bool | Indica si el lector RTPS espera QoS en línea enviada junto con los datos. |
Tabla 8.63 - Operaciones de RTPS Reader
| Nombre de Operación | Lista de Parámetros | Tipo |
|---|---|---|
new |
attribute values |
Reader |
8.4.10.1.1 Valores Predeterminados Relacionados con el Tiempo
Los siguientes valores relacionados con el tiempo se utilizan como configuraciones predeterminadas para facilitar la interoperabilidad "listo para usar" entre implementaciones:
heartbeatResponseDelay.sec = 0;
heartbeatResponseDelay.nanosec = 500 * 1000 * 1000; // 500 milisegundos
heartbeatSuppressionDuration.sec = 0;
heartbeatSuppressionDuration.nanosec = 0;
8.4.10.1.2 Operación new
Esta operación crea un nuevo lector RTPS. El lector recién creado se inicializa con los valores proporcionados durante la construcción, incluyendo su GUID, listas de localizadores, nivel de fiabilidad, tipo de tema, expectativa de QoS en línea y el retraso de respuesta de heartbeat. También se inicializa una nueva HistoryCache para su reader_cache.
8.4.10.2 RTPS StatelessReader
Un RTPS StatelessReader es una especialización de RTPS Reader que no mantiene estado sobre la cantidad de escritores coincidentes ni sobre cada escritor RTPS individual. En la implementación de referencia actual, StatelessReader es idéntico a Reader en términos de propiedades y operaciones, heredando todas las de su superclase. Las interacciones de la máquina virtual con StatelessReader se definen en la Tabla 8.64.
Tabla 8.64 - Operaciones de StatelessReader
| Nombre de Operación | Lista de Parámetros | Tipo |
|---|---|---|
new |
attribute values |
StatelessReader |
8.4.10.2.1 Operación new
Esta operación crea un nuevo RTPS StatelessReader. Su inicialización es idéntica a la realizada en la superclase RTPS Reader.
8.4.10.3 RTPS StatefulReader
Un RTPS StatefulReader es una especialización de RTPS Reader que mantiene el estado de cada escritor RTPS coincidente, gestionado a través de la clase WriterProxy.
Tabla 8.65 - Propiedades de RTPS StatefulReader
| Propiedad | Tipo | Significado |
|---|---|---|
matched_writers |
WriterProxy[*] |
Mantiene el estado de los escritores remotos que coinciden con el lector. |
Las operaciones para interactuar con StatefulReader se muestran en la Tabla 8.66.
Tabla 8.66 - Operaciones de StatefulReader
| Nombre de Operación | Lista de Parámetros | Tipo |
|---|---|---|
new |
attribute values |
StatefulReader |
matched_writer_add |
a_writer_proxy |
void |
matched_writer_remove |
a_writer_proxy |
void |
matched_writer_lookup |
a_writer_guid |
WriterProxy |
8.4.10.3.1 Operación new
Esta operación crea un nuevo RTPS StatefulReader. Se inicializa con los atributos proporcionados y una lista vacía de matched_writers.
8.4.10.3.2 Operación matched_writer_add
Añade un WriterProxy a la colección matched_writers del StatefulReader.
8.4.10.3.3 Operación matched_writer_remove
Elimina un WriterProxy de la colección matched_writers y lo elimina.
8.4.10.3.4 Operación matched_writer_lookup
Busca un WriterProxy en matched_writers basándose en su GUID.
8.4.10.4 RTPS WriterProxy
Un RTPS WriterProxy representa la información que un RTPS StatefulReader mantiene para cada escritor RTPS coincidente. Sus propiedades se describen en la Tabla 8.67.
Tabla 8.67 - Propiedades de RTPS WriterProxy
| Propiedad | Tipo | Significado |
|---|---|---|
remoteWriterGuid |
GUID_t | Identifica al escritor coincidente. |
unicastLocatorList |
Locator_t[*] | Lista de localizadores unicast para enviar mensajes al escritor. |
multicastLocatorList |
Locator_t[*] | Lista de localizadores multicast para enviar mensajes al escritor. |
dataMaxSizeSerialized |
long | Tamaño máximo opcional para SerializedPayload del escritor. |
changes_from_writer |
CacheChange[*] |
Lista de cambios CacheChange recibidos o esperados del escritor. |
remoteGroupEntityId |
EntityId_t | Identifica el grupo al que pertenece el lector remoto. |
Las operaciones para interactuar con WriterProxy se detallan en la Tabla 8.68.
Tabla 8.68 - Operaciones de WriterProxy
| Nombre de Operación | Lista de Parámetros | Tipo |
|---|---|---|
new |
attribute values |
WriterProxy |
available_changes_max |
SequenceNumber_t |
|
irrelevant_change_set |
a_seq_num |
void |
lost_changes_update |
first_available_seq_num |
void |
missing_changes |
SequenceNumber_t[] |
|
missing_changes_update |
last_available_seq_num |
void |
received_change_set |
a_seq_num |
void |
8.4.10.4.1 Operación new
Esta operación crea un nuevo RTPS WriterProxy. Se inicializa con los atributos proporcionados. La colección changes_from_writer se inicializa conceptualmente para contener todos los samples pasados y futuros del escritor representado. El estado de cada CacheChange se inicializa a UNKNOWN.
8.4.10.4.2 Operación available_changes_max
Devuelve el SequenceNumber_t más alto de los cambios en changes_from_writer que son accesibles para el DDS DataReader. Un cambio es accesible si todos los cambios anteriores de ese escritor RTPS son RECEIVED o LOST.
seq_num := MAX { change.sequenceNumber SUCH-THAT ( change IN this.changes_from_writer AND ( change.status == RECEIVED OR change.status == LOST) ) };
return seq_num;
8.4.10.4.3 Operación irrelevant_change_set
Establece el estado de un CacheChange a RECEIVED y marca is_relevant como FALSE, indicando que el cambio es irrelevante para el lector RTPS.
FIND change FROM this.changes_from_writer SUCH-THAT (change.sequenceNumber == a_seq_num);
change.status := RECEIVED;
change.is_relevant := FALSE;
8.4.10.4.4 Operación lost_changes_update
Actualiza el estado de los cambios en changes_from_writer que están en estado UNKNOWN o MISSING y cuyo número de secuencia es menor que first_available_seq_num, marcándolos como LOST.
FOREACH change IN this.changes_from_writer SUCH-THAT ( change.status == UNKNOWN OR change.status == MISSING AND seq_num < first_available_seq_num ) DO {
change.status := LOST;
}
8.4.10.4.5 Operación missing_changes
Devuelve un subconjunto de cambios en changes_from_writer que están en estado MISSING.
return { change IN this.changes_from_writer SUCH-THAT change.status == MISSING};
8.4.10.4.6 Operación missing_changes_update
Actualiza el estado de los cambios en changes_from_writer que están en estado UNKNOWN y cuyo número de secuencia es menor o igual a last_available_seq_num, cambiándolos a MISSING.
FOREACH change IN this.changes_from_writer SUCH-THAT ( change.status == UNKNOWN AND seq_num <= last_available_seq_num ) DO {
change.status := MISSING;
}
8.4.10.4.7 Operación received_change_set
Establece el estado de un CacheChange especificado por a_seq_num a RECEIVED.
FIND change FROM this.changes_from_writer SUCH-THAT change.sequenceNumber == a_seq_num;
change.status := RECEIVED
8.4.10.5 RTPS ChangeFromWriter
RTPS ChangeFromWriter es una clase asociativa que mantiene información sobre un CacheChange en el RTPS Reader HistoryCache, en relación con el RTPS Writer representado por un WriterProxy. Sus propiedades se detallan en la Tabla 8.69.
Tabla 8.69 - Propiedades de RTPS ChangeFromWriter
| Propiedad | Tipo | Significado |
|---|---|---|
status |
ChangeFromWriterStatusKind |
Representa el estado del CacheChange con respecto al RTPS Writer. |
is_relevant |
bool | Indica si el cambio es relevante para el RTPS Reader. |
8.4.11 Comportamiento del Lector RTPS sin Estado
8.4.11.1 Comportamiento del Lector sin Estado Best-Effort
El comportamiento de un lector RTPS sin estado best-effort no depende de ningún escritor. La máquina de estados se describe en la Figura 8.22 y sus transiciones en la Tabla 8.70.
Tabla 8.70 - Transiciones del Comportamiento del Lector sin Estado Best-Effort
| Transición | Evento | Siguiente Estado |
|---|---|---|
| T1 | Initial RTPS Reader is created |
Waiting |
| T2 | DATA message received in Waiting state |
Waiting |
| T3 | RTPS Reader is destroyed in Waiting state |
Final |
8.4.11.1.1 Transición T1
Activada por la creación de un lector RTPS sin estado. No hay acciones lógicas en la máquina virtual.
8.4.11.1.2 Transición T2
Activada por la recepción de un mensaje DATA. Dado que el lector es sin estado, puede presentar cambios duplicados o desordenados. Sin embargo, si el DDS DataReader está configurado para ordenar por "timestamp de origen", los datos accesibles se presentarán en orden. Las implementaciones pueden intentar mitigar esto manteniendo información temporalmente.
a_change := new CacheChange(DATA);
the_rtps_reader.reader_cache.add_change(a_change);
8.4.11.1.3 Transición T3
Activada por la destrucción del lector RTPS. No hay acciones lógicas en la máquina virtual.
8.4.11.2 Comportamiento del Lector sin Estado Fiable
El protocolo RTPS no soporta esta combinación, ya que un lector RTPS fiable requiere mantener algún estado sobre cada escritor coincidente.
8.4.12 Comportamiento del Lector RTPS con Estado
8.4.12.1 Comportamiento del Lector con Estado Best-Effort
El comportamiento de un lector RTPS con estado best-effort, con respecto a cada escritor coincidente, se describe en la Figura 8.23. Las transiciones de la máquina de estados se listan en la Tabla 8.71.
Tabla 8.71 - Transiciones del Comportamiento del Lector con Estado Best-Effort para cada Escritor Coincidente
| Transición | Estado Actual | Evento | Siguiente Estado |
|---|---|---|---|
| T1 | Initial | RTPS Reader configured with matching RTPS Writer | Waiting |
| T2 | Waiting | Received DATA message from matching writer | Waiting |
| T3 | Waiting | RTPS Reader configured to no longer match RTPS Writer | Terminating |
| T4 | Waiting | Received GAP message | Waiting |
8.4.12.1.1 Transición T1
Activada cuando un lector RTPS se configura con un escritor RTPS coincidente, típicamente a través del protocolo de descubrimiento. Se crea un WriterProxy y se añade al lector.
a_writer_proxy := new WriterProxy(remoteWriterGuid, remoteGroupEntityId, unicastLocatorList, multicastLocatorList);
the_rtps_reader.matched_writer_add(a_writer_proxy);
8.4.12.1.2 Transición T2
Activada por la recepción de un mensaje DATA. El lector verifica que el número de secuencia del cambio sea mayor que el número de secuencia máximo recibido previamente de este escritor para evitar duplicados y desorden.
a_change := new CacheChange(DATA);
writer_guid := {Receiver.SourceGuidPrefix, DATA.writerId};
writer_proxy := the_rtps_reader.matched_writer_lookup(writer_guid);
expected_seq_num := writer_proxy.available_changes_max() + 1;
if ( a_change.sequenceNumber >= expected_seq_num ) {
the_rtps_reader.reader_cache.add_change(a_change);
writer_proxy.received_change_set(a_change.sequenceNumber);
if ( a_change.sequenceNumber > expected_seq_num ) {
writer_proxy.lost_changes_update(a_change.sequenceNumber);
}
}
Postcondición: writer_proxy.available_changes_max() >= a_change.sequenceNumber.
8.4.12.1.3 Transición T3
Activada cuando un lector RTPS se desconfigura de un escritor RTPS coincidente. El WriterProxy se elimina y se destruye.
the_rtps_reader.matched_writer_remove(the_writer_proxy);
delete the_writer_proxy;
8.4.12.1.4 Transición T4
Activada por la recepción de un mensaje GAP. Marca los cambios correspondientes como irrelevantes.
FOREACH seq_num IN [GAP.gapStart, GAP.gapList.base-1] DO {
the_writer_proxy.irrelevant_change_set(seq_num);
}
FOREACH seq_num IN GAP.gapList DO {
the_writer_proxy.irrelevant_change_set(seq_num);
}
8.4.12.2 Comportamiento del Lector con Estado Fiable
El comportamiento de un lector RTPS con estado fiable para cada escritor RTPS coincidente se describe en la Figura 8.24. Las transiciones de la máquina de estados se detallan en la Tabla 8.72.
Tabla 8.72 - Transiciones del Comportamiento del Lector con Estado Fiable para Escritores Coincidentes
| Número Transición | Estado Actual | Evento | Siguiente Estado |
|---|---|---|---|
| T1 | Initial1 | RTPS Reader configured with matching RTPS Writer | Waiting |
| T2 | Waiting | Received HEARTBEAT message | If (HB.FinalFlag == NOT_SET) Must Send Ack, else If (HB.LivelinessFlag == NOT_SET) May Send Ack, else remain Waiting |
| T3 | May Send Ack | Guard Condition: WP::missing_changes() == <empty> | Waiting |
| T4 | May Send Ack | Guard Condition: WP::missing_changes() != <empty> | Must Send Ack |
| T5 | Must Send Ack | after(R::heartbeatResponseDelay) | Waiting |
| T6 | Initial2 | RTPS Reader configured with matching RTPS Writer | Ready |
| T7 | Ready | Received HEARTBEAT message | Ready |
| T8 | Ready | Received DATA message | Ready |
| T9 | Ready | Received GAP message | Ready |
| T10 | Any State | RTPS Reader configured to no longer match RTPS Writer | Terminating |
8.4.12.2.1 Transición T1
Similar a T1 en el comportamiento best-effort, esta transición configura el lector con un escritor coincidente y crea un WriterProxy.
8.4.12.2.2 Transición T2
Activada por la recepción de un mensaje HEARTBEAT. No hay acciones lógicas directas, pero puede desencadenar transiciones concurrentes (T7) y cambios de estado basados en la presencia de cambios faltantes.
8.4.12.2.3 Transición T3
Activada cuando no hay cambios faltantes para el escritor, permitiendo la transición a Waiting.
8.4.12.2.4 Transición T4
Activada cuando hay cambios faltantes, lo que requiere el envío de una confirmación (ACKNACK).
8.4.12.2.5 Transición T5
Tras el retraso especificado por heartbeatResponseDelay, se envía un mensaje ACKNACK para solicitar los cambios faltantes.
missing_seq_num_set.base := the_writer_proxy.available_changes_max() + 1;
missing_seq_num_set.set := <empty>;
FOREACH change IN the_writer_proxy.missing_changes() DO
ADD change.sequenceNumber TO missing_seq_num_set.set;
send ACKNACK(missing_seq_num_set);</empty>
8.4.12.2.6 Transición T6
Similar a T1, esta transición configura el lector con un escritor coincidente, pero inicia el estado Ready.
8.4.12.2.7 Transición T7
Activada por la recepción de un mensaje HEARTBEAT. Actualiza el estado de los cambios faltantes y perdidos basándose en la información del heartbeat.
the_writer_proxy.missing_changes_update(HEARTBEAT.lastSN);
the_writer_proxy.lost_changes_update(HEARTBEAT.firstSN);
8.4.12.2.8 Transición T8
Activada por la recepción de un mensaje DATA. El cambio se añade a la caché del lector y se marca como recibido por el escritor.
a_change := new CacheChange(DATA);
the_reader.reader_cache.add_change(a_change);
the_writer_proxy.received_change_set(a_change.sequenceNumber);
8.4.12.2.9 Transición T9
Activada por la recepción de un mensaje GAP. Marca los cambios correspondientes como irrelevantes.
FOREACH seq_num IN [GAP.gapStart, GAP.gapList.base-1] DO {
the_writer_proxy.irrelevant_change_set(seq_num);
}
FOREACH seq_num IN GAP.gapList DO {
the_writer_proxy.irrelevant_change_set(seq_num);
}
8.4.12.2.10 Transiicón T10
Similar a T3, esta transición ocurre cuando un lector RTPS se desconfigura de un escritor RTPS coincidente.
the_rtps_reader.matched_writer_remove(the_writer_proxy);
delete the_writer_proxy;
8.4.12.3 Explicación de ChangeFromWriter
ChangeFromWriter rastrea el estado de comunicación y relevancia de un CacheChange en relación con un escritor RTPS remoto. La Figura 8.25 ilustra la máquina de estados para el atributo status de ChangeFromWriter para lectores con estado fiable.
Los estados posibles para status incluyen: Unknown, Missing, Requested, Received y Lost. Las transiciones entre estos estados son provocadas por eventos como la recepción de mensajes HEARTBEAT, ACKNACK, GAP o DATA, así como por la inicialización o el fin de la comunicación.
8.4.13 Protocolo de Vitalidad del Escritor
El protocolo de vitalidad del escritor asegura la interoperabilidad al definir el intercambio de información necesario para afirmar la vitalidad de los escritores dentro de un participante. Utiliza puntos de acceso incorporados específicos: BuiltinParticipantMessageWriter y BuiltinParticipantMessageReader.
8.4.13.1 Método Genarel
El protocolo de vitalidad del escritor emplea puntos de acceso incorporados predefinidos. Una vez que un participante conoce la existencia de otro, puede asociarse con los puntos de acceso incorporados coincidentes y usar el mismo protocolo de comunicación que para los puntos de acceso definidos por la aplicación.
8.4.13.2 Puntos de Acceso Incorporados Requeridos
Los puntos de acceso requeridos son BuiltinParticipantMessageWriter y BuiltinParticipantMessageReader, con EntityId_t reservados como ENTITYID_P2P_BUILTIN_PARTICIPANT_MESSAGE_WRITER y ENTITYID_P2P_BUILTIN_PARTICIPANT_MESSAGE_READER.
8.4.13.3 QoS para BuiltinParticipantMessageWriter y BuiltinParticipantMessageReader
Para la interoperabilidad, estos puntos de acceso deben usar las siguientes QoS: TRANSIENT_LOCAL_DURABILITY, KEEP_LAST_HISTORY_QOS con una profundidad de 1. El escritor debe ser RELIABLE_RELIABILITY_QOS, mientras que el lector puede ser fiable o best-effort.
8.4.13.4 Tipos de Datos Asociados
Cada punto de acceso incorpora una HistoryCache. El tipo de datos ParticipantMessageData se define para los temas DCPSParticipantMessage, utilizado por estos puntos de acceso incorporados.
8.4.13.5 Implementación del Protocolo de Vitalidad del Escritor
La vitalidad de los escritores se declara escribiendo muestras en el BuiltinParticipantMessageWriter. La frecuencia de escritura está ligada a la duración mínima del lease de vitalidad de los escritores del participante. Se utilizan instancias separadas para diferentes tipos de QoS de vitalidad (ej. AUTOMATIC vs. MANUAL_BY_PARTICIPANT).
8.4.14 Comportamiento Opcional
8.4.14.1 Datos Grandes
El protocolo RTPS soporta datos grandes mediante fragmentación, utilizando submensajes como DataFrag, HeartbeatFrag y NackFrag. Esto es necesario cuando el tamaño de los datos excede las limitaciones del transporte subyacente (ej. UDP).
8.4.14.1.1 Cómo Seleccionar el Tamaño del Fragmento
El tamaño del fragmento es determinado por el escritor y debe ser fijo para todos los lectores remotos. Debe ser menor o igual a 65536 bytes y ser compatible con todas las transportes disponibles.
8.4.14.1.2 Cómo Enviar Fragmentos
Los datos se fragmentan en submensajes DataFrag enviados secuencialmente. La fragmentación solo ocurre si es necesario. La QoS en línea se incluye en el primer fragmento DataFrag si se utiliza.
8.4.14.1.3 Cómo Reensamblar Fragmentos
Los fragmentos se reensamblan usando la información dentro de los submensajes DataFrag. El comportamiento es similar al manejo de submensajes Data estándar una vez que todos los fragmentos son recibidos.
8.4.14.1.4 Comunicación Fiable
Los submensajes Heartbeat y AckNack siguen semánticas similares a las de los datos no fragmentados. Sin embargo, NackFrag se usa para solicitar fragmentos específicos de un cambio. HeartbeatFrag puede usarse para notificar sobre la disponibilidad parcial de fragmentos.
8.4.15 Guías de Implementación
8.4.15.1 Implementación de ReaderProxy y WriterProxy
Se sugieren optimizaciones para la implementación de ReaderProxy y WriterProxy. En lugar de mantener asociaciones explícitas con cada CacheChange, se pueden usar números de secuencia de alto nivel (ej. highestSeqNumSent) y ventanas deslizantes de números de secuencia para rastrear eficientemente los cambios enviados, solicitados o confirmados.
8.4.15.2 Uso Eficiente de Submensajes Gap y AckNack
Se recomienda fusionar múltiples submensajes Gap y AckNack en uno solo para optimizar el uso del ancho de banda y la CPU, aprovechando la capacidad de estos submensajes para contener conjuntos de números de secuencia.
8.4.15.3 Fusión de Múltiples Submensajes Data
La fusión de múltiples submensajes Data (incluso de diferentes escritores RTPS) en un único mensaje RTPS puede mejorar significativamente la eficiencia del uso del ancho de banda y la CPU debido a la reducción de la sobrecarga del encabezado.
8.4.15.4 "Aprovechamiento" de Submensajes Heartbeat
Se aconseja adjuntar submensajes Heartbeat a los submensajes Data para reducir el tráfico de red al solicitar confirmaciones de los cambios enviados.
8.4.15.5 Envío a readerId Desconocido
Se anima a las implementaciones a utilizar la capacidad de enviar mensajes RTPS sin un readerId especificado (ENTITYID_UNKNOWN), especialmente para transmisiones multicast, para minimizar el uso de ancho de banda.
8.4.15.6 Recuperación de Recursos Limitados de Lectores No Respondedores
Los escritores deben implementar mecanismos para identificar lectores no respondedores y recuperar recursos asociados con los cambios confirmados por lectores activos. Esto previene la degradación del rendimiento debido a lectores inactivos, aunque la fiabilidad estricta puede verse comprometida para dichos lectores.
8.4.15.7 Establecimiento de Contadores en Mensajes Heartbeat, Heartbeat Frag, AckNack y NackFrag
Los contadores en estos submensajes deben usarse para distinguir latidos lógicos y evitar sesiones de reparación duplicadas. Deben incrementarse y compararse usando aritmética modular para manejar el desbordamiento de enteros.