Ekt, from our R&D department, ran into some issues with the image viewers because his backend forbids to invoke OpenCV’s imshow & friends from other threads but the main one. Indeed, OpenCV supports several UI backends (e.g. GTK, Qt, OpenGL, Win32) and maybe some support multi-thread interaction but this is not well supported in general.
If you are on Windows, you should be on the safe side, however this is not the case for Ekt and in this post we’ll find a solution.
Hosting a message loop in the main thread
First of all, we should move all the UI housekeeping into the main thread. To this end, a message loop is a common idiom. Basically, the main thread extracts messages one by one from a synchronized queue (or whatever) and performs the corresponding operations on the UI. Other agents (or threads, more in general) do not call UI functions directly, instead they encode UI requests into messages and send them to that queue.
A message chain is the perfect choice for modeling the message queue, letting us write code in a declarative way:
while (!st.stop_requested())
{
receive(from(chain).handle_n(1),
[](const some_message& msg) {
// handler for a message
},
//... handler for another message
);
}
Indeed, receive accepts an arbitrary number of message handlers.
In the code above, we loop until the stop token st is triggered. However, you might have spotted a subtle issue already: if there are no messages, receive won’t return.
We need a slightly different behavior: if the chain contains a message, handle it, otherwise just return and give the loop a chance to finish. Message chains are flexible enough to support this kind of customizations. In general, the semantic of the receive operation can be refined by calling other functions (commonly called modificators) in cascade. For example, we can use no_wait_on_empty() to get exactly that behavior:
while (!st.stop_requested())
{
receive(from(chain).handle_n(1).no_wait_on_empty(),
[](const some_message& msg) {
// handler for a message
},
//... handler for another message
);
}
However, we now get a possibly unintended issue: if the chain is empty, the receive operation becomes too fast and the loop might put pressure on the CPU. What about setting a timeout instead? Something like “before interrupting the receive operation while the chain is empty, wait for a new message by and not later than 100 milliseconds”. empty_timeout() comes to the rescue:
while (!st.stop_requested())
{
receive(from(chain).handle_n(1).empty_timeout(100ms),
[](const some_message& msg) {
// handler for a message
},
//... handler for another message
);
}
In this case, receive returns if no message is received by 100 milliseconds.
Now, it’s time to create the counterparts of smart_image_viewer and responsive_image_viewer that use this new way of drawing and handling windows.
Sending imshow messages
First of all, we should define the message that encodes an “imshow request”. Since cv::imshow accepts the name of the window and the image, this message can be simply declared as follows:
struct imshow_message
{
std::string window;
cv::Mat image;
};
Bear in mind that replacing std::string with std::string_view is futile because imshow needs a std::string in any case (remember that imshow is just intended for troubleshooting or demos).
Also, instead of supporting a waitkey_message, we might simply call cv::waitKey(1) at the end of each iteration. This way, all existing windows will stay responsive. Here is the message loop:
while (!st.stop_requested())
{
receive(from(chain).handle_n(1).empty_timeout(100ms),
[](const imshow_message& mex) {
imshow(cmd.window, cmd.image);
}
);
cv::waitKey(1);
}
At this point, image_viewer_live‘s alter ego is straightforward to develop:
class image_viewer_live final : public so_5::agent_t
{
public:
image_viewer_live(so_5::agent_context_t ctx, so_5::mbox_t channel, so_5::mchain_t ui_queue)
: agent_t(std::move(ctx)), m_channel(std::move(channel)), m_message_queue(std::move(ui_queue))
{
}
void so_define_agent() override
{
so_subscribe(m_channel).event([this](cv::Mat mat) {
so_5::send<imshow_message>(m_message_queue, m_title, std::move(mat));
});
}
private:
static inline int global_id = 0;
so_5::mbox_t m_channel;
so_5::mchain_t m_message_queue;
std::string m_title = std::format("image viewer live {}", global_id++);
};
ui_queue is passed as a message chain but we can also pass it as message box (as_mbox()). However, the advantage of passing it as a message chain is that it can’t be used in the wrong way (e.g. we can’t accidentally subscribe to it).
No rocket science involved here. We just send frames as soon as they arrive. The message loop will handle the rest.
Sending destroyWindow messages
image_viewer‘s alter ego needs to close (cv::destroyWindow) the window. Thus, we add a new message to the party:
struct close_window_message
{
std::string window;
};
Handling this command is straightforward:
while (!st.stop_requested())
{
receive(from(message_queue).handle_n(1).empty_timeout(100ms),
[](const imshow_message& mex) {
imshow(mex.window, mex.image);
},
[](const close_window_message& mex) {
cv::destroyWindow(mex.window);
}
);
cv::waitKey(1);
}
Then, image_viewer‘s alter ego comes out very naturally since it just the matter of replacing OpenCV calls with message sending:
class image_viewer final : public so_5::agent_t
{
so_5::state_t st_handling_images{ this };
so_5::state_t st_stream_down{ this };
public:
image_viewer(so_5::agent_context_t ctx, so_5::mbox_t channel, so_5::mchain_t ui_queue)
: agent_t(std::move(ctx)), m_channel(std::move(channel)), m_ui_queue(std::move(ui_queue))
{
}
void so_define_agent() override
{
st_handling_images
.event(m_channel, [this](cv::Mat image) {
so_5::send<imshow_message>(m_message_queue, m_title, std::move(image));
st_handling_images.time_limit(500ms, st_stream_down);
}).on_exit([this] { so_5::send<close_window_message>(m_message_queue, m_title); });
st_stream_down
.transfer_to_state<cv::Mat>(m_channel, st_handling_images);
st_stream_down.activate();
}
private:
static inline int global_id = 0;
so_5::mbox_t m_channel;
so_5::mchain_t m_message_queue;
std::string m_title = std::format("smart image viewer {}", global_id++);
};
As a side effect, these new viewers are independent of OpenCV.
Putting all together
Let’s see this in action:
int main()
{
auto ctrl_c = get_ctrl_c_token();
const so_5::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());
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<remote_control>(commands_channel);
c.make_agent<image_viewer_live>(main_channel, message_queue);
c.make_agent<image_viewer>(main_channel, message_queue);
}
// ui message loop
while (!ctrl_c .stop_requested())
{
receive(from(message_queue).handle_n(1).empty_timeout(100ms),
[](const imshow_message& mex) {
imshow(mex.window, mex.image);
},
[](const close_window_message& mex) {
cv::destroyWindow(mex.window);
}
);
cv::waitKey(1);
}
}
As you might imagine, we don’t need wait_for_stop(st) anymore because the program terminates automatically when the loop finishes (it’s just a choice, as it would be supporting a “quit” message or whatever). Also, consider that values passed to empty_timeout(100ms) and waitKey(1) are arbitrary and might be chosen more wisely.
What about remote_control?
Some agents might need to get the output value of waitKey. For example our remote_control that is based on OpenCV. The full implementation can be found in this previous episode.
However, this particular agent differs from the others as it requires receiving something from the message loop, in particular it needs the result of cv::waitKey to check if the user pressed a key. How can we address this situation?
Let’s see a possible solution, but consider that some alternatives exist.
The interesting part here is that the communication seems to be directed from the message loop to the agent, and not the other way around as before.
Also, waitKey is a function we should call only once per iteration (at the end) and possibly several agents might be interested in its return value. That said, the idea is to use a message box that will host the results of waitKey after every iteration. All the agents interested in such messages can just subscribe to that. After all, this is message passing!
Here is the idea:
// somewhere...
auto waitkey_out = env.environment().create_mbox("waitkey_out");
// ... as before
// ui message loop
while (!st.stop_requested())
{
receive(from(message_queue)...
// ... as before
);
const auto key = cv::waitKey(1);
so_5::send<waitkey_message>(waitkey_out, key);
}
The message type involved is intuitive:
struct waitkey_message
{
int pressed;
};
Thus, the implementation of the agent boils down to:
class remote_control final : public so_5::agent_t
{
public:
remote_control(so_5::agent_context_t ctx, so_5::mbox_t commands, so_5::mchain_t ui_queue)
: agent_t(std::move(ctx)), m_channel(std::move(commands)), m_message_queue(std::move(ui_queue)), m_waitkey_channel(so_environment().create_mbox("waitkey_out"))
{
}
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 });
so_5::send<imshow_message>(m_message_queue, "Remote control", std::move(frame));
}
void so_define_agent() override
{
so_subscribe(m_waitkey_channel)
.event([this](const waitkey_message& waitkey_result) {
switch (waitkey_result.pressed)
{
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;
}
});
}
private:
so_5::mbox_t m_channel;
so_5::mchain_t m_message_queue;
so_5::mbox_t m_waitkey_channel;
};
Here above, we leveraged the power of named message boxes by obtaining the waitkey channel from inside the agent instead of injecting it from the constructor. It’s a design choice. Usually, when relying on this feature, the name of the channel is either configurable (e.g. we have a sort of registry or whatever) or declared as a constant somewhere (in calico, you will find it in a file "constants.h" that has just been added).
It’s worth mentioning that the message loop might avoid sending “no press key” responses to the subscribers:
// somewhere...
auto waitkey_out = env.environment().create_mbox("waitkey_out");
// ...
// ui message loop
while (!st.stop_requested())
{
receive(from(chain).handle_n(1).empty_timeout(100ms),
// ... as before
if (const auto key = cv::waitKey(1); key != -1)
{
so_5::send<waitkey_message>(waitkey_out, key);
}
}
It’s just a design choice.
Here is an example of usage:
int main()
{
auto ctrl_c = get_ctrl_c_token();
const so_5::wrapped_env_t sobjectizer;
const auto main_channel = sobjectizer.environment().create_mbox("main");
const auto commands_channel = sobjectizer.environment().create_mbox("commands");
const auto ui_commands = create_mchain(env.environment());
const auto waitkey_out = env.environment().create_mbox("waitkey_out");
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<remote_control>(commands_channel, ui_commands);
c.make_agent<image_viewer_live>(main_channel, ui_commands);
c.make_agent<image_viewer>(main_channel, ui_commands);
}
// ui message loop
while (!ctrl_c.stop_requested())
{
receive(from(chain).handle_n(1).empty_timeout(100ms),
[](const imshow_command& cmd) {
imshow(cmd.window, cmd.image);
},
[](const close_window_command& cmd) {
cv::destroyWindow(cmd.window);
}
);
if (const auto key = cv::waitKey(1); key != -1)
{
so_5::send<waitkey_message>(waitkey_out, key);
}
}
}
Takeaway
In this episode we have learned:
- the semantic of
receiveon message chains can be refined by calling some additional functions (modificators); -
no_wait_on_empty()interrupts waiting when the chain is empty; -
empty_timeout(timeout)interrupts waiting when the chain is empty and no more messages arrive by a timeout; -
receiveaccepts any number of message handlers; - an advantage of passing a message chain as a message chain (not as a message box) is that it can’t be used in the wrong way (e.g. we can’t subscribe to it).
As usual, calico is updated and tagged.
What’s next?
Ekt is finally able to use calico and he is going to give us some feedback about new features and possible bugs. In the meantime, we can finally think about another big topic we have not discussed yet: testing. How can we introduce some tests to calico?
We’ll work on that in the next installment!
Thanks to Yauheni Akhotnikau for having reviewed this post.
Top comments (0)