Каналы mpsc в Rust
Каналы - второй способ организовать многопоточность: вместо общего состояния передавать владение. Спрашивают, что означает mpsc, когда завершается цикл приёма и чем ограниченный канал отличается от безграничного.
Стержень: канал переносит владение сообщением, а закрытие определяется тем, живы ли ещё отправители.
// Формулировки: «что такое mpsc?», «когда recv вернёт Err?», «почему цикл приёма не завершается?»
Кто отправляет и кто принимает
std::sync::mpsc - multi-producer, single-consumer: Sender клонируется и раздаётся воркерам, Receiver существует в одном экземпляре. Такая форма закрывает типовой сбор результатов. Когда получателей нужно несколько, берут внешние крейты вроде crossbeam или flume.
Канал знает, сколько живых отправителей осталось. Когда умирает последний и очередь пуста, recv() возвращает Err, а цикл for msg in rx просто завершается. Отправка в канал с мёртвым получателем тоже даёт Err - обе стороны узнают о судьбе друг друга.
// Отсюда самая частая ошибка: оригинальный tx клонировали воркерам, но саму переменную не уронили. Воркеры закончились, их копии умерли, а последняя живёт в главном потоке - канал открыт, цикл ждёт вечно. Лечится drop(tx) перед циклом.
let (tx, rx) = mpsc::channel();
for i in 0..4 { let tx = tx.clone(); /* spawn */ }
drop(tx); // иначе цикл ниже не кончится
for msg in rx { /* .. */ }- mpsc
- много отправителей, один получатель
- закрытие канала
- смерть всех отправителей; recv возвращает Err
Буфер и обратное давление
channel() безграничен: отправитель никогда не ждёт. Звучит удобно ровно до момента, когда получатель не успевает: очередь растёт, задержка растёт, память кончается. Проблема проявляется под нагрузкой, а не в тестах.
sync_channel(n) вводит ограничение: при заполненной очереди send блокируется, и производитель естественным образом замедляется до темпа потребителя. Это и есть обратное давление. Частный случай sync_channel(0) - рандеву: отправитель ждёт, пока сообщение заберут.
// Выбор между каналом и общим Mutex простой: канал хорош, когда работа передаётся по конвейеру и владение уезжает вместе с задачей; общий замок - когда состояние по сути общее и долгоживущее, вроде кэша или счётчиков.
- обратное давление
- замедление производителя при полной очереди
- рандеву
- канал с нулевым буфером: передача из рук в руки
Как отвечать: «Почему цикл for msg in rx может не завершиться?»
Потому что канал закрывается только тогда, когда умирает последний Sender. Типовая ситуация: оригинальный tx склонировали воркерам, воркеры завершились, их копии уничтожились, а сам tx остался живым в главном потоке. Для канала это значит, что отправитель ещё есть, и получатель честно ждёт. Лечится явным drop(tx) перед циклом или тем, что все копии передаются внутрь потоков и в вызывающем коде не остаётся ни одной. Ровно так же работает и в асинхронных каналах - там ошибка встречается ещё чаще.
Это конкретный баг из практики, а не пересказ документации: ты называешь причину, симптом и два способа починки.
На чём валятся
- −Держат лишнюю копию Sender и получают зависший цикл приёма.
- −Берут безграничный канал под нагрузкой и вместо замедления получают рост памяти.
- −Считают, что recv паникует при закрытии канала - он возвращает Err.
- −Пробуют завести несколько получателей на std-канале, не зная про crossbeam и flume.
- −Гоняют по каналу большое общее состояние вместо доступа под замком.
Проверьте себя
Пять вопросов из банка по этой подтеме. Всего их 12, остальные разбираются в тренажёре.
- Чем sync_channel(10) отличается от channel()?A)Он синхронный: отправка блокируется до приёма получателемB)Он передаёт сообщения по ссылке без перемещения владенияC)Ограниченный буфер: отправитель ждёт на полной очередиD)Он сохраняет порядок сообщений, а обычный канал — нет
показать ответ и разбор
+C)Ограниченный буфер: отправитель ждёт на полной очереди// разбор: channel безграничен: отправитель никогда не ждёт, и при медленном получателе очередь растёт, пока не съест память. sync_channel(n) вводит обратное давление — send блокируется на полном буфере. Частный случай sync_channel(0) — рандеву: отправитель ждёт, пока сообщение заберут.
- Когда канал предпочтительнее Arc<Mutex<T>>?A)Когда потоков больше двух: мьютекс работает только для парыB)Когда работа идёт по конвейеру: владение уезжает с задачейC)Когда нужна максимальная скорость: канал быстрее блокировкиD)Когда данных много: канал не копирует их при передаче
показать ответ и разбор
+B)Когда работа идёт по конвейеру: владение уезжает с задачей// разбор: Канал переносит владение: отправил задачу — она больше не твоя, и разделяемого состояния попросту нет. Это убирает целый класс ошибок с порядком блокировок. Общий замок уместнее, когда состояние по сути общее и живёт долго — кэш, счётчики, конфиг: гонять его копии по каналу бессмысленно.
- Воркеры пишут результаты в канал, главный поток читает в цикле for. Цикл не заканчивается. Почему?A)В главном потоке остался живой Sender — канал считается открытымB)Получатель обязан вызывать recv_timeout, иначе не увидит закрытияC)Цикл for по получателю не завершается по определению, нужен breakD)Воркеры не вызвали flush после отправки последнего сообщения
показать ответ и разбор
+A)В главном потоке остался живой Sender — канал считается открытым// разбор: Классическая ловушка: оригинальный tx клонировали воркерам, но саму переменную не уронили. Воркеры завершились, их копии умерли, а последний экземпляр живёт в главном потоке — канал открыт, цикл ждёт. Лечится явным drop(tx) до входа в цикл или передачей всех копий внутрь потоков.
- Что делает recv() у получателя, если в канале пусто, а отправители живы?A)Возвращает Err и предлагает вызывающему повторить попытку позже самостоятельноB)Крутит цикл проверки, занимая процессорное время до первого сообщенияC)Возвращает Ok со значением по умолчанию для типа сообщенияD)Блокирует поток до появления сообщения
показать ответ и разбор
+D)Блокирует поток до появления сообщения// разбор: recv усыпляет поток до появления сообщения или до закрытия канала — активного ожидания нет. Если ждать нельзя, берут try_recv (Err(Empty) сразу) или recv_timeout (Err(Timeout) по истечении срока). Err от recv означает только одно: живых отправителей не осталось, и ждать больше нечего.
- Что произойдёт при send, если получателя уже уничтожили?A)Сообщение останется в буфере до появления нового получателя каналаB)Отправка запаникует: писать в закрытый канал считается ошибкой программыC)send вернёт Err с исходным сообщением внутриD)Сообщение будет молча отброшено, а send вернёт Ok
показать ответ и разбор
+C)send вернёт Err с исходным сообщением внутри// разбор: Канал закрывается с обеих сторон: пропал получатель — send отдаёт Err(SendError(msg)), причём значение возвращается внутри ошибки, чтобы его можно было переиспользовать или залогировать. Это штатный сигнал остановки для воркера: получатель ушёл, работать больше не на кого. Паники нет, реакцию выбирает вызывающий.
дальше
Теорию прочитали. Навык ставится повторением
В Сеньорчике эта подтема идёт в ежедневных сессиях: движок возвращает её, пока ответы не станут уверенными, и ведёт прогресс отдельно по каждой подтеме. Теория внутри тоже бесплатна, лимит только на количество вопросов в день.