SObjectizer Tales - 25. Runtime telemetry
Tags: cpp, sobjectizer, softwareengineering, actormodel
Published: 2024-03-28 09:04:01

SObjectizer features a basic mechanism for gathering and distributing runtime telemetry data regarding its internals, such as the number of registered cooperations, the quantity of delayed or periodic messages, the length of event queues, the time spent processing events for each handler, and so on.

This feature is essential for getting insights into the **performance and health of our applications at runtime**, avoiding the necessity to develop our own monitoring system.

Following up our recent discussion about performance metrics, Helen is back with her discoveries about this feature of SObjectizer that is called [Runtime monitoring](https://github.com/Stiffstream/sobjectizer/wiki/SO-5.8-InDepth-Runtime-Monitoring). In this article we’ll learn how to use this tool effectively and we’ll apply it to `calico`.

#### Runtime monitoring in a nutshell

Runtime monitoring is seamlessly integrated into SObjectizer through conventional mechanisms: **message passing and subscriptions**.

Essentially, when the feature is enabled, SObjectizer automatically generates certain **data sources** that observe specific internal aspects of SObjectizer and generate corresponding metrics. For example, there is a data source for counting the number of cooperations registered within the environment. These data sources transmit their information to a dedicated component of the telemetry system known as the **stats controller**. The stats controller manages the activation and deactivation of telemetry data collection and distribution. When the telemetry is enabled, the stats controller runs in a separate thread. To distribute data, the stats controller grants access to a specialized message box that operates in the standard manner via subscriptions. It’s worth noting that telemetry samples are not continuously distributed; instead, the **distribution period** can be configured, as we’ll see in a moment.

Practically speaking, to collect telemetry data, we must explicitly turn on the stats controller and, optionally, set a distribution period – otherwise the default value is 2 seconds.

The stats controller is accessed through the environment:

```cpp
// turn on telemetry
so_environment().stats_controller().turn_on();
// send telemetry data every second
so_environment().stats_controller().set_distribution_period(1s);
```

Since these options can be changed anytime, we introduce an agent that enables or disables telemetry collection depending on the current state, when a specific command is received (e.g. pressing `t`):

```cpp
struct enable_telemetry_command final : so_5::signal_t {};

class telemetry_agent final : public so_5::agent_t
{
public:
	telemetry_agent(so_5::agent_context_t ctx)
		: agent_t(ctx), m_commands(so_environment().create_mbox("commands"))
	{
	}

	void so_define_agent() override
	{
		st_turned_off.event(m_commands, [this](so_5::mhood_t<enable_telemetry_command>) {		
			so_environment().stats_controller().turn_on();
			so_environment().stats_controller().set_distribution_period(1s);
			st_turned_on.activate();
		});
		st_turned_on.event(m_commands, [this](so_5::mhood_t<enable_telemetry_command>) {		
			so_environment().stats_controller().turn_off();
			st_turned_off.activate();
		});

		st_turned_off.activate();
	}
private:
	state_t st_turned_on{ this }, st_turned_off{ this };
	so_5::mbox_t m_commands;
};
```

At this point, we are ready to subscribe for telemetry data.

#### How to subscribe for telemetry data

Runtime telemetry is delivered as messages defined into the namespace `stats::messages`. In particular, this namespace defines a message type representing a generic **quantity** of data:

```cpp
stats::messages::quantity<T>
```

This message type is intended to convey information regarding quantities of a certain telemetry entity, such as the number of working threads for a dispatcher or the quantity of registered cooperations. In essence, when subscribing for telemetry data, we will receive a `quantity<T>`.

You might expect that `T` represents the type of telemetry information we are interested in. After all, one of the key advantages of SObjectizer is “message pattern matching”. However, `T` solely denotes the “unit type” associated with such a quantity, such as `size_t`. For instance, to obtain information about the number of working threads for a dispatcher, we receive a `quantity<size_t>`. This type is essentially hardcoded, indicating the type of data being transmitted, and should be retrieved by the documentation. At the time of writing, the only unit type available is `size_t`.

At this point, you might wonder **how to differentiate between various telemetry types**, like the count of registered cooperations and the number of working threads for a dispatcher, if they both arrive as `quantity<size_t>`. The answer lies in the quantity type itself: it includes not only the numerical “value” of the quantity but also carries information about the **data** **source** that generated it.

In particular, data sources have **unique names**, each consisting of two strings:

*   a **prefix**, indicating a “group” of related data sources;
*   a **suffix**, indicating the specific data source of the group which produced the quantity.

For example, here below:

mbox_repository/named_mbox.count

`mbox_repository` is the prefix, whereas `/named_mbox.count` is the suffix.

Prefixes could be considered “dynamic” in nature, indicating that they may reference entities generated at runtime. For instance, consider these two distinct prefixes identifying two different `one_thread` dispatchers:

```
disp/ot/DEFAULT
disp/ot/0x3be520
```

`0x3be520` uniquely identifies a particular dispatcher and can only be generated – then referenced – at runtime.

Conversely, suffixes are “fixed” strings, signifying that they pertain to predetermined **categories** of information within SObjectizer. For example:

```
/agent.count
/demands.count
/named_mbox.count
```

Therefore, when receiving a `quantity<T>`, we might filter the “type” of information by examining its prefix and suffix. To simplify this process, SObjectizer furnishes all “predefined” prefixes and suffixes in `stats::prefixes` and `stats::suffixes` namespaces.

Although prefixes for dispatcher-related information are not supplied since they are dynamic, the [rules](https://github.com/Stiffstream/sobjectizer/wiki/SO-5.8-InDepth-Runtime-Monitoring#information-from-dispatchers) regulating their generation are rather straightforward:

*   every prefix starts with `disp`;
*   then it follows an abbreviated form based on the specific dispatcher, such as:
    *   `/ot` — for one_thread dispatcher;
    *   `/ao` — for active_obj dispatcher;
    *   `/ag` — for active_group dispatcher;
    *   `/tp` — for thread_pool dispatcher;
    *   `/atp` — for adv_thread_pool dispatcher;
*   finally, the last part is a **unique identification** for a dispatcher instance that typically includes the dispatcher name (when specified) or an hex representation of the address of the dispatcher object.

It’s worth noting that all standard dispatcher factories (e.g. `make_dispatcher()`) accept a parameter to specify the name of the dispatcher.

Now, suppose we enhance our `telemetry_agent` to record statistics concerning our `active_obj` dispatcher, which is bound to the majority of our agents. Hence, we should subscribe to the `stats_controller`‘s message box and then we should only consider quantities of a specified `active_obj` dispatcher. First of all, we add a name to the dispatcher when we create the cooperation:

```cpp
sobjectizer.environment().introduce_coop(
	so_5::disp::active_obj::make_dispatcher(sobjectizer.environment(), "calico_active_object").binder(), [&](so_5::coop_t& c) {
	// ...
});
```

Then, we install a **delivery filter** to discard any other telemetry data (one motivation for introducing delivery filters in SObjectizer was to facilitate discarding of uninteresting telemetry quantities):

```cpp
so_set_delivery_filter(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::quantity<size_t>& q) {
	return std::string_view(q.m_prefix.c_str()).contains("/ao/calico_active_object");
});
```

We opted for wrapping the prefix into a `string_view` since SObjectizer provides prefix and suffix as raw char pointers. Update: [SObjectizer 5.8.2](https://github.com/Stiffstream/sobjectizer/wiki/v.5.8.2) provides a function `as_string_view()` for both `prefix_t` and `suffix_t`. Finally, we add the subscription:

```cpp
st_turned_on.event(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::quantity<size_t>& q) {
	std::osyncstream(std::cout) << q.m_prefix << q.m_suffix << "=" << q.m_value << "\n";
});
```

Another common way to filter quantities involves the suffix, allowing retrieval of data from all sources that produce a specific category of telemetry information. For instance, suppose we monitor the queue size of each dispatcher (the number of events awaiting processing). The suffix to verify against can be located in `stats::suffixes::work_thread_queue_size()`:

```cpp
so_set_delivery_filter(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::quantity<size_t>& q) {
	return q.m_suffix == so_5::stats::suffixes::work_thread_queue_size();
});
```

Note that if we install multiple filters, only the last one is taken. Thus, to get `/ao/calico_active_object` OR any information about queue sizes, we might merge the two conditions as follows:

```cpp
so_set_delivery_filter(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::quantity<size_t>& q) {
	return
		std::string_view(q.m_prefix.c_str()).contains("/ao/calico_active_object")
		|| q.m_suffix == so_5::stats::suffixes::work_thread_queue_size();
});
```

Monitoring the queue size of each dispatcher is a significant metric to detect potential bottlenecks, akin to assessing throughput, as discussed in the [initial post about performance](https://dev.to/ilpropheta/sobjectizer-tales-20-performance-part-1-4gla). Especially noteworthy is the potential need for further investigation if the queue size shows a gradual increase over time.

The complete list of all available suffixes and prefixes is [published](https://github.com/Stiffstream/sobjectizer/wiki/SO-5.8-InDepth-Runtime-Monitoring#run-time-monitoring-in-sobjectizer-58) on the official documentation. We recommend referring to it to explore the additional telemetry data provided by SObjectizer.

#### Identifying start and end of each telemetry batch

You may have observed that the `telemetry_agent` receives one quantity for each telemetry data sent from the data sources. Then we typically set up some filters to refine the data to suit our specific interests. For example, suppose we have this setup of `calico`:

```cpp
int main()
{
	const auto ctrl_c = utils::get_ctrlc_token();

	const wrapped_env_t sobjectizer;
	const auto main_channel = sobjectizer.environment().create_mbox("main");
	const auto commands_channel = sobjectizer.environment().create_mbox("commands");
	const auto message_queue = create_mchain(sobjectizer.environment());

	sobjectizer.environment().introduce_coop(disp::active_obj::make_dispatcher(sobjectizer.environment(), "calico_active_object").binder(), [&](coop_t& c) {
		c.make_agent<image_producer_recursive>(main_channel, commands_channel);
		c.make_agent<maint_gui::remote_control>(commands_channel, message_queue);		
		c.make_agent<telemetry_agent>();
	});
	
	do_gui_message_loop(ctrl_c, message_queue, sobjectizer.environment().create_mbox(constants::waitkey_channel_name));		
}
```

Once the program is started and the ‘`t`‘ key is pressed in the `remote_control` window, logs similar to the following begin to appear:

```
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=5
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=4
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=4
```

Every line is a different quantity delivered to the `telemetry_agent` that passes through our filter. Thus, the corresponding message handler will be called 5 times. Nothing unexpected here.

Essentially, the stats controller generates a telemetry “batch” (called also “cycle” or “period”) occurring every second (as we configured before). At this point, it might be convenient to identify when a single batch starts and finishes, perhaps to enhance the clarity of the log. SObjectizer provides the necessary components to accomplish this task in an effective way. Specifically, the stats controller dispatches two “status” messages: one before sending the first quantity and another after dispatching the last quantity of each batch:

*   `messages::distribution_started`
*   `messages::distribution_finished`

Agents can subscribe to these two messages and do whatever they need. For example, we add these two subscriptions to our `telemetry_agent`:

```cpp
st_turned_on.event(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::distribution_started&) {
	std::osyncstream(std::cout) << "telemetry batch starts----\n";
});
st_turned_on.event(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::distribution_finished&) {
	std::osyncstream(std::cout) << "telemetry batch ends----\n";
});
```

With this modification, the log above might change to:

```
telemetry batch starts----
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=5
telemetry batch ends----
telemetry batch starts----
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=4
telemetry batch ends----
telemetry batch starts----
disp/ot/DEFAULT/wt-0/demands.count=0
disp/ao/calico_active_object/agent.count=3
disp/ao/calico_active_object/wt-0x1baa413f9f0/demands.count=0
disp/ao/calico_active_object/wt-0x1baa67c6170/demands.count=0
disp/ao/calico_active_object/wt-0x1baa68b27d0/demands.count=4
telemetry batch ends----
```

You might wonder why these are messages and not signals. Well, this is to allow SObjectizer developers to add some content to the message in the future.

It’s important to highlight that this follows a classical pattern in message passing-styled applications. We provided another example of this pattern in the previous post regarding the `image_cache`.

This covers nearly everything you need to begin working with telemetry! In the upcoming section, we’ll introduce another telemetry capability of SObjectizer that provides information about the time spent in dispatcher threads, essential to get a snapshot of the exhibited performance of one or more agents.

#### Observing thread activity

An additional category of telemetry data available, not provided as `quantity`, pertains to the time spent inside event-handlers and the time spent waiting for the next event in threads managed by dispatchers. This feature is not enabled by default to prevent any negative impact on dispatchers’ performance. Enabling this support must be done individually for a specific dispatcher or for the entire environment. For instance:

```cpp
auto disp = so_5::disp::active_obj::make_dispatcher( env, "my_disp",
  // Collecting of work thread activity is enabled
  // in the dispatcher's parameters
  disp_params_t{}.turn_work_thread_activity_tracking_on() 
);
```

Or globally:

```cpp
so_5::launch(
   []( so_5::environment_t & env ) {
      // ...
   }
   []( so_5::environment_params_t & params ) {
      // Enable work thread activity statistics collection explicitly
      params.turn_work_thread_activity_tracking_on();
      // ...
   } 
);
``` 

To enable support for all but certain dispatchers, we typically enable the feature globally while selectively disable it for some:

```cpp
so_5::launch(
   []( so_5::environment_t & env ) {
      // ...
      
	  // disable this one
      auto my_disp = so_5::disp::one_thread::make_dispatcher(
            env, "my_disp",
            so_5::disp::one_thread::disp_params_t{}
                 .turn_work_thread_activity_tracking_off() );
      // ...
   []( so_5::environment_params_t & params ) {
      // enable all the others
      params.turn_work_thread_activity_tracking_on();
      // ...
   }
);
```

The telemetry data related to thread activity are distributed using the message type:

stats::messages::work_thread_activity

In addition to prefix and suffix strings, this type contains three information:

*   `m_thread_id` that is valued with the thread identifier related to the data;
*   `m_working_stats` that contains information on the time spent in event handlers;
*   `m_waiting_stats` that contains information on the time spent waiting for new events.

`m_working_stats` and `m_waiting_stats` have type `so_5::stats::activity_stats_t` that contains:

*   `m_count` that is the number of events collected so far (this value won’t decrease);
*   `m_total_time` that is the total time spent for events so far (this value won’t decrease);
*   `m_avg_time` that is the average time spent for events so far.

In essence, the “working stats” object includes metrics regarding the time during which the specific thread was engaged in processing work, akin to the concept of “service time” we explored in a [previous post](https://dev.to/ilpropheta/sobjectizer-tales-20-performance-part-1-4gla). Conversely, the “waiting stats” object contains details about the duration spent awaiting new events (that is **not** the “waiting time” discussed before).

We add this subscription to our `telemetry_agent`:

```cpp
st_turned_on.event(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::work_thread_activity& q) {
			std::osyncstream(std::cout) << q.m_prefix.c_str() << q.m_suffix.c_str() << "[" << q.m_thread_id << "]\n"
					<< "  working: " << q.m_stats.m_working_stats << "\n"
					<< "  waiting: " << q.m_stats.m_waiting_stats << "\n";
		});
```

Let’s see this in action. Suppose we have this setup:

```cpp
int main()
{
	const auto ctrl_c = utils::get_ctrlc_token();

	const wrapped_env_t sobjectizer;
	const auto main_channel = sobjectizer.environment().create_mbox("main");
	const auto commands_channel = sobjectizer.environment().create_mbox("commands");
	const auto message_queue = create_mchain(sobjectizer.environment());

	sobjectizer.environment().introduce_coop(disp::active_obj::make_dispatcher(sobjectizer.environment()).binder(), [&](coop_t& c) {
		c.make_agent<image_producer_recursive>(main_channel, commands_channel);
		c.make_agent<maint_gui::remote_control>(commands_channel, message_queue);
		
		c.make_agent_with_binder<face_detector>(
			make_dispatcher(sobjectizer.environment(), "face_detector", 
				so_5::disp::active_obj::disp_params_t{}.turn_work_thread_activity_tracking_on().binder()), 
			main_channel);
			
		c.make_agent<telemetry_agent>();
	});
	
	do_gui_message_loop(ctrl_c, message_queue, sobjectizer.environment().create_mbox(constants::waitkey_channel_name));		
}
```

This way, we name only `face_detector`‘s dispatcher `face_detector`. Also, to keep “ordinary” telemetry data about our `face_detector` dispatcher, we change `telemetry_agent'`s filter this way:

```cpp
so_set_delivery_filter(so_environment().stats_controller().mbox(), [](const so_5::stats::messages::quantity<size_t>& q) {
	return std::string_view(q.m_prefix.c_str()).contains("/ao/face_detector");
});
```

Then we launch the program and we press ‘`t`‘ to start collecting. This is a possible log:

```
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x131d8f30020/demands.count=0
disp/ao/face_detector/wt-0x131d8f30020/thread.activity[27952]
  working: [count=1;total=0.0164ms;avg=0.0164ms]
  waiting: [count=2;total=543.154ms;avg=271.577ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x131d8f30020/demands.count=0
disp/ao/face_detector/wt-0x131d8f30020/thread.activity[27952]
  working: [count=1;total=0.0164ms;avg=0.0164ms]
  waiting: [count=2;total=1543.58ms;avg=771.79ms]
telemetry batch ends----
```

As evident, there are no events (`demands.count=0`) and the total time spent on active work remains **constant**. Also, both the working and waiting counters do not increase, but only the waiting time increases. In practical terms, this suggests that the thread is effectively idle. This aligns with our scenario, where we expect no active work from the `face_detector` until frames are received.

Then we press ‘Enter’ to acquire frames and the log changes a bit:

```
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/demands.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/thread.activity[11832]
  working: [count=268;total=5199.48ms;avg=18.944ms]
  waiting: [count=251;total=7389.55ms;avg=22.8689ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/demands.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/thread.activity[11832]
  working: [count=298;total=5748.59ms;avg=18.7555ms]
  waiting: [count=281;total=7840.36ms;avg=20.8269ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/demands.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/thread.activity[11832]
  working: [count=328;total=6322.38ms;avg=18.8271ms]
  waiting: [count=311;total=8266.64ms;avg=19.0942ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/demands.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/thread.activity[11832]
  working: [count=358;total=6885.93ms;avg=18.8066ms]
  waiting: [count=341;total=8703.1ms;avg=17.8755ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=1
disp/ao/face_detector/wt-0x1c0bb525cd0/demands.count=0
disp/ao/face_detector/wt-0x1c0bb525cd0/thread.activity[11832]
  working: [count=387;total=7428ms;avg=18.7564ms]
  waiting: [count=371;total=9161.09ms;avg=17.1918ms]
telemetry batch ends----
```

Now, `demands.count` fluctuates between 0 and 1. This means, in practice, the thread is fast enough to dequeue and process events from the queue. Also, other metrics are increasing. This means, the thread is doing some work. It appears that the `face_detector` is consuming, on average, 18ms per event. It’s worth noting that this average is computed across the **entire set of events from the start**, not just the last batch. This detail is crucial because libraries often have warm-up times that can influence this calculation. Hence, for precise profiling, it’s advisable to conduct tests in a more controlled environment. Nonetheless, this provides a general insight into the situation: the face detector’s work is sustainable within the system, with the total waiting time surpassing the total working time!

Suppose at this point, we simply add another `face_detector` to the system, using the same dispatcher in order to include it to the telemetry log:

```cpp
int main()
{
	const auto ctrl_c = utils::get_ctrlc_token();

	const wrapped_env_t sobjectizer;
	const auto main_channel = sobjectizer.environment().create_mbox("main");
	const auto commands_channel = sobjectizer.environment().create_mbox("commands");
	const auto message_queue = create_mchain(sobjectizer.environment());

	sobjectizer.environment().introduce_coop(disp::active_obj::make_dispatcher(sobjectizer.environment()).binder(), [&](coop_t& c) {
		c.make_agent<image_producer_recursive>(main_channel, commands_channel);
		c.make_agent<maint_gui::remote_control>(commands_channel, message_queue);
		
		const auto disp = make_dispatcher(env.environment(), "face_detector", so_5::disp::active_obj::disp_params_t{}.turn_work_thread_activity_tracking_on());
		c.make_agent_with_binder<face_detector>(disp, mbox);
		c.make_agent_with_binder<face_detector>(disp, mbox);
			
		c.make_agent<telemetry_agent>();

	});
	
	do_gui_message_loop(ctrl_c, message_queue, sobjectizer.environment().create_mbox(constants::waitkey_channel_name));		
}
```

As before, before starting the camera acquisition, the log is quite predictable:

```
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=1;total=0.0008ms;avg=0.0008ms]
  waiting: [count=1;total=1763.31ms;avg=1763.31ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=1;total=0.0004ms;avg=0.0004ms]
  waiting: [count=1;total=1763.27ms;avg=1763.27ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=1;total=0.0008ms;avg=0.0008ms]
  waiting: [count=1;total=2763.14ms;avg=2763.14ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=1;total=0.0004ms;avg=0.0004ms]
  waiting: [count=1;total=2763.1ms;avg=2763.1ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=1;total=0.0008ms;avg=0.0008ms]
  waiting: [count=1;total=3763.81ms;avg=3763.81ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=0
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=1;total=0.0004ms;avg=0.0004ms]
  waiting: [count=1;total=3763.77ms;avg=3763.77ms]
telemetry batch ends----
```

We have 2 agents and threads are still idle.

Things get more interesting after starting the camera:

```
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=16
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=37;total=1647.67ms;avg=44.5316ms]
  waiting: [count=1;total=4118.41ms;avg=4118.41ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=7
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=46;total=1510.06ms;avg=32.8274ms]
  waiting: [count=11;total=4255.96ms;avg=386.905ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=10
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=73;total=2647.73ms;avg=36.2703ms]
  waiting: [count=1;total=4118.41ms;avg=4118.41ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=15
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=68;total=2510.12ms;avg=36.9135ms]
  waiting: [count=11;total=4255.96ms;avg=386.905ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=14
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=98;total=3648.62ms;avg=37.2308ms]
  waiting: [count=1;total=4118.41ms;avg=4118.41ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=14
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=98;total=3511.01ms;avg=35.8266ms]
  waiting: [count=11;total=4255.96ms;avg=386.905ms]
telemetry batch ends----
telemetry batch starts----
disp/ao/face_detector/agent.count=2
disp/ao/face_detector/wt-0x1a3c8825280/demands.count=27
disp/ao/face_detector/wt-0x1a3c8825280/thread.activity[30292]
  working: [count=115;total=4648.92ms;avg=40.8211ms]
  waiting: [count=1;total=4118.41ms;avg=4118.41ms]
disp/ao/face_detector/wt-0x1a3c8825780/demands.count=11
disp/ao/face_detector/wt-0x1a3c8825780/thread.activity[15092]
  working: [count=131;total=4511.31ms;avg=34.035ms]
  waiting: [count=11;total=4255.96ms;avg=386.905ms]
telemetry batch ends----
```

As you see, two `face_detector`s working concurrently within the system are suffering a bit: each requires double the time taken to process a single frame when operating individually! Also, look at the waiting event counters: at times, they remain constant, indicating that the thread is consistently saturated and it rarely gets a break. Moreover, pay attention to the trend of both `demands.count`: initially, it appears that thread `30292` can dequeue events more rapidly, but then it starts **buffering**. Conversely, thread `15092` follows an opposite trend.

Thus, enabling this additional telemetry collection on thread activity is convenient to spot possible issues related to event handlers. However, as the documentation remembers, the impact of this feature largely depends on the type of dispatcher and the load profile. For `one_thread` dispatchers handling heavy message streams, this impact is scarcely noticeable. However, for `active_obj` dispatchers dealing with occasional messages, performance can decrease by 3-4 times. Consequently, it’s challenging to provide precise numbers on cumulative performance loss due to the significant variability in results.

Nevertheless, it’s important to emphasize that this mechanism is considered secondary, primarily intended for debugging and profiling purposes. Therefore, it’s imperative to assess the performance impact of work thread activity tracking on your application. Only after thorough evaluation should a decision be made regarding whether to enable this mechanism in the production environment.

#### Design considerations

The first aspect we discuss in this brief section regards how to integrate telemetry support into our applications (existing or new).

First, for “ordinary” telemetry (not related to threading activity), we can easily integrate it by utilizing agents similar to `telemetry_agent`. Runtime monitoring can be toggled on and off dynamically without much additional effort, simply by accessing the environment instance. It’s common to integrate the application with commands, as demonstrated with `telemetry_agent`, and to have multiple agents that process specific telemetry information.

When telemetry is enabled, we can introduce specific agents based on the type of issue we’re investigating or the data we’re interested in. These agents might be always up and running, like `telemetry_agent`, or created on demand. The approach depends on the specific requirements and architecture of the application.

On the contrary, thread activity monitoring is more intrusive and necessitates modifications to how a dispatcher is instantiated. Ideally, we suggest incorporating this setting into the application’s configuration. However, in cases where the application needs to allow interactive toggling of activity tracking, the only option is to recreate the dispatchers. This typically involves deregistering some cooperations and agents, potentially causing temporary interruptions in their operation.

The second aspect regards **how to add our own telemetry data into the telemetry system**. It’s quite easy, as we only need to implement the interface `stats::source_t` that consists of a single function. For example:

```cpp
struct my_data_source : so_5::stats::source_t
{
	void distribute(const so_5::mbox_t& box) override
	{
		so_5::send<so_5::stats::messages::quantity<size_t>>(box, "mydata", "/some-category", 125);
	}
};
```

The stats controller invokes this function of all the registered data sources when it’s time to “sample” telemetry data. It’s important to mention a couple of things regarding prefixes and suffixes:

*   `prefix` is limited to 47 characters + null-terminator (exceeding characters will be discarded by `quantity`‘s constructor);
*   `suffix` just copies the given pointer and not the data. This means, the given string should be either statically allocated or kept alive until the quantity is referenced. Typically, we use statically allocated strings or string literals in this case. It’s important to note that comparing two suffixes is the same as comparing their two pointers and not the content they point to.

To register a new data source, we add an instance of it to the `stats_repository`, a special object that collects all the data sources. It’s accessible from the environment:

```cpp
my_data_source my_source;
env.environment().stats_repository().add(my_source);
```

It’s mandatory to keep the instance alive until the system is up and running. Typically, dispatchers implement the data source interface or keep a data source instance alive. Thus, whenever we develop a custom dispatcher intended for production, it’s good practice to add telemetry support this way. Additionally, it’s important to manually remove the data source to prevent potential issues where the stats controller attempts to call `distribute()` while the data source is being destructed.

To simplify this process, SObjectizer offers a couple of wrappers that handle the registration and deregistration of a data source automatically in a RAII fashion:

*   `auto_registered_source_holder_t`
*   `manually_registered_source_holder_t`

The key distinction between the two lies in the manual control over registration and deregistration provided by the latter, while still automating the removal during destruction if necessary. A minimal example is provided here below:

```cpp
class some_agent final : public so_5::agent_t
{
	so_5::stats::auto_registered_source_holder_t<my_data_source> m_ds_holder;

public:
	some_agent(agent_context_t ctx ) 
		: so_5::agent_t(std::move(ctx)), m_ds_holder(so_5::outliving_mutable( so_environment().stats_repository() ))
	{}
};

int main()
{
	so_5::launch( [](so_5::environment_t & env) {
			env.introduce_coop([](so_5::coop_t& coop) {
					// ... other agents
					coop.make_agent<some_agent>();
				} );
		});
}
```

On the other hand, if we need to defer adding the data source to the stats controller, we might opt for `manually_registered_source_holder_t`:

```cpp
class some_agent final : public so_5::agent_t
{
	so_5::stats::manually_registered_source_holder_t<my_data_source> m_ds_holder;

public:
	some_agent(agent_context_t ctx ) 
		: so_5::agent_t(std::move(ctx))
	{}

	void so_evt_start() override
	{
		m_ds_holder.start(so_5::outliving_mutable( so_environment().stats_repository() ) );
	}
};
```

Additionally, `manually_registered_source_holder_t` provides a `stop()` function to remove the data source before destruction.

### Takeaway

In this episode we have learned:

*   SObjectizer provides a feature to collect and distribute **runtime monitoring information** that are telemetry data regarding its internals;
*   runtime monitoring is **disabled by default** as it might have a negative performance impact;
*   when enabled, telemetry data are collected and distributed by the **stats controller**, an entity that runs in a dedicated thread and is accessible from the environment using `env.stats_controller()`;
*   to **enable** runtime monitoring, we call `env.stats_controller().turn_on()`;
*   to **disable** runtime monitoring, we call `env.stats_controller().turn_off()`;
*   the stats controller **distributes data at intervals** configured using `set_distribution_period(number-of-secs)`;
*   distribution of telemetry data happens as usual: **subscriptions to a certain message box**;
*   the stats controller provides access to the standard message box for telemetry data through `mbox()`;
*   telemetry data are represented by `messages::quantities<size_t>` which, in addition to the **value**, includes the **data source name** in the form of **prefix** and **suffix**;
*   the data source **prefix** identifies a **group** of related data sources (e.g. a particular dispatcher);
*   the data source **suffix** identifies the particular **category** of telemetry information (e.g. size of the event queue);
*   typically, we use **delivery filters** to keep only telemetry data we are interested in;
*   at every interval, all the telemetry quantities are delivered; to **identify batches**, two special messages are sent by the stats controller: `distribution_started` and `distribution_finished`;
*   it’s possible to enable – per dispatcher and also globally – additional telemetry data regarding **thread activity** such as the time spent processing events;
*   thread activity statistics are delivered as messages of type `messages::work_thread_activity`;
*   to add **custom data sources**, we implement the interface `source_t` which provides a single function.

As usual, [`calico`](https://github.com/ilpropheta/calico) is updated and tagged.

### What’s next?

With the recent addition of Ronnie, a new developer unfamiliar with SObjectizer, we’ve encountered a common dilemma: choosing the appropriate dispatchers to bind agents to. While Ronnie grasps the concept of decomposing business logic into agents that exchange messages, selecting the right dispatchers presents a challenge. To address this, we’ve scheduled pair programming sessions with Ronnie to share our guidelines and considerations on dispatcher binding.

Stay tuned for the next installment, where we delve into this topic in detail.

* * *

_Thanks to [Yauheni Akhotnikau](https://www.linkedin.com/in/eao197/) for having reviewed this post._