Асинхронность

Какие задачи держит работающий сервис, как через них проходит одна доставка и в каком порядке их останавливает завершение работы.

Схема текстом

1. Рантайм приложения: Tokio, многопоточный: по умолчанию одна задача на всё; workers(n) задача цикла раздаёт; threads(n) задача подачи кладёт в кольца; n воркеров, ящик на одну; n колец по восемь ячеек; Отслеживаемые задачи: and_after, хуки, таймеры retry_after
2. Подписки подключённых брокеров: потоки доставок → по умолчанию одна задача на всё: poll_next
3. Подписки подключённых брокеров: потоки доставок → workers(n) задача цикла раздаёт
4. Подписки подключённых брокеров: потоки доставок → threads(n) задача подачи кладёт в кольца
5. workers(n) задача цикла раздаёт → n воркеров, ящик на одну: send
6. threads(n) задача подачи кладёт в кольца → n колец по восемь ячеек: push
7. n колец по восемь ячеек → n потоков ОС у каждого свой рантайм Tokio: pop
8. по умолчанию одна задача на всё → Отслеживаемые задачи: and_after, хуки, таймеры retry_after: spawn
9. n воркеров, ящик на одну → Отслеживаемые задачи: and_after, хуки, таймеры retry_after

Сервис работает на Tokio. #[ruststream::app] строит многопоточный рантайм. flavor = "current_thread" выбирает однопоточный, а worker_threads = n задаёт число потоков.

Подписчик - это Stream доставок, и обратное давление возникает из самого опроса. Цикл по умолчанию запрашивает следующую доставку, только когда предыдущая подтверждена. workers(n) держит в работе не больше n доставок. workers(n, by_key) отдаёт все доставки одного ключа одному воркеру, поэтому порядок внутри ключа сохраняется. threads(n) читает вперёд не больше n × 9 доставок: по одной в руках каждого потока и по восемь в его кольце.

Выделенный поток проводит доставку целиком, от декодирования до подтверждения, на собственном однопоточном рантайме. Работа, которую обработчик передаёт в MainRuntime через Ctx<MainRuntime>, выполняется в рантайме приложения.

Одна доставка

Схема текстом

1. Задача цикла: опрос потока; опрос следующей
2. Задача обработки: сборка контекста; слои приёма; декодирование; тело обработчика; публикация ответа; подтверждение
3. Позже, под учётом: продолжения и таймеры
4. опрос потока → сборка контекста
5. сборка контекста → слои приёма
6. слои приёма → декодирование
7. декодирование → тело обработчика
8. тело обработчика → публикация ответа
9. публикация ответа → подтверждение
10. подтверждение → опрос следующей: новый круг
11. подтверждение → продолжения и таймеры: запуск

По умолчанию задача обработки - это сама задача цикла, при workers(n) - задача-воркер, при threads(n) - выделенный поток. Продолжение and_after, хук после подтверждения и таймер retry_after запускаются после подтверждения как задачи, которые приложение отслеживает. Они выполняются не больше одного раза: доставка к этому моменту уже подтверждена.

Завершение работы

Схема текстом

1. Сигнал, fail-fast или RunningApp::shutdown → Хуки on_shutdown, пока брокеры подключены
2. Хуки on_shutdown, пока брокеры подключены → Циклы перестают брать новые доставки
3. Циклы перестают брать новые доставки → Доставки на руках обрабатываются и подтверждаются
4. Доставки на руках обрабатываются и подтверждаются → Отслеживаемые продолжения и таймеры повторов завершаются
5. Отслеживаемые продолжения и таймеры повторов завершаются → Брокеры закрываются в обратном порядке регистрации
6. Брокеры закрываются в обратном порядке регистрации → Хуки after_shutdown при закрытых брокерах

Тайм-аут завершения ограничивает ожидание. Когда он истекает, ещё работающие циклы прерываются, а незавершённые отслеживаемые задачи оставляются доделывать своё самостоятельно. Без тайм-аута завершение ждёт их все. Задача, запущенная обычным tokio::spawn на выделенном потоке, не отслеживается и заканчивается вместе с потоком.