31.8 Writing Data

The write() operation informs Connext that there is a new value for a data-instance to be published for the corresponding Topic. By default, calling write() will send the data immediately over the network (assuming that there are matched DataReaders). However, you can configure and execute operations on the DataWriter’s Publisher to buffer the data so that it is sent in a batch with data from other DataWriters or even to prevent the data from being sent. Those sending “modes” are configured using the 46.6 PRESENTATION QosPolicy as well as the Publisher’s suspend/resume_publications() operations. The actual transport-level communications may be done by a separate, lower-priority thread when the Publisher is configured to send the data for its DataWriters. For more information on threads, see Part 11: Connext Threading Model.

When you call write(), Connext automatically attaches a stamp of the current time that is sent with the DDS data sample to the DataReader(s). The timestamp appears in the source_timestamp field of the DDS_SampleInfo structure that is provided along with your data using DataReaders (see 41.6 The SampleInfo Structure).

DDS_ReturnCode_t  write (const Foo &instance_data, 
    const DDS_InstanceHandle_t &handle)

You can use an alternate DataWriter operation called write_w_timestamp(). This performs the same action as write(), but allows the application to explicitly set the source_timestamp. This is useful when you want the user application to set the value of the timestamp instead of the default clock used by Connext. 

DDS_ReturnCode_t  write_w_timestamp (
	const Foo &instance_data,
	const DDS_InstanceHandle_t &handle, 
	const DDS_Time_t &source_timestamp)

Note that, in general, the application should not mix these two ways of specifying timestamps. That is, for each DataWriter, the application should either always use the automatic timestamping mechanism (by calling the normal operations) or always specify a timestamp (by calling the “w_timestamp” variants of the operations). Mixing the two methods may result in not receiving sent data.

You can also use an alternate DataWriter operation, write_w_params(), which performs the same action as write(), but allows the application to explicitly set the fields contained in the DDS_WriteParams structure, see Table 31.15 DDS_WriteParams_t.

Table 31.15 DDS_WriteParams_t

Type

Field Name

Description

DDS_Boolean

replace_auto

Allows retrieving the actual value of those fields that were automatic.

When this field is set to true, the fields that were configured with an automatic value (for example, DDS_AUTO_SAMPLE_IDENTITY in identity) receive their actual value after write_w_params is called.

DDS_
SampleIdentity_t

identity

Identity of the DDS sample being written. The identity consists of a pair (Virtual Writer GUID, Virtual Sequence Number).

When the value DDS_AUTO_SAMPLE_IDENTITY is used, the write_w_params() operation will determine the DDS sample identity as follows:

  • The Virtual Writer GUID (writer_guid) is the virtual GUID associated with the DataWriter writing the DDS sample. This virtual GUID is configured using the member virtual_guid in 47.5 DATA_WRITER_PROTOCOL QosPolicy (DDS Extension).
  • The Virtual Sequence Number (sequence_number) is increased by one with respect to the previous value.

The virtual sequence numbers for a given virtual GUID must be strictly monotonically increasing. If you try to write a DDS sample with a sequence number smaller or equal to the last sequence number, the write operation will fail.

A DataReader can inspect the identity of a received DDS sample by accessing the fields original_publication_virtual_guid and original_publication_virtual_sequence_number in 41.6 The SampleInfo Structure.

DDS_
SampleIdentity_t

related_sample_
identity

The identity of another DDS sample related to this one.

The value of this field identifies another DDS sample that is logically related to the one that is written.

For example, the DataWriter created by a Replier (sets 61.1 Introduction to the Request-Reply Communication Pattern) uses this field to associate the identity of the DDS request sample to reponse sample.

To specify that there is no related DDS sample identity use the value DDS_UNKNOWN_SAMPLE_IDENTITY,

A DataReader can inspect the related DDS sample identity of a received DDS sample by accessing the fields related_original_publication_virtual_guid and related_original_publication_virtual_sequence_number in 41.6 The SampleInfo Structure.

DDS_Time

source_timestamp

Source timestamp that will be associated to the DDS sample that is written.

If source_timestamp is set to DDS_TIMER_INVALID, the middleware will assign the value.

A DataReader can inspect the source_timestamp value of a received DDS sample by accessing the field source_timestamp 41.6 The SampleInfo Structure.

DDS_
InstanceHandle_t

handle

The instance handle.

This value can be either the handle returned by a previous call to register_instance() or the special value DDS_HANDLE_NIL.

DDS_Long

priority

Positive integer designating the relative priority of the DDS sample, used to determine the transmission order of pending transmissions.

To use publication priorities, the DataWriter’s 47.20 PUBLISH_MODE QosPolicy (DDS Extension) must be set for asynchronous publishing and the DataWriter must use a FlowController with a highest-priority first scheduling_policy.

For Multi-channel DataWriters, the publication priority of a DDS sample may be used as a filter criteria for determining channel membership.

For more information, see 34.4.4 Prioritized DDS Samples.

DDS_Long

flag

Flags for the DDS sample, represented as a 32-bit integer, of which only the 16 least-significant bits are used.

RTI reserves least-significant bits [0-7] for middleware-specific usage. The application can use least significant bits [8-15].

An application can inspect the flags associated with a received DDS sample by checking the flag field in 41.6 The SampleInfo Structure.

For details about the reserved bits see 41.6 The SampleInfo Structure.

Default 0 (no flags are set).

struct DDS_GUID_t

source_guid

Identifies the application logical data source associated with the sample being written.

struct DDS_GUID_t

related_source_guid

Identifies the application logical data source that is related to the sample being written.

struct DDS_GUID_t

related_reader_guid

Identifies a DataReader that is logically related to the sample that is being written.

When using the C API, a newly created variable of type DDS_WriteParams_t should be initialized by setting it to DDS_WRITEPARAMS_DEFAULT.

The write() operation also asserts liveliness on the DataWriter, the associated Publisher, and the associated DomainParticipant. It has the same effect with regards to liveliness as an explicit call to assert_liveliness(), see 31.17 Asserting Liveliness and the 47.15 LIVELINESS QosPolicy. Maintaining liveliness is important for DataReaders to know that the DataWriter still exists and for the proper behavior of the 47.17 OWNERSHIP QosPolicy.

See also: 16.3.15 Configuring the Clock per DomainParticipant.

31.8.1 Blocking During a write()

The write() operation may block if the 47.21 RELIABILITY QosPolicy kind is set to Reliable, the send window is full, or the modification would cause data to be lost. Specifically, write() may block in the following situations (note that the list may not be exhaustive):

This operation may also block when using BEST_EFFORT Reliability 47.21 RELIABILITY QosPolicy) and ASYNCHRONOUS Publish Mode (47.20 PUBLISH_MODE QosPolicy (DDS Extension)) QoS settings. In this case, the DataWriter will queue DDS samples until they are sent by the asynchronous publishing thread. The number of DDS samples that can be stored is determined by the 47.12 HISTORY QosPolicy. If the asynchronous thread does not send DDS samples fast enough (such as when using a slow FlowController (34.4 FlowControllers (DDS Extension))), the queue may fill up. In that case, subsequent write calls will block.

If this operation does block for any of the above reasons, the 47.21 RELIABILITY QosPolicy's max_blocking_time configures the maximum time the write operation may block (waiting for space to become available). If max_ blocking_time elapses before the DataWriter can store the modification without exceeding the limits, the operation will fail and return RETCODE_TIMEOUT for KEEP_ALL configurations.

31.8.2 write() behavior with KEEP_LAST and KEEP_ALL

Following is how the write operation behaves when KEEP_LAST (in the 47.12 HISTORY QosPolicy) and RELIABLE (in the 47.21 RELIABILITY QosPolicy) are used:

  • The send window size is determined by the max/min_send_window_size fields in the DDS_RtpsReliableWriterProtocol_t structure in the 47.5 DATA_WRITER_PROTOCOL QosPolicy (DDS Extension). If a send window is specified (max_send_window_size is not UNLIMITED) and the window is full, the write operation will block until one of the samples in the send window is protocol-acknowledged (ACKed) (Note 1) or until the max_blocking_time in the 47.21 RELIABILITY QosPolicy (writer_qos.reliability.max_blocking_time) expires.
  • Then, the DataWriter will try to add the new sample to the writer history.
  • If the instance associated with the sample is present in the writer history and there are depth (in the 47.12 HISTORY QosPolicy) samples in the instance, the DataWriter will replace the oldest sample of that instance independently of that sample’s acknowledged status, and the write operation will return DDS_RETCODE_OK. Otherwise, no sample will be replaced and the write operation will continue.
  • If the instance associated with the sample is not present in the writer history and max_instances (in the 47.22 RESOURCE_LIMITS QosPolicy) is exceeded, the DataWriter will try to replace an existing instance (and its samples) according to the value of the instance_replacement field in the 47.6 DATA_WRITER_RESOURCE_LIMITS QosPolicy (DDS Extension) (see 47.6.1 Configuring DataWriter Instance Replacement).
    • If no instance can be replaced, the write operation returns a DDS_RETCODE_OUT_OF_RESOURCES error.
  • If max_samples (in the 47.22 RESOURCE_LIMITS QosPolicy) is exceeded, the DataWriter will try to drop a sample from a different instance as follows:
    • The DataWriter will try first to remove a fully ACKed (Note 2) sample from a different instance 'I' as long as that sample is not the last remaining sample for the instance 'I'. To find this sample, the DataWriter starts iterating from the oldest sample in the writer history to the newest sample.
    • If no such sample is found, the DataWriter will replace the oldest sample in the writer history.
  • The sample is added to the writer history, and the write operation returns DDS_RETCODE_OK.

Following is how the write operation behaves when KEEP_ALL (in the 47.12 HISTORY QosPolicy) and RELIABLE (in the 47.21 RELIABILITY QosPolicy) are used:

  • The send window size is determined by the max/min_send_window_size fields in the DDS_RtpsReliableWriterProtocol_t structure in the 47.5 DATA_WRITER_PROTOCOL QosPolicy (DDS Extension). If a send window is specified (max_send_window_size is not UNLIMITED) and the window is full, the write operation will block until one of the samples in the send window is protocol-ACKed (Note 1) or until the max_blocking_time in the 47.21 RELIABILITY QosPolicy (writer_qos.reliability.max_blocking_time) expires.
    • If writer_qos.reliability.max_blocking_time expires, the write operation returns DDS_RETCODE_TIMEOUT.
  • When a sample is protocol-ACKed (Note 1) before max_blocking_time expires, the DataWriter will try to add the sample to the writer history as follows:
    • If the instance associated with the sample is not present in the writer history and max_instances is exceeded, the DataWriter will try to replace an existing instance (and its samples) according to the value of the instance_replacement field in the 47.6 DATA_WRITER_RESOURCE_LIMITS QosPolicy (DDS Extension) (see 47.6.1 Configuring DataWriter Instance Replacement).
      • If no instance can be replaced, the write operation returns a DDS_RETCODE_OUT_OF_RESOURCES error.
    • If max_samples is exceeded, the DataWriter will go through the samples in the order in which they were added, and it will replace the first sample that is fully ACKed (Note 2).
      • If no fully ACKed sample is found, the DataWriter will block (Note 3) until a sample is fully ACKed and can be replaced or writer_qos.reliability.max_blocking_time expires. If writer_qos.reliability.max_blocking_time expires, the write operation will return DDS_RETCODE_TIMEOUT.
    • If max_samples_per_instance is exceeded, the DataWriter will go through the samples of the instance in the order in which they were added, and it will replace the first sample that is fully ACKed.
      • If no fully ACKed sample is found, the DataWriter will block (Note 3) until a sample is fully ACKed and can be replaced or writer_qos.reliability.max_blocking_time expires. If writer_qos.reliability.max_blocking_time expires, the write operation will return DDS_RETCODE_TIMEOUT.
    • The sample is added to the writer history, and the write operation returns DDS_RETCODE_OK.

See 31.12.1 Application Acknowledgment Kinds for more information on the following notes:

Note 1:

A sample in the writer history is considered “protocol ACKed” when the sample has been individually ACKed at the RTPS protocol level by each one of the DataReaders that matched the DataWriter at the moment the sample was added to the DataWriter queue.

  • Late joiners do not change the protocol ACK state of a sample. If a sample is marked as protocol ACKed because it has been acknowledged by all the matching DataReaders and a DataReader joins later on, the historical sample is still considered protocol ACKed even if it has not been received by the late joiner.
  • If a sample 'S1' is protocol ACKed and a TopicQuery is received, triggering the publication of 'S1', the sample is still considered protocol ACKed. If a sample 'S1' is not ACKed and a TopicQuery is received triggering the publication of 'S1', the DataWriter will require that both the matching DataReaders on the live RTPS channel and the DataReader on the TopicQuery channel individually protocol ACK the sample in order to consider the sample protocol ACKed.

Note 2:

A sample in the writer history is considered “fully ACKed” when all of the following conditions are met:

  • The sample is protocol-ACKed.
  • The sample has been “application-level ACKed” by all the DataReaders matching the DataWriter that have their reader_qos.reliability.acknowledgment_kind set to AUTO_ACKNOWLEDGMENT_MODE or EXPLICIT_ACKNOWLEDGMENT_MODE. Once the sample is application-level ACKed, it cannot change its status to not ACked after new DataReaders are matched. (Application-level ACK occurs when the application acknowledges receipt of a sample.)
  • If required subscriptions are enabled (see 47.1 AVAILABILITY QosPolicy (DDS Extension)), the sample must also be ACKed by all the required subscriptions configured on the DataWriter.

Note 3:

It is possible within a single call to the write operation for a DataWriter to block both when the send window is full and then again when max_samples or max_samples_per_instance is exceeded. This can happen because blocking on the send window only considers protocol-ACKed samples, while blocking based on resource limits considers fully-ACKed samples. In any case, the total max blocking time of a single call to the write operation will not exceed writer_qos.reliability.max_blocking_time.

The write operation on a DataWriter configured to use batching may also block if the sample being written cannot be added to the existing outstanding batch and the batch has to be synchronously flushed within the context of the write thread (see 47.2.1 Synchronous and Asynchronous Flushing). The flushing operation may block under the same scenarios described above for individual samples, taking into account that the send window is applied per batch and not per sample.

The unregister_instance() and dispose() operations, with regards to KEEP_LAST and KEEP_ALL, behave the same as for the write() operation. See 31.14.2 Registering Instances, 31.14.4 Unregistering Instances, and 31.14.3 Disposing Instances.