[prog.c++.sobjectizer] Разбор примера простой реализации дедлайнов для сообщений

В продолжение темы, начатой заметками "Реализация дедлайнов для сообщений своими руками" и "Агент-collector + агент-performer = паттерн для нескольких ситуаций", в SObjectizer добавлен еще один пример simple_message_deadline. Оказалось, что в данном примере задействуется много ключевых возможностей SObjectizer-а, поэтому имеет смысл разобрать пример подробнее, объясняя, по ходу рассказа, что к чему и почему. Рассказ я постарался сделать максимально простым и подробным, ориентированным на читателя, который либо вообще не имеет представления о SObjectizer-е, либо имеет самое поверхностное.

Что делает пример? Один агент-generator инициирует запросы, у каждого из которых есть дедлайн. Т.е. для запроса задано время, до наступления которого обработка запроса должна быть начата. Если этого не случилось, то запрос должен быть отвергнут.

Агент-collector, который собирает запросы от агента-generator-а в очередь и по одному отдает их агенту-performer-у. Агент-performer имитирует обработку запросов, затрачивая по четыре секунды на каждый полученный им запрос. За то время, пока агент-performer "занимается" обработкой, у части сохраненных агентом-collector-ом запросов может наступить дедлайн. Это контролируется агентом-collector-ом, который выбрасывает просроченные запросы из очереди.

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

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

Первое сообщение -- это запрос, который идет от generator-а к collector-у, а затем и к performer-у:

// Запрос, который будет отсылаться генератором. struct msg_request : public so_5::rt::message_t {    std::string m_id;    std::time_t m_deadline;    const so_5::rt::mbox_t m_reply_to;    msg_request(       std::string id,       std::time_t deadline,       const so_5::rt::mbox_t & reply_to )       :  m_id( std::move( id ) )       ,  m_deadline( deadline )       ,  m_reply_to( reply_to )    {} };

Сообщение содержит минимальное количество полей, необходимых для работы примера. Некий идентификатор, который поможет сопоставлять запросы и ответы. Дедлайн, до наступления которого запрос должен поступить на обработку. И обратный адрес, на который должен быть отослан ответ на запрос (в SObjectizer сообщения отсылаются в "почтовые ящики", называемые mbox-ами, а ссылки на эти почтовые ящики хранятся в виде объектов mbox_t).

Далее идет определение удобного псевдонима, который позволит в некоторых местах писать более лаконично:

// Удобный псевдоним умного указателя на экземпляр запроса. using msg_request_smart_ptr_t = so_5::intrusive_ptr_t< msg_request >;

Нужен этот псевдоним вот для чего. Сообщения в SObjectizer отсылаются в виде динамически созданных объектов. Время жизни экземпляра сообщения контролируется SObjectizer-ом посредством использования интрузивного умного указателя intrusive_ptr_t. Экземпляр сообщения автоматически уничтожается после того, как исчезнут все ссылки на него. Т.е. когда сообщение будет обработано всеми подписчиками.

В данном примере агент-collector получив сообщение msg_request от generator-а должен сохранить экземпляр сообщения у себя на какое-то время. Сделать это можно либо просто скопировав сообщение (т.е. создав новый экземпляр), либо же сохранив у себя ссылку на уже имеющийся экземпляр посредством умного указателя. Второй подход более эффективен, т.к. не требует повторного выделения памяти под msg_request. Именно поэтому он используется агентом-collector-ом. А чтобы не писать intrusive_ptr_t<msg_request> вводится более удобное имя msg_request_smart_ptr_t.

Далее необходимы ответные сообщения, которые будут сообщать generator-у результат обработки запроса. Таких сообщений два. Одно используется для информирования об успешной обработке запроса. Второе -- о неудачной.

// Результат успешной обработки запроса. struct msg_positive_reply : public so_5::rt::message_t {    std::string m_id;    std::string m_result;    std::time_t m_started_at;    msg_positive_reply(       std::string id,       std::string result,       std::time_t started_at )       :  m_id( std::move( id ) )       ,  m_result( std::move( result ) )       ,  m_started_at( started_at )    {} }; // Уведомление о том, что запрос обработан не был. struct msg_negative_reply : public so_5::rt::message_t {    std::string m_id;    std::time_t m_deadline;    msg_negative_reply(       std::string id,       std::time_t deadline )       :  m_id( std::move( id ) )       ,  m_deadline( deadline )    {} };

В сообшении msg_positive_reply поле m_started_at указывает время, когда сообщение поступило на обработку. В сообщении msg_negative_reply поле m_deadline -- это копия значения поля m_deadline из msg_request. Оно нужно лишь для того, чтобы проще было убедиться в том, что дедлайн для сообщения действительно наступил.

То, что все показанные выше сообщения являются наследниками so_5::rt::message_t пока оставим без объяснений, а чуть ниже разберем этот вопрос чуть подробнее, когда появится использование еще одного базового типа so_5::rt::signal_t.

После объявления всех необходимых прикладных сообщений можно определять агента-generator-а. В данном примере у него тривиальная логика поведения: при старте он отсылает агенту-collector-у серию сообщений msg_request с разными дедлайнами, а затем ждет ответов на свои вопросы. Когда все ответы получены, работа примера завершается.

Для отсылки запросов collector-у нужно знать mbox collector-а. Этот mbox передается генератору в конструктор и сохраняется в виде атрибута агента-generator-а:

class a_generator_t : public so_5::rt::agent_t { public :    a_generator_t(       so_5::rt::environment_t & env,       const so_5::rt::mbox_t & processor_mbox )       :  so_5::rt::agent_t( env )       ,  m_processor_mbox( processor_mbox )    {}

Название processor_mbox, а не, скажем, collector_mbox, было выбрано потому, что агент-generator не знает, что за обработку его запросов будет отвечать пара агентов (collector+performer). Для генератора есть только mbox, на который нужно отослать запрос. Скрывается ли за этим mbox-ом один обработчик, два, три или больше -- уже неважно. Поэтому агент-generator ситает, что есть некий processor для его запросов, mbox которого и передается агенту-generator-у в конструкторе.

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

virtual void    so_define_agent() override    {       so_default_state()             .event( &a_generator_t::evt_positive_reply )             .event( &a_generator_t::evt_negative_reply );    }

Агент-generator обладает очень простой логикой, поэтому ему не нужно менять свое состояние и оба своих события агент подписывает для состояния по-умолчанию, которое есть у всех SObjectizer-овских агентов. Метод so_default_state() возвращает ссылку на соответствующий объект-состояние, а методы event() этого объекта выполняют подписку событий, реализованных методами evt_positive_reply() и evt_negative_reply().

В SObjectizer существует несколько вариантов метода event() для подписки событий агента. В данном случае используется самый минималистичный и простой вариант -- события агента подписываются на сообщения, которые адресованы персонально этому агенту-generator-у. Тип сообщения, которое должно приводить к возникновению события, определяется автоматически на основании сигнатуры метода-события. В данном случае сигнатуры следующие:

void evt_positive_reply( const msg_positive_reply & evt ); void evt_negative_reply( const msg_negative_reply & evt );

Что позволяет SObjectizer-у самостоятельно определить, что событие evt_positive_reply должно возникать когда агенту присылают сообщение msg_positive_reply, а событие evt_negative_reply -- когда присылают сообщение msg_negative_reply.

Для упрощения примера все заявки агент-generator рассылает в начале своей работы. У агентов в SObjectizer-е есть метод so_evt_start(), который SObjectizer автоматически вызывает при запуске агента. Агент-generator использует этот метод для того, чтобы отослать серию запросов на processor_mbox:

virtual void so_evt_start() override {    // Информация для вычисления дедлайнов отсылаемых сообщений.    unsigned int delays[] = { 1, 4, 5, 3, 9, 15, 12 };    const std::time_t now = std::time(nullptr);    int i = 0;    for( auto d : delays )    {       // Создаем идентификатор сообщения.       std::ostringstream idstream;       idstream << "i=" << i << ";d=" << d;       const std::string id = idstream.str();       // Вычисляем дедлайн сообщения.       const std::time_t deadline = now + d;       // Отсылаем запрос на обработку.       so_5::send< msg_request >( m_processor_mbox,             id,             deadline,             so_direct_mbox() );       std::cout << "sent: [" << id << "], deadline: "             << time_to_string( deadline ) << std::endl;       // Количество отосланных запросов увеличилось,       // должно увеличится количество ожидаемых ответов.       ++m_expected_replies;       ++i;    } }

Внутри метода so_evt_start() к SObjectizer-у имеет отношение только шаблонная функция so_5::send. Она используется для конструирования и отсылки сообщения типа msg_request. Первым параметром для send-а является mbox, на который сообщение должно быть отослано. Остальные параметры передаются в конструктор нового экземпляра msg_request. Этот экземпляр будет сконструирован внутри send, пользователю не нужно делать этого вручную.

Последним параметром конструктора msg_request (и, соответственно, последним аргументом при вызове send) является mbox, на который должен придти ответ. В качестве этого mbox-а указывается персональный mbox агента-generator-а. Каждый агент имеет свой личный mbox, доступ к этому mbox-у осуществляется посредством метода so_direct_mbox(). Название direct_mbox используется потому, что в SObjectizer есть два типа mbox-ов. Самый простой из них реализует модель multi-producer/single-consumer, т.е. сообщение в mbox может отослать кто угодно, а вот получить сообщение из mbox-а может только единственный агент-владелец. Таким образом mbox подобного типа является прямым каналом для связи с агентом, отсюда и название direct_mbox. Есть так же и тип multi-producer/multi-consumer mbox-а, но в этом примере он не используется.

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

void    count_reply()    {       --m_expected_replies;       if( !m_expected_replies )          so_deregister_agent_coop_normally();    }

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

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

В данном примере регистрируется всего одна кооперация. Когда она дерегистрируется, SObjectizer понимает, что работающих агентов больше не осталось, и завершает свою работу.

Самый сложный агент в данном примере -- это агент-collector. Перед тем, как рассматривать его код, имеет смысл объяснить логику его работы.

Агент-collector получает запросы от generator-а и переадресует их агенту-performer-у. Т.к. запросы могут приходить быстрее, чем performer будет их обрабатывать, агенту-collector-у приходится сохранять запросы у себя. Для чего используется приоритетная очередь. Запросы с наиболее близким дедлайном получают наивысший приоритет и оказываются в начале очереди.

Запросы пересылаются агенту-performer-у. Когда performer завершает обработку очередного запроса, он информирует агента-collector-а об этом посредством уведомления msg_select_next_job. Когда агент-collector получает msg_select_next_job, он берет первую заявку из очереди и пересылает ее агенту-performer-у.

У каждого сообщения есть дедлайн и агент-collector должен контролировать дедлайны. Для этого при сохранении сообщения в очереди агент-collector взводит таймер. При срабатывании таймера collector получит сигнал и сможет выбросить из очереди сообщение, дедлайн которого натупил (если оно еще есть в очереди).

При получении msg_request агент-collector может оказаться в двух ситуациях:

• агент-performer свободен и новый msg_request может быть сразу отослан на обработку;
• агент-performer занят обработкой ранее переданного ему запроса и новый msg_request должен быть сохранен в очереди.

В коде агента-collector-а эти две ситуации выражаются явно путем объявления двух состояний для агента:

const so_5::rt::state_t st_performer_is_free = so_make_state();    const so_5::rt::state_t st_performer_is_busy = so_make_state();

Состояние определяет, какие сообщения посредством каких событий будут обрабатываться, когда агент находится в этом состоянии. Это показывается в методе so_define_agent() агента-collector-а при подписке агента на сообщения:

virtual void    so_define_agent() override    {       this >>= st_performer_is_free;       st_performer_is_free          .event( &a_collector_t::evt_first_request );       st_performer_is_busy          .event( &a_collector_t::evt_yet_another_request )          .event< msg_select_next_job >( &a_collector_t::evt_select_next_job )          .event< msg_check_deadline >( &a_collector_t::evt_check_deadline );    }

Здесь видно, что в состоянии st_perfomer_is_free агент реагирует только на одно событие -- получение нового msg_request. Обработка этого события выполняется методом evt_first_request.

А вот в состоянии st_performer_is_busy агенту нужно обрабатывать больше событий. Событие evt_yet_another_request отвечает за обработку очередного msg_request. Событие evt_select_next_job -- за выдачу первого msg_request из внутренней очереди освободившемуся агенту-performer-у. Событие evt_check_deadline -- за проверку наступления дедлайнов для хранящихся во внутренней очереди запросов.

Почему в состоянии st_performer_is_busy выполняется больше действий, чем в состоянии st_performer_is_free? Потому, что когда агент-performer свободен, очередь заявок в collector-е пуста из-за чего нет надобности реагировать на просьбу выдать следующий запрос (очередь пуста, выдавать нечего). Как и нет надобности проверять дедлайны -- очередь пуста, в ней нет заявок. Поэтому msg_select_next_job и msg_check_deadline в состоянии st_performer_is_free просто игнорируются.

Первой же операцией в so_define_agent() является перевод агента-collector-а в состояние st_performer_is_free. Это обязательно нужно сделать, т.к. изначально агент находится в состоянии по-умолчанию. И, если агента не перевести из него в состояние st_performer_is_free, то агент-collector не будет работать, т.к. подписки сделаны для st_performer_is_free и st_performer_is_busy, а не для состояния по-умолчанию.

Глядя на код so_define_agent() можно увидеть, что в нем используются две формы методов event() для подписки событий. В одном случае event() получает только указатель на метод-обработчик события. Этот случай уже рассматривался выше для агента-generator-а. Во втором случае параметром шаблона задается еще и имя типа сообщения. Этот вариант event() нужен для случая, когда по сигнатуре метода-обработчика нельзя определить, какое сообщение приводит к возникновению события. Действительно, методы evt_select_next_job() и evt_check_deadline() не имеют аргументов вообще:

void evt_select_next_job(); void evt_check_deadline();

Все дело в том, что в SObjectizer-е различаются два типа сообщений. Первый тип -- это собственно полноценные сообщения, которые переносят в себе информацию. Примерами являются сообщения msg_request, msg_positive_reply и msg_negative_reply. Методы-обработчики таких сообщений должны получать экземпляр сообщения в качестве параметра, иначе смысла в обработчике нет. Поэтому при подписке обработчика полноценного сообщения в методе event() нет необходимости указывать тип сообщения явным образом -- он выводится автоматически.

Второй тип -- это сигналы. Т.е. это сообщения, которые не переносят никакой дополнительной информации кроме факта собственного существования. Пример -- это уведомление msg_select_next_job от агента-performer-а. Важен сам факт того, что агент закончил с текущей заявкой и готов перейти к следующей. Поэтому больше никакой информации с msg_select_next_job не связано.

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

st_performer_is_busy    .event( &a_collector_t::evt_yet_another_request )    .event< msg_select_next_job >( &a_collector_t::evt_select_next_job )    .event< msg_check_deadline >( &a_collector_t::evt_check_deadline );

Различия между сообщениями и сигналами еще и в том, что тип для сообщения с данными должен быть наследником so_5::rt::message_t. А тип для сигнала должен наследоваться от so_5::rt::signal_t. Что можно видеть в коде агента-collector-а:

struct msg_select_next_job : public so_5::rt::signal_t {}; struct msg_check_deadline : public so_5::rt::signal_t {};

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

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

void evt_positive_reply( const msg_positive_reply & evt ); void evt_negative_reply( const msg_negative_reply & evt );

Тогда как во втором случае -- ссылку на вспомогательный SObjectizer-овский объект event_data_t:

void evt_first_request(     const so_5::rt::event_data_t< msg_request > & evt ) void evt_yet_another_request(     const so_5::rt::event_data_t< msg_request > & evt )

Первая форма является наиболее общеупотребительной и удобной -- обработчик события просто получает ссылку на экземпляр сообщения-инцидента. Вторая форма используется, в основном, в случаях, когда нужно иметь возможность сохранить указатель на экземпляр сообщения-инцидента. Именно это и нужно агенту-collector-у. Для того, чтобы не пересоздавать сообщение msg_request заново, агент-collector использует тот же самый экземпляр сообщения (при перепосылке его агенту-performer-у и при сохранении msg_request-а в своей очереди). Для получения умного указателя на экземпляр сообщения-инцидента используется метод make_reference() класса event_data_t.

Реакция на msg_request в состоянии st_performer_is_free (т.е. когда агент-perfomer не занят обработкой запросов) тривиальна -- нужно просто сменить состояние на st_performer_is_busy и переслать этот же экземпляр сообщения агенту-performer-у:

void evt_first_request( const so_5::rt::event_data_t< msg_request > & evt ) {    this >>= st_performer_is_busy;    m_performer_mbox->deliver_message( evt.make_reference() ); }

Здесь для перепосылки того же самого экземпляра сообщения используется метод deliver_message(), а не функция send(). В текущей версии SObjectizer-а шаблонные функции send() не поддерживают операцию перепосылки, поэтому задействуется метод mbox-а. Вообще-то шаблонные функции send() -- это всего лишь простые вспомогательные обертки, которые конструируют новый экземпляр сообщения, а затем передают его в mbox-овский deliver_message(). Изначально функций send() вообще не было, но со временем они были добавлены для упрощения жизни разработчикам, в особенности при написании кода внутри C++ных шаблонов.

Реакция на msg_request в состоянии st_performer_is_busy поинтереснее:

void evt_yet_another_request(    const so_5::rt::event_data_t< msg_request > & evt ) {    // Агент-performer занят, поэтому запрос должен быть    // сохранен в очереди, а дедлайн запроса нужно проконтролировать.    const std::time_t now = std::time(nullptr);    if( now < evt->m_deadline )    {       m_pending_requests.push( evt.make_reference() );       // Для простоты на каждый сохраненный в очереди запрос       // генерируем таймерный сигнал.       so_5::send_delayed_to_agent< msg_check_deadline >(             *this,             std::chrono::seconds( evt->m_deadline - now ) );    }    else    {       // Дедлайн для сообщения наступил пока сообщение шло к нам.       // Поэтому вместо обработки сразу же отсылается       // отрицательный ответ.       send_negative_reply( *evt );    } }

Новый запрос сохраняется в очереди. И сразу же агент-collector отсылает себе отложенное сообщение-сигнал для контроля дедлайна сообщения. На всякий случай делается проверка того, что дедлайн еще впереди, дабы предотвратить ситуацию, когда дедлайн уже прошел, но мы этого не заметили и получили отрицательный результат (evt->m_deadline-now).

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

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

void evt_check_deadline() {    const std::time_t now = std::time(nullptr);    while( !m_pending_requests.empty() )    {       auto & request = m_pending_requests.top();       if( now >= request->m_deadline )       {          send_negative_reply( *request );          m_pending_requests.pop();       }       else          break;    } }

Когда агент-performer завершает обработку заявки, он присылает сигнал msg_select_next_job. При обработке этого сигнала нужно либо отослать performer-у очередное сообщение, либо вернутся в состояние st_performer_is_free, если заявок для обработки больше нет.

void evt_select_next_job() {    if( m_pending_requests.empty() )       // Поскольку ожидающих своей очереди запросов нет,       // то остается просто вернуться в начальное состояние.       this >>= st_performer_is_free;    else    {       // Нужно оставаться в состоянии st_performer_is_busy       // и отосылать агенту-performer-у следующую заявку       // из очереди.       auto & request = m_pending_requests.top();       m_performer_mbox->deliver_message( request );       m_pending_requests.pop();    } }

Агент-performer самый простой. Его код можно привести сразу целиком:

class a_performer_t : public so_5::rt::agent_t { public :    a_performer_t(       so_5::rt::environment_t & env,       const so_5::rt::mbox_t & collector_mbox )       :  so_5::rt::agent_t( env )       ,  m_collector_mbox( collector_mbox )    {}    virtual void    so_define_agent() override    {       so_default_state().event( &a_performer_t::evt_request );    } private :    const so_5::rt::mbox_t m_collector_mbox;    void    evt_request( const msg_request & evt )    {       const std::time_t started_at = std::time(nullptr);       // Имитация какой-то напряженной работы.       // Текущая рабочая нить будет заблокирована на 4 секунды.       std::this_thread::sleep_for( std::chrono::seconds(4) );       // Ответ на запрос отсылается тому, кто инициировал запрос.       so_5::send< msg_positive_reply >(             evt.m_reply_to,             evt.m_id,             "-=<" + evt.m_id + ">=-",             started_at );       // Агент-collector должен узнать о нашей готовности к обработке       // следующей заявки.       so_5::send< a_collector_t::msg_select_next_job >( m_collector_mbox );    } };

В реализации performer-а можно отметить один момент, который будет важен при обсуждении диспетчеров чуть ниже: блокирование рабочей нити, на которой запускается evt_request посредством выполнения sleep_for. Это означает, что пока агент-performer спит на этом вызове, рабочая нить не может использоваться для выполнения других операций.

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

SObjectizer запускается посредством функции so_5::launch(). В SObjectizer-е есть несколько ее вариантов, в примере используется вариант, получающий два аргумента. Первый аргумент -- это функция за запуска стартовой кооперации. Второй аргумент -- это функция для настройки параматров SObjectizer-а. Вот как запуск SObjectizer выглядит в main-е:

int main( int argc, char ** argv ) {    try    {       so_5::launch(          create_coop,          []( so_5::rt::environment_params_t & params )          {             params.add_named_dispatcher( "thread_pool",                   so_5::disp::thread_pool::create_disp( 3 ) );          } );       return 0;    }    catch( const std::exception & x )    {       std::cerr << "*** Exception caught: " << x.what() << std::endl;    }    return 2; }

Информирование об ошибках в SObjectizer базируется на исключениях, поэтому обращение к launch() обрамлено блоком try-catch.

Первым аргументом в launch() передается функция create_coop(), которая будет вызвана SObjectizer-ом в нужный момент. Подробнее create_coop() рассматривается ниже. Вторым аргументом в launch() передается лямбда-функция, осуществляющая настройку конкретного экземпляра SObjectizer. Возврат из launch() осуществляется когда запущенный внутри launch() экземпляр SObjectizer-а завершит свою работу.

В приложении можно создать и одновременно запустить множество экземпляров SObjectizer-а. Каждый из них может иметь разные настройки, разный набор коопераций и агентов. Экземпляр SObjectizer-а -- это объект, реализующий интерфейс environment_t. Такой объект создается внутри launch() и далее ссылка на него будет передаваться в стартовую функцию, а оттуда во все создаваемые пользователем агенты. Каждый агент должен быть привязан к одному environment-у, именно поэтому в конструкторах всех агентов примера первый аргумент -- это ссылка на environment.

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

SObjectizer предоставляет несколько готовых типов диспетчеров. В данном примере используется диспетчер, который создает пул рабочих нитей и распределяет агентов по ним. Поскольку в примере используется три агента, причем агент-performer выполняет длительные операции, блокирующие рабочую нить, thread_pool-диспетчеру предписывается создать три рабочие нити. Это позволит иметь две свободные нити для событий агентов generator и collector пока агент-performer спит на третьей рабочей нити.

Теперь можно перейти к т.н. "стартовой функции", которая в данном примере представлена функцией create_coop.

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

Пришло время посмотреть, что из себя представляет стартовая функция в этом примере:

void create_coop( so_5::rt::environment_t & env ) {    using namespace so_5::disp::thread_pool;    auto c = env.create_coop( so_5::autoname,          create_disp_binder( "thread_pool",             []( params_t & p ) { p.fifo( fifo_t::individual ); } ) );    std::unique_ptr< a_collector_t > collector{ new a_collector_t{ env } };    std::unique_ptr< a_performer_t > performer{          new a_performer_t{ env, collector->so_direct_mbox() } };    collector->set_performer_mbox( performer->so_direct_mbox() );    std::unique_ptr< a_generator_t > generator{          new a_generator_t{ env, collector->so_direct_mbox() } };    c->add_agent( std::move( collector ) );    c->add_agent( std::move( performer ) );    c->add_agent( std::move( generator ) );    env.register_coop( std::move( c ) ); }

Ничего особенного, обычные рутинные операции.

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

auto c = env.create_coop( so_5::autoname,          create_disp_binder( "thread_pool",             []( params_t & p ) { p.fifo( fifo_t::individual ); } ) );

Каждая коперация в SObjectizer должна иметь уникальное имя. Это имя можно придумать самому, а можно попросить SObjectizer сделать это вместо нас. Здесь мы как раз просим об этом SObjectizer -- первый аргумент so_5::autoname -- это как раз такая просьба.

А вот второй аргумент для create_coop() несколько сложнее. Есть в SObjectizer такое понятие, как dispatcher binding. Т.е. связь агента с диспетчером, который будет обслуживать агента, т.е. предоставлять рабочие нити для агента.

Каждый агент должен быть связан с каким-то диспетчером. Назначать эту связь можно разными способами. Можно вообще не указывать явно никакой связи. Тогда SObjectizer сам свяжет агента с диспетчером по-умолчанию (все такие агенты будут работать на одной общей рабочей нити). Можно явным образом задавать dispatcher binding для каждого агента. Но, если все агенты кооперации должны работать на одном и том же диспетчере, то явное указание связи для каждого агента кооперации -- это утомительно. Поэтому можно задать связь для кооперации. И эта связь распространится на агентов кооперации.

Именно этот подход и используется здесь. Т.к. все агенты должны работать на thread_pool-диспетчере, то связь с этим диспетчером задается при создании кооперации. А все агенты, которые затем добавляются в кооперацию, будут использовать эту связь.

Небольшая техническая подробность, связанная с вызовом create_disp_binder: в обсуждаемом примере для каждого агента кооперации должна поддерживаться собственная очередь сообщений, что позволит диспетчеру распределять агентов по разным нитям из своего пула. Поэтому при создании binding-а это явным образом указывается при задании параметра fifo в виде значения fifo_t::individual.

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

Отметить можно разве лишь то, что агенты должны быть динамически созданными объектами, время жизни которых будет контролироваться SObjectizer-ом (SObjectizer сам определяет, когда агентов можно безопасно уничтожать после полной дерегистрации кооперации, которой они принадлежат). И это единственное место в обсуждаемом коде, где пользователь вынужден вручную вызвать оператор new. Да и то только потому, что в C++11 нет вспомогательной функции std::make_unique, которую добавили в C++14. Так что после перевода SObjectizer на C++14 необходимости в ручном вызове new будет еще меньше.

Почему внимание было заострено на количестве вызовов new в данном примере? Потому, что управление динамической памятью -- это одна из самых больших проблем в C++. Еще острее эта проблема становится в многопоточных приложениях. А вот в чем большой плюс SObjectizer-а, так это в том, что очень многие вопросы по работе с созданными динамически объектами, вроде агентов и сообщений, SObjectizer берет на себя, тем самым существенно упрощая жизнь пользователю.

---

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

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

Исходный код примера находится сейчас в репозитории SObjectizer-а в рабочей ветке. Для его работы необходима версия 5.5.2, которую можно загрузить с SourceForge. В официальный релиз пример simple_message_deadline будет включен при выпуске версии 5.5.3 (или 5.6.0), ориентировочно в феврале-марте 2015.