В комментариях к последней статье про шишки, которые нам довелось набить за 15 лет использования акторов в C++, вновь всплыла тема отсутствия в SObjectizer-5 распределенности «из коробки». Мы уже отвечали на эти вопросы множество раз, но очевидно, что этого недостаточно.

В SObjectizer-5 нет распределенности потому, что в SObjectizer-4 поддержка распределенности была, но по мере того, как расширялся спектр решаемых на SObjectizer задач и росли нагрузки на SObjectizer-приложения, нам пришлось выучить несколько уроков:

• Под каждый тип задачи желательно иметь свой специализированный протокол. Потому что обмен большим количеством мелких сообщений, потеря части которых не страшна, сильно отличается от обмена большими бинарными файлами;
• Реализация back-pressure для асинхронных агентов — это сама по себе непростая штука. А когда сюда еще и примешивается общение по сети, ситуация становится гораздо хуже;
• Сегодня какие-то куски распределенного приложения обязательно будут написаны на других языках программирования, а не на C++. Поэтому требуется интероперабильность и наш собственный протокол, заточенный под C++ и SObjectizer, мешает разработке распределенных приложений.

Далее в статье попробуем раскрыть тему подробнее.

Преамбула

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

Прозрачная отсылка и получение сообщений

Распределенность «из коробки» хороша тогда, когда она прозрачна. Т.е. когда отсылка сообщения на удаленную сторону ничем не отличается от отсылки сообщения внутри приложения. Т.е. когда агент вызывает send<Msg>(target,...), а уже фреймворк определяет, что target — это какой-то другой процесс или даже другая нода. После чего сериализует, ставит сообщение в очередь на отправку, отсылает в какой-то канал.

Если такой прозрачности нет и API для отсылки сообщения на удаленную сторону заметно отличается от API для отсылки сообщения внутри того же процесса, то в чем выгода от использования встроенного в фреймворк механизма? Чем это лучше использования какого-нибудь готового MQ-брокера, ZeroMQ, RESTful API или, скажем, gRPC? Собственный велосипед на эту тему вряд ли будет лучше уже готового стороннего инструмента, который развивается давным-давно и гораздо большими силами.

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

Прозрачность топологии приложения

Хорошо, когда все узлы распределенного приложения связаны друг с другом:

В этом случае вообще нет проблемы отослать сообщение от узла A узлу E.

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

В этом случае узел A не может отправить сообщение узлу E непосредственно, для этого нужно будет воспользоваться услугами промежуточного узла D. Естественно, хочется, чтобы программист, который использует акторный фреймворк, вообще ничего не знал про выбор маршрутов передачи сообщений. Поскольку если программисту придется явно указывать, что сообщение от A к E должно идти через D, то толку от такой распределенности «из коробки» будет не так много, как хотелось бы.

Минимизация количества каналов и их типов

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

Только на практике это не удобно. Чем меньше каналов, тем лучше. Если достаточно всего одного канала для передачи всех трех типов трафика, то и самому программисту будет проще, а уж DevOps-ы точно скажут спасибо.

Минимизация количества каналов так же важна и для решения проблем топологии распределенного приложения. Так, если взаимодействовать нужно узлам A и E, то вряд ли будет удобно создавать под это взаимодействие три разных канала между A и D, а затем еще три между E и D.

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

Ну и, собственно, сами подводные камни

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

Back-pressure между прикладными и траспортными агентами

Асинхронный обмен сообщениями плохо дружит с back-pressure. Агент, который выполняет send<Msg>(target,...) не может быть просто так приостановлен на вызове send, если в очереди получателя скопилось достаточно много необработанных сообщений. В случае, когда target — это удаленная сторона, у нас получается следующее:

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

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

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

• Размеры этих буферов хорошо бы тюнить для того, чтобы терять как можно меньше и спокойно переживать небольшие затыки. Появляется головная боль по поводу настройки этих параметров. Еще веселее, когда данные параметры должны адаптироваться к текущим условиям в run-time;
• Прикладная логика приложения может требовать как-то фиксировать факты выбрасывания сообщений. Грубо говоря, если сообщение A с идентификатором Id1 было выброшено из-за затыков в сети, то следы этого должны появиться в логах. Иначе затем будет не понять, почему транзакция X началась, дошла до стадии отсылки A, а потом «подвисла» без движения.

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

Причем понятие «длительный» может варьироваться от приложения к приложению. Где-то «длительный» — это несколько минут или даже десятков минут. А где-то, где сообщения передаются с темпом 10000 в секунду, длительным будет считаться разрыв всего в 5 секунд.

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

Back-pressure между транспортными и прикладными агентами

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

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

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

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

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

Смешение трафика с разными приоритетами в одном канале

Когда один и тот же канал используется для передачи трафика разного типа, то рано или поздно возникает задача приоритизации этого самого трафика. Например, если из узла A на узел E был отправлен большой сверочный файл размером в несколько десятков мегабайт, то пока он идет, может возникнуть необходимость передать несколько сообщений с новыми транзакциями. Поскольку транзакционный трафик для приложения приоритетнее, чем передача файлов со сверками, то хотелось бы приостановить передачу файла сверок, отослать несколько коротких сообщений с транзакциями и вернуться вновь к приостановленной передаче большого файла.

Проблема здесь в том, что тривиальные реализации транспортного слоя так не сумеют :(

Fire-and-forget or not-to-forget?

Думаю, что не ошибусь, если скажу, что использование политики fire-and-forget является общепринятой практикой при построении приложений на базе агентов/акторов. Агент-отправитель просто асинхронно отсылает сообщение агенту-получателю и не заботится о том, дошло ли сообщение или потерялось где-то по дороге (например, было выброшено механизмом защиты агентов от перегрузки).

Все это хорошо до тех пор, пока не возникает необходимость передачи больших бинарных блоков (BLOB-ов) через транспортный канал. Особенно, если этот канал имеет свойство часто рваться. Ведь никому не понравится, если мы начали передавать блок в 100MiB, передали 50MiB, канал порвался, затем восстановился, а мы не стали ничего досылать. Или начали перепосылать все заново.

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

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

Итого

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

Ни в коем случае не хочу сказать, что невозможно сделать транспортный слой, который бы успешно закрывал все или большую часть обозначенных проблем. Насколько я знаю, в Ice от ZeroC что-то подобное как раз и реализовано. Или в nanomsg. Так что это возможно.

Проблемы здесь две:

• Сможете ли вы настолько вложиться в разработку своего продвинутого транспортного слоя? Тот же Ice от ZeroC находится в активной разработке и эксплуатации больше полутора десятков лет. А автор nanomsg перед этим «собаку съел» на реализации ZeroMQ. Будут ли у вас такие ресурсы и такая компетенция?
• Как при этом обеспечить еще и интероперабильность с другими языками программирования? Замкнутым системам, вроде Erlang-а и Akka, хорошо, их мало волнует то, что происходит вне их экосистем. В случае же с самодельным и малоизвестным акторным фреймворком для C++ ситуация совершенно другая.

В SObjectizer-4 со всеми этими проблемами мы столкнулись. Какие-то решили, какие-то обошли, какие-то остались. А вот при создании SObjectizer-5 мы взглянули на проблему распределенности еще раз и поняли, что наших ресурсов на то, чтобы успешно справиться еще и с ней, у нас не хватит. Мы сосредоточились на том, чтобы SObjectizer-5 стал классным инструментом для упрощения жизни при создании многопоточных приложений. Поэтому SO-5 закрывает проблемы диспетчеризации сообщений внутри одно процесса. Ну а общение между процессами...

Как оказалось, такое общение довольно удобно делать с использованием готовых протоколов и готовых MQ-брокеров. Например, посредством MQTT. Конечно же, при этом нет никакой прозрачной распределенности. Но практика показывает, что когда речь заходит о высоких нагрузках или сложных сценариях работы, от прозрачной распределенности пользы мало.

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