Когда асинхронная операция может завершиться в Rust?
Контекст
Я читаю этот блог. предоставлен некоторый код:
async fn parse_line(socket: &TcpStream) -> Result<String, Error> {
let len = socket.read_u32().await?;
let mut line = vec![0; len];
socket.read_exact(&mut line).await?;
let line = str::from_utf8(line)?;
Ok(line)
}
loop {
select! {
line_in = parse_line(&socket) => {
if let Some(line_in) = line_in {
broadcast_line(line_in);
} else {
// connection closed, exit loop
break;
}
}
line_out = channel.recv() => {
write_line(&socket, line_out).await;
}
}
}
И автор утверждает, что parse_line может оказаться в поврежденном состоянии, если channel получит сообщение во время выполнения parse_line.
Вопрос
В каких моментах parse_line можно прервать? Это в какой-то момент? Насколько я понимаю (что может быть ошибочным), Rust может переключать задачи в потоке с помощью операторов ожидания, но в эти моменты состояние сохраняется, чтобы работу можно было возобновить.
Что я представляю
Я представляю, что в parse_line Rust загружает байты в переменную line. После чтения некоторого количества байтов (а для некоторого символа ASCII может быть только половина байтов) и ожидания поступления большего количества channel получает что-то, и контекст переключается.
После выполнения задачи channel.recv() Rust возвращается к чтению входных данных, однако пользователь, предоставивший байты, отменил запрос, и теперь читать больше нечего.
Теперь str::from_utf8(line)? выдает ошибку UTF-8, поскольку line имеет неполный символ ASCII.
Ответы 1
TL;DR: Ни в какой момент, только в .awaitс.
асинхронный код опускается в конечные автоматы, реализующие Будущее . .await вызывает Future::poll() в цикле, приостанавливая вызывающую программу, когда она возвращается Poll::Pending, и завершая Poll::Ready. По сути, это доводит внутреннее будущее до завершения.
Однако что произойдет, если мы позвоним poll() несколько раз, а затем остановимся, не завершив будущее? В этом случае те несколько раз, которые мы вызывали poll(), продвинули будущее до некоторой точки его исполнения, и на этом мы останавливаемся. Мы отменили будущее.
Ключевое наблюдение: эта точка исполнения должна находиться внутри .await. Это потому, что синхронный код без-.await в будущем преобразуется в прямой синхронный код в реализации poll(), и мы не можем на этом останавливаться. Мы можем остановиться только после того, как вернулись из poll(), и это происходит либо в середине .await, либо по завершении.
Однако не все фьючерсы безопасны при отмене. Некоторые, как ваш parse_line(), теряют работу, если ее отменяют. Если мы отменим второй .await, длина уже будет прочитана (и отброшена) из сокета, но тело еще не прочитано, поэтому мы потеряем эту длину. Мы не можем его восстановить, и при следующем вызове функции она увидит испорченные данные из сокета (или просто пропустит одну запись).
select! отменяет все фьючерсы, кроме первого завершенного, поэтому в этом коде есть ошибка.
Решение состоит в том, чтобы никогда не терять будущее, а сохранять его на потом:
let mut parse_line_fut = std::pin::pin!(parse_line(&socket));
loop {
select! {
line_in = parse_line_fut.as_mut() => {
if let Ok(line_in) = line_in {
broadcast_line(line_in);
parse_line_fut.set(parse_line(&socket));
} else {
// connection closed, exit loop
break;
}
}
line_out = channel.recv() => {
write_line(&socket, line_out).await;
}
}
}
@financial_ Physician Значение было прочитано из потока и больше не существует. Переменная удаляется. Значение нигде не сохраняется.
Как узнать, сохранится ли переменная после переключения контекста или она будет удалена?
@financial_ Physician Переменные в будущем не удаляются при переключении контекста. Они отбрасываются, когда отбрасывается само будущее.
Уточняющий вопрос: Что вы подразумеваете под словом «длина была отброшена»? Я предполагаю, что переменная все еще инициализирована.