Корни SObjectizer берут свое начало в теме автоматизированных систем управления технологическими процессами (АСУТП). Но использовали мы SObjectizer в далеких от АСУТП областях. Поэтому иногда возникает ностальгия из категории «эх, давно не брал в руки шашек...» Однажды из-за этого в составе SObjectizer появился один из самых объемных примеров — machine_control. Уж очень тогда захотелось «тряхнуть стариной», смоделировать задачку управления оборудованием на современном SObjectizer-е. Ну и под шумок запихнуть в пример разные вкусные фичи SObjectizer-а вроде фильтров доставки, шаблонных агентов и диспетчера с поддержкой приоритетов. Сегодня попробуем рассказать и показать, как это все работает.

Photo by Mike Boening

Что мы моделируем?

Предположим, что мы имеем дело с какой-то машиной или со станком на производстве, у которого внутри есть двигатель. Ну, скажем, привод ленты конвейера. Или это насос, качающий воду. Не суть важно. Важно то, что когда двигатель работает, он нагревается. А когда он нагревается, его нужно охлаждать. Поэтому рядом с двигателем установлен охлаждающий вентилятор. Этот вентилятор нужно включить, если двигатель разогрелся свыше 70 градусов. Вентилятор следует выключить, если в результате охлаждения двигатель остыл до 50 градусов. Если же двигатель, не смотря на охлаждение, продолжает нагреваться и его температура достигает 95 градусов, то двигатель вовсе нужно выключить и подождать, пока он не остынет до 50 градусов.

Естественно, что нам бы хотелось видеть, как все это происходит в динамике. Нам нужно видеть работает ли сейчас двигатель, какова температура у двигателя, включен ли охлаждающий вентилятор. Для этого в реализации machine_control мы используем простую периодическую печать всей этой информации на консоль.

Ну и для того, чтобы нам было интереснее, сделаем работу не с одним двигателем, а с несколькими. Все они работают одинаковым образом, но их свойства отличаются. Какой-то из них нагревается быстрее, какой-то медленнее. Поэтому если наблюдать за работой, скажем, четырех двигателей, то будет казаться, что каждый из них живет своей жизнью.

Поехали!

Пара общих слов о выбранном решении

Понятно, что любую задачу можно решить несколькими способами. Описываемый ниже способ всего лишь один из возможных. Его выбрали не столько из-за соображений простоты и практичности, сколько из-за возможности продемонстрировать разные фичи SObjectizer-а. Поэтому лучше относиться к последующему тексту как к демонстрации. Тем более, что в продакшен все будет гораздо серьезнее и страшнее ;)

Как именно работает пример?

Есть несколько агентов-машин, есть специальный общий почтовый ящик. Агенты-машины время от времени посылают в этот общий почтовый ящик сообщения о своем текущем статусе.

Из общего почтового ящика сообщения о статусе агента-машины попадают к двум совершенно разным агентам. Первый из них, total_status_dashboard, собирает информацию о статусах агентов-машин и периодически отображает информацию о происходящем на стандартный поток вывода. Благодаря чему запустив пример мы видим приблизительно вот такую картинку:

Второй агент — это statuses_analyser. Он получает информацию о статусах для того, чтобы контролировать происходящее с машинами и определять моменты, когда какой-то из машин требуется внешнее воздействие. Когда воздействие требуется, агент statuses_analyser отсылает еще одно сообщение в этот же общий почтовый ящик. На это сообщения реагируют агенты machine_controller. Они уже решают какое именно управляющее воздействие должно быть применено к конкретной машине и отсылают соответствующее сообщение напрямую соответствующему агенту-машине.

Вот, собственно, и все. Остальные моменты в деталях постараемся рассмотреть ниже по тексту.

Агент a_machine_t

Начнем разбор реализации нашего примера «от печки», т.е. от самой машины с двигателем и охлаждающим вентилятором. В реальной жизни у нас было бы какое-то настоящее «железо», к которому были бы подключены какие-то датчики и управляторы. И со всей этой кухней нужно было бы работать из программы через тот или иной интерфейс, а может быть и через несколько разных интерфейсов.

Но у нас вымышленный пример, поэтому нам не нужно опрашивать никаких портов и считывать данные с реального оборудования. Нам нужно что-то изображающее из себя работающую машину. Для этого у нас есть агент a_machine_t, который:

• периодически выдает во внешний мир информацию о своем состоянии, как то: работает ли двигатель, работает ли вентилятор, какова текущая температура двигателя;
• принимает команды на включение/выключение двигателя и выполняет их;
• принимает команды на включение/выключение вентилятора и выполняет их.

Не смотря на то, что данный агент занимается имитацией, в свое время, в задачах АСУТП, доводилось использовать похожую логику. Агент по событию таймера обращался через некоторый интерфейс к оборудованию и собирал информацию с датчиков. После чего преобразовывал снятую информацию в нужный вид и отсылал ее тем, кто мог эту информацию должным образом обработать.

Посмотрим на то, как выглядит этот самый агент a_machine_t, но сперва введем несколько определений, которые нам понадобятся в реализации a_machine_t:

// Специальный тип для обозначения состояния двигателя.
enum class engine_state_t { on, off };
// Специальный тип для обозначения состояния вентилятора.
enum class cooler_state_t { on, off };
 
// Управляющие сообщения-сигналы для включения и выключения двигателя.
struct turn_engine_on : public so_5::signal_t {};
struct turn_engine_off : public so_5::signal_t {};
 
// Управляющие сообщения-сигналы для включения и выключения вентилятора.
struct turn_cooler_on : public so_5::signal_t {};
struct turn_cooler_off : public so_5::signal_t {};
 
// Описание текущего состояния машины.
struct machine_status
{
	// Уникальное название-идентификатор для машины.
	// Необходимо для того, чтобы отличать машины друг от друга.
	const std::string m_id;
	// Статус двигателя машины.
	const engine_state_t m_engine_status;
	// Статус вентилятора машины.
	const cooler_state_t m_cooler_status;
	// Текущая температура двигателя.
	const float m_engine_temperature;
};

Соответственно, наш агент a_machine_t будет получать управляющие команды в виде сообщений-сигналов turn_engine_on/turn_engine_off и turn_cooler_on/turn_cooler_off, а о своем состоянии будет извещать посредством отсылки сообщения machine_status.

Теперь уже можно перейти к рассмотрению самого агента a_machine_t. Начнем с самых потрохов:

class a_machine_t : public so_5::agent_t
{
	// Этот сигнал будет использоваться для периодического информирования
	// о текущем статусе машины.
	struct update_status : public so_5::signal_t {};
 
	// Агент-машина может находиться в двух состояниях:
	// состояние, когда двигатель включен,
	const state_t st_engine_on{ this, "on" };
	// состояние, когда двигатель выключен.
	const state_t st_engine_off{ this, "off" };
 
	// Уникальный идентификатор-название этой машины.
	const std::string m_id;
	// В этот почтовый ящик будет отсылаться сообщение machine_status.
	const so_5::mbox_t m_status_distrib_mbox;
 
	// Уникальные параметры конкретной машины:
	// начальная температура,
	const float m_initial_temperature;
	// шаг изменения температуры двигателя во время нагревания,
	const float m_engine_heating_step;
	// шаг изменения температуры двигателя во время охлаждения.
	const float m_cooler_impact_step;
 
	// Текущая температура двигателя.
	float m_engine_temperature;
 
	// Текущие состояния двигателя и вентилятора.
	engine_state_t m_engine_status = engine_state_t::off;
	cooler_state_t m_cooler_status = cooler_state_t::off;
 
	// ID таймера для периодического сообщения update_status.
	// В SO-5 для периодических сообщений ID таймера нужно сохранять
	// в течении всего времени работы, иначе произойдет автоматическая
	// отмена периодического сообщения.
	so_5::timer_id_t m_update_status_timer;

Агент a_machine_t представляет из себя очень простой конечный автомат с двумя состояниями: «двигатель включен» и «двигатель выключен». В каждом из них он реагирует на некоторые сообщения по-разному. Для того, чтобы представить агента в виде конечного автомата в SObjectizer нам и потребовались два отдельных атрибута st_engine_on и st_engine_off.

При своем старте агент инициирует периодическое сообщение update_status. Каждый раз, когда он получает это сообщение, агент пересчитывает значение m_engine_temperature с учетом того, работают ли сейчас двигатель и вентилятор или нет. После чего в почтовый ящик m_status_distrib_mbox отсылается сообщение machine_status с текущими показаниями.

Конструктор агента выглядит объемным из-за обилия начальных параметров, но на самом деле он тривиален, поэтому мы его рассматривать не будем.

А вот дальше уже интереснее. Во-первых, это специальный метод so_define_agent(), который используется для того, чтобы агент мог настроить себя для работы внутри SObjectizer. Нашему a_machine_t нужно перейти в свое начальное состояние и подписаться на нужные ему сообщения. Вот как это выглядит:

virtual void so_define_agent() override
{
	this >>= st_engine_off;
 
	st_engine_on
		.event< turn_engine_off >( &a_machine_t::evt_turn_engine_off )
		.event< turn_cooler_on >( &a_machine_t::evt_turn_cooler_on )
		.event< turn_cooler_off >( &a_machine_t::evt_turn_cooler_off )
		.event< update_status >( &a_machine_t::evt_update_status_when_engine_on );
	st_engine_off
		.event< turn_engine_on >( &a_machine_t::evt_turn_engine_on )
		.event< turn_cooler_on >( &a_machine_t::evt_turn_cooler_on )
		.event< turn_cooler_off >( &a_machine_t::evt_turn_cooler_off )
		.event< update_status >( &a_machine_t::evt_update_status_when_engine_off );
}

Можно обратить внимание, что на сигнал update_status агент в разных состояниях реагирует посредством разных обработчиков. Также можно увидеть, что в состоянии st_engine_on сигнал turn_engine_on игнорируется, поскольку нет смысла включать уже работающий двигатель. Аналогично и с turn_engine_off в состоянии st_engine_off.

По поводу метода so_define_agent() нельзя не сделать маленького лирического отступления: на первый взгляд этот метод выглядит избыточным и кажется, что без него можно было бы обойтись. Ведь можно все настройки агента выполнить прямо в конструкторе.

Действительно, можно. В простейших случаях так и делается. Но вот если используется наследование для агентов, то настройка агента в so_define_agent() оказывается более удобной, чем в конструкторе. Классы наследники получают простой и удобный способ вмешательства в настройки базового класса (например, можно просто не вызывать so_define_agent() для базового класса, тогда все настройки будут сделаны производным классом).

Кроме того, по нашему опыту, отдельный метод so_define_agent() начинает себя оправдывать в больших проектах, написанных разными людьми: открываешь чужой код, сразу же заглядываешь в so_define_agent() чужого агента и видишь все в одном месте. Сильно экономит время и силы.

Следующий важный метод — это so_evt_start(). SObjectizer автоматически вызывает его у всех агентов, которые начинают работать внутри SObjectizer-а. Наш a_machine_t использует so_evt_start() для того, чтобы начать отсылку периодического сообщения update_status:

virtual void so_evt_start() override
{
	// Запускаем периодическое сообщение типа update_status и сохраняем
	// ID таймера, чтобы сообщение продолжало отсылаться до тех пор,
	// пока агент существует.
	m_update_status_timer = so_5::send_periodic< update_status >(
			// Указываем себя в качестве получателя сообщения.
			*this,
			// Нет задержки перед появлением сообщения в первый раз.
			std::chrono::milliseconds(0),
			// Далее сообщение будет отсылаться каждые 200ms.
			std::chrono::milliseconds(200) );
}

Далее идут обработчики событий агента a_machine_t. Обработчиком события называется метод, который SObjectizer вызовет когда агент получит соответствующее сообщение-инцидент. Соответствие между сообщением-инцидентом и обработчиком задается при подписке на сообщение. Так, подписка вида:

st_engine_on
	.event< turn_engine_off >( &a_machine_t::evt_turn_engine_off )

указывает SObjectizer-у, что когда агенту приходит сообщение типа turn_engine_off, то у агента нужно вызвать метод evt_turn_engine_off().

У агента a_machine_t есть четыре простых обработчика, пояснять работу которых нет смысла:

void evt_turn_engine_off()
{
	// Меняем текущее состояние агента.
	this >>= st_engine_off;
	// Обновляем соответствующий статус.
	m_engine_status = engine_state_t::off;
}
 
void evt_turn_engine_on()
{
	this >>= st_engine_on;
	m_engine_status = engine_state_t::on;
}
 
void evt_turn_cooler_off()
{
	// Состояние агента менять не нужно.
	// Поэтому просто обновляем соответствующий статус.
	m_cooler_status = cooler_state_t::off;
}
 
void evt_turn_cooler_on()
{
	m_cooler_status = cooler_state_t::on;
}

А вот по поводу реакций на периодическое сообщение update_status нужно будет дать несколько пояснений. Сперва посмотрим на реакцию на update_status когда двигатель работает:

void evt_update_status_when_engine_on()
{
	m_engine_temperature += m_engine_heating_step;
	if( cooler_state_t::on == m_cooler_status )
		m_engine_temperature -= m_cooler_impact_step;
 
	distribute_status();
}

Поскольку двигатель у нас работает, а значит и нагревается, то мы должны увеличить текущую температуру. Однако, если включен охлаждающий вентилятор, то температуру нужно откорректировать с учетом влияния охлаждения.

Когда сигнал update_status обрабатывается в состояние st_engine_off, т.е. когда двигатель отключен, нам нужно только учесть влияние охлаждение, если оно сейчас включено:

void evt_update_status_when_engine_off()
{
	if( cooler_state_t::on == m_cooler_status )
	{
		m_engine_temperature -= m_cooler_impact_step;
		// В данной имитации мы не позволяем температуре двигателя
		// опуститься слишком низко.
		if( m_engine_temperature < m_initial_temperature )
			m_engine_temperature = m_initial_temperature;
	}
 
	distribute_status();
}

Ну а вспомогательный метод distribute_status() имеет совсем тривиальную реализацию, т.к. его единственная задача — это отсылка сообщения machine_status в специально предназначенный для этого почтовый ящик:

void distribute_status()
{
	// Внутри send-а создается экземпляр типа machine_status,
	// который инициализируется параметрами из вызова send.
	// После чего этот экземпляр будет отослан в почтовый ящик
	// m_status_distrib_mbox.
	so_5::send< machine_status >(
			// Куда направляется сообщение.
			m_status_distrib_mbox,
			// Все остальные аргументы используются для конструирования
			// экземпляра сообщения machine_status.
			m_id,
			m_engine_status,
			m_cooler_status,
			m_engine_temperature );
}

Куда же улетают сообщения machine_status?

Возможно, у кого-то из читателей возник вопрос: а почему же a_machine_t сам не принимает решения по поводу включения/выключения вентилятора и двигателя? Почему вместо того, чтобы самостоятельно анализировать происходящее и предпринимать соответствующие действия, a_machine_t лишь периодически отсылает куда-то сообщения о своем текущем состоянии?

Это потому, что декомпозиция :)

Когда мы проектируем какую-то систему на основе объектно-ориентированного подхода, то мы стремимся к тому, чтобы каждый объект отвечал за свою задачу и нужный нам эффект достигался бы за счет их комбинирования. Точно так же происходит и при проектировании системы на базе акторов (агентов): пусть каждый актор отвечает за свою задачу, а мы сформируем решение за счет их комбинации.

Поэтому агент a_machine_t решает только одну задачу: реализует интерфейс с оборудованием (в нашем примере он имитирует этот интерфейс). В реальной задаче a_machine_t бы занимался выставлением/сбросом битиков, чтением-записью байтиков в какой-то коммуникационный порт с проверками успешности операций ввода-вывода, контролем скорости передачи данных, тайм-аутов и пр. низкоуровневыми вещами.

Так что задача a_machine_t — это получить от устройства осмысленную информацию, пригодную для дальнейшей обработки, отдать эту информацию кому-то наверх, принять сверху команду для устройства, преобразовать эту команду в последовательность понимаемых устройством воздействий. Что a_machine_t в меру ограничений конкретной имитации и делает.

Агент a_total_status_dashboard_t

Агент a_total_status_dashboard_t должен собирать сообщения machine_status, агрегировать информацию из этих сообщений и периодически отображать агрегированную информацию в стандартный поток вывода.

Для выполнения своей работы a_total_status_dashboard_t подписывается на два сообщения:

virtual void so_define_agent() override
{
	so_subscribe( m_status_distrib_mbox )
		.event( &a_total_status_dashboard_t::evt_machine_status );
 
	so_subscribe_self().event< show_dashboard >(
			&a_total_status_dashboard_t::evt_show_dashboard );
}

Первое сообщение, machine_status, агент ожидает из специального почтового ящика, в который агенты a_machine_t отсылают свои machine_status-сообщения. А второе сообщение, show_dashboard, агент a_total_status_dashboard_t отсылает себе сам в виде периодического сообщения:

virtual void so_evt_start() override
{
	// Запускаем периодический сигнал для отображения
	// текущей информации на консоль.
	const auto period = std::chrono::milliseconds( 1500 );
	m_show_timer = so_5::send_periodic< show_dashboard >( *this,
			period, period );
}

Здесь a_total_status_dashboard_t использует тот же самый подход, как и агент a_machine_t — инициирует периодическое сообщение в своем методе so_evt_start(), который SObjectizer автоматически вызывает в самом начале работы a_total_status_dashboard_t.

Обработка machine_status очень простая: нужно всего лишь сохранить очередную порцию информации в ассоциативный контейнер:

void evt_machine_status( const machine_status & status )
{
	m_machine_statuses[ status.m_id ] = one_machine_status_t{
			status.m_engine_status, status.m_cooler_status,
			status.m_engine_temperature
		};
}

Да и обработчик show_dashboard не содержит ничего сложного: всего лишь итерация по содержимому ассоциативного контейнера с печатью на стандартный поток вывода:

void evt_show_dashboard()
{
	auto old_precision = std::cout.precision( 5 );
	std::cout << "=== The current status ===" << std::endl;
 
	for( const auto & m : m_machine_statuses )
	{
		show_one_status( m );
	}
 
	std::cout << "==========================" << std::endl;
	std::cout.precision( old_precision );
}

Поскольку агент a_total_status_dashboard_t не представляет большого интереса и его реализация весьма проста, то углубляться дальше в его реализацию мы не будем. Кому интересно, полный исходный код a_total_status_dashboard_t можной увидеть здесь.

Агенты a_statuses_analyser_t и a_machine_controller_t

В данной реализации примера machine_control анализом информации от агентов a_machine_t и выдачей управляющих команд агентам-машинам занимается связка из нескольких агентов.

Во-первых, это агент a_statuses_analyser_t, который получает сообщения machine_status, анализирует их и определяет, что конкретный a_machine_t нуждается в каком-то воздействии.

Во-вторых, есть группа агентов-шаблонов типа a_machine_controller_t, которые реагируют на сигналы от a_statuses_analyser_t и выдают то или иное воздействие на конкретную машину. Так, один агент a_machine_controlle_t реагирует на ситуацию, когда следует включить охлаждающий вентилятор и отсылает сообщение turn_cooler_on соответствующему агенту a_machine_t. Другой агент a_machine_controller_t реагирует на ситуацию, когда следует отключить двигатель и отсылает сообщение turn_engine_off. И т.д.

Вообще говоря, такое деление на a_statuses_analyser_t и a_machine_controller_t — это явное усложнение нашего примера. Можно было бы вполне обойтись всего одним агентом a_statuses_analyser_t, который бы сам мог и информацию анализировать, и управляющие команды отсылать. Хотя, скорее всего, агент a_statuses_analyser_t сильно бы увеличился в объеме при этом.

Изначально в machaine_control хотелось показать разнообразные фичи SObjectizer-а, в частности, использование агентов-шаблонов и приоритетов агентов, поэтому мы пошли на разделение логики между a_statuses_analyser_t и a_machine_controller_t.

Итак, суть взаимодействия a_machine_t, a_statuses_analyser_t и a_machine_controller_t в следующем:

• a_machine_t периодически рассылает информацию о себе в виде сообщения machine_status;
• a_statuses_analyser_t собирает информацию из machine_status и определяет, нужно ли какое-то воздействие на конкретный a_machine_t. Если воздействие нужно, то a_status_analyser_t отсылает сообщение machine_needs_attention с информацией о том, какой машине какое воздействие требуется;
• агенты a_machine_controller_t реагируют на сообщения machine_needs_attention и генерируют сообщения turn_engine_on/turn_engine_off и/или turn_cooler_on/turn_cooler_off для конкретного агента a_machine_t.

Сообщение machine_needs_attention имеет следующий вид:

// Перечисление, которое определяет какую именно ситуацию диагностировали.
enum class attention_t 
{
	none,
	engine_cooling_done,
	engine_cooling_needed,
	engine_overheat_detected
};
 
// Сообщение о том, что с конкретной машиной что-то произошло.
struct machine_needs_attention
{
	// Уникальный идентификатор-название машины.
	const std::string m_id;
	// Что именно диагностировали.
	const attention_t m_attention;
	// Текущий статус двигателя у машины (включен/выключен).
	const engine_state_t m_engine_status;
	// Текущий статус охлаждающего вентилятора (включен/выключен).
	const cooler_state_t m_cooler_status;
};

Агент a_statuses_analyser_t хранит прошлую информацию о каждой машине и сравнивает ее с новой информацией, поступающей с сообщением machine_status. Если обнаруживается, что двигатель требует охлаждения или что двигатель перегрелся, или что двигатель достиг безопасной температуры, то a_statuses_analyser_t генерирует сообщение machine_needs_attention. Это сообщение подхватывается соответствующим a_machine_controller_t и нужному агенту a_machine_t будет отослана нужная команда.

Подробнее об агенте a_statuses_analyser_t

Потроха у агента a_statuses_analyser_t довольно объемные. Но большая их часть относится к тому, чтобы хранить и анализировать текущее состояние агентов-машин. Разбирать эту часть в деталях мы не будем (если возникнут какие-то вопросы, то я на них отвечу в комментариях), просто поясним в двух словах:

• агент a_statuses_analyser_t хранит ассоциативный контейнер с информацией о последнем известном статусе каждого агента-машины. Ключем в этом контейнере является уникальное имя-идентификатор машины;
• при получении очередного сообщения machine_status агент сверяет свежую информацию с тем, что у него было сохранено ранее. Если обнаруживаются ситуации, когда требуется воздействие на агента-машину, то генерируется сообщение machine_needs_attention;
• после чего агент a_statuses_analyser_t обновляет данные в своем ассоциативном контейнере.

В коде все это занимает приличное количество строк, но ничего сложного там нет.

А вот та часть агента a_statuses_analyser_t, которая относится к взаимодействию с SObjectizer-ом, вообще минимальна: всего лишь подписка на одно-единственное сообщение в so_define_agent() и одно-единственное событие для этого сообщения:

virtual void so_define_agent() override
{
	so_subscribe( m_status_distrib_mbox ).event(
			&a_statuses_analyzer_t::evt_machine_status );
}
void evt_machine_status( const machine_status & status )
{
	auto it = m_last_infos.find( status.m_id );
	if( it == m_last_infos.end() )
		// Об этой машине мы еще информацию не получали.
		// Добавим информацию в наше хранилище.
		it = m_last_infos.insert( last_info_map_t::value_type {
				status.m_id,
				last_machine_info_t {
					attention_t::none,
					status.m_engine_temperature
				} } ).first;
 
	handle_new_status( status, it->second );
}

Где метод handle_new_status(), который вызывается внутри evt_machine_status(), это уже часть той прикладной логики контроля за статусом агента-машины, о которой мы коротко рассказали ранее.

Подробнее об агентах a_machine_controller_t

В чем вообще смысл a_machine_controller_t?

В наличии нескольких экземпляров агентов a_machine_controller_t есть следующий смысл: когда агент a_statuses_analyser_t определяет, например, что двигатель у машины остыл до безопасной температуры, то в нам может потребоваться сделать несколько разных действий. Во-первых, выключить охлаждающий вентилятор. Во-вторых, может оказаться, что двигатель был предварительно отключен и нам нужно включить его вновь.

Если бы мы разместили всю эту логику внутри a_statuses_analyser_t, то его код бы разросся и понимать его было бы труднее. Вместо этого мы просто заставили a_statuses_analyser_t объявлять на что именно следует обратить внимание в работе агента-машины. А уже реагировать на это объявление будут агенты a_machine_controller_t. И каждый a_machine_controller_t сам определяет, нужно ли ему реагировать или нет. Если нужно, то a_machine_contoller_t получает сообщение machine_needs_attention и инициирует соответствующую команду агенту-машине. Если не нужно, то a_machine_controller_t сообщение machine_needs_attention просто игнорирует.

Агент a_machine_controller_t

Код агента-шаблона a_machine_controller_t небольшой, поэтому для простоты работы с ним приведем весь код этого агента полностью:

template< class LOGIC >
class a_machine_controller_t : public so_5::agent_t
{
public :
	a_machine_controller_t(
		context_t ctx,
		so_5::priority_t priority,
		so_5::mbox_t status_distrib_mbox,
		const machine_dictionary_t & machines )
		:	so_5::agent_t( ctx + priority )
		,	m_status_distrib_mbox( std::move( status_distrib_mbox ) )
		,	m_machines( machines )
		,	m_logic()
	{}
 
	virtual void so_define_agent() override
	{
		so_set_delivery_filter( m_status_distrib_mbox,
			[this]( const machine_needs_attention & msg ) {
				return m_logic.filter( msg );
			} );
 
		so_subscribe( m_status_distrib_mbox )
			.event( [this]( const machine_needs_attention & evt ) {
					m_logic.action( m_machines, evt );
				} );
	}
 
private :
	const so_5::mbox_t m_status_distrib_mbox;
 
	const machine_dictionary_t & m_machines;
 
	const LOGIC m_logic;
};

Итак, это шаблонный класс, который параметризуется одним параметром: типом прикладной логики, которую должен иметь конкретный a_machine_controller_t. Этот тип LOGIC должен быть типом с двумя методами следующего вида:

struct LOGIC
{
	bool filter( const machine_needs_attention & msg ) const;
 
	void action(
		const machine_dictionary_t & machines,
		const machine_needs_attention & evt ) const;
};

Если бы в C++11 были концепты, то можно было бы объявить соответствующий концепт для того, чтобы проще было определять, какой тип может быть параметром для шаблона a_machine_controller_t, а какой — нет. Но, т.к. в C++11 концепты не завезли, то приходится полагаться на утиную типизацию.

Агент a_machine_controller_t создает у себя внутри экземпляр типа LOGIC и делегирует все свои действия этому экземпляру. И этот экземпляр есть не что иное, как контроллер, который выдает управляющие воздействия агенту-машине.

Действий же у контроллера всего два:

Во-первых, нужно отфильтровать сообщения, которые конкретному контроллеру не интересны. Для этого a_machine_controller_t назначает фильтр доставки сообщений:

so_set_delivery_filter( m_status_distrib_mbox,
	[this]( const machine_needs_attention & msg ) {
		return m_logic.filter( msg );
	} );

Фильтр доставки — это специальный механизм SObjectizer-а. Он нужен в ситуациях, когда агент подписывается на некоторое сообщение типа T из почтового ящика M, но хочет получать не все сообщения типа T, а только те, внутри которых содержится интересная агенту информация.

В этих случаях агент посредством своего метода so_set_delivery_filter() задает фильтр, лямбда-функцию, которую почтовый ящик M будет автоматически вызывать каждый раз, когда в него приходит сообщение типа T. Если фильтр возвращает true, значит экземпляр сообщения агенту интересен и доставка сообщения до агента выполняется. Если же фильтр возвращает false, то сообщение этому агенту не доставляется, тем самым агент не отвлекается на то, что ему не интересно.

Наглядная схема работы фильтра доставки. Агент Subscriber-1 не использует фильтр доставки и поэтому получает все сообщения из mbox-а. Тогда как агент Subscriber-2 установил фильтр доставки и получает только те сообщения, которые проходят через фильтр.

Вот метод LOGIC::filter() и должен играть роль фильтра доставки, он пропускает только те экземпляры machine_needs_attention, которые интересны контроллеру.

Во-вторых, контроллер должен что-то предпринять при получении сообщения machine_needs_attention, в котором уже точно лежит информация, интересная для контроллера.

За это отвечает метод LOGIC::action(). Агент a_machine_controller_t вызывает этот метод в своем обработчике machine_needs_attention:

so_subscribe( m_status_distrib_mbox )
	.event( [this]( const machine_needs_attention & evt ) {
		m_logic.action( m_machines, evt );
	} );

Примеры нескольких controller-ов

Ну а для того, чтобы было понятнее, как же именно работают a_machine_controller_t, приведем пару примеров контроллеров. Первый очень простой, он отвечает за выдачу команды turn_engine_off, если обнаружили перегрев двигателя:

struct engine_stopper_t
{
	bool filter( const machine_needs_attention & msg ) const
	{
		return msg.m_attention == attention_t::engine_overheat_detected;
	}
 
	void action(
		const machine_dictionary_t & machines,
		const machine_needs_attention & evt ) const
	{
		so_5::send< turn_engine_off >( machines.find_mbox( evt.m_id ) );
	}
};

Т.е. фильтруются только те сообщения machine_needs_attention, которые говорят о перегреве двигателя. Когда же такое сообщение поступает, то соответствующему агенту-машине отсылается сигнал turn_engine_off.

А вот этот контроллер уже посложнее:

struct cooler_starter_t
{
	bool filter( const machine_needs_attention & msg ) const
	{
		return (msg.m_attention == attention_t::engine_overheat_detected ||
				msg.m_attention == attention_t::engine_cooling_needed) &&
				msg.m_cooler_status == cooler_state_t::off;
	}
 
	void action(
		const machine_dictionary_t & machines,
		const machine_needs_attention & evt ) const
	{
		so_5::send< turn_cooler_on >( machines.find_mbox( evt.m_id ) );
	}
};

Он отвечает за включение охлаждающего вентилятора. Этот вентилятор нужно включать только если на данный момент вентилятор отключен, а двигатель либо просто нуждается в охлаждении, либо уже перегрелся.

Всего таких контроллеров в примере четыре:

• engine_stopper_t отвечает за останов двигателя;
• engine_starter_t отвечает за запуск двигателя;
• cooler_starter_t отвечает за запуск вентилятора;
• cooler_stopper_t отвечает за останов вентилятора.

Это означает, что и агентов a_machine_controller_t будет четыре — по одному на каждый тип контроллера.

Приоритеты для агентов a_machine_controller_t

При работе примера machine_control может случиться так, что один и тот же экземпляр machine_needs_attention будет получен несколькими контроллерами. Например, если в machine_needs_attention передается attention_t::engine_overheat_detected и cooler_status_t::off, то такое сообщение будет получено двумя контроллерами: и контроллером engine_stopper_t, и контроллером cooler_starter_t.

И тут нам важно, в каком порядке эти контроллеры обработают сообщение. Логично же, чтобы в ситуации engine_overheat_detected сперва была дана команда turn_engine_off, а уже затем команда turn_cooler_on. Ведь выполнение команды будет занимать какое-то время, скажем несколько секунд. Если сперва потратить эти секунды на включение вентилятора, то двигатель перегреется еще больше. Поэтому лучше сразу выключить двигатель, а потом уже тратить время на все остальное.

Каким же образом в SObjectizer-е можно управлять порядком обработки сообщений?

Вопрос совсем непростой. Есть опасение, что попытка развернуто ответить на него потребует написания еще одной статьи. Если говорить коротко, то гарантии очередности обработки сообщений могут быть только для диспетчеров с одной рабочей нитью. Но и этого еще недостаточно. Нужно, чтобы очередь сообщений на этой рабочей нити упорядочивалась с учетом приоритетов агентов, которым сообщения адресуются.

В SObjectizer есть такой тип диспетчера: он запускает всех привязанных к нему агентов на одной общей рабочей нити, при этом агенты с более высоким приоритетом обрабатывают свои сообщения перед агентами с более низкими приоритетами. Это диспетчер so_5::disp::prio_one_thread::strictly_ordered. Именно к нему привязываются агенты a_machine_controller_t в данном примере.

При этом приоритеты для агентов-контроллеров распределены следующим образом:

• самый высокий приоритет у агента с контроллером типа engine_stopper_t, т.к. команды на выключение двигателя должны выполняться в первую очередь;
• далее следует агент с контроллером типа cooler_starter_t, поскольку включение охлаждающего вентилятора — это вторая по значимости команда;
• далее идет агент с контроллером типа engine_starter_t;
• ну и самый низкий приоритет у агента с контроллером типа cooler_stopper_t.

Следы того, как задаются приоритеты для агентов a_machine_controller_t можно найти в конструкторе:

a_machine_controller_t(
	...,
	so_5::priority_t priority,
	... )
	:	so_5::agent_t( ctx + priority )
	,	...
{}

И при непосредственном создании таких агентов:

coop.make_agent_with_binder< a_machine_controller_t< engine_stopper_t > >(
		disp->binder(),
		so_5::prio::p4,
		status_distrib_mbox,
		machines );
 
coop.make_agent_with_binder< a_machine_controller_t< cooler_starter_t > >(
		disp->binder(),
		so_5::prio::p3,
		status_distrib_mbox,
		machines );

Здесь значения so_5::prio::p4 и so_5::prio::p3 — это и есть приоритеты агентов.

Агент-стартер

Можно было обратить внимание, что агенты a_machine_t начинают свою работу в состоянии st_engine_off, т.е. двигатель считается выключенным. Но запустив пример можно увидеть, что у машин двигатель сразу же включается. Как же так происходит, если ни a_machine_t, ни a_machine_controller_t изначально никаких команд на включение двигателя не выдают?

Для того, чтобы у агентов-машин заработал двигатель используется отдельный агент-стартер, единственной задачей которого является выдача команды turn_engine_on каждому агенту-машине.

Этот агент настолько прост, что для него мы даже не делаем отдельный C++ класс, как это происходило у нас с a_machine_t, a_total_status_dashboard_t, a_statuses_analyser_t и a_machine_controller_t. Вместо этого мы используем такую штуку, как ad-hoc агент. Т.е. агент, который создается из набора заданных пользователем лямбда-функций. В примере machine_control это делается следующим образом:

coop.define_agent().on_start( [&dict] {
		dict.for_each(
			[]( const std::string &, const so_5::mbox_t & mbox ) {
				so_5::send< turn_engine_on >( mbox );
			} );
	} );

Метод define_agent() создает пустого ad-hoc агента, которого пользователь может наполнить нужной функциональностью. Здесь мы задаем всего лишь реакцию на начало работы ad-hoc агента внутри SObjectizer-а: при старте всем агентам просто отсылается turn_engine_on.

Распределение агентов по диспетчерам

Одной из самых важных возможностей SObjectizer-а, которая постоянно используется на практике, является возможность задать рабочую нить для агента через привязку агента к тому или иному диспетчеру.

Каждому агенту нужно обрабатывать доставленные ему сообщения. Делать это нужно на контексте какой-то рабочей нити. В SObjectizer рабочую нить агентам предоставляют диспетчеры. Поэтому программист при создании своего агента должен привязать агента к тому диспетчеру, который обеспечит агенту нужный режим работы.

В примере machine_control используются несколько диспетчеров, благодаря чему происходит следующее распределение агентов по рабочим нитям:

• все агенты-машины работают на одной общей рабочей нити, для чего эти агенты привязываются к отдельному диспетчеру типа one_thread;
• агент a_total_status_dashboard_t работает на собственной рабочей нити (привязан к отдельному диспетчеру типа one_thread);
• агент a_statuses_analyser_t работает на собственной рабочей нити (привязан к отдельному диспетчеру типа one_thread);
• все агенты a_machine_controller_t работают на одной общей рабочей нити отдельного диспетчера prio_one_thread::strictly_ordered, который обслуживает события агентов с учетом приоритетов;
• агент-стартер отрабатывает свое единственное событие на диспетчере по умолчанию.

Что еще осталось «за кадром»?

В статье мы не стали рассматривать одну важную для реализации вещь, которая, однако, никак не влияет на поведение агентов в примере: это machine_dictionary. Специальный ассоциативный словарь, в котором сохранено соответствие между уникальным названием-идентификатором агента-машины и его персональным почтовым ящиком.

Этот словарь используется для того, чтобы по имени агента-машины (которое присутствует в сообщениях machine_status и machine_needs_attention) получить доступ к почтовому ящику соответствующего агента. Ведь сообщения в SObjectizer отсылаются в почтовые ящики, поэтому, для того, чтобы отослать turn_engine_on, нужно по имени получить почтовый ящик. Что и делается посредством machine_dictionary.

Ну а что же самое важное в этом примере?

Теперь же, как в любом уважающем себя произведении, вслед за долгим прологом должна последовать основная часть, никакого отношения к прологу не имеющая.

Если читатель подумал, что статья была написана для того, чтобы показать, как можно разрабатывать агентов, похожих на a_machine_t или a_machine_controller_t, то он оказался не совсем прав.

Демонстрация конкретных фич SObjectizer-а, вроде привязки агентов к разным диспетчерам или использование ad-hoc агентов для простых действий — это лишь одна из задач данной статьи. Мы надеемся, что знакомство с тем, как может выглядеть более-менее объемный код на SObjectizer, поможет читателям лучше понять, нравится ли им то, что они видят или нет, возникнет ли у них желание разрабатывать свои программы в таком стили или же лучше пойти «другим путем». В конце-концов, примеры чуть посложнее, чем ставший классическим и совершенно бесполезным ping-pong, дают намного лучшее представление о том, как будет выглядеть продакшен код. И захочется ли связываться после подобного знакомства с таким фреймворком, как SObjectizer.

Однако, есть у данного примера и другая задача, не менее важная. Эта задача состоит в том, чтобы показать, как посредством Actor Model и Publish/Subscribe организовать взаимодействие между независимыми сущностями в программе.

Сделаем акцент на взаимодействии

Ключевой момент в рассмотренном примере machine_control в том, что взаимодействующие сущности (т.е. агенты) не имеют никакого понятия друг о друге и о взаимосвязях между собой.

Действительно, агент a_machine_t просто отсылает сообщение machine_status в некоторый почтовый ящик и понятия не имеет, кто стоит за этим ящиком, куда реально попадает сообщение и что с этим сообщением происходит дальше. Просто агент a_machine_t выполняет свою работу (имитирует периодический опрос оборудования) и выдает результат своей работы во внешний мир. Все, больше его ничего не интересует.

Даже когда к агенту a_machine_t приходит внешняя команда (вроде turn_engine_on), он не представляет, кто именно эту команду ему прислал. От a_machine_t требуется лишь выполнить поступившую команду, если в этом есть смысл. А уж из каких соображений эта команда возникла, через что она прошла по дороге — все это для a_machine_t не имеет никакого значения.

Кстати говоря, когда кто-то выдает команду вроде turn_engine_on конкретному агенту a_machine_t, то это классическое взаимодействие в рамках Actor Model, где акторы общаются в режиме 1:1.

Но вот в случае с сообщениями machine_status и machine_needs_attention мы уже оказываемся за рамками Actor Model, т.к. у нас образуется взаимодействие 1:N. И тут мы пользуемся моделью Publish/Subscribe. Специальный почтовый ящик, куда агенты отсылают сообщения machine_status и machine_needs_attention, играет роль брокера. Отсылка сообщения в этот ящик — это просто-напросто операция Publish. А для получения сообщения из ящика необходимо выполнить операцию Subscribe, что, собственно, агенты a_total_status_dashboard_t, a_statuses_analyser_t и a_machine_controller_t и делают в своих so_define_agent().

И тут можно увидеть еще одну важную штуку, которую дает разработчику SObjectizer: возможность использовать один и тот же экземпляр сообщения для совершенно разных целей. Происходит это как раз благодаря тому, что в SObjectizer есть multi-producer/multi-consumer почтовые ящики, через которые и происходит общение агентов в режиме 1:N.

Действительно, сообщение machine_status, например, получают и обрабатывают два разных агента, решающих совершенно разные задачи и ничего не знающие друг о друге. Агент a_total_status_dashboard_t использует machine_status для периодического отображения хода работы примера на консоль. Тогда как a_statuses_analyser_t использует machine_status для контроля за агентами a_machine_t.

В этом заложена большая гибкость. Мы совершенно спокойно можем добавить еще одного агента, который мог бы собирать информацию о температуре двигателей для построения графиков изменения температуры. И наличие этого агента никак не повлияет на работу a_total_status_dashboard_t и a_statuses_analyser_t. Или же мы можем заменить агента a_total_status_dashboard_t на какого-нибудь a_gui_status_dashboard_t, который будет отображать ход работы примера не на std::cout, а в графическое окно. И это, опять же, никак не повлияет на других агентов приложения, которые работают с сообщениями machine_status.

Так что разработка приложений на SObjectizer — это все про создание агентов, каждый из которых выполняет свою независимую часть работы, и про каналы связи между агентами. А каналы связи в SObjectizer представляют из себя почтовые ящики и сообщения, посредством которых информация распространяется (хотя есть и CSP-шные каналы).

И как раз основная сложность разработки приложений на SObjectizer состоит в том, чтобы выделить в предметной области те сущности, которые в программе могут быть выражены в виде агентов. А также в выделении и формировании каналов связи (т.е. наборов сообщений и почтовых ящиков), посредством которых агенты будут взаимодействовать друг с другом.

Ну а уже внешний вид самих агентов, их привязка к диспетчерам и пр. технические детали — это всего лишь детали. Но детали, которые оказывают важное влияние на привлекательность конкретного фреймворка. Ну вот не понравится кому-то, что для объявления сообщения-сигнала нужно сделать структуру, отнаследованную от типа с именем so_5::signal_t, и ничего уже не поделать. Хотя суть вовсе не в том, нужно ли наследоваться от чего-то или что наследоваться нужно от типа, название которого не отвечает чьим-то эстетическим представлениям. А в том, чтобы выделить в предметной области сущности, для какого-то взаимодействия между которыми вот такое сообщение-сигнал будет необходимо.

Вместо заключения

В финале статьи хочется поблагодарить самых настойчивых читателей за терпение, прекрасно понимаем, что дочитать до заключения было непросто.

Заодно хотим похвастаться тем, что для SObjectizer-а появился сопутствующий проект so_5_extra, который будет содержать дополнения и расширения для SObjectizer. Смысл в том, чтобы в SObjectizer включать только самую базовую функциональность, без которой ну никак не обойтись, и которая будет нужна большинству пользователей. Тогда как so_5_extra может содержать в том числе и экзотические вещи, необходимые лишь узкому кругу пользователей. Так что если вам чего-то не хватает в SObjectizer, то скажите чего именно. Если мы не увидим смысла добавления этого в SObjectizer, то место вполне может найтись в so_5_extra.

https://habrahabr.ru/post/332166/