SObjectizer Tales – 9. Splitting work
Tags: cpp, actormodel, softwareengineering, sobjectizer
Published: 2023-12-07 08:27:00

Last time we left with a new feature request: **saving images to the hard drive**. The specification is quite vague but we can imagine the user requires an agent that persists compressed frames (e.g. `jpg`) to disk.

OpenCV provides [`imwrite`](https://docs.opencv.org/3.4/d4/da8/group__imgcodecs.html#gabbc7ef1aa2edfaa87772f1202d67e0ce) that compresses a specified image and writes the resulting bytes to a specified path. The compression is based on the filename extension (e.g. `c:/saved/frame_1.jpg` means `jpg`). Additional details are not important here, let’s assume to pass only image types that this function is able to handle.

A working implementation of that agent is here below:

```cpp
class save_agent final : public so_5::agent_t
{
public:
	save_agent(so_5::agent_context_t ctx, so_5::mbox_t input, std::filesystem::path root_folder)
		: agent_t(std::move(ctx)), m_input(std::move(input)), m_root_folder(std::move(root_folder))
	{
	}
	
	void so_define_agent() override
	{
		so_subscribe(m_input).event([this](const cv::Mat& image) {
			imwrite((m_root_folder / std::format("image_{}.jpg", m_counter++)).string(), image);
		});
	}
	
	void so_evt_start() override
	{
		std::error_code ec;
		if (create_directories(m_root_folder, ec); ec)
		{
			throw std::runtime_error(std::format("save_agent can't create root folder: {}", ec.message()));
		}		
	}

private:
	so_5::mbox_t m_input;
	std::filesystem::path m_root_folder;
	int m_counter = 0;
};
```

Throwing an exception in case the directory tree can’t be created is just an option and other choices are possible, but this is not the point here.

Images are individually saved as they are received, following a typical pattern of executing a relatively resource-intensive operation for each message. Specifically, the `imwrite` function utilizes both CPU power for compression and the disk for storage. Most likely, employing this agent wouldn’t significantly impact the performance of your machine.

Here is an example of usage:

```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>(main_channel, commands_channel);
		c.make_agent<remote_control>(commands_channel);
		c.make_agent<responsive_image_viewer>(main_channel);
		c.make_agent<save_agent>(main_channel, "c:/images");
	}
		
	wait_for_stop(ctrl_c);
}
```

However, giving the agent too much work could quickly make performance worse. For example, if we enlarge the images significantly, such as adding an additional `image_resizer` with a scale factor greater than 2, it could lead to a substantial increase in memory consumption. This is because `imwrite` would require more time, consequently delaying the dequeuing of the subsequent image and causing noticeable message buffering.

**As these operations are self-contained** and do not rely on one another, there’s a chance to **distribute these tasks among several workers**, potentially harnessing the computational capabilities of our machine. For instance, while _worker1_ handles _image1_, _worker2_ can simultaneously process _image2_, and this sequence continues. Given that we assume CPU-dependent compression, processing multiple images concurrently _might_ yield some advantages and efficiencies.

Bear in mind that this article is not about finding the best way to compress and save images to disk nor the ideal topology to distribute data. Rather, it serves as an introduction to demonstrate how SObjectizer can be utilized to divide tasks among multiple agents.

#### Can we have contention on message boxes?

This was the [very first question](https://github.com/Stiffstream/sobjectizer/issues/23) I asked to SObjectizer folks back in 2021.

You have learned that message boxes (often called “channels” throughout the series) are carriers for delivering messages and signals to agents or, in general, to _subscribers_. Any messages dispatched to a message box will be distributed to all the subscribers connected to it.

However, when distributing work to do, we have a very different need: every message must be delivered to exactly one agent for processing. For example, if `agent1` and `agent2` take work from a certain buffer, only one of them must receive `message1`, only one of them receive `message2`, and so on. Maybe while `agent1` is working on `message1`, `agent2` is working on `message2`, and so on.

In general, assigning tasks to workers follows a sophisticated logic, they might have priorities, and things can be extremely complicated. After all, **scheduling** is one of the main topics in computer science.

Returning to our simple example, it’s clear that workers shouldn’t directly subscribe to message boxes as they would receive all the messages intended for that subscription. Hence, we can introduce a “scheduler” agent in between that subscribes to pertinent messages and allocates them to workers based on specific logic. For instance, the logic might be “send the next message to the first available agent”. In this arrangement, the scheduler needs to be aware of which workers are currently occupied.

While this is achievable, there’s no urgency to rush into the implementation right now. Fortunately, SObjectizer already offers a fundamental structure that facilitates this pattern in just a few lines of code. This concept is somewhat analogous to a message box and is known as a **message chain**.

Message chains share similarities with asynchronous queues and do not provide a subscription-based interface like message boxes. Similar to message boxes, chains accept data via the function `so_5::send`. However, unlike message boxes, **chains cannot be directly subscribed to**; instead, we need to explicitly use the `so_5::receive` function to extract data. The `so_5::receive` function can be invoked from an agent or any other part of the application, **even outside of SObjectizer**. Because of this flexibility, message chains were primarily introduced to facilitate data exchange between non-SObjectizer components. It’s feasible for multiple threads to concurrently retrieve data from the same chain.

The unique aspect of message chains lies in how messages are extracted from them: **when a message is received and handled, it’s essentially extracted from the chain**. This implies that if one thread extracts the next message and another thread attempts to receive from the chain immediately afterward, that specific message won’t be extracted again by the second thread. You’ve likely understood our point: **message chains embody the contention pattern we require**.

In practice, we can give workers the same chain and let them receive from it. When one extracts a message and starts working on it, the others will have a chance of extracting **the next** message.

Message chains are even more sophisticated than a message queue, and provide other outstanding features we won’t get into at this time. First of all, there are several overloads to create a chain but we can just use the following one for the moment:

```cpp
auto chain = create_mchain(so_environment());
```

Differently from message boxes, message chains can be **closed** explicitly:

```cpp
auto chain = create_mchain(so_environment());
// ...
// possibly on another thread
close_drop_content(so_5::terminate_if_throws, chain);
```

Closing a chain can either drop leftovers (`close_drop_content`) or not (`close_retain_content`). This choice has effects on receivers, as we’ll see in a moment. Ignore `terminate_if_throws` for now, it just means that if closing a chain throws an exception, the program will be terminated (like throwing from a destructor).

At this point, one or more threads can receive from the chain by using the **receive/from/handle idiom**:

```cpp
receive(from(chain).handle_n(1), [](const cv::Mat& image){
	// ... handle one image
});
```

Here above, `receive` blocks the execution until either the chain is closed or exactly one message of type `cv::Mat` is received, extracted and handled. As you see, `receive` accepts a message handler that is a function object.

As you might expect, using either `close_retain_content` or `close_drop_content` has important effects on the queue if the receiver is still waiting for new messages. Both ultimately interrupt wait, however, before that, the former lets remaining messages to be handled first, the latter does not. To receive all the messages until a chain is closed we use `handle_all()`:

```cpp
receive(from(chain).handle_all(), [](const cv::Mat& image){
	// ... handle one image
});
```

In this case, `receive` does not return until the chain is closed.

#### What a worker looks like

At this point, we have enough information to sketch a save agent worker:

```cpp
class image_save_worker final : public so_5::agent_t
{
public:
	image_save_worker(so_5::agent_context_t ctx, so_5::mchain_t input, std::filesystem::path root_folder)
		: agent_t(std::move(ctx)), m_input(std::move(input)), m_root_folder(std::move(root_folder))
	{
	}

	void so_evt_start() override
	{		
		receive(from(m_input).handle_all(), [this](const cv::Mat& image) {
			imwrite((m_root_folder / std::format("image_{}_{}.jpg", m_worker_id, m_counter++)).string(), image);
		});
	}

private:
	so_5::mchain_t m_input;
	std::filesystem::path m_root_folder;
	int m_counter = 0;
	static inline int global_id = 0;
	int m_worker_id = global_id++;
};
```

A few details:

*   `mchain_t` represents the message chain type (it’s actually a more abstract type);
*   `m_worker_id` is used to avoid generate unique image names across multiple agents;
*   as usual, we have not defended against specific errors (e.g. `imwrite` fails) to keep things simple;
*   the worker processes messages until the chain is closed, otherwise it never terminates.

The last point is essential: someone has to close the chain at the end of the program. We can do it after CTRL+C is detected.

Before getting to that, there is another question: who is sending images to the message chain? The question is on-point since we currently send images to the main channel only. Again, here we can leverage the power of **message sinks**. Before SObjectizer 5.8, the only way was to craft a proxy agent as discussed in a [previous episode](https://dev.to/ilpropheta/sobjectizer-tales-7-agent-composition-8ij). Instead, we can now introduce a **binding** that takes images from the main channel and sends them to the chain:

```cpp
auto binding = c.take_under_control(std::make_unique<so_5::single_sink_binding_t>());
binding->bind<cv::Mat>(resized, wrap_to_msink(chain->as_mbox()));
```

Actually, a similar use case motivated the introduction of message sinks and bindings. It was partially inspired by an [issue](https://github.com/Stiffstream/sobjectizer/issues/30#issuecomment-821180821) I opened and discussed with Yauheni in 2021. Once more, it’s worth emphasizing that the SObjectizer community is eager and willing to assist as well as incorporate new features if necessary.

A relevant detail: message chains can be represented as message boxes through `as_mbox()` but, in this case, they can’t accept subscriptions (if we try adding a subscription, an exception will be thrown at runtime).

At this point, we have collected all the pieces of the puzzle (we resize the images beforehand to increase the chances to press the CPU):

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

	const wrapped_env_t sobjectizer;
	const auto main_channel = sobjectizer.environment().create_mbox("main");
	auto commands_channel = sobjectizer.environment().create_mbox("commands");
	auto save_buffer = 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<remote_control>(commands_channel);
		c.make_agent<image_viewer_live>(main_channel);
		const auto resized = c.make_agent<image_resizer>(main_channel, 3.0)->output();

		const auto binding = c.take_under_control(make_unique<single_sink_binding_t>());
		binding->bind<cv::Mat>(resized, wrap_to_msink(save_buffer->as_mbox()));

		c.make_agent<image_save_worker>(save_buffer, "c:/images");
		c.make_agent<image_save_worker>(save_buffer, "c:/images");
	});

	wait_for_stop(ctrl_c);
	close_retain_content(terminate_if_throws, save_buffer);
}
```

Here is an picture of the situation:

[![](https://marcoarena.files.wordpress.com/2023/09/image-4.png?w=1024)](https://marcoarena.files.wordpress.com/2023/09/image-4.png)

In addition, SObjectizer provides a tiny RAII-wrapper to guarantee that chains are automatically closed at the end of the scope. It supports both the closing policies. For example:

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

	//... as before
	auto chain = create_mchain(sobjectizer.environment());
	auto closer = auto_close_retain_content(chain);
	
	// ... as before
		
	wait_for_stop(ctrl_c);	
} // chain closed automatically here
```

Have you noticed that we forgot one subtle detail in this implementation? We didn’t invoke `create_directories`…where is the best place to put this responsibility?

#### A little step forward

Often, data distribution is an implementation detail hidden underneath another entity. For instance, a “root” node might take on the responsibility of spawning, managing, and overseeing workers, potentially gathering their results at the conclusion. Similarly, the previously mentioned “scheduler” represents another instance of such an entity that manages the allocation of tasks.

In our scenario, things are a bit simpler but we can take the opportunity to introduce a sort of “saver agent” that creates the destination directory tree, hides data distribution, and spawns workers. In particular, it should be responsible for:

*   invoking `create_directories`;
*   making worker agents;
*   creating and closing the message chain;
*   creating the binding object that forwards data to the message chain.

Clearly, new agents have to be added to the cooperation. However, we can’t add new agents to “running cooperations” (it wouldn’t make sense because they are “transactional”, remember?). Thus, the workers need their own cooperation.

**We might be tempted to create a child cooperation and this wouldn’t be a problem.** However, we would miss an opportunity. Parent-child relationships add a constraint: parents won’t be destroyed until children have done. In our case, the children (workers) will be perpetually waiting for new images on the chain. Then it’s mandatory that someone else closes that chain when it’s time to terminate. Well, we can take advantage of coordinator’s `so_evt_finish()`, can’t we? Unfortunately no, because the parent-child relationship would make it impossible to call coordinator’s `so_evt_finish()` before the workers have done…Deadlock!

On the other hand, without any relationships, the coordinator would be destroyed independently of the workers at some point (strictly speaking, this is true only if they have a different worker thread – that is the case). Deal!

Here is the code:

```cpp
class image_saver final : public so_5::agent_t
{
public:
	image_saver (so_5::agent_context_t ctx, so_5::mbox_t input, std::filesystem::path root_folder)
		: agent_t(std::move(ctx)), m_input(std::move(input)), m_root_folder(std::move(root_folder))
	{
		m_chain = create_mchain(so_environment());
	}

	void so_evt_start() override
	{
		std::error_code ec;
		if (create_directories(m_root_folder, ec); ec)
		{
			throw std::runtime_error(std::format("image_saver can't create root folder: {}", ec.message()));
		}
		
		so_environment().introduce_coop(so_5::disp::active_obj::make_dispatcher(so_environment()).binder(), [this](so_5::coop_t& c) {
			const auto binding = c.take_under_control(std::make_unique<so_5::single_sink_binding_t>());
			binding->bind<cv::Mat>(m_input, wrap_to_msink(m_chain->as_mbox()));
			c.make_agent<image_save_worker>(m_chain, m_root_folder);
			c.make_agent<image_save_worker>(m_chain, m_root_folder);
		});
	}

	void so_evt_finish() override
	{
		close_retain_content(so_5::terminate_if_throws, m_chain);
	}

private:
	so_5::mbox_t m_input;
	so_5::mchain_t m_chain;
	std::filesystem::path m_root_folder;
};
```

In calico, you will find that `image_save_worker` has been hidden as `image_saver`‘s implementation detail but, if needed, there is nothing wrong with keeping it public.

Then the whole initialization boils down to:

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

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

	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<remote_control>(commands_channel);
		c.make_agent<image_viewer_live>(main_channel);
		const auto resized = c.make_agent<image_resizer>(main_channel, 3.0)->output();
		c.make_agent<image_saver>(resized, "c:/images");		
	});

	wait_for_stop(ctrl_c);	
}
```

As you can well imagine, the saver agent might be sophisticated at will. Workers can be bound to a thread pool dispatcher and the assignment of tasks can be quite complex. For now, let’s just settle for the general pattern that is at the foundations of splitting work among multiple agents.

### Takeaway

In this episode we have learned:

*   **message chains** are advanced **asynchronous queues**;
*   message chains receives messages sent via the usual `so_5::send`;
*   message chains **do not accept subscriptions**, instead they can be received from any code – whether SObjectizer-based or not – via the `receive/from/handle` idiom;
*   message chains **can be closed** to interrupt `receive`, either dropping or retaining leftovers.

### What’s next?

The customer is happy with the new feature and takes some time to test it properly. In the meantime, Ekt from our R&D department would like to use the program for some top secret projects…However, he ran into some issues with the image viewers because his backend forbids to invoke `imshow` & friends from other threads but main.

In the next post we propose a SObjectizer-based solution to this problem!

* * *

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