Telemetry Subscriber

Overview

The purpose of the Telemetry Subscriber is to receive samples published from one or more Telemetry Republishers for various DDS Topics, correlate these samples, and then write them into a shared memory queue for subsequent consumption by Data Tasks on the same node. Samples from the various DDS topics are correlated based on their sample ID, i.e. DDS samples with the same sample ID are matched together. The payloads for the matched samples are extracted and combined into a single, user-defined data structure using a user-supplied function, called the blender function. Each entry in the shared memory queue therefore contains data captured at the same loop cycle.

Since the topic data structure for the shared memory queue must be provided by the RTC Toolkit user, it is not possible to have a fully working, pre-compiled and linked executable for the Telemetry Subscriber. The RTC Toolkit provides a template version of the Telemetry Subscriber in a library. A working Telemetry Subscriber must be instantiated using the user-provided topic structure, and compiled and linked against the librtctkTelemetrySubscriber.so library.

The idea is to provide an almost ready application that is a SRTC component and delivers all the boilerplate code for reading from DDS and correlating samples. The user then only needs to concentrate on defining the shared memory topic structure and a blender function that constructs the topic from correlated DDS samples.

The following section describes a Telemetry Subscriber that needs to be instantiated into an actual working application, how the application can be configured and how it is expected to behave.

Prerequisites

The Telemetry Subscriber library has the following external dependencies:

FastDDS - Provides the communication services (DDS implementation) to receive data from a Telemetry Republisher.
ipcq - Provides the shared memory queue into which correlated DDS samples are written for the downstream Data Tasks.
NUMA++ - This utility library is used to optionally control the NUMA node where shared memory is allocated, thread scheduling and pinning policies. This allows runtime performance optimisations.

If the RTC Toolkit has been successfully built with the Waf build system, the above dependencies should already be fulfilled.

It is assumed that the installation of the RTC Toolkit is performed into the INTROOT directory and that the environment variable INTROOT is defined (see section Installation). If this is not the case, any reference to INTROOT in the following sections needs to be properly adjusted.

Although a Telemetry Subscriber is able to start with default settings for FastDDS, to properly use any Telemetry Subscriber instance together with a Telemetry Republisher, one should use the dedicated FastDDS XML QoS file supplied by the RTC Toolkit. The default file is telemDataPathDdsQos.xml installed in $INTROOT/resource/config/rtctk/dds/. Actually the file can be put anywhere under rtctk/dds/ in $CFGPATH. The file name can be set in the component’s configuration dds_qos_file (see Configuration section for details).

Note

FASTDDS_DEFAULT_PROFILES_FILE environment variable shall not be used (set) and used in particular if the file that the variable points to contains the same QoS profiles.

In addition, the Telemetry Subscriber configuration must be prepared so that it uses the correct profile that is defined in the XML file. These configuration settings for dds_qos_profile are indicated in the Configuration section below.

Customisation

The only customisation currently available for a Telemetry Subscriber instance is to specify the shared memory topic type and provide a blender function that will appropriately construct the shared memory super-topic from correlated input DDS samples. It is in fact mandatory to instantiate a Telemetry Subscriber with a user-supplied topic and blender method to get a working Telemetry Subscriber. It is also the user’s responsibility to ensure that the topic is the appropriate one, as expected by downstream Data Tasks, and that the blender function assembles the topic in shared memory from the individual DDS samples correctly.

Shared Memory Super-Topic

A user has almost full freedom to define the shared memory super-topic structure. The structure is primarily driven by the input data format requirements of the Data Tasks and/or Telemetry Recorders.

It is mandatory to include a Sample ID field with the name sample_id, which is checked at compile-time, so that this information is propagated downstream.

It is also important to maintain each sample’s time information throughout the entire telemetry path, from its source to consumption in the SRTC, e.g. Data Task and/or Telemetry Recorder. In the case of the Telemetry Subscriber, the timestamps received from DDS topics need to be preserved. Therefore it is mandatory that all Shared Memory Topics contain at least one timestamp for all the DDS topics, or a list of timestamps (e.g. std::array), one for each DDS topic.

A simple example for a topic structure could be as follows:

struct MyTopic {
    uint32_t sample_id;   // mandatory field: for sample ID needs to be named time_stamp
    taiclock::TaiClock::time_point time_stamp;    // mandatory field: for timestamp (at least one)
    float vector_data[1024];
};

Note

The topic structure must be a continuous flat structure in memory. It cannot have pointers to other memory blocks. This will not be readable in the Data Tasks and most likely cause segmentation violations or corruption. Any complex structure with pointers to other memory regions will have to be serialised into a flat topic structure.

Ideally this should be declared in a header file that is shared with the corresponding Data Task that will process the topic’s data.

The Blender Function

A function used to construct the user-defined topic in shared memory must be provided with the following signature:

// Signature as a normal function:
std::error_code Blender(const rtctk::telSub::DataSamplesView& dds_samples,
                        MyTopic& shm_sample) noexcept {
    // ... user code goes here ...
}

// Signature as a lambda function:
auto blender = [](const rtctk::telSub::DataSamplesView& dds_samples,
                  MyTopic& shm_sample) noexcept -> std::error_code {
    // ... user code goes here ...
}

Two alternative declarations are shown above, the first using the typical function declaration syntax and the second showing the blender function declared as a lambda. Choosing one style over the other is at the user’s discretion.

The first input argument must be the list of correlated DDS samples. Refer to the API reference documentation for details about the structure of DataSamplesView.

The second output argument must be the topic structure previously declared by the user. This is a reference to the region in the shared memory that the data should be written to.

The return type of the blender function must be an error code. When the user topic structure has been successfully constructed it must return a zero or empty error code, indicating success, e.g. return {};. In the case that the topic could not be constructed an appropriate error code should be returned. The suggested error code to use at the moment is std::errc::bad_message, to indicate there is something wrong with the format of the DDS samples.

Note

The blender function must not throw any exceptions. Throwing an exception will cause the processing thread to call std::terminate, which will cause the Telemetry Subscriber component to abort. This means that all error conditions must be indicated by returning an appropriate error code. If any code is executed that could throw an exception, this must be caught in a try..catch block and converted to an error code instead. E.g. this includes functions like std::map::at which can throw std::out_of_range.

The code for the body of the blender function must be provided by the user, since it will depend on the input DDS sample topics and the user provided shared memory topic structure. Typically this only involves copying the data from the DDS sample buffers to the correct location within the shared memory. In the most trivial case this can simply be a call to std::memcpy.

Note

Since the blender function is executed in the DDS reading thread, which is time critical, minimal amount of work should be performed in the blender function. Only basic sanity checks should be applied to the input data and the minimal code to copy the data into the shared memory topic should be provided. Any complex computation should be done in a Data Task.

Instantiation

To instantiate a Telemetry Subscriber with a user-defined super-topic, one needs to prepare a custom application. The minimal wscript build configuration file to instantiate an application called myTelSub would be similar to the following:

from wtools.module import declare_cprogram

declare_cprogram(
    target="myTelSub",
    use="rtctk.reusableComponents.telSub.lib"
)

As can be seen, it only needs to depend on rtctk.reusableComponents.telSub.lib. If the user-defined shared memory topic is defined in a separate module, then this additional module would also have to be added to the use argument.

The entrypoint declaration for this example myTelSub application, i.e. the contents of the main.cpp file, should look similar to the following:

#include <rtctk/telSub/main.hpp>
#include "myTopic.hpp"

void RtcComponentMain(const rtctk::componentFramework::Args& args) {
    auto blender = [](const rtctk::telSub::DataSamplesView& dds_samples,
                      MyTopic& shm_sample) noexcept -> std::error_code {
        // ... user code for the blender goes here ...
        return {};
    };
    rtctk::telSub::Main<MyTopic>(args, std::move(blender));
}

Since any Telemetry Subscriber is an SRTC component, we must declare the entrypoint with RtcComponentMain, rather than with just int main(int argc, const char* argv[]). We also rely on the template function rtctk::telSub::Main to perform the setup specific to a Telemetry Subscriber and enter the processing loop.

Refer to the Customise a Telemetry Subscriber tutorial to see a complete working example of a Telemetry Subscriber instantiation.

Configuration

All Telemetry Subscriber components will accept the configuration parameters described in this section. These are all read from the Runtime Configuration Repository during initialisation when the Init command is received.

The configuration parameters are divided into groups as indicated in the sections below. Optional parameters are indicated in the description.

DDS Parameters

Configuration Path	Type	Description
`static/dds/domain_id`	`!cfg.type:int32`	(optional) DDS domain identifier to subscribe to. If not specified, domain id 0 is used.
`static/dds/qos_profile`	`cfg.type:string`	(optional) The name of the profile to use from the QoS configuration XML file. If not specified, default `RtcTk_Default_Profile` FastDDS QoS parameters are used.
`static/dds/qos_file`	`cfg.type:string`	(optional) Name of DDS QoS XML file. The file should be found in $CFGPATH under rtctk/dds/. If not specified the default name is used: telemDataPathDdsQos.xml.
`static/dds/interface_white_list`	`cfg.type:vector_string`	(optional) list of network interfaces to be used by DDS. If not specified any interface may be used.
`static/dds_topics/topic_name/multicast_address`	`cfg.type:string`	multicast_address to add to locator list for DDS. Can be empty string. The topic_name in the path needs to correspond to the dds_topic. This setting has to be present for each topic.

The following is an example of the above configuration parameters in a YAML file:

static:
    dds:
        domain_id: !cfg.type:int32 123
        qos_profile: !cfg.type:string RtcTk_Default_Profile
        qos_file: !cfg.type:string telemDataPathDdsQos.xml
    dds_topics: !cfg.type:vector_string
        SlopesTopic:
            multicast_address: !cfg.type:string 224.0.0.9
        IntensitiesTopic:
            multicast_address: !cfg.type:string ""
        CommandsTopic:
            multicast_address: !cfg.type:string "224.0.0.5"

Shared Memory Queue Parameters

Configuration Path	Type	Description
`static/shm_topic_name`	`RtcString`	The name of the shared memory queue, i.e. this corresponds to the file name `/dev/shm/ipcq-<topic-name>`.
`static/shm_capacity`	`RtcInt64`	Maximum number of samples that can be stored in the shared memory queue.
`static/shm_memory_policy_mode`	`RtcString`	(optional) Memory allocation policy to apply to the shared memory. It can be one of the values indicated in Table 2.
`static/shm_memory_policy_nodes`	`RtcString`	(optional) Mask of NUMA nodes to which the memory policy is applied. See the `numa_parse_nodestring` function in the numa(3) manpage for details about the correct format of this string.

The following is an example of the above configuration parameters in a YAML file:

static:
    shm_topic_name: !cfg.type:string mytopic
    shm_capacity: !cfg.type:int64 1024
    shm_memory_policy_mode: !cfg.type:string Bind
    shm_memory_policy_nodes: !cfg.type:string "1,2"

Operational Logic Parameters

Configuration Path	Type	Description
`static/close_detach_delay`	`RtcInt32`	(optional) Delay in milliseconds before receiving the signal to stop writing to the shared memory queue and closing, i.e. destroying the shared memory writer. This provides an opportunity for the shared memory queue readers to detach gracefully. If not specified, the default value is 0 ms.
`static/correlator_poll_timeout`	`RtcInt32`	(optional) Time in milliseconds to wait for receiving and correlating DDS input samples, before considering the operation to have timed out. If a timeout occurs, the error will be indicated, but the Telemetry Subscriber component will not enter the error state. If not specified, the default value is 100 ms.
`static/monitor_report_interval`	`RtcInt32`	(optional) Time in milliseconds, which determines how often the monitoring thread wakes up to report errors from the processing thread. If not specified, the default value is 500 ms.
`static/processing_thread_policies`	NUMA Policies	(optional) Defines NUMA policies for the processing thread.
`static/monitoring_thread_policies`	NUMA Policies	(optional) Defines NUMA policies for the monitoring thread.
`static/expected_sample_id_increment`	`RtcUInt32`	(optional) Expects a specific sample_id increment (useful for sub-sampled telemetry topics). If not specified, the default value is 1.

The following is an example of the above configuration parameters in a YAML file:

static:
    close_detach_delay: !cfg.type:int32 1000
    correlator_poll_timeout: !cfg.type:int32 200
    monitor_report_interval: !cfg.type:int32 1000
    processing_thread_policies:
        cpu_affinity: !cfg.type:string "1-4"
        scheduler_policy: !cfg.type:string Fifo
        scheduler_priority: !cfg.type:int32 10
        memory_policy_mode: !cfg.type:string Bind
        memory_policy_nodes: !cfg.type:string "1-4"
    monitoring_thread_policies:
        # Monitoring thread is executed with lower priority
        scheduler_policy: !cfg.type:string Other
        scheduler_priority: !cfg.type:int32 -1

Commands

Telemetry Subscriber instances can be steered by sending commands to their provided CII MAL interfaces.

The set of commands is indicated in the following table, together with the expected behaviour:

Command	Behaviour
Init	Constructs objects to read DDS input samples; constructs shared memory writer; spawns low level processing and monitoring threads; starts reading and correlating DDS samples (no publishing to SHM yet, plus read errors and timeouts are ignored).
Exit	Terminates the process.
Stop	Stops the initialisation procedure started by the Init command.
Reset	Resets the state of the component to the state before the Init command was received. This means that all DDS objects are destroyed, shared memory is released, and the processing and monitoring threads are also destroyed.
Enable	No particular actions are performed here.
Disable	No particular actions are performed here.
Update	Reloads dynamic component configuration.
Run	Requests transition to On::Operational::Running which: Turns on writing to the shared memory queue. Clears error counters and last error in transition to Running. Errors are counted and reported by the monitoring thread, but attempts to correlate and write to shared memory will continue. This means, errors lead to missing samples in the shared memory queue.
Idle	Requests transition to On::Operational::Idle which: Turns off writing to the shared memory queue. Ignores errors and monitoring thread will stop logging. Restarts correlation with new data samples if bad data samples are encountered automatically.

Errors

Not Operational

The main errors that can currently occur during component initialisation when it receives the Init command are due to missing mandatory configuration parameters. The following is an example of the log message for such an error when dds_topics is missing:

[13:52:21:730][ERROR][rtctk] Failed to load configuration from the Runtime Repository: Path '/tel_sub_1/static/dds_topics' does not exist.
[13:52:21:730][ERROR][rtctk] Activity.Initialising: failed, exception: Path '/tel_sub_1/static/dds_topics' does not exist.

For cases where the wrong data type was used in a type field in the YAML file, the following example error message would be seen in the logs:

[00:02:49:597][ERROR][rtctk] Activity.Initialising: failed, exception: Wrong type used to read data point.

If the YAML file format is not correct, e.g. a value field has the wrong format, the following example error message would be seen in the logs:

[00:03:54:061][ERROR][rtctk] Activity.Initialising: failed, exception: File '/home/eltdev/test_install/run/exampleTelSub/runtime_repo/tel_sub_1.yaml' has an invalid data format.

Operational

While the Telemetry Subscriber is running, the main sources of errors are:

DDS subscription.
Problem with received data samples.

If timeout errors occur, it may indicate that the DDS publisher is no longer working, or the timeout threshold is too low for the sample rate and needs to be adjusted (see Operational Logic Parameters). In such cases, the following error messages will be seen in the logs:

[13:49:59:480][ERROR][rtctk] Detected errors during wait for DDS: [Last error code = 10: Timeout Error. Total number of errors = 1]

If data is received but cannot be correlated because the sample id is wrong, it is indicated with the following log message:

[11:37:27:388][ERROR][rtctk] Unexpected Sample Id. Expected 9 but got 10 for topic 'example_topic'. Total number of errors = 43

In both cases, the Telemetry Subscriber attempts to auto-recover by resetting the correlator and starting over again.

Note

This automatic re-synchronisation leads to missing samples in the shared memory queue and should be avoided by carefully dimensioning the network and configuring the DDS QoS file.

For failures when writing to the shared memory queue, messages about “Detected errors in SHM Publisher” will be logged with the error code received from the sub-system, similar to the following:

[13:48:44:448][ERROR][rtctk] Detected errors in SHM Publisher: error code = 74: Bad message. [Total number of errors = 1]

When the blender function returns a non-default error_code, the last raised error_code will be logged with the following message:

[14:43:24:385][ERROR][rtctk] Detected errors in Blender: error code: 74: Bad message. [Total number of errors = 5]

Alerts

The following alerts are used in Telemetry Subscriber.

Note

Latched means that the alert will be set active if the respective problem occurs in the Running state. The alert remains active until it is explicitly cleared together with other alerts using ClearAlerts command or a Run -> Idle or Idle -> Run transition is made.

Table 10 Alert conditions
Alert Id	Latched	Description
`dds`	Yes	Indicates a problem on the DDS subscription side, meaning problems to receive the telemetry data over DDS.
`correlator`	Yes	Indicates a problem when correlating the telemetry streams.
`blender`	Yes	The blender function returned an error code.
`shm`	Yes	Indicates a problem writing to shared memory (for example when the queue is closed).

Component Metrics

For each Telemetry Subscriber the following performance counter metrics are periodically published under the OLDB path COMPONENT_ID/metrics/counter/.

Table 11 Performance Counter Metrics
OLDB Path	Type	Description
`frequency_estimate`	`RtcDouble`	Estimated frequency of the processing loop [Hz].
`correlator/duration_estimate`	`RtcInt64`	Duration of DDS reception and correlation [us].
`dds_samples_correlated`	`RtcInt64`	Number of DDS samples successfully received and correlated.
`dds_errors`	`RtcInt64`	Number of DDS reception errors.
`correlation_errors`	`RtcInt64`	Number of correlation errors.
`shm_samples_published`	`RtcInt64`	Number of samples successfully blended and published to SHM.
`shm_publication_errors`	`RtcInt64`	Number of SHM publication errors.
`blender_errors`	`RtcInt64`	Number of blender errors.
`last_sample_id_published`	`RtcInt64`	Last sample id that was successfully published to SHM queue.