Обновился инструмент для работы с агентами в C++: SObjectizer 5.5.5

Пока мало что понятно

Можно спрашивать, не стесняться :)

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

По самой задачке с pipeline. Ща как раз делаю наброски... Пока кода получается слишком много инфраструктурного. Но может это из-за того, что шаблонные навороты в C++ — это не мое :)

Нет желание обсудить это дело в привате? Или можно прямо здесь продолжить, может кому-то из читателей будет интересно.

Интересно, завтра поразбираю примеры и пощупаю SO5, и можно будет здесь пообсуждать как организовать пайп.

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

А что мешает весь этот конвейер запихать в одного агента.

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

Хотя, на практике, как раз внутри агента весь пайп или его куски и нужно будет располагать. Но тогда от разработчиков SO5 особого участия не требуется :)

Все это напоминает популярный сейчас reactive programming.

В Akka это называется stream.

Интересно, завтра поразбираю примеры и пощупаю SO5, и можно будет здесь пообсуждать как организовать пайп.

В первом приближении получилось вот так.

Для простоты реализации и дабы не привлекать слишком много шаблоновой магии каждая стадия пайпа должна представляться в виде объекта std::function. Формат функции, которая находится внутри должен быть таким:

intrusive_ptr_t<OUT>(const IN&);

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

Отдельная стадия пайпа создается вот таким образом:

stage_handler_t< first, second >
first_stage( const std::string & border )
{
	return [border]( const first & msg ) {
		return make_message< second >( border + msg.m_data + border );
	};
}

Сам пайп организуется вот так:

auto first_pipeline = make_pipeline( *this,
	src | first_stage( "=" ) | second_stage( "-" ) | third_stage( self ) );

Вообще в примере создается два пайпа из одинаковых стадий, но с разными параметрами:

virtual void so_evt_start() override
{
	const auto self = so_direct_mbox();
	auto first_pipeline = make_pipeline( *this,
		src | first_stage( "=" ) | second_stage( "-" ) | third_stage( self ) );

	auto second_pipeline = make_pipeline( *this,
		src | first_stage( "#" ) | second_stage( "*" ) | third_stage( self ) );

	send< first >( first_pipeline, " one " );
	send< first >( second_pipeline, " two " );
}

Это самый первый более-менее приличный набросок формирования пайпов. В нем не решена следующая проблемка. По сути, последний элемент пайпа (т.е. sink) может иметь прототип void(const IN&). Но сейчас его таким образом нельзя задать. Даже в таком виде нельзя задать:

intrusive_ptr_t<void>(const IN&)

Полагаю, что с этим можно справиться, если придумать, где именно добавить специализации для void.

PS. Проверялось это все под MSVC++ 12, GCC 4.9.2, GCC 5.1.0, Clang 3.6.0. Как раз из-за MSVC++ 12 не были задействованы возможности C++14, могло бы быть чуток поприятнее и поэффективнее из-за изменения правил lambda captures.

Еще один важный момент: за счет шаблонов проверяется совместимость стадий по аргументам. Т.е. OUT от предыдущей стадии должен быть IN-ом для следующей. Поэтому, если вдруг по ошибке записать вот так:

auto first_pipeline = make_pipeline( *this,
	src | first_stage( "=" ) | third_stage( self ) );

Полагаю, что с этим можно справиться, если придумать, где именно добавить специализации для void.

Вот вариант, который позволяет sink-стадии пайпа иметь возвращаемое значение типа void.

Пришлось сделать дополнительный handler_traits_t со специализацией для случая <IN,void>. И stage_handler_t теперь является не псевдонимом для std::function, а самостоятельным объектом.

Так же потребовалось сделать специализацию для a_stage_point_t для случая <IN,void>, т.к. этот агент не может иметь m_next и ему не нужно при получении очередного сообщения куда-то передавать результат обработки.

В конструировании же пайпа ничего не изменилось. Синтаксис остался точно таким же.

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

src | first_stage() | second_stage() | third_stage()

src | first_stage() | second_stage() | third_stage() | fourth_stage()

Одна из важнейших тем при работе с SO5 — это использование исключений. Долго не удавалось выкроить время и подготовить более-менее подробный рассказ на эту тему. Сейчас удалось. Очередную часть серии Dive into SObjectizer-5.5, полностью посвященную исключениям, можно найти здесь (или на SlideShare).

Еще один важный момент: за счет шаблонов проверяется совместимость стадий по аргументам.

Это наоборот плюс, в Erlang'e приходится в receive делать catch-all. Спасибо за пример, сегодня поразбираюсь что и как у вас там работает. Заодно попробую написать ранее оговоренный пайп на SO5.

Это наоборот плюс

Честно говоря, я не очень понял, к чему относится «наоборот плюс» — к контролю типов во время компиляции в C++ или к отсутствию оного в Erlang.

Мое личное мнение очень простое: в Erlang строгая типизация, хотя и динамическая. Поэтому, если там мы ждем, скажем, {encrypt, Data}, а приходит {verify_sign,Data,Signature}, то возникнет ошибка в run-time, но порчи данных или каких-то иных фатальных неприятностей не будет. А вот в С++ и, особенно, в C, типизация хоть и статическая, но слабая. Поэтому, если мы получаем некий void* и думаем, что это указатель на encrypt_request, а на самом деле там verify_sign_request, то мы можем запортить чего-нибудь в памяти процесса и это аукнется не сразу, а когда-нибудь потом и в неожиданном месте.

Поэтому в C++ наличие контроля за типами во время компиляции — это очень и очень серьезное подспорье при написании надежного кода.

Честно говоря, я не очень понял, к чему относится «наоборот плюс» — к контролю типов во время компиляции в C++ или к отсутствию оного в Erlang.

К контролю типов во время компиляции. Изменив формат сообщения в одном из элементов пайпа, нужно не забыть обновить receive-выражение в следующем элементе. Я у себя нагородил для пайпов аналог gen_server'a, который всякие плюшки добавляет к пайпу, вроде обратной связи для того чтобы Producer не завалил Consumer'a сообщениями если тот не успевает их обработать и т.д. Так факапы случаются значительно реже.

К контролю типов во время компиляции.

Как раз это и есть одно из основных преимуществ C++ над C при разработке софта. Благо, шаблоны в C++11 стали мощнее и чуть удобнее, чем в C++03. Так что даже мне удается иногда немного шаманить с определением типов аргументов/возвратов во время компиляции.

вроде обратной связи для того чтобы Producer не завалил Consumer'a сообщениями если тот не успевает их обработать и т.д.

Как раз для упрощения создания механизма overload control не так давно в SO5 были добавлены такие штуки, как лимиты на сообщения. А в следующую версию 5.6.0 хотим добавить еще и приоритеты доставки, которые должны помочь Consumer-ам оперативно реагировать на постановку к ним в очередь слишком большого количества сообщений для обработки.

Благо, шаблоны в C++11 стали мощнее и чуть удобнее, чем в C++03. Так что даже мне удается иногда немного шаманить с определением типов аргументов/возвратов во время компиляции.

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

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

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

|-------|    |-------|    |-------|
| ALSA  |--->| GSM   |--->| UDPdst|
|-------| |  |-------|    |-------|
   ^      |
   |      |  |-------|    |-------|    |-------|
   |      |->|  EC   |<---| GSM   |<---| UDPsrc|
   |         |-------|    |-------|    |-------|
   |-------------|

EC - echo cancellation

А что скрывается за echo cancellation?

И роль ALSA в чем? Воспроизведение звука, оцифровка звука?

Тебе было лень плюсики ставить в схеме?

А что скрывается за echo cancellation?

Кратко в вики - https://ru.wikipedia.org/wiki/Эхоподавление

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

Просто я затупил.

Понятно, я так и подумал.

Очень похоже на реализацию голосовой почты для GSM-абонентов.

Очень похоже на реализацию голосовой почты для GSM-абонентов.

Это простая схема работы VoIP. =) Деджиттера только нету.

Это простая схема работы VoIP

Никогда не приходилось иметь дело с этим зверем :)

Это, имхо, идеальный юзкейз для акторов =) Саму идею юзанья вот таких вот пайпов позаимствовали из либы mediastreamer2 - оказалось что это ну очень удобно. всякие эфекты, фильтры и плагины просто ставятся в пайп и вуаля =)

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

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

Но тогда, похоже, схема должна выглядеть чуть иначе. Выход из ALSA должен быть сначала в EC, затем уже в GSM, затем в UDPdst:

+------+    +----+    +-----+    +--------+
| ALSA |--->| EC |--->| GSM |--->| UDPdst |
+------+    |    |    +-----+    +--------+
   ^        |    |
   |        |    |    +-----+    +--------+
   `--------|    |<---| GSM |<---| UDPsrc |
            +----+    +-----+    +--------+

Лорвики вестимо. Будет православно. или менее православно в файлике на шитхабе.

в этот раз я припозднился со своим «ненужно», но, похоже:

Но тогда от разработчиков SO5 особого участия не требуется :)

до тебя начинает доходить

Изменив формат сообщения в одном из элементов пайпа, нужно не забыть

нужно не забыть использовать где надо auto и decltype

Как раз для упрощения создания механизма overload control не так давно в SO5 были добавлены такие штуки, как лимиты на сообщения. А в следующую версию 5.6.0 хотим добавить еще и приоритеты доставки

смотрю, ты на своей поделке наконец собрался сделать что-то сложнее hello-world-а...

+------+    +-----+    +--------+
| ALSA |--->| GSM |--->| UDPdst |
+------+ |  +-----+    +--------+
   ^     |  +----+
   |     `->|    |    +-----+    +--------+
   `--------| EC |<---| GSM |<---| UDPsrc |
            |    |    +-----+    +--------+
            +----+ 

то, что снято с микрофона идет независимо в GSM для кодирования и, одновременно, в EC для того, чтобы затем этот сигнал можно было вырезать от того, что GSM принял в ответ на вход

This

нужно не забыть использовать где надо auto и decltype

В Erlang'e? Речь там идет о нем.

Поэтому, если там мы ждем, скажем, {encrypt, Data}, а приходит {verify_sign,Data,Signature}, то возникнет ошибка в run-time

если ты в ресиве ждёшь первое, а приходит второе, то нифига не происходит.

то возникнет ошибка в run-time

Нет. Из mailbox будет извлечено следующее сообщение и так далее до тех пор, пока сообщений не останется.

Тогда, вероятно, вся конструкция выглядела бы как-то так:

// Echo canceler будет динамическим объектом, который будет
// использоваться из разных стадий в разных контейнерах.
auto echo_canceler = make_shared< echo_canceler >(...);

// ALSA нужно будет делать отдельным агентом, с наследованием
// от agent_t и т.д. Это нужно, во-первых, чтобы включить его
// сразу в несколько конвейеров.
// Во-вторых, он является источником данных для конвейера с
// исходящими данными.
auto alsa = coop.make_agent< a_alsa_t >( ... );

// Конвейер для исходящих данных.
auto gsm_to_udp = make_pipeline( coop,
  src | broadcast(
          src | stage(gsm_encoder) | stage(udp_dest),
          src | stage( [echo_canceler]( const recorded_voice & v ) {
                  echo_canceler->save_recorded_voice(v);
                } )
        )
  );
// Сообщаем об этом конвейере агенту alsa.
alsa->set_out_pipeline( gsm_to_udp );

// Конвейер для входящих данных.
auto udp_to_alsa = make_pipeline( coop,
  src | stage(gsm_decoder)
      | stage( [echo_canceler]( const received_voice & v ) {
          return echo_canceler->handle_received_voice(v);
        } )
      | stage( [alsa]( const cleared_voice & v ) {
          send< cleared_voice >( alsa->so_direct_mbox(), v );
        } )
  );

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

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

смотрю, ты на своей поделке наконец собрался сделать что-то сложнее hello-world-а...

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

Что конкретно до overload control и приоритетов доставки, то здесь ситуация с обычными форумными болтунами видна еще более явно, чем где-то еще. Тут даже не многие могут на абстраткно-сферическо-вакуумном уровне потрындеть.

Если такая ветка есть, то сообщения, для которого мы не написали match будет потеряно, что, вероятнее всего, баг.

Конечно будет потеряно. Но рантайм-ошибки не будет.

Конечно будет потеряно. Но рантайм-ошибки не будет.

Я думаю он имел в виду что на такой случай иногда делают так:

receive
    {myawesomedata, Data} -> process(Data);
    Unknown -> wtf = Unknown
end

Но рантайм-ошибки не будет.

Трудности перевода :)

Подразумевалось вот что: баг будет диагностирован только в run-time, при этом никаких пагубных последствий, скажем, для VM, не будет. Но сам баг в приложении проявится только в run-time.

Тогда как в C/C++ можно нарваться не только на баг в приложении, но еще и спровоцировать порчу памяти в процессе.

Если только корректность работы с типами не проверяется в compile-time. Что в ряде случаев возможно в C++, но вряд ли возможно в Erlang из-за его динамической типизации.

даже список штатных примеров включает далеко не только ping-pong.

ну да, их там целых три штуки. и ещё два философа :)

Что конкретно до overload control и приоритетов доставки, то здесь ситуация с обычными форумными болтунами видна еще более явно

я уже давно перестал отрицать твою тупость и неадекватность. и не надо со мной спорить...

без

overload control и приоритетов доставки

и много чего ещё компетентный менеджер проект на пушечный выстрел к деплою не пустит, а ты только через 15 лет допёр

ну да, их там целых три штуки. и ещё два философа :)

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

я уже давно перестал отрицать твою тупость и неадекватность. и не надо со мной спорить...

Зачем же спорить, если вы своими словами подтверждаете то, о чем я и говорил: ничего конкретного и адекватного на тему overload control и приоритетов доставки форумные болтуны сказать не могут. Проверено неоднократно.

и много чего ещё компетентный менеджер проект на пушечный выстрел к деплою не пустит

Видимо, ПМ-ы, руководящие запуском проектов на Erlang-е или Akka, компетентностью не отличаются. Куда им до анонимных тимлидов с LOR-а.

Видимо, ПМ-ы, руководящие запуском проектов на Erlang-е или Akka, компетентностью не отличаются.

ты, видимо, пытаешься их самостоятельно оценить...

ты, видимо, пытаешься их самостоятельно оценить...

Да куда уж мне до многоопытных анонимных тимлидов. Я хотя бы чего-нибудь про overload control от столь эрудированных экспертов услышать. А то опять 15 лет доходить буду до того, что избранным уже ведомо.

ничего конкретного и адекватного на тему overload control и приоритетов доставки форумные болтуны сказать не могут. Проверено неоднократно.

Кстати, интересно как вы собираетесь реализовать эти функции (в плане архитектуры). Квоты на размер ящика? Или какие-то более хитрые эвристические методы? А приоритеты доставки как?

Поскольку текста оказалось много, то разместил его у себя в блоге. Вот ссылка.

Должен сказать, что пока все это на очень ранней стадии проектирования, так что можно все переиграть чуть меньше, чем полностью :)

Как всегда, все дело за хорошими идеями и/или примерами из реальной жизни.

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

Плохая затея. Event-stream'ы с высоким приоритетом могут заставить вечно «голодать» в ожидании низкоприоритетные потоки и в конце концов просто выстрелит overload control для них (низкоуровневых потоков). Шедульте их хотя бы по элементарным формулам вроде: stream ticks = stream priority + 1, т.е если ести три очереди с приоритетами 0, 1 и 2. то сначала исполняется 3 таска из высокоприоритетной, 2 из среднеприоритетной, и 1 из дефолтной. Добавьте сюда немного эвристики только, предложенный метод совсем уж тупой.

Есть такое опасение. Но эта идеология идет из систем реального времени, где приоритет — это приоритет без всяких оговорок.

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

Гм, заведите пул dirty-scheduler'ов. Как только диспетчер засечет что очереди голодают, то он просто отдает в dirty-scheduler низкоприоритетные таски (или таск).

А зачем?

Допустим, у нас есть агент A с заявками e1 и e0 (цифра означает приоритет). Нет смысла запускать e0 для A до тех пор, пока для A не отработает e1.

Если же речь идет о том, что есть e1 для агента B и e0 для агента C, то тут будет зависеть от того, к какому диспетчеру B и C привязаны:

- если к диспетчеру с одной рабочей нитью, то порядок должен быть таким: e1 для B, затем e0 для C;

- если к диспетчеру с пулом рабочих потоков, то, вероятно, на одном потоке будет работать e1 для B, на втором e0 для C, если других заявок с приоритетом 1 нет.