FlowController Use Cases

Working with flow controllers. More...

Working with flow controllers.

Creating a flow controller

• Set up participant

• Create a flow controller

  FlowController controller = null;
  FlowControllerProperty_t property = new FlowControllerProperty_t();
  
  retcode = participant.get_default_flowcontroller_property(property);
  
  // optionally modify flow controller property values
  
  try {
      controller = participant.create_flowcontroller(
          "my flow controller name", property);
  } catch (RETCODE_ERROR err) {
      // handle exception
  }

Flow controlling a data writer

• Set up participant

• Create flow controller

• Create an asynchronous data writer, FooDataWriter, of user data type Foo:

  DataWriterQos writer_qos = new DataWriterQos();
  
  // MyWriterListener is user defined and 
  // implements DataWriterListener
  MyWriterListener writer_listener = new MyWriterListener(); // or = null
  
  publisher.get_default_datawriter_qos(writer_qos);
  
  /* Change the writer QoS to publish asnychronously */
  writer_qos.publish_mode.kind = PublishModeQosPolicyKind.ASYNCHRONOUS_PUBLISH_MODE_QOS;
  
  /* Setup to use the previously created flow controller */
  writer_qos.publish_mode.flow_controller_name = "my flow controller name";
  
  /* Samples queued for asynchronous write are subject to the History Qos policy */
  writer_qos.history.kind = HistoryQosPolicyKind.KEEP_ALL_HISTORY_QOS;
  
  FooDataWriter writer = null;
  try {
      writer = (FooDataWriter) publisher.create_datawriter(topic, writer_qos,
                                                           writer_listener,
                                                           StatusKind.STATUS_MASK_ALL);
  
  
      /* Send data asynchronously... */
      
      /* Wait for asynchronous send completes, if desired */
      writer.wait_for_asynchronous_publishing(timout);
  
  } catch (RETCODE_ERROR err) {
      // handle exception
  }

Using the built-in flow controllers

RTI Connext provides several built-in flow controllers.

The com.rti.dds.publication.FlowController.DEFAULT_FLOW_CONTROLLER_NAME built-in flow controller provides the basic asynchronous writer behavior. When calling com.rti.ndds.example.FooDataWriter.write, the call signals the com.rti.dds.publication.Publisher asynchronous publishing thread (com.rti.dds.publication.PublisherQos.asynchronous_publisher) to send the actual data. As with any com.rti.dds.infrastructure.PublishModeQosPolicyKind.PublishModeQosPolicyKind.ASYNCHRONOUS_PUBLISH_MODE_QOS com.rti.dds.publication.DataWriter, the com.rti.ndds.example.FooDataWriter.write call returns immediately afterwards. The data is sent immediately in the context of the com.rti.dds.publication.Publisher asynchronous publishing thread.

When using the com.rti.dds.publication.FlowController.FIXED_RATE_FLOW_CONTROLLER_NAME flow controller, data is also sent in the context of the com.rti.dds.publication.Publisher asynchronous publishing thread, but at a regular fixed interval. The thread accumulates samples from different com.rti.dds.publication.DataWriter instances and generates data on the wire only once per com.rti.dds.publication.FlowControllerTokenBucketProperty_t.period.

In contrast, the com.rti.dds.publication.FlowController.ON_DEMAND_FLOW_CONTROLLER_NAME flow controller permits flow only when com.rti.dds.publication.FlowController.trigger_flow is called. The data is still sent in the context of the com.rti.dds.publication.Publisher asynchronous publishing thread. The thread accumulates samples from different com.rti.dds.publication.DataWriter instances (across any com.rti.dds.publication.Publisher) and sends all data since the previous trigger.

The properties of the built-in com.rti.dds.publication.FlowController instances can be adjusted.

• Set up participant

• Lookup built-in flow controller

  FlowController controller = null;
  
  try {
      controller = participant.lookup_flowcontroller(
          FlowController.DEFAULT_FLOW_CONTROLLER_NAME);
  } catch (RETCODE_ERROR err) {
      // This should never happen, built-in flow controllers are always created
      // handle exception
  }

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

  FlowControllerProperty_t property = new FlowControllerProperty_t();
  
  /* Get the property of the looked-up default flow controller */
  controller.get_property(property);
  
  /* Change the property value as desired */
  property.token_bucket.period.sec = 2;
  property.token_bucket.period.nanosec = 0;
  
  /* Update the flow controller property */
  controller.set_property(property);

• 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 com.rti.dds.publication.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).