FlowController Use Cases

Working with flow controllers.

Working with flow controllers.

Creating a flow controller

• Set up participant
• Create a flow controller

  DDS_ReturnCode_t retcode;
  DDS_FlowController *controller = NULL;
  struct DDS_FlowControllerProperty_t property = DDS_FlowControllerProperty_t_INITIALIZER;
   
  retcode = DDS_DomainParticipant_get_default_flowcontroller_property(
      participant, &property);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to get default flow controller property\n");
  }
   
  /* optionally modify flow controller property values */
   
  controller = DDS_DomainParticipant_create_flowcontroller(
      participant, "my flow controller name", &property);
   
  if (controller == NULL) {
      printf("***Error: failed to create flow controller\n");
  }

Flow controlling a data writer

• Set up participant
• Create flow controller
• Create an asynchronous data writer, FooDataWriter, of user data type Foo:

  struct DDS_DataWriterQos writer_qos = DDS_DataWriterQos_INITIALIZER;
  DDS_DataWriter* writer;
  struct DDS_DataWriterListener writer_listener =
      DDS_DataWriterListener_INITIALIZER;
   
  /* MyWriterListener_* functions are user defined to match
     DDS_DataWriterListener functions */
  writer_listener.on_offered_deadline_missed =
      MyWriterListener_OfferedDeadlineMissed;
  writer_listener.on_offered_incompatible_qos =
      MyWriterListener_OfferedIncompatibleQos;
  writer_listener.on_liveliness_lost = MyWriterListener_LivelinessLost;
  writer_listener.on_publication_matched = MyWriterListener_PublicationMatch;
   
  retcode = DDS_Publisher_get_default_datawriter_qos(publisher, &writer_qos);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to get default datawriter qos\n");
  }
   
  /* Change the writer QoS to publish asnychronously */
  writer_qos.publish_mode.kind = DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS;
   
  /* Setup to use the previously created flow controller */
  writer_qos.publish_mode.flow_controller_name = 
      DDS_String_dup("my flow controller name");
   
  /* Samples queued for asynchronous write are subject to the History Qos policy */
  writer_qos.history.kind = DDS_KEEP_ALL_HISTORY_QOS;
   
  writer = DDS_Publisher_create_datawriter(publisher, 
                                           topic, 
                                           &writer_qos, 
                                           &writer_listener /* or NULL */,
                                           DDS_STATUS_MASK_ALL);
   
  if (writer == NULL) {
      printf("***Error: failed to create writer\n");
  }
   
  /* Send data asynchronously... */
   
  /* Wait for asynchronous send completes, if desired */
  retcode = DDS_DataWriter_wait_for_asynchronous_publishing(writer, &timout);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to wait for asynchronous publishing\n");
  }

Using the built-in flow controllers

RTI Connext provides several built-in flow controllers.

The DDS_DEFAULT_FLOW_CONTROLLER_NAME built-in flow controller provides the basic asynchronous writer behavior. When calling FooDataWriter_write, the call signals the DDS_Publisher asynchronous publishing thread (DDS_PublisherQos::asynchronous_publisher) to send the actual data. As with any DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS DDS_DataWriter, the FooDataWriter_write call returns immediately afterwards. The data is sent immediately in the context of the DDS_Publisher asynchronous publishing thread.

When using the DDS_FIXED_RATE_FLOW_CONTROLLER_NAME flow controller, data is also sent in the context of the DDS_Publisher asynchronous publishing thread, but at a regular fixed interval. The thread accumulates samples from different DDS_DataWriter instances and generates data on the wire only once per DDS_FlowControllerTokenBucketProperty_t::period.

In contrast, the DDS_ON_DEMAND_FLOW_CONTROLLER_NAME flow controller permits flow only when DDS_FlowController_trigger_flow is called. The data is still sent in the context of the DDS_Publisher asynchronous publishing thread. The thread accumulates samples from different DDS_DataWriter instances (across any DDS_Publisher) and sends all data since the previous trigger.

The properties of the built-in DDS_FlowController instances can be adjusted.

• Set up participant
• Lookup built-in flow controller

  DDS_FlowController *controller = NULL;
   
  controller = DDS_DomainParticipant_lookup_flowcontroller(
      participant, DDS_DEFAULT_FLOW_CONTROLLER_NAME);
   
  /* This should never happen, built-in flow controllers are always created */
  if (controller == NULL) {
      printf("***Error: failed to lookup flow controller\n");
  }

• Change property of built-in flow controller, if desired

  DDS_ReturnCode_t retcode;
  struct DDS_FlowControllerProperty_t property = DDS_FlowControllerProperty_t_INITIALIZER;
   
  /* Get the property of the flow controller */
  retcode = DDS_FlowController_get_property(controller, &property);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to get flow controller property\n");
  }
   
  /* Change the property value as desired */
  property.token_bucket.period.sec = 2;
  property.token_bucket.period.nanosec = 0;
   
  /* Update the flow controller property */
  retcode = DDS_FlowController_set_property(controller, &property);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to set flow controller property\n");
  }

• Create a data writer using the correct flow controller name

Shaping the network traffic for a particular transport

• Set up participant
• Create the transports
• Create a separate flow controller for each transport
• Configure DDS_DataWriter instances to only use a single transport
• Associate all data writers using the same transport to the corresponding flow controller
• For each transport, the corresponding flow controller limits the network traffic based on the token bucket properties

Coalescing multiple samples in a single network packet

• Set up participant
• Create a flow controller with a desired token bucket period
• Associate the data writer with the flow controller
• Multiple samples written within the specified period will be coalesced into a single network packet (provided that tokens_added_per_period and bytes_per_token permit).