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;
  DDSFlowController *controller = NULL;
  DDS_FlowControllerProperty_t property;
   
  retcode = participant->get_default_flowcontroller_property(property);
   
  if (retcode != DDS_RETCODE_OK) {
      printf("***Error: failed to get default flow controller property\n");
  }
   
  // optionally modify flow controller property values
   
  controller = participant->create_flowcontroller(
      "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:

  DDS_DataWriterQos writer_qos;
  DDS_ReturnCode_t retcode;
   
  // MyWriterListener is user defined and 
  // extends DDSDataWriterListener
  MyWriterListener* writer_listener = new MyWriterListener(); // or = NULL
   
  retcode = publisher->get_default_datawriter_qos(writer_qos);
   
  if (retcode != DDS_RETCODE_OK) {
      // ... error 
  }
   
  /* 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;
   
  FooDataWriter* writer = publisher->create_datawriter(topic,
                                             writer_qos, 
                                             writer_listener,
                           DDS_STATUS_MASK_ALL);
  if (writer == NULL) {
      // ... error 
  };
   
  /* Send data asynchronously... */
   
  /* Wait for asynchronous send completes, if desired */
  retcode = writer->wait_for_asynchronous_publishing(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 DDSPublisher asynchronous publishing thread (DDS_PublisherQos::asynchronous_publisher) to send the actual data. As with any DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS DDSDataWriter, the FooDataWriter::write call returns immediately afterwards. The data is sent immediately in the context of the DDSPublisher asynchronous publishing thread.

When using the DDS_FIXED_RATE_FLOW_CONTROLLER_NAME flow controller, data is also sent in the context of the DDSPublisher asynchronous publishing thread, but at a regular fixed interval. The thread accumulates samples from different DDSDataWriter 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 DDSFlowController::trigger_flow is called. The data is still sent in the context of the DDSPublisher asynchronous publishing thread. The thread accumulates samples from different DDSDataWriter instances (across any DDSPublisher) and sends all data since the previous trigger.

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

• Set up participant
• Lookup built-in flow controller

  DDSFlowController *controller = NULL;
   
  controller = participant->lookup_flowcontroller(
      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;
  DDS_FlowControllerProperty_t property;
   
  /* Get the property of the flow controller */
  retcode = controller->get_property(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 = controller->set_property(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 DDSDataWriter 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).