SObjectizer Tales - 5. Sending commands
Tags: cpp, softwareengineering, actormodel, sobjectizer
Published: 2023-11-09 08:19:46

In our previous session, we delved into the process of **managing commands** to oversee the acquisition. This journey was notably extensive, given the necessity of updating all our producers along the way. The positive takeaway from this endeavor is that the process of sending commands is now significantly simplified.

To illustrate this improvement, we manually initiated start and stop commands from the main function, which, admittedly, lacks excitement. However, in this upcoming post, we will explore various instances of generating commands from different segments within the program.

Our producers are now primed to accept commands via the designated “commands” message box. The responsibility lies with various components of the program to transmit signals through this channel. By sending the `start_acquisition_command` and `stop_acquisition_command`, any of our producers will respond as required. This opens up a **wide array of potential possibilities**. We could develop a keyboard listener that translates keystrokes into signals, or create a REST API capable of receiving commands from the network, among other options.

For example, here is a minimal keyboard listener based on OpenCV that shows a tiny frame where you can press Enter or Escape to start and stop the acquisition:

```cpp
class remote_control : public so_5::agent_t
{
	struct keep_on final : so_5::signal_t {};
public:
	remote_control(so_5::agent_context_t ctx, so_5::mbox_t commands)
		: agent_t(std::move(ctx)), m_channel(std::move(commands))
	{
	}

	void so_evt_start() override
	{
		cv::Mat frame = cv::Mat::ones(100, 200, CV_8UC3);
		putText(frame, "Start (Enter)", { 2, 20 }, cv::FONT_HERSHEY_COMPLEX_SMALL, 1, { 240, 200, 1 });
		putText(frame, "Stop (Escape)", { 2, 50 }, cv::FONT_HERSHEY_COMPLEX_SMALL, 1, { 240, 200, 1 });
		imshow("Remote Control", frame);		
		so_5::send<keep_on>(*this);
	}

	void so_define_agent() override
	{
		so_subscribe_self()
			.event([this[(so_5::mhood_t<keep_on>) {
			switch (cv::waitKey(1))
			{
			case 13: // Enter
				so_5::send<start_acquisition_command>(m_channel);
				break;
			case 27: // Escape
				so_5::send<stop_acquisition_command>(m_channel);
				break;			
			default:
				break;
			}
			so_5::send_delayed<keep_on>(*this, 100ms);
		});
	}
private:
	so_5::mbox_t m_channel;
};
```

Technical detail: the frame must be kept responsive by iteratively calling either `waitKey` or `pollKey`, for this reason we use that `keep_on` signal. As said in a previous discussion, the command channel might be retrieved by name but here we prefer to inject it.

As usual, we add that agent to the party as follows:

```cpp
int main()
{
	auto ctrl_c = get_ctrl_c_token();

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

	auto dispatcher = so_5::disp::active_obj::make_dispatcher(sobjectizer.environment());
	sobjectizer.environment().introduce_coop(dispatcher.binder(), [&[(so_5::coop_t& c) {
		c.make_agent<image_producer_recursive>(main_channel, commands_channel);
		c.make_agent<image_viewer>(main_channel);
		c.make_agent<remote_control>(commands_channel);
	}
		
	wait_for_stop(ctrl_c);
}
```

Here is a live demo:

[![](https://marcoarena.files.wordpress.com/2023/11/calico-5.gif?w=1024)](https://marcoarena.files.wordpress.com/2023/11/calico-5.gif)

Now, let’s consider adding some excitement to the situation. Imagine we need to capture frames for a specific duration, such as 10 seconds. How can we achieve this? One approach is to initiate the process with a “start” command, wait for 10 seconds, and then issue a “stop” command:

```cpp
// somewhere
send<start_acquisition_command>(commands);
this_thread::sleep_for(10s);
send<stop_acquisition_command>(commands);
```

This method is functional, and as you may be aware, the “stop” command will not be executed until at **least 10 seconds have passed** (potentially even longer due to scheduling and other operating system intricacies). However, this approach can be somewhat inconvenient because it might require relocating the operation to a different context, like a separate thread or handling it within a coroutine, to prevent it from blocking the current thread.

An alternative way is using [SObjectizer’s timers](https://github.com/Stiffstream/sobjectizer/wiki/SO-5.8-InDepth-Timers).

#### Introducing timed send functions

Typical applications developed with SObjectizer use timers quite extensively and for this reason SObjectizer uses a dedicated _timer thread_ that can efficiently process a big amount of timers (tens and hundreds of million, even billions of timers).

Timers come in the form of two additional ways of sending messages:

*   **delayed messages**
*   **periodic messages**

The former is a manner of **postponing the delivery of a message**. For example:

```cpp
so_5::send_delayed<stop_acquisition_command>(commands, 10s);
```

sends `stop_acquisition_command` after 10 seconds.

Messages sent by `send_delayed` can’t be cancelled. On the other hand, messages sent by `send_periodic` can. We’ll see how in a moment.

As one would anticipate, **periodic messages** provide a means of transmitting the same message multiple times at specific intervals. In practice, the message instance is not deallocated but, instead, it is delivered again and again, until cancellation happens (either automatic or explicit).

For instance, consider a scenario in which our program is employed for video surveillance, and we require capturing images for a duration of 5 minutes every half an hour. In such a case, we can dispatch a pair of periodic messages:

```cpp
auto receipt_start = so_5::send_periodic<start_acquisition_command>(commands,
      30m, // first delivery delay
      30m); // repetition period
 
auto receipt_stop = so_5::send_periodic<stop_acquisition_command>(commands,
      35m, // first delivery delay
      30m); // repetition period
```

`receipt_start` and `receipt_stop` are instances of `so_5::timer_id_t`, a class that identifies a timed message. Often such instances are simply referred to as **timers**.

Storing `receipt_start` and `receipt_stop` is crucial, indeed `timer_id_t` works like a shared pointer and destroying the last instance will cause the periodic message to be cancelled. This means, the message won’t be sent again according to the period. In other words, this code:

```cpp
so_5::send_periodic<stop_acquisition_command>(commands, 35m, 30m);
```

causes the message to be immediately cancelled!

Also, `send_periodic` supports delayed messages in case the third parameter (`period`) is 0. In this special case, the message will be delivered only once, after the specified `delay`:

```cpp
auto delayed = so_5::send_periodic<start_acquisition_command>(commands,
      30m, // delivery delay
      0m); // Repetition period is zero -> it's a delayed message
```

Clearly, in this case destroying the receipt causes the delayed message to be cancelled.

#### Cancelling a message

As said, periodic messages can be cancelled, whether they are created with a `period` greater than zero or not. Cancellation happens automatically when the last instance of `timer_id_t` returned from `send_periodic` is destroyed. However, we can even cancel a timer explicitly by means of `release`:

```cpp
auto receipt = so_5::send_periodic<start_acquisition_command>();
...
receipt.release(); // cancellation
```

This method cancels the message regardless of the count of remaining `timer_id_t` objects pointed to that instance.

Often timers are stored into agents as member variables. For example:

```cpp
class some_agent : public so_5::agent_t 
{
	so_5::timer_id_t m_timer;
	//...
	void some_function() 
	{
		m_timer = so_5::send_periodic<some_message>(...); 
		// ... 
	}
};
```

In this case, when the agent is destroyed, so is `m_timer`.

Have you ever tried to cancel a timer on Alexa…while it was about to ring?! In SObjectizer, **message cancellation is not strictly guaranteed**. Indeed, **a delayed message won’t be revoked if it left the timer thread already**. This happens if you try cancelling the message too late, in that case there are good chances it will be delivered anyway. For example, if you send a message with delay 100ms and you try cancelling after 100ms then it might have left the timer thread already. There are ways to prevent this problem but won’t be discussed here. If you are interested, have a look at [this](https://github.com/stiffstream/so5extra/wiki/so5extra-1.6-Revocable-Timers).

#### What about `virtual_image_producer`?

You make a good observation! In the previous post, we inadvertently overlooked upgrading `virtual_image_producer` to react to commands properly. This occurred because **we needed timed messages**. Let’s do it now.

The state machine closely resembles that of `image_producer_recursive` with two key distinctions:

*   the images are retrieved from the file system;
*   `grab_image` signal is **dispatched as a delayed message** to replicate the delay between two successive frames.

Here is the full implementation:

```cpp
class virtual_image_producer final : public so_5::agent_t
{
	struct grab_image final : so_5::signal_t {};
public:
	virtual_image_producer(so_5::agent_context_t ctx, so_5::mbox_t channel, so_5::mbox_t commands, std::filesystem::path path)
		: agent_t(std::move(ctx)), m_channel(std::move(channel)), m_commands(std::move(commands)), m_path(std::move(path))
	{
	}

	void so_define_agent() override
	{
		st_started.event([this[(so_5::mhood_t<grab_image>) {
			cv::Mat mat;
			m_current_it = std::find_if(m_current_it, {}, [[(const auto& e) {
				return e.is_regular_file();
			});
			so_5::send<cv::Mat>(m_channel, cv::imread(m_current_it->path().string()));
			so_5::send_delayed<grab_image>(*this, std::chrono::milliseconds(20));
			if (++m_current_it == std::filesystem::directory_iterator{})
			{
				m_current_it = std::filesystem::directory_iterator{ m_path };
			}
		}).event(m_commands, [this[(so_5::mhood_t<stop_acquisition_command>) {
			st_stopped.activate();
		});

		st_stopped.event(m_commands, [this[(so_5::mhood_t<start_acquisition_command>) {
			st_started.activate();
			so_5::send<grab_image>(*this);
		});

		st_stopped.activate();
	}

	void so_evt_start() override
	{
		if (!is_directory(m_path))
		{
			throw std::runtime_error("Can't open virtual device directory");
		}

		m_current_it = std::filesystem::directory_iterator{ m_path };
	}
private:
	so_5::state_t st_stopped{ this };
	so_5::state_t st_started{ this };

	so_5::mbox_t m_channel;
	so_5::mbox_t m_commands;
	std::filesystem::path m_path;
	std::filesystem::directory_iterator m_current_it;
};
```

To provide a more accurate adjustment, we should modify the pause duration by subtracting the milliseconds that were “consumed” while processing the last frame, such as the time required to read a frame from the disk:

```cpp
// ... as before

void so_define_agent() override
{
	st_started.event([this[(so_5::mhood_t<grab_image>) {
		const auto tic = std::chrono::steady_clock::now();
		cv::Mat mat;
		m_current_it = std::find_if(m_current_it, {}, [[(const auto& e) {
			return e.is_regular_file();
		});
		so_5::send<cv::Mat>(m_channel, cv::imread(m_current_it->path().string()));
        const auto elapsed = std::min(20ms, std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - tic));
		so_5::send_delayed<grab_image>(*this, 20ms - elapsed);
		if (++m_current_it == std::filesystem::directory_iterator{})
		{
			m_current_it = std::filesystem::directory_iterator{ m_path };
		}
	}). //... as before
```

While there are more sophisticated strategies available, we can stick with this simplified implementation for the time being. After all, this is merely a mock device!

### Takeaway

In this episode we have learned:

*   SObjectizer supports timers through **timed send functions**;
*   `send_delayed` delivers a message **after a delay**, and such a message **can’t be revoked**;
*   `send_periodic` delivers a message **again and again**, until **cancelled**;
*   **cancellation** of periodic messages can happen **implicitly** or **explicitly**;
*   `send_periodic` with **0-period** is like delayed message but can be cancelled;
*   cancelling a message **immediately after the deadline** might prevent the revocation.

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

### What’s next?

We can unleash our imagination and craft any command sources. We can also create functions and components to send commands automatically on the basis of time.

However, playing a bit with start and stop we notice a subtle problem in the behavior of `image_viewer`: it seems that after stopping the acquisition, the window hangs and stays unresponsive. When we start the acquisition again then it reawakens…what’s happening?

In the next post we’ll deal with that issue and provide a solution.

* * *

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