Restoring authorship annotation for Cthulhu <cthulhu@yandex-team.ru>. Commit 1 of 2.

contrib/libs/grpc/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h

@@ -22,7 +22,7 @@
 #include <grpc/support/port_platform.h>
 
 #include <string.h>
-#include <util/system/compiler.h> // Y_NO_SANITIZE
+#include <util/system/compiler.h> // Y_NO_SANITIZE 
 
 #include <grpc/support/alloc.h>
 
@@ -353,7 +353,7 @@ template <typename SubchannelListType, typename SubchannelDataType>
 SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList(
     LoadBalancingPolicy* policy, TraceFlag* tracer, ServerAddressList addresses,
     LoadBalancingPolicy::ChannelControlHelper* helper,
-    const grpc_channel_args& args) Y_NO_SANITIZE("undefined")
+    const grpc_channel_args& args) Y_NO_SANITIZE("undefined") 
     : InternallyRefCounted<SubchannelListType>(tracer),
       policy_(policy),
       tracer_(tracer) {

library/cpp/actors/README.md

@@ -1,107 +1,107 @@
-## Actor library
-
-### Часть первая, вводная.
-Иногда приходится разрабатывать асинхронные, существенно параллельные, местами распределённые программы. Иногда еще и внутренняя логика нетривиальна, разнородна, пишется разными командами не один год. Всё как мы любим. Человечеством придумано не так много способов внутренней организации структуры и кода таких программ. Большинство из них плохие (и именно из-за плохих подходов разработка асинхронных, многопоточных программ приобрела дурную славу). Некоторые получше. А серебряной пули как обычно нет.
-
-Когда мы начинали разработку Yandex Database (тогда еще KiKiMR), сразу было понятно что простыми наколеночными поделиями обойтись (и сделать при этом хорошо, так что бы не было стыдно) не получится. В качестве базиса мы выбрали мессадж-пассинг и модель акторов. И не пожалели. Постепенно этот подход распространился на смежные проекты.
-
-### Базовые концепции.
-Если отбросить шелуху – представляем сервис (программу в случае запуска изолированного бинарника) как ансамбль независимых агентов, взаимодействующих через отправку асинхронных сообщений внутри общего окружения. Тут все слова важны:
-
-Независимых – не разделяют состояние и поток выполнения.
-Передача сообщений – формализуем протоколы, а не интерфейсы.
-
-Асинхронная – не блокируемся на отправке сообщений.
-Общее окружение – все агенты разделяют общий пул ресурсов и каждый из них, зная адрес, может послать сообщение каждому.
-
-В более хайповых терминах – очень похоже на колокейтед микросервисы, только уровнем ниже. И да, мы заведомо не хотели прятать асинхронщину и параллелизм от разработчика, показывая прям самое мясо.
-
-### IActor.
-https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/core/actor.h?rev=5315854#L105
-Базовый класс всех агентов, напрямую обычно не используется. Инстанцируется либо TActor, либо TActorBootstrapped. Фактически весь полезный код программы размещается в акторах.
-(важное замечание – в коде увидите ручки с TActorContext и без него, схожие по названию и назначению. На данный момент вариант с TActorContext является устаревшим, новый код стоит писать без его использования).
-Важные методы:
-
-PassAway – единственный корректный способ зарегистрированному актору умереть. Может вызываться только находясь внутри обработчика сообщения.
-Send – отправка сообщения, зная адрес получателя. В акторе доступен хелпер, принимающий непосредственно сообщение. Базовый вызов, принимающий полный event handle – доступен в контексте.
-
-Become – установить функцию-обработчик сообщений, которая будет использована при получении следующего сообщения.
-
-Register – зарегистрировать новый актор в акторсистеме, с выделением нового мейлбокса. Важно – с момента вызова владение актором передается акторсистеме, т.е. уже к моменту выхода актор может начать выполняться на другом потоке, нельзя к нему ни обращаться прямыми вызовами, ни даже предполагать что он еще жив.
-
-Schedule – зарегистрировать сообщение, которое будет отправлено не менее чем через запрошенную задержку. В акторе доступен хелпер, декорирующий сообщение хендлом отправки самому себе, в контексте можно передать полный хендл.
-
-SelfId – узнать собственный адрес. Возвращаемый объект TActorIdentity можно передавать если требуется делегировать отправку сообщений от имени актора (например если пишете полезный код пользуясь пассивными объектами).
-Посылка сообщений дешёвая, не нужно на ней чрезмерно экономить (но не бесплатная – поэтому посылать сообщения только ради посылки сообщений то же не стоит).
-
-Инстанцирование акторов так же дёшево, актор на запрос или фазу запроса – вполне нормальная практика. Мультиплексировать обработку разных запросов в одном акторе – так же вполне нормально. В нашем коде много примеров и первого, и второго. Пользуйтесь здравым смыслов и собственным вкусом.
-Т.к. на время обработки сообщения актор занимает тред из пула акторсистемы – уходить в длинные вычисления лучше на отдельном отселённом акторе (и либо отселять в отдельный пол акторсистемы, либо контролировать параллельность брокером ресурсов), блокирующие вызовы делать почти всегда ошибка. Стремление написать мютекс - ересь и от лукавого.
-Идентифицируются акторы своим TActorID-ом, который уникален и вы не должны его придумывать из воздуха, а только получить из регистрации (для порождённых акторов) или его вам должен рассказать кто-то, законно его знающий.
-
-Отправка на несуществующий актор (уже умерший) безопасна, сообщение будет просто выброшено в момент обработки (как обрабатывать недоставку сообщений в протоколах расскажу ниже).
-
-Кроме нормальных TActorID существуют еще и сервисные (составленные из строчки и номера ноды). Под ними может быть зарегистрирован реальный актор и фактически при получении сообщения по сервисному адресу – попробует переправить его текущему фактическому. Это позволяет размещать хорошо известные сервисы по хорошо известному адресу, не выстраивая параллельную машинерию поиска.
-
-Строить из актора конечный автомат при помощи переключений функции-обработчика – выбор в каждом конкретном случае, иногда удобнее да, иногда сваливать всё в одно состояние, а иногда – применять гибридное решение (когда часть жизненного цикла – обычно инициализации и завершение – выражены в переходах, а часть – нет).
-Меньше слов, больше дела – этого уже достаточно что бы прочитать самый простой пример. https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/examples/01_ping_pong
-Здесь можно увидеть образец самого простого актора, занимающегося переброской сообщений и использующего все основные вызовы. Заодно покрутив за разные ручки (количество тредов в тредпуле, количество пар перебрасывающихся акторов) можно посмотреть на изменение поведения системы (hint: в таких простых сценариях максимум перфоманса достигается при одном треде в тредпулах).
-
-### Event и Event Handle.
-Полезную нагрузку сообщений заворачиваем в наследника IEventBase, у которого два важных метода – сериализация и загрузка. Сериализация виртуальная, а вот загрузка – нет, и для разбора сообщения из байтовой последовательности – необходимо на стороне получателя сматчить число-идентификатор типа ивента с С++ типом. Именно это делают макросы из hfunc.h. На практике ивенты создаются либо как наследник TEventLocal<> (для строго локальных сообщений) либо как наследник TEventPB<> (для потенциально пересылаемых по сети сообщений, типизируются protobuf-мессаджем).
-
-Кроме непосредственно ивента (в виде структуры либо в виде байтовой строки) для пересылки сообщения необходим набор дополнительных полей
-
-Адресат
-
-Отправитель
-
-Тип сообщения
-
-Кука
-
-Флаги
-
-Сообщение + дополнительные поля = IEventHandle. Именно хендлами акторсистема и оперирует. <event-type>::TPtr – в примере выше – это и есть указатель на типизированный хендл.
-
-Технически типом сообщения может быть любое число, которое получатель и отправитель договорились понимать как идентификатор сообщения. Сложившаяся практика – выделять диапазон идентификаторов макросом EventSpaceBegin (фактически блоками по 64к), начиная с блока ES_USERSPACE.
-Кука – неинтерпретируемое ui64 число, передаваемое с хендлом. Хорошей практикой является в ответе сервиса на сообщение выставлять куку в куку исходного сообщения, особенно для сервисов, потенциально используемых конкурентно.
-
-В флагах несколько бит зарезервировано под флаги, декларирующие как необходимо обрабатывать особые ситуации и 12 бит – под номер канала интерконнекта, в котором будет пересылаться сообщение (для локальных сообщений в имеющихся реализациях номер канала не имеет значения - хотя можно представить реализацию где для каналов будут независимые очереди).
-
-### Тредпулы и мейлбоксы.
-В рамках одной акторсистемы может сосуществовать несколько независимых тредпулов, каждый актор регистрируется на конкретном и в процессе жизни не может мигрировать (но может создавать новые акторы на произвольном тредпуле). Используется для крупноблочного разделения ресурсов, либо между разными активностями (вот здесь – обрабатываем один класс запросов, а вот здесь - другой), либо между разными профилями активности (вот здесь обрабатываем быстрые запросы, здесь – медленные, а вот там – вообще батчёвые). Например в YDB работает системный тредпул (в котором запускаются акторы, необходимые для функционирования YDB, и для которого мы следим что бы не было длительной блокировки в обработчиках), пользовательский тредпул (в котором обрабатываются запросы и потенциально обработчики могут уходить в себя подольше, но это не повлияет на инфраструктуру), батчёвый тредпул (куда отгружается длительная обработка – компакшены дисков, сканы таблиц и подобное) и, в жирных нодах – тредпул интерконнекта (как наиболее чувствительного к задержкам).
-Пересылка сообщений между акторами разных тредпулов но одной локальной акторсистемы остаётся локальной, принудительной сериализации сообщения не происходит.
-
+## Actor library 
+ 
+### Часть первая, вводная. 
+Иногда приходится разрабатывать асинхронные, существенно параллельные, местами распределённые программы. Иногда еще и внутренняя логика нетривиальна, разнородна, пишется разными командами не один год. Всё как мы любим. Человечеством придумано не так много способов внутренней организации структуры и кода таких программ. Большинство из них плохие (и именно из-за плохих подходов разработка асинхронных, многопоточных программ приобрела дурную славу). Некоторые получше. А серебряной пули как обычно нет. 
+ 
+Когда мы начинали разработку Yandex Database (тогда еще KiKiMR), сразу было понятно что простыми наколеночными поделиями обойтись (и сделать при этом хорошо, так что бы не было стыдно) не получится. В качестве базиса мы выбрали мессадж-пассинг и модель акторов. И не пожалели. Постепенно этот подход распространился на смежные проекты. 
+ 
+### Базовые концепции. 
+Если отбросить шелуху – представляем сервис (программу в случае запуска изолированного бинарника) как ансамбль независимых агентов, взаимодействующих через отправку асинхронных сообщений внутри общего окружения. Тут все слова важны: 
+ 
+Независимых – не разделяют состояние и поток выполнения. 
+Передача сообщений – формализуем протоколы, а не интерфейсы. 
+ 
+Асинхронная – не блокируемся на отправке сообщений. 
+Общее окружение – все агенты разделяют общий пул ресурсов и каждый из них, зная адрес, может послать сообщение каждому. 
+ 
+В более хайповых терминах – очень похоже на колокейтед микросервисы, только уровнем ниже. И да, мы заведомо не хотели прятать асинхронщину и параллелизм от разработчика, показывая прям самое мясо. 
+ 
+### IActor. 
+https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/core/actor.h?rev=5315854#L105 
+Базовый класс всех агентов, напрямую обычно не используется. Инстанцируется либо TActor, либо TActorBootstrapped. Фактически весь полезный код программы размещается в акторах. 
+(важное замечание – в коде увидите ручки с TActorContext и без него, схожие по названию и назначению. На данный момент вариант с TActorContext является устаревшим, новый код стоит писать без его использования). 
+Важные методы: 
+ 
+PassAway – единственный корректный способ зарегистрированному актору умереть. Может вызываться только находясь внутри обработчика сообщения. 
+Send – отправка сообщения, зная адрес получателя. В акторе доступен хелпер, принимающий непосредственно сообщение. Базовый вызов, принимающий полный event handle – доступен в контексте. 
+ 
+Become – установить функцию-обработчик сообщений, которая будет использована при получении следующего сообщения. 
+ 
+Register – зарегистрировать новый актор в акторсистеме, с выделением нового мейлбокса. Важно – с момента вызова владение актором передается акторсистеме, т.е. уже к моменту выхода актор может начать выполняться на другом потоке, нельзя к нему ни обращаться прямыми вызовами, ни даже предполагать что он еще жив. 
+ 
+Schedule – зарегистрировать сообщение, которое будет отправлено не менее чем через запрошенную задержку. В акторе доступен хелпер, декорирующий сообщение хендлом отправки самому себе, в контексте можно передать полный хендл. 
+ 
+SelfId – узнать собственный адрес. Возвращаемый объект TActorIdentity можно передавать если требуется делегировать отправку сообщений от имени актора (например если пишете полезный код пользуясь пассивными объектами). 
+Посылка сообщений дешёвая, не нужно на ней чрезмерно экономить (но не бесплатная – поэтому посылать сообщения только ради посылки сообщений то же не стоит). 
+ 
+Инстанцирование акторов так же дёшево, актор на запрос или фазу запроса – вполне нормальная практика. Мультиплексировать обработку разных запросов в одном акторе – так же вполне нормально. В нашем коде много примеров и первого, и второго. Пользуйтесь здравым смыслов и собственным вкусом. 
+Т.к. на время обработки сообщения актор занимает тред из пула акторсистемы – уходить в длинные вычисления лучше на отдельном отселённом акторе (и либо отселять в отдельный пол акторсистемы, либо контролировать параллельность брокером ресурсов), блокирующие вызовы делать почти всегда ошибка. Стремление написать мютекс - ересь и от лукавого. 
+Идентифицируются акторы своим TActorID-ом, который уникален и вы не должны его придумывать из воздуха, а только получить из регистрации (для порождённых акторов) или его вам должен рассказать кто-то, законно его знающий. 
+ 
+Отправка на несуществующий актор (уже умерший) безопасна, сообщение будет просто выброшено в момент обработки (как обрабатывать недоставку сообщений в протоколах расскажу ниже). 
+ 
+Кроме нормальных TActorID существуют еще и сервисные (составленные из строчки и номера ноды). Под ними может быть зарегистрирован реальный актор и фактически при получении сообщения по сервисному адресу – попробует переправить его текущему фактическому. Это позволяет размещать хорошо известные сервисы по хорошо известному адресу, не выстраивая параллельную машинерию поиска. 
+ 
+Строить из актора конечный автомат при помощи переключений функции-обработчика – выбор в каждом конкретном случае, иногда удобнее да, иногда сваливать всё в одно состояние, а иногда – применять гибридное решение (когда часть жизненного цикла – обычно инициализации и завершение – выражены в переходах, а часть – нет). 
+Меньше слов, больше дела – этого уже достаточно что бы прочитать самый простой пример. https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/examples/01_ping_pong 
+Здесь можно увидеть образец самого простого актора, занимающегося переброской сообщений и использующего все основные вызовы. Заодно покрутив за разные ручки (количество тредов в тредпуле, количество пар перебрасывающихся акторов) можно посмотреть на изменение поведения системы (hint: в таких простых сценариях максимум перфоманса достигается при одном треде в тредпулах). 
+ 
+### Event и Event Handle. 
+Полезную нагрузку сообщений заворачиваем в наследника IEventBase, у которого два важных метода – сериализация и загрузка. Сериализация виртуальная, а вот загрузка – нет, и для разбора сообщения из байтовой последовательности – необходимо на стороне получателя сматчить число-идентификатор типа ивента с С++ типом. Именно это делают макросы из hfunc.h. На практике ивенты создаются либо как наследник TEventLocal<> (для строго локальных сообщений) либо как наследник TEventPB<> (для потенциально пересылаемых по сети сообщений, типизируются protobuf-мессаджем). 
+ 
+Кроме непосредственно ивента (в виде структуры либо в виде байтовой строки) для пересылки сообщения необходим набор дополнительных полей 
+ 
+Адресат 
+ 
+Отправитель 
+ 
+Тип сообщения 
+ 
+Кука 
+ 
+Флаги 
+ 
+Сообщение + дополнительные поля = IEventHandle. Именно хендлами акторсистема и оперирует. <event-type>::TPtr – в примере выше – это и есть указатель на типизированный хендл. 
+ 
+Технически типом сообщения может быть любое число, которое получатель и отправитель договорились понимать как идентификатор сообщения. Сложившаяся практика – выделять диапазон идентификаторов макросом EventSpaceBegin (фактически блоками по 64к), начиная с блока ES_USERSPACE. 
+Кука – неинтерпретируемое ui64 число, передаваемое с хендлом. Хорошей практикой является в ответе сервиса на сообщение выставлять куку в куку исходного сообщения, особенно для сервисов, потенциально используемых конкурентно. 
+ 
+В флагах несколько бит зарезервировано под флаги, декларирующие как необходимо обрабатывать особые ситуации и 12 бит – под номер канала интерконнекта, в котором будет пересылаться сообщение (для локальных сообщений в имеющихся реализациях номер канала не имеет значения - хотя можно представить реализацию где для каналов будут независимые очереди). 
+ 
+### Тредпулы и мейлбоксы. 
+В рамках одной акторсистемы может сосуществовать несколько независимых тредпулов, каждый актор регистрируется на конкретном и в процессе жизни не может мигрировать (но может создавать новые акторы на произвольном тредпуле). Используется для крупноблочного разделения ресурсов, либо между разными активностями (вот здесь – обрабатываем один класс запросов, а вот здесь - другой), либо между разными профилями активности (вот здесь обрабатываем быстрые запросы, здесь – медленные, а вот там – вообще батчёвые). Например в YDB работает системный тредпул (в котором запускаются акторы, необходимые для функционирования YDB, и для которого мы следим что бы не было длительной блокировки в обработчиках), пользовательский тредпул (в котором обрабатываются запросы и потенциально обработчики могут уходить в себя подольше, но это не повлияет на инфраструктуру), батчёвый тредпул (куда отгружается длительная обработка – компакшены дисков, сканы таблиц и подобное) и, в жирных нодах – тредпул интерконнекта (как наиболее чувствительного к задержкам). 
+Пересылка сообщений между акторами разных тредпулов но одной локальной акторсистемы остаётся локальной, принудительной сериализации сообщения не происходит. 
+ 
 При регистрации актор прикрепляется к мейлбоксу (в типичном случае на собственном мейлбоксе, но по особой нужде можно находясь внутри обработки сообщения прикрепить порождённый актор к текущему активному мейлбоксу – см. RegisterWithSameMailbox (ранее RegisterLocal) – в этом случае будет гарантироваться отсутствие конкурентной обработки сообщений). Собственно Send – это и есть заворачивание ивента в хендл, помещение хендла в очередь мейлбокса и добавление мейлбокса в очередь активации тредпула. В рамках одного мейлбокса – обработка FIFO, между мейлбоксами таких гарантий нет, хотя и стараемся активировать мейлбоксы примерно в порядке появления в них сообщений.
-
-При регистрации актора можно выбрать тип мейлбокса, они немного отличаются стоимость добавления – либо дёшево, но похуже под контеншеном, либо почти wait-free, но подороже. См. комментарии к TMailboxType за актуальными подсказками что-как.
-
-Полезные хелперы.
-
-STFUNC – декларация стейт-функции, рекомендую всегда использовать именно такую форму для декларации, т.к. потом проще искать.
-
-hFunc – макрос выбора хендлера, передающий ивент в обработчик.
-
-cFunc – макрос выбора хендлера, не передающий ивент в обработчик.
-
-### Обработка сбоев.
-В рамках локальной акторсистемы доставка сообщений гарантирована, если по какой-то причине сообщение не доставлено (важно! Именно не доставлено, факт обработанности сообщения уже на совести принимающего актора) – то произойдёт одно из:
-
-Если выставлен флаг FlagForwardOnNondelivery – сообщение будет переправлено на актор, переданный как forwardOnNondelivery при конструировании хендла. Полезно например если какие-то сервисы создаются по требованию и для несозданных сервисов – желаем зароутить в роутер. Работает только в рамках локальной акторсистемы.
-
-Иначе при выставленном флаге FlagTrackDelivery – для отправителя будет сгенерирован ивент TEvUndelivered от имени недоступного актора. Получение такого сообщения гарантирует что исходный ивент не был обработан и никакие эффекты не произошли. Генерация и доставка нотификации в рамках локальной акторсистемы гарантирована, в распределённой – как повезёт, может и потеряться.
-
-Иначе, если никакие флаги не выставлены – сообщение будет выброшено.
-
-Т.к. в распределённой системе доставка нотификаций о недоставке не гарантируется, то для надёжной обработки сбоев необходим дополнительный механизм – по флагу FlagSubscribeOnSession при пересечении границ ноды происходит подписка отправителя на нотификацию о разрыве сетевой сессии, в рамках которой сообщение было отправлено. Теперь при разрыве сетевой сессии отправитель узнает что сообщение могло быть недоставлено (а могло и быть доставлено – мы не знаем) и сможет отреагировать. Нужно не забывать отписываться от нотификации о разрыве сессий – иначе будут копиться вплоть до ближайшего разрыва (который может и не скоро произойти).
-
-Резюмируя: при необходимости контролировать доставку внутри локальной акторсистемы – выставляем флаг FlagTrackDelivery и обрабатываем TEvUndelivered. Для распределённой – добавляем FlagSubscribeOnSession и дополнительно обрабатываем TEvNodeDisconnected не забывая отписываться от более не нужных подписок.
-
-### Интерконнект.
-Локальная акторсистема – это только половина пирога, возможность объединить их в распределённую – вторая половина. Реализация интерконнекта доступна из коробки и умеет
-Передавать сообщения через одно tcp-соединение
-Мультиплексировать разные потоки (ака каналы) в рамках одного соединения, гарантируя порядок в рамках канала
-Старается делать это хорошо.
-В рамках распределённой системы требуется каждой локальной акторсистеме назначить уникальный номер (например табличкой или реализовав динамическую раздачу номеров ноды) и запустить в рамках каждой локальной акторсистемы локальный неймсервис (например по табличке ремапинга номера ноды в сетевой адрес либо как кеш опорного неймсервиса).
-Смотрим на второй пример https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/examples/02_discovery
-Тут у нас конфигурируется распределённая акторсистема (в примере все пять запускаются в одном бинарнике, но точно так же – можно запускать и частями) на пять нод. На каждой ноде запускается реплика для паблишинга строчек и актор-эндпоинт (каждый со своим портом). Эндпоинты с помощью актора-паблишера публикуют свои явки/пароли на распределённый сторадж (с обработкой нештатных ситауций и поддержанием в актуальном состоянии). И рядом лежит реализация запроса к стораджу на листинг опубликованого по мажорити. Собственно это упрощённый и почищенный от специфики код, используемый в YDB для публикации и нахождения актуальных эндпоинтов пользовательской базы.
+ 
+При регистрации актора можно выбрать тип мейлбокса, они немного отличаются стоимость добавления – либо дёшево, но похуже под контеншеном, либо почти wait-free, но подороже. См. комментарии к TMailboxType за актуальными подсказками что-как. 
+ 
+Полезные хелперы. 
+ 
+STFUNC – декларация стейт-функции, рекомендую всегда использовать именно такую форму для декларации, т.к. потом проще искать. 
+ 
+hFunc – макрос выбора хендлера, передающий ивент в обработчик. 
+ 
+cFunc – макрос выбора хендлера, не передающий ивент в обработчик. 
+ 
+### Обработка сбоев. 
+В рамках локальной акторсистемы доставка сообщений гарантирована, если по какой-то причине сообщение не доставлено (важно! Именно не доставлено, факт обработанности сообщения уже на совести принимающего актора) – то произойдёт одно из: 
+ 
+Если выставлен флаг FlagForwardOnNondelivery – сообщение будет переправлено на актор, переданный как forwardOnNondelivery при конструировании хендла. Полезно например если какие-то сервисы создаются по требованию и для несозданных сервисов – желаем зароутить в роутер. Работает только в рамках локальной акторсистемы. 
+ 
+Иначе при выставленном флаге FlagTrackDelivery – для отправителя будет сгенерирован ивент TEvUndelivered от имени недоступного актора. Получение такого сообщения гарантирует что исходный ивент не был обработан и никакие эффекты не произошли. Генерация и доставка нотификации в рамках локальной акторсистемы гарантирована, в распределённой – как повезёт, может и потеряться. 
+ 
+Иначе, если никакие флаги не выставлены – сообщение будет выброшено. 
+ 
+Т.к. в распределённой системе доставка нотификаций о недоставке не гарантируется, то для надёжной обработки сбоев необходим дополнительный механизм – по флагу FlagSubscribeOnSession при пересечении границ ноды происходит подписка отправителя на нотификацию о разрыве сетевой сессии, в рамках которой сообщение было отправлено. Теперь при разрыве сетевой сессии отправитель узнает что сообщение могло быть недоставлено (а могло и быть доставлено – мы не знаем) и сможет отреагировать. Нужно не забывать отписываться от нотификации о разрыве сессий – иначе будут копиться вплоть до ближайшего разрыва (который может и не скоро произойти). 
+ 
+Резюмируя: при необходимости контролировать доставку внутри локальной акторсистемы – выставляем флаг FlagTrackDelivery и обрабатываем TEvUndelivered. Для распределённой – добавляем FlagSubscribeOnSession и дополнительно обрабатываем TEvNodeDisconnected не забывая отписываться от более не нужных подписок. 
+ 
+### Интерконнект. 
+Локальная акторсистема – это только половина пирога, возможность объединить их в распределённую – вторая половина. Реализация интерконнекта доступна из коробки и умеет 
+Передавать сообщения через одно tcp-соединение 
+Мультиплексировать разные потоки (ака каналы) в рамках одного соединения, гарантируя порядок в рамках канала 
+Старается делать это хорошо. 
+В рамках распределённой системы требуется каждой локальной акторсистеме назначить уникальный номер (например табличкой или реализовав динамическую раздачу номеров ноды) и запустить в рамках каждой локальной акторсистемы локальный неймсервис (например по табличке ремапинга номера ноды в сетевой адрес либо как кеш опорного неймсервиса). 
+Смотрим на второй пример https://a.yandex-team.ru/arc/trunk/arcadia/library/actors/examples/02_discovery 
+Тут у нас конфигурируется распределённая акторсистема (в примере все пять запускаются в одном бинарнике, но точно так же – можно запускать и частями) на пять нод. На каждой ноде запускается реплика для паблишинга строчек и актор-эндпоинт (каждый со своим портом). Эндпоинты с помощью актора-паблишера публикуют свои явки/пароли на распределённый сторадж (с обработкой нештатных ситауций и поддержанием в актуальном состоянии). И рядом лежит реализация запроса к стораджу на листинг опубликованого по мажорити. Собственно это упрощённый и почищенный от специфики код, используемый в YDB для публикации и нахождения актуальных эндпоинтов пользовательской базы. 

library/cpp/actors/core/actorsystem.cpp

@@ -1,6 +1,6 @@
 #include "defs.h"
 #include "actorsystem.h"
-#include "callstack.h"
+#include "callstack.h" 
 #include "cpu_manager.h"
 #include "mailbox.h"
 #include "events.h"
@@ -64,10 +64,10 @@ namespace NActors {
         if (Y_UNLIKELY(!ev))
             return false;
 
-#ifdef USE_ACTOR_CALLSTACK
+#ifdef USE_ACTOR_CALLSTACK 
         ev->Callstack.TraceIfEmpty();
-#endif
-
+#endif 
+ 
         TActorId recipient = ev->GetRecipientRewrite();
         const ui32 recpNodeId = recipient.NodeId();
 

library/cpp/actors/core/callstack.cpp

@@ -1,14 +1,14 @@
-#include "callstack.h"
-#include <util/thread/singleton.h>
-
-#ifdef USE_ACTOR_CALLSTACK
-
-namespace NActors {
+#include "callstack.h" 
+#include <util/thread/singleton.h> 
+ 
+#ifdef USE_ACTOR_CALLSTACK 
+ 
+namespace NActors { 
     namespace {
         void (*PreviousFormatBackTrace)(IOutputStream*) = 0;
         ui32 ActorBackTraceEnableCounter = 0;
     }
-
+ 
     void ActorFormatBackTrace(IOutputStream* out) {
         TStringStream str;
         PreviousFormatBackTrace(&str);
@@ -16,19 +16,19 @@ namespace NActors {
         TCallstack::DumpCallstack(str);
         *out << str.Str();
     }
-
+ 
     void EnableActorCallstack() {
         if (ActorBackTraceEnableCounter == 0) {
             Y_VERIFY(PreviousFormatBackTrace == 0);
             PreviousFormatBackTrace = SetFormatBackTraceFn(ActorFormatBackTrace);
         }
-
+ 
         ++ActorBackTraceEnableCounter;
     }
 
     void DisableActorCallstack() {
         --ActorBackTraceEnableCounter;
-
+ 
         if (ActorBackTraceEnableCounter == 0) {
             Y_VERIFY(PreviousFormatBackTrace);
             SetFormatBackTraceFn(PreviousFormatBackTrace);
@@ -42,12 +42,12 @@ namespace NActors {
         , LinesToSkip(0)
     {
     }
-
+ 
     void TCallstack::SetLinesToSkip() {
         TTrace record;
         LinesToSkip = BackTrace(record.Data, TTrace::CAPACITY);
     }
-
+ 
     void TCallstack::Trace() {
         size_t currentIdx = (BeginIdx + Size) % RECORDS;
         if (Size == RECORDS) {
@@ -59,18 +59,18 @@ namespace NActors {
         record.Size = BackTrace(record.Data, TTrace::CAPACITY);
         record.LinesToSkip = LinesToSkip;
     }
-
+ 
     void TCallstack::TraceIfEmpty() {
         if (Size == 0) {
             LinesToSkip = 0;
             Trace();
         }
-    }
-
+    } 
+ 
     TCallstack& TCallstack::GetTlsCallstack() {
         return *FastTlsSingleton<TCallstack>();
-    }
-
+    } 
+ 
     void TCallstack::DumpCallstack(TStringStream& str) {
         TCallstack& callstack = GetTlsCallstack();
         for (int i = callstack.Size - 1; i >= 0; --i) {
@@ -86,8 +86,8 @@ namespace NActors {
                 FormatBackTrace(&str, record.Data, size);
             }
             str << Endl;
-        }
-    }
+        } 
+    } 
 }
-
-#endif
+ 
+#endif 

library/cpp/actors/core/callstack.h

@@ -1,24 +1,24 @@
-#pragma once
-
-#ifndef NDEBUG
-//#define ENABLE_ACTOR_CALLSTACK
-#endif
-
-#ifdef ENABLE_ACTOR_CALLSTACK
+#pragma once 
+ 
+#ifndef NDEBUG 
+//#define ENABLE_ACTOR_CALLSTACK 
+#endif 
+ 
+#ifdef ENABLE_ACTOR_CALLSTACK 
 #include "defs.h"
 #include <util/system/backtrace.h>
 #include <util/stream/str.h>
 #include <util/generic/deque.h>
 #define USE_ACTOR_CALLSTACK
-
-namespace NActors {
+ 
+namespace NActors { 
     struct TCallstack {
         struct TTrace {
             static const size_t CAPACITY = 50;
             void* Data[CAPACITY];
             size_t Size;
             size_t LinesToSkip;
-
+ 
             TTrace()
                 : Size(0)
                 , LinesToSkip(0)
@@ -30,29 +30,29 @@ namespace NActors {
         static const size_t RECORDS_TO_SKIP = 2;
         TTrace Record[RECORDS];
         size_t BeginIdx;
-        size_t Size;
-        size_t LinesToSkip;
-
+        size_t Size; 
+        size_t LinesToSkip; 
+ 
         TCallstack();
         void SetLinesToSkip();
         void Trace();
         void TraceIfEmpty();
         static TCallstack& GetTlsCallstack();
         static void DumpCallstack(TStringStream& str);
-    };
-
+    }; 
+ 
     void EnableActorCallstack();
     void DisableActorCallstack();
-
+ 
 }
-
-#else
-
-namespace NActors {
+ 
+#else 
+ 
+namespace NActors { 
     inline void EnableActorCallstack(){};
-
+ 
     inline void DisableActorCallstack(){};
-
+ 
 }
-
-#endif
+ 
+#endif 

library/cpp/actors/core/event.h

@@ -2,7 +2,7 @@
 
 #include "defs.h"
 #include "actorid.h"
-#include "callstack.h"
+#include "callstack.h" 
 #include "event_load.h"
 
 #include <library/cpp/actors/wilson/wilson_trace.h>
@@ -117,9 +117,9 @@ namespace NActors {
         static const size_t ChannelBits = 12;
         static const size_t ChannelShift = (sizeof(ui32) << 3) - ChannelBits;
 
-#ifdef USE_ACTOR_CALLSTACK
+#ifdef USE_ACTOR_CALLSTACK 
         TCallstack Callstack;
-#endif
+#endif 
         ui16 GetChannel() const noexcept {
             return Flags >> ChannelShift;
         }
@@ -133,7 +133,7 @@ namespace NActors {
             Y_VERIFY(flags < (1 << ChannelShift));
             return (flags | (channel << ChannelShift));
         }
-
+ 
     private:
         THolder<IEventBase> Event;
         TIntrusivePtr<TEventSerializedData> Buffer;
@@ -165,7 +165,7 @@ namespace NActors {
         TActorId GetForwardOnNondeliveryRecipient() const {
             return OnNondeliveryHolder.Get() ? OnNondeliveryHolder->Recipient : TActorId();
         }
-
+ 
         IEventHandle(const TActorId& recipient, const TActorId& sender, IEventBase* ev, ui32 flags = 0, ui64 cookie = 0,
                      const TActorId* forwardOnNondelivery = nullptr, NWilson::TTraceId traceId = {})
             : Type(ev->Type())

library/cpp/actors/core/executor_thread.cpp

@@ -1,6 +1,6 @@
 #include "executor_thread.h"
 #include "actorsystem.h"
-#include "callstack.h"
+#include "callstack.h" 
 #include "mailbox.h"
 #include "event.h"
 #include "events.h"
@@ -148,10 +148,10 @@ namespace NActors {
                     TActorContext ctx(*mailbox, *this, hpprev, recipient);
                     TlsActivationContext = &ctx;
 
-#ifdef USE_ACTOR_CALLSTACK
+#ifdef USE_ACTOR_CALLSTACK 
                     TCallstack::GetTlsCallstack() = ev->Callstack;
                     TCallstack::GetTlsCallstack().SetLinesToSkip();
-#endif
+#endif 
                     CurrentRecipient = recipient;
                     CurrentActorScheduledEventsCounter = 0;
 

library/cpp/actors/core/executor_thread.h

@@ -4,7 +4,7 @@
 #include "event.h"
 #include "actor.h"
 #include "actorsystem.h"
-#include "callstack.h"
+#include "callstack.h" 
 #include "probes.h"
 #include "worker_context.h"
 

library/cpp/actors/core/log.cpp

@@ -1,5 +1,5 @@
 #include "log.h"
-#include "log_settings.h"
+#include "log_settings.h" 
 
 #include <library/cpp/monlib/service/pages/templates.h>
 
@@ -211,7 +211,7 @@ namespace NActors {
 
     void TLoggerActor::Log(TInstant time, NLog::EPriority priority, NLog::EComponent component, const char* c, ...) {
         Metrics->IncDirectMsgs();
-        if (Settings && Settings->Satisfies(priority, component, 0ull)) {
+        if (Settings && Settings->Satisfies(priority, component, 0ull)) { 
             va_list params;
             va_start(params, c);
             TString formatted;
@@ -233,14 +233,14 @@ namespace NActors {
         if (!OutputRecord(now, NActors::NLog::EPrio::Error, Settings->LoggerComponent, message)) {
             BecomeDefunct();
         }
-    }
-
+    } 
+ 
     void TLoggerActor::HandleIgnoredEvent(TLogIgnored::TPtr& ev, const NActors::TActorContext& ctx) {
         Y_UNUSED(ev);
         LogIgnoredCount(ctx.Now());
-        IgnoredCount = 0;
-        PassedCount = 0;
-    }
+        IgnoredCount = 0; 
+        PassedCount = 0; 
+    } 
 
     void TLoggerActor::HandleIgnoredEventDrop() {
         // logger backend is unavailable, just ignore
@@ -248,7 +248,7 @@ namespace NActors {
 
     void TLoggerActor::WriteMessageStat(const NLog::TEvLog& ev) {
         Metrics->IncActorMsgs();
-
+ 
         const auto prio = ev.Level.ToPrio();
 
         switch (prio) {
@@ -281,7 +281,7 @@ namespace NActors {
                 ++IgnoredCount;
                 PassedCount = 0;
                 return;
-            }
+            } 
             PassedCount++;
         } else {
             // Enable of disable throttling depending on the load
@@ -289,8 +289,8 @@ namespace NActors {
                 AtomicSet(IsOverflow, 1);
             else if (delayMillisec <= (i64)Settings->TimeThresholdMs && AtomicGet(IsOverflow))
                 AtomicSet(IsOverflow, 0);
-        }
-
+        } 
+ 
         const auto prio = ev->Get()->Level.ToPrio();
         if (!OutputRecord(ev->Get()->Stamp, prio, ev->Get()->Component, ev->Get()->Line)) {
             BecomeDefunct();
@@ -337,7 +337,7 @@ namespace NActors {
                         auto name = Settings->ComponentName(i);
                         if (!*name)
                             continue;
-                        NLog::TComponentSettings componentSettings = Settings->GetComponentSettings(i);
+                        NLog::TComponentSettings componentSettings = Settings->GetComponentSettings(i); 
 
                         TABLER() {
                             TABLED() {
@@ -410,8 +410,8 @@ namespace NActors {
 
         TStringStream str;
         if (hasComponent && !hasPriority && !hasSamplingPriority && !hasSamplingRate) {
-            NLog::TComponentSettings componentSettings = Settings->GetComponentSettings(component);
-            ui32 samplingRate = componentSettings.Raw.X.SamplingRate;
+            NLog::TComponentSettings componentSettings = Settings->GetComponentSettings(component); 
+            ui32 samplingRate = componentSettings.Raw.X.SamplingRate; 
             HTML(str) {
                 DIV_CLASS("row") {
                     DIV_CLASS("col-md-12") {
@@ -682,17 +682,17 @@ namespace NActors {
                 OutputDebugString(x.c_str());
             }
 #endif
-            bool isOk = false;
-            do {
-                try {
+            bool isOk = false; 
+            do { 
+                try { 
                     TRecordWithNewline r(rec);
                     Cerr.Write(r.Buf.Data(), r.Buf.Filled());
-                    isOk = true;
-                } catch (TSystemError err) {
-                    // Interrupted system call
+                    isOk = true; 
+                } catch (TSystemError err) { 
+                    // Interrupted system call 
                     Y_UNUSED(err);
-                }
-            } while (!isOk);
+                } 
+            } while (!isOk); 
         }
 
         void ReopenLog() override {
@@ -743,10 +743,10 @@ namespace NActors {
         return new TLineFileLogBackend(fileName);
     }
 
-    TAutoPtr<TLogBackend> CreateNullBackend() {
-        return new TNullLogBackend();
-    }
-
+    TAutoPtr<TLogBackend> CreateNullBackend() { 
+        return new TNullLogBackend(); 
+    } 
+ 
     TAutoPtr<TLogBackend> CreateCompositeLogBackend(TVector<TAutoPtr<TLogBackend>>&& underlyingBackends) {
         return new TCompositeLogBackend(std::move(underlyingBackends));
     }

library/cpp/actors/core/log.h

@@ -12,7 +12,7 @@
 
 #include <util/generic/vector.h>
 #include <util/string/printf.h>
-#include <util/string/builder.h>
+#include <util/string/builder.h> 
 #include <library/cpp/logger/all.h>
 #include <library/cpp/monlib/dynamic_counters/counters.h>
 #include <library/cpp/monlib/metrics/metric_registry.h>
@@ -24,14 +24,14 @@
 // TODO: limit number of messages per second
 // TODO: make TLogComponentLevelRequest/Response network messages
 
-#define IS_LOG_PRIORITY_ENABLED(actorCtxOrSystem, priority, component)                           \
-    (static_cast<::NActors::NLog::TSettings*>((actorCtxOrSystem).LoggerSettings()) &&            \
-     static_cast<::NActors::NLog::TSettings*>((actorCtxOrSystem).LoggerSettings())->Satisfies(   \
-             static_cast<::NActors::NLog::EPriority>(priority),                                  \
-             static_cast<::NActors::NLog::EComponent>(component),                                \
-             0ull)                                                                               \
-    )
-
+#define IS_LOG_PRIORITY_ENABLED(actorCtxOrSystem, priority, component)                           \ 
+    (static_cast<::NActors::NLog::TSettings*>((actorCtxOrSystem).LoggerSettings()) &&            \ 
+     static_cast<::NActors::NLog::TSettings*>((actorCtxOrSystem).LoggerSettings())->Satisfies(   \ 
+             static_cast<::NActors::NLog::EPriority>(priority),                                  \ 
+             static_cast<::NActors::NLog::EComponent>(component),                                \ 
+             0ull)                                                                               \ 
+    ) 
+ 
 #define LOG_LOG_SAMPLED_BY(actorCtxOrSystem, priority, component, sampleBy, ...)                                               \
     do {                                                                                                                       \
         ::NActors::NLog::TSettings* mSettings = static_cast<::NActors::NLog::TSettings*>((actorCtxOrSystem).LoggerSettings()); \
@@ -49,9 +49,9 @@
         logStringBuilder << stream;                                                    \
         return static_cast<TString>(logStringBuilder);                                 \
     }().data())
-
-#define LOG_LOG(actorCtxOrSystem, priority, component, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, priority, component, 0ull, __VA_ARGS__)
-#define LOG_LOG_S(actorCtxOrSystem, priority, component, stream) LOG_LOG_S_SAMPLED_BY(actorCtxOrSystem, priority, component, 0ull, stream)
+ 
+#define LOG_LOG(actorCtxOrSystem, priority, component, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, priority, component, 0ull, __VA_ARGS__) 
+#define LOG_LOG_S(actorCtxOrSystem, priority, component, stream) LOG_LOG_S_SAMPLED_BY(actorCtxOrSystem, priority, component, 0ull, stream) 
 
 // use these macros for logging via actor system or actor context
 #define LOG_EMERG(actorCtxOrSystem, component, ...) LOG_LOG(actorCtxOrSystem, NActors::NLog::PRI_EMERG, component, __VA_ARGS__)
@@ -69,11 +69,11 @@
 #define LOG_CRIT_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_CRIT, component, stream)
 #define LOG_ERROR_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_ERROR, component, stream)
 #define LOG_WARN_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_WARN, component, stream)
-#define LOG_NOTICE_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_NOTICE, component, stream)
+#define LOG_NOTICE_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_NOTICE, component, stream) 
 #define LOG_INFO_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_INFO, component, stream)
 #define LOG_DEBUG_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_DEBUG, component, stream)
 #define LOG_TRACE_S(actorCtxOrSystem, component, stream) LOG_LOG_S(actorCtxOrSystem, NActors::NLog::PRI_TRACE, component, stream)
-
+ 
 #define LOG_EMERG_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_EMERG, component, sampleBy, __VA_ARGS__)
 #define LOG_ALERT_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_ALERT, component, sampleBy, __VA_ARGS__)
 #define LOG_CRIT_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_CRIT, component, sampleBy, __VA_ARGS__)
@@ -83,7 +83,7 @@
 #define LOG_INFO_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_INFO, component, sampleBy, __VA_ARGS__)
 #define LOG_DEBUG_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_DEBUG, component, sampleBy, __VA_ARGS__)
 #define LOG_TRACE_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, ...) LOG_LOG_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_TRACE, component, sampleBy, __VA_ARGS__)
-
+ 
 #define LOG_EMERG_S_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, stream) LOG_LOG_S_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_EMERG, component, sampleBy, stream)
 #define LOG_ALERT_S_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, stream) LOG_LOG_S_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_ALERT, component, sampleBy, stream)
 #define LOG_CRIT_S_SAMPLED_BY(actorCtxOrSystem, component, sampleBy, stream) LOG_LOG_S_SAMPLED_BY(actorCtxOrSystem, NActors::NLog::PRI_CRIT, component, sampleBy, stream)
@@ -167,11 +167,11 @@ namespace NActors {
     };
 
     class TLogIgnored: public TEventLocal<TLogIgnored, int(NLog::EEv::Ignored)> {
-    public:
-        TLogIgnored() {
-        }
-    };
-
+    public: 
+        TLogIgnored() { 
+        } 
+    }; 
+ 
     ////////////////////////////////////////////////////////////////////////////////
     // LOGGER ACTOR
     ////////////////////////////////////////////////////////////////////////////////
@@ -212,7 +212,7 @@ namespace NActors {
 
         void StateFunc(TAutoPtr<IEventHandle>& ev, const TActorContext& ctx) {
             switch (ev->GetTypeRewrite()) {
-                HFunc(TLogIgnored, HandleIgnoredEvent);
+                HFunc(TLogIgnored, HandleIgnoredEvent); 
                 HFunc(NLog::TEvLog, HandleLogEvent);
                 HFunc(TLogComponentLevelRequest, HandleLogComponentLevelRequest);
                 HFunc(NMon::TEvHttpInfo, HandleMonInfo);
@@ -298,7 +298,7 @@ namespace NActors {
                                               bool logPError, bool logCons);
     TAutoPtr<TLogBackend> CreateStderrBackend();
     TAutoPtr<TLogBackend> CreateFileBackend(const TString& fileName);
-    TAutoPtr<TLogBackend> CreateNullBackend();
+    TAutoPtr<TLogBackend> CreateNullBackend(); 
     TAutoPtr<TLogBackend> CreateCompositeLogBackend(TVector<TAutoPtr<TLogBackend>>&& underlyingBackends);
 
     /////////////////////////////////////////////////////////////////////