Итак, текущая реализация синхронного взаимодействия SObjectizer-агентов в новой, разрабатываемой сейчас версии 5.3 приобрела стабильные очертания. Что дает возможность рассказать о том, что это такое, откуда растут ноги, и показать, как этим можно пользоваться. Поскольку тема специфическая и содержит большое количество текста, то она упрятана под кат, дабы не мешать неинтересующимся.
История вопроса уходит далеко в древность :) Практически сразу после возникновения SObjectizer-4 весной 2002 года почти у всех, кто знакомился и пробовал использовать SObjectizer, возникал вопрос: а почему нет возможности синхронно обратиться к другому агенту?
Ответ на этот вопрос был прост и некрасив до безобразия: да потому!
Действительно, идея SObjectizer, как одного из вариантов реализации Actor Model, заключалась в асинхронном обмене сообщениями между агентами. Фактически, краеугольных камней в SObjectizer было три: асинхронный обмен сообщениями, явно декларируемые состояния агентов и привязка агентов к диспетчерам. Вот три столпа, на которых все строилось. Поэтому вопрос о синхронном взаимодействии всего-то навсего подвергал сомнению базовые принципы, положенные в основу SObjectizer :)
Объективно многие задачи, которые решались с помощью SObjectizer, вполне успешно укладывались в схему асинхронных сообщений. Работа распределялась между несколькими агентами, каждый из которых выполнял свою роль или был очередной стадией потока обработки данных. Некий агент получал новый объем работы в виде входящего сообщения, выполнял его обработку, в процессе которой генерировал одно или несколько новых сообщений для других агентов. После чего получал очередное сообщение и т.д. (стадийность обработки данных ярко выражена в фреймворке SEDA). Так же корни SObjectizer растут из АСУТП-ных задач, во многих из которых асинхронность возникновения внешних воздействий, а так же инициирования нескольких ответных действий -- это вполне естественная, привычная всем специфика. Поэтому при разработке сначала SCADA Objectizer, а затем его наследника -- SObjectizer-4, речи о синхронности и не было, по сути-то.
С появлением SObjectizer-4 и его понятия "диспетчер" важность асинхронности стала еще выше. Каждый агент в SObjectizer привязывается к своему диспетчеру. На данный момент есть три основных диспетчера. На одном из них все агенты работают на одной общей рабочей нити. Этакая кооперативная многозадачность, если кто-то "заснет", то "заснут" и все остальные агенты на этом диспетчере. На втором диспетчере каждый агент получает свою собственную рабочую нить -- т.е. агенты на этом диспетчере превращаются в чистой воды активные объекты, методы которых работают исключительно на своей отдельной нити. Третий диспетчер позволяет выдавать отдельную нить группе агентов. Внутри этой группы будет "кооперативная многозадачность" и если кто-то "заснет", то заснут и остальные члены этой группы, однако агенты на других рабочих нитях продолжат свою работу.
Т.е. в SObjectizer-4, а потом и в SObjectizer-5, получилось так, что у агента есть собственный единственный контекст исполнения. И все свои действия агент выполняет только на этом контексте. Асинхронный обмен сообщениями означает, что агенты используют архитектуру share nothing, т.е. каждый владеет своей порцией данных и не может просто так обратиться к данным другого агента. Если хочешь воздействовать на другого агента, то отошли ему сообщение. Он его получит и на собственном контексте разберется со своими данными. Нет необходимости в синхронизации доступа к данным, нужно гораздо меньше думать о блокировках, гонках и прочих прелестях многопоточного программирования.
Писать многопоточные приложения на основе обмена сообщениями намного проще, чем с использованием событий, семафоров, мьютексов, условных переменных или даже более высокоуровневых механизмов, вроде рандеву. В одном из нагруженных приложений, написанных мной на SObjectizer, я как-то обнаружил больше 90 параллельно работающих потоков на которых работало пару сотен агентов. Все это работало как часы, тогда как написать ручное управление таким количеством плотно взаимодействующих между собой нитей на мьютексах -- это то еще занятие, в особенности на C/C++.
Еще один интересный побочный эффект от тотальной асинхронности -- надежность :) Действительно, обмен сообщениями -- это совершенно ничего не гарантирующий механизм :) Сообщение может не дойти. Может задержаться где-то по дороге. Агент-получатель может сломаться при его обработке и рестартовать после этого "с чистого листа". Поэтому приложение сразу разрабатывается так, чтобы спокойно переживать все это. Основной способ здесь -- это тайм-ауты и повторы. Простые средства, но очень действенные.
И, что еще интереснее, когда все взаимодействие строится на асинхронности, то остается совсем маленький шажок к тому, чтобы приложение стало распределенным. Действительно, share nothing уже внутри. Сообщения и так уже ходят. Так какая разница, ходят ли они в одном процессе или между несколькими процессами? А то и между несколькими сетевыми узлами? Конечно же разница есть: и время отклика будет чуть больше, и голые указатели внутри сообщений передавать нельзя. Но это уже частности. Принципиально-то ничего не меняется. Да и предыдущий абзац про надежность здесь так же актуален: если приложение спокойно переживает потерю сообщения у себя внутри, то оно так же спокойно переживет и потерю сообщения где-то на пути от одного узла к другому.
В общем, у асинхронного обмена сообщениями есть много достоинств. Что и позволило столько времени успешно пользоваться SObjectizer-ом не предоставлявшем других механизмов общения агентов, в том числе и для построения распределенных приложений из десятков компонент только средствами SObjectizer. Однако, чистота модели -- это вещь хорошая, но случается, что она разбивается о реалии окружающей нас действительности :)
Бывают случаи, когда ответ нужно получить здесь и сейчас. И ничего не делать до тех пор, пока он не получен. Например, нужно вам начать транзакцию в БД, неудобно же отсылать сообщение на ее начало, ждать подтверждения и лишь после его получения выполнять действия над БД. А потом таким же образом коммитить транзакцию в БД. Или другой пример: нужно вам зашифровать или подписать блок данных. Можно, конечно, отослать его агенту-криптопровайдеру и ждать результата. Но, опять же, проще сделать синхронный вызов и тут же получить результат, чем заниматься отсылкой/получением сообщений, сменой состояний, отслеживанием тайм-аутов и пр.
На мой взгляд, идеальный баланс нашли разработчики Erlang-а. Но для этого им пришлось написать свой собственный язык и свой собственный, сильно специализированный run-time для него. В Erlang-е можно послать сообщение агенту-криптопровайдеру и сразу, следующей же инструкцией, повиснуть на ожидании ответа. Поскольку run-time свой собственный, такой заснувший на ожидании ответа процесс может быть снят с процессора, а на его место будет распределен другой готовый к выполнению процесс. Происходит это за счет собственной реализации вытесняющей многозадачности в run-time Erlang-а.
Ну а как быть, если мы находимся в C++ и рабочая нить -- это нить ОС, которая диспетчируется не нами, а самой операционкой? Допустим, мы пишем в C++ной функции:
|
crypto_provider->send_msg( encrypt_data(data) ); encryption_result = crypto_provider->receive_msg< encryption_result >(); |
Где внутри вызова receive_msg происходит засыпание на каком-то внутреннем condition_variable. Рабочая нить заснет, операционка ее снимет с процессора и даст процессор другой нити. И все, нить для нас потеряна. Заснут все агенты, которые работают на ней.
А что, если на этой же нити работает тот самый криптопровайдер, которому мы отослали сообщение и от которого мы ждем ответа? Или на этой нити работает какой-то вспомогательный агент, которому криптопровайдер отошлет свой запрос в процессе обработки нашего сообщения?
На C/C++ не получится программировать так же, как на Erlang-е или каком-то другом языке с виртуальной машиной, в работу которой можно вмешаться. Умельцы, конечно же, и для C++ умудряются писать собственные шедулеры на основе разных механизмов (будь то coroutines, fibers или что-то еще). Но это хорошо, если есть возможность заточиться под определенную прикладную задачу и, особенно, под конкретную платформу. А если хочется иметь переносимый фреймворк, который не стремится поместить разработчика в какую-то особую среду исполнения?
Добавление синхронности в SObjectizer обдумывалось очень давно. Следы этого вопроса обнаруживаются даже в первой публичной статье про SObjectizer в RSDN Magazine. Если не ошибаюсь, первая серьезная попытка найти решение была предпринята в 2004-м году. С тех пор, с интервалом в 1.5-2 года мысли возвращались к этой проблеме, но серьезных сдвигов не происходило.
Пожалуй, главным сухим остатком от всех предыдущих попыток оказалась терминология. Для обсуждения синхронного взаимодействия агентов удобным оказалось использование словосочетания service_request для обозначения синхронного обращения к другому агенту (в противовес deliver_message/send_message). А агент, у которого синхронно можно что-то спросить, назывался service_handler или же говорилось, что агент предоставляет какой-то сервис. Почему именно такие названия -- сейчас уже не вспомнить :) Но они используются в коде и комментариях, так что пусть будут такие.
Как бы то ни было, вопросы про синхронность в SObjectizer с регулярным постоянством всплывали, а хорошего ответа на него не было. Это напрягало. Так же изменился статус SObjectizer-а: ранее это была внутренняя разработка компании Интервэйл, которую в свое время открыли чтобы дать толчок его развитию. Теперь же это самостоятельный OpenSource проект, результаты которого используются в Интервэйле, но будущее проекта должно определяться уже другими нуждами и целями. Поэтому, во многом из маркетинговых соображений желания сделать SObjectizer полезным более широкому кругу разработчиков, была предпринята очередная попытка реализовать синхронность. Которая таки увенчалась успехом.
Ключевых факторов, обеспечивших появление синхронности в SObjectizer, два:
Во-первых, произошел полный отказ от попыток как-то помочь пользователю в диагностике потенциальных проблем с синхронностью. Сейчас все отдано на откуп пользователю: хочет синхронно вызывать агентов -- пусть делает это, на свой страх и риск. Такое взаимодействие чревато дедлоками. И, наверное, SObjectizer мог бы это обнаруживать. Но в попытке сделать такую диагностику можно было бы утонуть в технических деталях или застрять в поиске наилучшего по ресурсоемкости/скорости алгоритма. Поэтому SObjectizer ничем здесь пользователю не помогает. Спровоцирует пользователь дедлок, пусть сам его и ищет. Ведь говорили же, что сообщениями обмениваться нужно, а не синхронными вызовами :)
Во-вторых, в качестве основы взят механизм std::promise/std::future, который появился в C++11 и с которым должны быть знакомы все более-менее нормальные разработчики, использующие современный С++. Это, с одной стороны, дает пользователю SObjectizer уже известный и понятный ему инструмент, только слегка иначе упакованный. С другой -- сильно снижает трудозатраты на реализацию синхронности в SObjectizer, т.к. можно полагаться на то, что уже очень хорошо реализовано другими людьми.
Итак, механизм синхронного взаимодействия в SObjectizer-5.3 прост: добавлен еще один способ отослать сообщение агенту, при котором отправитель сообщения получает std::future для результата обработки сообщения. Затем на этом future можно заснуть, вызвав get(), и получить результат. Либо же исключение, если результата не было.
При этом агент-получатель сообщения получит сообщение обычным образом, на контексте своей рабочей нити. Т.е. для получателя вообще ничего не меняется. Он как работал на своей нити, думая, что к нему приходят только асинхронные сообщения, так и продолжает работать. Строго говоря, так оно и есть, ведь для доставки сообщения-запроса агенту-получателю используется тот же самый механизм диспетчеризации сообщений, который уже был в SObjectizer.
Поскольку механизм доставки сообщения-запроса остался прежним, все параметры для синхронного запроса должны быть оформлены в виде C++ класса, наследника so_5::rt::message_t.
Для примера допустим, что мы хотим написать сервис для конвертации int в string. Этот сервис должен получать параметр -- число для конвертации. Соответственно, данный параметр должен быть оформлен в виде структуры:
|
struct msg_convert : public so_5::rt::message_t { int m_value; msg_convert( int value ) : m_value( value ) {} }; |
Далее нам нужен service_handler -- агент, который будет обрабатывать сообщения этого типа. Обрабатывать он их будет практически точно так же, как и раньше -- посредством метода-обработчика (или метода-события, или просто события). Изменилось совсем немногое: раньше метод-событие должен был иметь void в качестве возвращаемого значения. А сейчас возвращаемое значение может быть любым. Для нашего сервиса конвертации это будет std::string:
|
std::string a_convert_service_t::svc_convert( const so_5::rt::event_data_t< msg_convert > & evt ) { std::ostringstream s; s << evt->m_value; return s.str(); } |
Метод-событие должен быть подписан на инициирующее сообщение. Здесь все делается точно так же, как и в предыдущих версиях SObjectizer-5, удалось обойтись без изменения синтаксиса методов подписки:
|
virtual void a_convert_service_t::so_define_agent() { so_subscribe( m_self_mbox ) .event( &a_convert_service_t::svc_convert ); } |
Т.е. для агента-получателя вообще почти ничего не изменилось за исключением того, что методы-события теперь могут иметь возвращаемые значения.
Имхо, это очень здорово, т.к. агенту, который предоставляет какую-то функциональность, не нужно заботится о том, как его будут вызывать -- синхронно или асинхронно. Для него работа все время будет выглядеть как асинхронная. Т.е. архитектура share nothing может и дальше использоваться в полный рост.
Синхронность/асинхронность обращения к агенту видна только клиенту, т.е. тому , кто инициирует это обращение. Он либо отсылает сообщение обычным образом и не заботится о результате. Либо же использует новую форму обращения, о которой речь пойдет ниже. Но сначала нужно пояснить пару фокусов, которые лежат в основе синхронного взаимодействия.
Первый фокус в том, какое сообщение реально летит агенту-получателю. Так, если мы асинхронно отсылаем msg_convert, то именно msg_convert будет лежать в очереди сообщений агента. Но вот если msg_convert отсылается синхронно, то на самом деле в очередь ставится другое сообщение, внутреннее. Это сообщение содержит указатель на экземпляр msg_convert и объект std::promise, в который будет сохранен результат, возвращенный методом-событием. Т.е. если попытаться расписать синхронную отсылку сообщения в псевдо-коде, то она будет выглядеть как-то так:
|
// Это функция скрывается за синхронной отсылкой сообщения msg_convert. std::future< std::string > send_msg_convert( mbox, msg_convert * request ) { // Объект этого типа будет поставлен в очередь к агенту. struct actual_message { std::promise< std::string > promise; msg_convert * request; }; // Создаем то сообщение, которое пойдет в очередь к агенту. auto msg = new actual_message(); // Сохраняем в нем актуальный запрос. msg->request = request; // Из promise берем future для возврата из операции отсылки сообщения. auto f = msg->promise.get_future(); // Ставим сообщение в очередь к получателю. mbox->store_to_message_queue( actual_message ); // Отдаем отправителю future в качестве результата. return f; } |
Второй же фокус в том, что есть две функции, связанные с обработкой входящего сообщения. Первая -- это написанный пользователем код обработчика события. Вроде того, что был показан выше. А вторая функция -- это лямбда, генерируемая при выполнении подписки события. Эта лямбда и разбирается с тем, что пришло к агенту: обычное асинхронное сообщение или же синхронный запрос. На уровне псевдокода работу этой лямбды можно представить так:
|
void on_msg_convert_handler( delivery_type, message_t * msg ) { // Если обычное асинхроное сообщение, то просто вызываем событие. if( async_msg == delivery_type ) agent->evt_convert( event_data_t( dynamic_cast< msg_convert * >(msg) ) ); else { // А вот если это синхронный запрос... // ...то нужно вызывать метод и сохранить результат в promise. actual_message * sync_request = dynamic_cast< actual_message * >(msg); sync_request->promise.set_value( agent->evt_convert( event_data_t( actual_message->request ) ) ); } } |
В реальности, конечно, кода чуть побольше, но делает он именно такую работу.
Пока мы говорили лишь о стороне получателя, т.е. о том, как запросы исполняются. Это более простая тема, поэтому с нее и начали. Отсылка же синхронного запроса несколько посложнее.
Начнем с того, что сообщение отсылается не агенту, а в почтовый ящик, называемый в SObjectizer-е mbox-ом. Сообщения попадают в mbox, а оттуда доставляются всем подписчикам, которые оформили подписку на конкретный тип сообщений из этого mbox-а (именно это и делает цепочка вызовов so_subscribe(mbox).event(handler) из примера выше). Подписчиков может не быть вовсе. Может быть всего один подписчик. А может и больше.
Так вот в текущей версии SObjectizer синхронная доставка сообщения завершится успешно только если подписчик будет ровно один. Не больше, не меньше. В противном случае возникнет исключение. Т.е. текущая реализация нацелена на получение только одного результата от одного агента-получателя. Об этом говорит и синтаксис, который используется при синхронной отсылке сообщения:
| mbox->get_one< std::string >()./* что-то еще... */ |
Как раз шаблонный метод get_one у mbox-а и указывает на намерение пользователя сделать синхронный запрос с целью получения результата типа std::string.
Название get_one выбрано не случайно. Сейчас берется результат только от одного обработчика запроса. Однако, если в будущем кому-то потребуется собрать результаты от всех подписчиков на сообщение, то можно будет сделать и метод get_all:
mbox->get_all< std::string, std::vector< std::string > >()./* что-то еще... */
Т.е., в будущих версиях SObjectizer можно будет сделать так, что будут запущены все подписчики на сообщение, а результаты их работы будут собраны в один std::vector, который и станет результатом get_all. Если это кому-нибудь потребуется, конечно :)
Далее у отправителя сообщения есть выбор. Он может либо получить объект-future, либо же захотеть получить конкретный результат, спрятав ожидание на future где-то внутри SObjectizer-а.
Получение объекта-future удобно тем, что мы можем сначала инициировать запрос, потом сделать какие-то собственные действия, а затем уже взять результат операции из future:
|
// Сначала запрашиваем конвертацию... auto user_count = mbox->get_one< std::string >().async( new msg_convert( 10 ) ); auto request_processed = mbox->get_one< std::string >().async( new msg_convert( 20 ) ); // ... затем делаем что-то еще... ... // ...и лишь сейчас используем результат конвертации. status = user_count + " user(s) served with " + request_processed + " request(s) processed"; |
Более того, инициировать запрос и получить результат мы можем в разных событиях агента-инициатора запроса:
|
class a_client_t : public so_5::rt::agent_t { public : void evt_1(...) { convert_result = mbox->get_one< std::string >().async( new msg_convert(1) ); ... /* еще какие-то действия */ } void evt_2(...) { ... /* какие-то действия */ // И вот только сейчас обращаемся к результату. std::cout << convert_result.get() << std::endl; ... } ... private : // Объект-future для результата запроса. // Этот объект должен быть атрибутом объекта, дабы сохранять // свое значение между вызовами обработчиков событий. std::future< std::string > convert_result; }; |
Т.е. работа через future являтся самой гибкой и дает пользователю максимум возможностей. Он может получать результат когда ему заблогорассудится, может ожидать его появляения через wait_for или wait_until, может передавать future куда-то еще.
Но, если гибкость future не нужна, а нужно сразу получить значение, то и это можно сделать. Только сначала нужно указать, сколько времени SObjectizer будет ожидать готовности результата. Ждать можно либо "до посинения", т.е. бесконечно долго:
|
auto user_count = mbox->get_one< std::string >().wait_forever().sync_get( new msg_convert( 10 ) ); |
Либо же органичив тайм-аут ожидания:
|
// Ждать не более 100 миллисекунд. auto user_count = mbox->get_one< std::string >() .wait_for( std::chrono::milliseconds( 100 ) ) .sync_get( new msg_convert( 10 ) ); |
На самом деле конструкции wait_forever().sync_get() и wait_for().sync_get() -- это просто облегчающие разработчику жизнь обертки над async() и последующим вызовом std::future::get(). Никакой хитрой логики внутри них нет.
Хотя использовать wait_for().sync_get() более предпочтительно, чем wait_forever, даже не смотря на более длинный синтаксис вызова sync_get(). Ожидание на тайм-ауте не препятствует возникновению дедлоков. Но зато позволяет их разрушать, когда тайм-аут завершается. В конце-концов, посредством нехитрых приемов от лишней избыточности можно избавиться. А истечение тайм-аутов и разрушенные таким образом тупики позволят приложению не заснуть навечно.
В общем, главная задача пользователя при использовании синхронного взаимодействия -- обеспечение работы service_handler-а и клиента сервиса на разных нитях. В идеале, service_handler-ы должны быть либо активными объектами, либо же членами независимых от клиентов активных групп. Если пользователь обеспечивает это посредством распределения прикладных агентов по соответствующим диспетчерам, то проблем с синхронностью у него не будет.
Вот, в приниципе, и все. Из всего вышесказанного становится понятно, что речь идет не столько о чисто синхронном вызове одного агента другим (с прямым доступом к данным друг друга). И даже не о синхронной доставке сообщения-инцидента агенту-получателю. Более точно было бы говорить о синхронном получении результата операции, асинхронно выполненной другим агентом где-то на своей нити. Но какому маркетологу нужны такие технические тонкости? ;)
На данный момент реализация синхронного взаимодействия лежит в branch-е версии 5.3. В свет она выйдет вместе с релизом версии 5.3.0, где-то в конце июня или начале июля 2014, если все пойдет нормально. На мой взгляд, формат API синхронного взаимодействия стабилизировался и изменений не предвидится. Хотя, кто знает, может со временем еще что-то более интересное придумается.
Но рассказ был бы неполным без поверхностного затрагивания еще нескольких связанных с async()/sync_get() тем.
Во-первых, это тема исключений. Исключения могут возникнуть по инициативе SObjectizer-а. Например, если получателей сообщения-инцидента больше одного. Или, если агент-получатель не смог обработать исключение из-за нахождения в состоянии, где это сообщение не обрабатывается. Или же исключение может возникнуть в цепочке wait_for().sync_get() при истечении тайм-аута ожидания результата.
Так же исключения могут возникнуть внутри метода-события агента-обработчика. Такое исключение будет поймано SObjectizer-ом и сохранено в объекте std::promise. Следовательно, это же исключение возникнет у клиента при обращении к future::get() (или при вызове wait_*().sync_get()). Это дает еще один привычный для синхронных вызовов эффект: вылет исключения из вызванного чужого метода. Так что при работе с синхронными сообщениями пользователю нужно быть готовым к исключениям. Пример можно посмотреть в репозитории.
Во-вторых, это тема избыточности синтаксиса. Примеры выше показывали, что синхронные обращения требуют довольно много писанины. Если все делать с прицелом на преодоление дедлоков, то нужно писать get_one<RESULT>().wait_for(timeout).sync_get(new MSG(parameters)), что довольно длинно.
Сократить объем писанины можно за счет того, что get_one.wait_* возвращают промежуточные объекты. Которые можно сохранить и затем переиспользовать. Что позволяет писать компактнее:
|
auto svc = mbox->get_one< std::string >().wait_for( std::chrono::milliseconds(100) ); auto user_count = svc.sync_get( new msg_convert(10) ); auto request_count = svc.sync_get( new msg_convert(20) ); |
Так же, если компилятор поддерживает variadic templates, можно использовать make_async вместо async, и make_sync_get вместо sync_get. Разница получается приблизительно такая же, как между прямым использованием std::shared_ptr и обращением к std::make_shared. Посредством функций make_* приведенный только что пример может быть записан как:
|
auto svc = mbox->get_one< std::string >().wait_for( std::chrono::milliseconds(100) ); auto user_count = svc.make_sync_get< msg_convert >( 10 ); auto request_count = svc.make_sync_get< msg_convert>( 20 ); |
По количеству символов выигрыша не будет, но выглядит приятнее :)
В-третьих, это тема методов-событий, которые возвращают void. Т.е. иногда при синхронном взаимодействии важно не столько получить результат предоставляемого каким-то агентом сервиса, а дождаться завершения обработки запроса. Например, если это запрос на запись содержимого внутренных буферов с данными из ОП на диск. В таких случаях вполне можно писать и так:
|
// Возврат управления произойдет только когда завершится // обработка сигнала msg_flush (либо возникнет исключение). mbox->get_one< void >().wait_forever().make_sync_get< msg_flush >(); |
Но get_one<void> выглядит как-то дико :) Синонимом такого обращения является метод run_one:
|
// Возврат управления произойдет только когда завершится // обработка сигнала msg_flush (либо возникнет исключение). mbox->run_one().wait_forever().make_sync_get< msg_flush >(); |
Ну вот теперь уже точно все, что я хотел рассказать на эту тему. Спасибо всем, у кого хватило терпения дочитать до этого места :)
PS. Данный текст был написан за короткое время, поэтому его наверняка можно сделать лучше. Оставляйте в комментариях свои вопросы/замечания и я постараюсь их учесть в последующих редакциях этой заметки.
Комментариев нет:
Отправить комментарий