[ru] Изменение способа синхронного взаимодействия между агентами в SO-5.6

[Yauheni Akhotnikau]

Проблема и ее актуальность

Возможность синхронного взаимодействия между агентами в SObjectizer была добавлена в версии 5.3.0 летом 2014-го. В это время в SObjectizer-е еще не было ни семейства send-функций, ни возможности отсылки сообщений произвольных типов, ни mchain-ов. Кроме того, при реализации механизма синхронного взаимодействия не был должным образом учтен опыт других акторных фреймворков (в частности Akka). Поэтому первая реализация синхронного взаимодействия получилась громоздкой и неудобной.

Со временем SObjectizer перестал ориентироваться на компиляторы без поддержки variadic templates, что позволило добавить функции request_value и request_future для упрощения синхронного взаимодействия. Использовать синхронное взаимодействие стало проще. Но при этом внутри SObjectizer-5.5 все равно был тот же самый механизм доставки синхронных сообщений, реализованный в версии 5.3.0.

За прошедшее время сложилось устойчивое ощущение, что существовавший в версиях 5.3-5.5 механизм синхронного взаимодействия должен быть упразднен и версии 5.6 он должен быть заменен чем-то новым.

К существующему механизму есть ряд претензий:

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

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

3. Реакция на исключения, которые покинули обработчик сообщения агента, принципиально отличается для обычного сообщения и для синхронного запроса. Так, если агент подписался на сообщение типа M и при его обработке выпустил наружу исключение, то SObjectizer по умолчанию завершит работу всего приложения. Но если M пришло к этому же самому агенту как синхронный запрос, то исключение будет поймано и возвращено как результат синхронного запроса.

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

5. Механизм синхронного взаимодействия из версий 5.3-5.5 существенно усложняет разработку собственных mbox-ов.

Предложение по исправлению проблемы

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

Однако, поскольку ветка 5.6 нарушает совместимость с веткой 5.5, то есть возможность удалить из SObjectizer старый механизм синхронного взаимодействия и внедрить новый.

Новый механизм синхронного взаимодействия

Суть нового механизма взаимодействия состоит в следующем:

1. Да синхронного взаимодействия агент-клиент отсылает агенту-сервису специальное сообщение, например, имеющее следующий тип:

template<typename Request, typename Reply>

class request_reply_t final : public so_5::message_t {

public:

   ...

   const Request & request() const noexcept;

   ...

   template<typename... Args>

   void reply(Args && ...args);

};

Отсылка запроса и получение результата может происходить посредством вспомогательной функции request_value() по аналогии с тем, как это происходит сейчас.

2. Агент-сервис подписывается на сообщение request_reply_t<Req,Rep>, где в качестве Req и Rep будут использоваться конкретные типы данных. Например:

class service_provider_t : public so_5::agent_t {

   void on_request(mutable_mhood_t<request_reply_t<SomeRequest, SomeResponse>> cmd);

   ...

   void so_define_agent() override {

      so_subscribe_self().event(&service_provider_t::on_request);

      ...

   }

};

3. Агент-сервис отвечает на запрос вызывая метод reply() у полученного экземпляра request_reply_t:

void service_provider_t::on_request(

      mutable_mhood_t<request_reply_t<SomeRequest, SomeResponse>> cmd) {

   ... // Какая-то обработка.

   // Отсылка ответа.

   cmd->reply(... /* параметры для ответа */)

}

4. Под капотом этого механизма будет использоваться mchain. Когда вызывается request_value (или аналогичная функция), то создается уникальный mchain, ссылка на который передается в экземпляр request_reply_t. Когда агент-сервис вызывает request_reply_t::reply() в этот mchain записывается ответное сообщение. Соответственно, агент-клиент может ждать на этом mchain-е посредством обычных функций receive() или select().

Принцип работы нового механизма в двух словах

Предлагаемый механизм должен работать следующим образом:

* когда агент-клиент вызывает request_value() или другую аналогичную функцию, то создается специальный mchain для получения ответа от агента-сервиса и экземпляр сообщения request_reply_t (куда отдается ссылка на mchain);

* сообщение request_reply_t доставляется до агента-сервиса обычным образом (т.е. SObjectizer внутри не будет разбираться с тем, пришло ли обычное асинхронное сообщение или же синхронный запрос);

* агент-сервис должен записать ответ в ранее созданный mchain через вызов метода request_reply_t::reply();

* агент-клиент ожидает ответа на mchain-е (это ожидание может быть скрыто от агента-клиента функциями вроде request_value()).

При этом, вероятно, сообщение request_reply_t должно отсылаться как мутабельное сообщение.

Достоинства предлагаемого способа

Предполагается, что новый способ организации синхронного взаимодействия будет лишен недостатков старого механизма из версий 5.3-5.5, а именно:

* накладные расходы будут присутствовать только для отсылки request_reply_t, не будет никаких дополнительных проверок в обертках вокруг обработчиков сообщений;

* в реализациях mbox-ов не потребуется больше выполнять доставку обычных сообщений и синхронных запросов разными способами;

* агент-сервис явным образом декларирует, что он выполняет обработку синхронных запросов;

* сообщения request_reply_t можно переадресовывать как и любые другие сообщения;

* единообразный подход к обработке исключений, выпущенных из обработчиков событий агента-сервиса;

* возможность использования select() для ожидания и обработки ответов сразу от нескольких агентов-сервисов.

Степень проработанности нового способа

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

Детальная проработка нового решения начнется, предположительно, после 10 марта 2019-го.

Зачем этот вопрос вынесен на обсуждение?

Обсуждение замены механизма синхронного взаимодействия преследует две цели:

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

2. Выяснение того, можно ли сделать новый способ взаимодействия еще удобнее, гибче и мощнее.

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

Ну и если кто-то видит косяки и недостатки предлагаемого способа, то озвучьте свое мнение, пожалуйста. Это поможет сделать SObjectizer-5.6 лучше.

[Pavel Vainerman]

На первый взгляд выглядит вполне пригодным для использования.
Планируется ли "готовый сахар" скрывающий select() на mchain в виде "request_value()"?
Будет ли для таких вызовов понятие timeout?

[Yauheni Akhotnikau]

> Планируется ли "готовый сахар" скрывающий select() на mchain в виде "request_value()"?
> Будет ли для таких вызовов понятие timeout?

Да. Думаю, что существующий request_value() можно будет сохранить практически в его текущем виде.

А вот сохранить request_future() вряд ли получится. Наверное и можно было бы, если бы это сильно кому-то нужно было бы. Но пока планов на сохранение request_future() нет.

--
Вы получили это сообщение, поскольку подписаны на группу "SObjectizer".
Чтобы отменить подписку на эту группу и больше не получать от нее сообщения, отправьте письмо на электронный адрес sobjectizer...@googlegroups.com.
Чтобы отправлять сообщения в эту группу, отправьте письмо на электронный адрес sobje...@googlegroups.com.
Чтобы зайти в группу, перейдите по ссылке https://groups.google.com/group/sobjectizer.
Чтобы настроить другие параметры, перейдите по ссылке https://groups.google.com/d/optout.

--

Regards,
Yauheni Akhotnikau

[Pavel Vainerman]

понедельник, 4 марта 2019 г., 16:47:55 UTC+3 пользователь Yauheni Akhotnikau написал:

> Планируется ли "готовый сахар" скрывающий select() на mchain в виде "request_value()"?
> Будет ли для таких вызовов понятие timeout?

Да. Думаю, что существующий request_value() можно будет сохранить практически в его текущем виде.

А вот сохранить request_future() вряд ли получится. Наверное и можно было бы, если бы это сильно кому-то нужно было бы. Но пока планов на сохранение request_future() нет.

    Отлично. Спасибо.

[eao...@gmail.com]

Первая версия реализации синхронного взаимодействия появилась в новой версии so5extra-1.3. Выглядит это следующим образом:

#include <so_5_extra/sync/pub.hpp>

#include <so_5/all.hpp>

// Short alias for convenience.
namespace sync_ns = so_5::extra::sync;

using namespace std::chrono_literals;

// The type of service provider.
class service_provider_t final : public so_5::agent_t
{
public :
   using so_5::agent_t::agent_t;

   void so_define_agent() override
   {
      so_subscribe_self().event(
            []( sync_ns::request_mhood_t<int, std::string> cmd ) {
               // Transform the incoming value, convert the result
               // to string and send the resulting string back.
               cmd->make_reply( std::to_string(cmd->request() * 2) );
            } );
   }
};

// The type of service consumer.
class consumer_t final : public so_5::agent_t
{
   // Message box of the service provider.
   const so_5::mbox_t m_service;

public :
   consumer_t( context_t ctx, so_5::mbox_t service )
      :  so_5::agent_t{ std::move(ctx) }
      ,  m_service{ std::move(service) }
   {}

   void so_evt_start() override
   {
      // Issue a request and wait for the result no more than 500ms.
      auto result = sync_ns::request_reply_t<int, std::string>::request_value(
            // The destination for the request.
            m_service,
            // Max waiting time.
            500ms,
            // Request's value.
            4 );

      std::cout << "The result: " << result << std::endl;

      so_deregister_agent_coop_normally();
   }
};

int main()
{
   so_5::launch( [](so_5::environment_t & env) {
      env.introduce_coop(
         // Every agent should work on its own thread.
         so_5::disp::active_obj::make_dispatcher( env ).binder(),
         [](so_5::coop_t & coop) {
            auto service_mbox = coop.make_agent< service_provider_t >()
                  ->so_direct_mbox();
            coop.make_agent< consumer_t >( service_mbox );
         } );
   } );
}

Идея следующая: есть шаблонный класс request_reply_t<Request, Reply>, который параметризуется типом запроса (Request) и типом результата (Reply). Для того, чтобы сделать синхронный запрос, у класса request_reply_t есть два статических метода. Во-первых, показанный выше request_value:

Reply request_value(target, timeout, args...);

Этот метод бросает исключение, если ответ не был получен (т.е. запрос не был обработан вообще или истек тайм-аут ожидания).
Есть небросающий исключения метод request_opt_value:

optional<Reply> request_opt_value(target, timeout, args...);

Этот метод возвращает либо пустое значение (если ответа не получили в течении заданного времени), либо непустое (если значение получено).

Внутри этих методов отсылается экземпляр сообщения request_reply_t<Request, Reply>, который нужно получить в качесте обычного мутабельного сообщения. И при его обработке у него нужно вызвать метод make_reply. В принципе, выглядеть это может так:

void on_request(mhood_t< so_5::mutable_msg< sync_ns::request_reply_t<Request, Reply> > > cmd) {

   ...

   cmd->make_reply(args...);

}

либо так:

void on_request(mutable_mhood_t< sync_ns::request_reply_t<Request, Reply> > cmd) {

   ...

   cmd->make_reply(args...);

}

либо, чтобы не путаться с мутабельность, можно записать как в примере выше:

void on_request(sync_ns::request_mhood_t<Request, Reply> cmd) {

   ...

   cmd->make_reply(args...);

}

Первоначально методы request_value и request_opt_value не были методами request_reply_t, а были свободными функциями. Поэтому их можно было вызывать вот так:

auto result = sync_ns::request_value<int, std::string>(target, timeout, args...);

Но я от такого варианта отказался потому, что в SO-5.5 порядок параметров шаблона в request_value был другим. Ранее было request_value<Reply, Request>, что должно было читаться как "хочу получить экземпляр типа Reply в качестве обработки запроса типа Request". Тогда как в so5extra-1.3 используется обратный порядок: Request, Reply (т.е. сперва то, что нужно послать, затем то, что хочется получить). Сделано это было для того, чтобы не путаться в том, в каком порядке задаются параметры в request_reply_t, в каком в request_value().

А раз порядок поменялся, то при портировании существующего кода на SO-5.6 и so5extra-1.3 легко ошибиться, оставить старый вызов request_value и получить ошибку.

Итак, методы request_value и request_opt_value были перемещены в request_reply_t. Но в итоге получился слишком многословный вариант:

auto result = sync_ns::request_reply_t<Request, Reply>::request_value(...);

Хочется чего-то покороче и лаконичнее.

Пока есть такие варианты:

1. Использовать имена ask_value и ask_opt_value:

auto result = sync_ns::request_reply_t<Request, Reply>::ask_value(...);

Принципиально ничего не меняется, но читается проще, имхо.

2. Ввести свободные функции request_reply и request_opt_value:

auto result = sync_ns::request_reply<Request, Reply>(...);

auto opt_result = sync_ns::request_opt_reply<Request, Reply>(...);

Собственно вопросы:

* нужно ли менять существующий вариант с методами request_value/request_opt_value класса request_value_t на что-то другое?

* если нужно, то на что? Какой из вариантов (№1 или №2) предпочтительнее? Может есть еще какой-то вариант?

[Pavel Vainerman]

Собственно вопросы:

* нужно ли менять существующий вариант с методами request_value/request_opt_value класса request_value_t на что-то другое?

* если нужно, то на что? Какой из вариантов (№1 или №2) предпочтительнее? Может есть еще какой-то вариант?

 

  Привычнее выглядит вариант N2.  Но в целом, привыкнуть можно к любому.

[eao...@gmail.com]

То, что получилось в итоге можно увидеть в новом репозитории для so5extra:

* примеры здесь: https://bitbucket.org/sobjectizerteam/so5extra/src/default/dev/sample/so_5_extra/sync/

* документация на английском здесь: https://bitbucket.org/sobjectizerteam/so5extra/wiki/so5extra-1.3-docs/tutorials/sync.md