探索 Rust 異步簡化編程

譯者 | 彎月

譯者 | 彎月     責編 | 歐陽姝黎

出品 | CSDN（ID：CSDNnews）

Rust 的異步功能很強大，但也以晦澀難懂著稱。在本文中，我將總結之前提過的一些想法，並給出一些新的點子，看看這些想法放在一起能產生什麼效果。

本文只是一個思想實驗。對 Rust 進行大改造很麻煩，因此我們需要一個精確的方法來找出優缺點，並確定某個改動是否值得。我知道一些觀點會產生完全相反的看法，所以我建議你用一種開放的心態閱讀本文。

在對 Rust 中實現異步的不同方式進行探索之前，我們應該首先了解何時應該使用異步編程。畢竟，異步編程並不像僅僅使用線程那麼容易。那麼異步的好處是什麼？有人會說是性能原因，異步代碼更快，因爲線程的開銷太大了。實際情況更復雜。根據具體情況不同，在以 I/O 爲主的應用程序中使用線程有可能更快。例如，一個基於線程的 echo 服務器在併發數小於 100 的時候比異步更快。但在併發數超過 100 之後，線程的性能就會下降，但也不是急劇下降。

我認爲，使用異步的更好的理由是可以更有效地針對複雜的流程控制進行建模。例如，如果不適用異步編程，那麼暫停或取消一個正在進行的操作就會非常困難。而且，使用線程時，在各個連接之間進行協調需要使用同步原語，這就會導致競爭。使用異步編程，可以在同一個線程中對多個連接進行操作，從而避免了同步原語。

Rust 的異步模型能夠非常好地對複雜流程控制進行建模。例如，mini-redis 的 subscribe 命令（https://github.com/tokio-rs/mini-redis/blob/master/src/cmd/subscribe.rs#L94-L156）就非常精練、非常優雅。但異步也不是萬能靈藥。許多人都認爲異步 Rust 的學習曲線非常複雜。儘管入門很容易，但很快就會遇到陡峭的曲線。很多人付出了很多努力，儘管有幾個方面有待改進，但我相信，異步 Rust 最大的問題就在於會違反 “最小驚訝原則”。

舉個例子。同學 A 在學習 Rust 時閱讀了 Rust 的教科書和 Tokio 的指南，打算寫一個聊天服務器作爲練習。他選了一個基於行的簡單協議，將每一行編碼，添加前綴表示行的長度。解析行的函數如下：

    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)
}

這段代碼除了 async 和 await 關鍵字之外，跟阻塞的 Rust 代碼沒有什麼兩樣。儘管同學 A 從來沒有寫過 Rust，但閱讀並理解這個函數完全沒問題，至少從他自己的角度看如此。在本地測試時，聊天服務器似乎也能正常工作，於是他給同學 B 發送了一個鏈接。但很不幸，在進行了一些聊天后，服務器崩潰了，並返回了 “invalid UTF-8” 的錯誤。同學 A 很迷惑，他檢查了代碼，但並沒有發現什麼錯誤。

那麼問題在哪兒？似乎該任務在調用棧的更高層的位置使用了一個 select!：

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;
        }
    }
}

假設 channel 上收到了一條消息，而此時 parse_line 在等待更多數據，那麼 select! 就會放棄 parse_line 操作，從而導致丟失解析中的狀態。在後面的循環迭代中，parse_line 再次被調用，從一幀的中間開始，從而導致讀入了錯誤數據。

問題在此：任何 Rust 異步函數都可能被調用者隨時取消，而且與阻塞 Rust 不同，這裏的取消是一個常見的異步操作。更糟糕的是，沒有任何新手教程提到了這一點。

Future

如果能改變這一點，讓異步 Rust 每一步的行爲符合初學者預期呢？如果行爲必須根據預期得到，那麼必然有一個能接受的點，爲初學者指引正確的方向。此外，我們還希望最大程度地減少學習過程中的意料之外，特別是剛開始的時候。

我們先來改變意料之外的取消問題，即讓異步函數總是能夠完成執行。當 future 能夠保證完成後，同學 A 發現異步 Rust 的行爲跟阻塞 Rust 完全相同，只不過是多了兩個關鍵字 async 和 await 而已。生成新任務會增加併發，也會增加任務之間的協調通道數量。select! 不再能夠接受任意異步語句，而只能與通道或類似通道的類型（例如 JoinHandle）一起使用。

使用能保證完成的 future 後，同學 A 的聊天服務器如下：

async fn handle_connection(socket: TcpStream, channel: Channel) {
    let reader = Arc::new(socket);
    let writer = reader.clone();
    let read_task = task::spawn(async move {
        while let Some(line_in) in parse_line(&reader).await {
            broadcast_line(line_in);
        }
    });
    loop {
        // `channel` and JoinHandle are both "channel-like" types.
        select! {
            res = read_task.join() => {
                // The connection closed, exit loop
                break;
            }
            line_out = channel.recv() => {
                write_line(&writer, line_out).await;
            }
        }
    }
}

這段代碼與前面的示例很相似，但由於所有異步語句必然會完成，而且 select! 只接受類似於通道的類型，因此 parse_line() 的調用被移動到了一個生成的任務中。select 要求類似於通道的類型，這能夠保證放棄丟失的分支是安全的。通道可以存儲值，而且接收值是原子操作。丟失 select 的分支並不會導致取消時丟失數據。

取消

如果寫入時發生錯誤會怎樣？現狀下 read_task 會繼續執行。然而，同學 A 希望它能出錯，並優雅地關閉連接和所有任務。不幸的是，這裏就會遇到設計上的難題。如果我們能夠隨時放棄任何異步語句，那麼取消就非常容易了，只需要放棄 future 就可以。我們需要一種方法來取消正在執行的操作，因爲這是使用異步編程的主要目的之一。爲了實現這一點，JoinHandle 提供了 cancel() 方法：

async fn handle_connection(socket: TcpStream, channel: Channel) {
    let reader = Arc::new(socket);
    let writer = reader.clone();
    let read_task = task::spawn(async move {
        while let Some(line_in) in parse_line(&reader).await? {
            broadcast_line(line_in)?;
        }
        Ok(())
    });
    loop {
        // `channel` and JoinHandle are both "channel-like" types.
        select! {
            _ = read_task.join() => {
                // The connection closed or we encountered an error,
                // exit the loop
                break;
            }
            line_out = channel.recv() => {
                if write_line(&writer, line_out).await.is_err() {
                    read_task.cancel();
                    read_task.join();
                }
            }
        }
    }
}

但是 cancel()能做什麼呢？它並不能立即終止任務，因爲現在異步語句是保證能夠執行完成的。但我們的確需要停止處理並儘快返回。相反，被取消的任務中的所有資源類型都應該停止執行，並返回 “被中斷” 的錯誤。進一步的嘗試也應該返回錯誤。這種策略與 Kotlin 很相似，只不過 Kotlin 會拋出異常而已。如果在任務取消時，read_task 正在 parse_line 中等待 socket.read_u32()，那麼 read_u32()函數會立即返回 Err(io::ErrorKind::Interrupted)。操作符? 會在任務中向上冒泡，導致整個任務中斷。

乍一看，這種行爲非常像其他任務停止的行爲，但其實不一樣。對於同學 A 而言，當前的異步 Rust 的終止行爲看起來就像任務不確定地發生掛起一樣。如果能強制資源（例如套接字）在取消時返回錯誤，就能跟蹤取消的流程。同學 A 可以添加 println! 語句或使用其他調試策略來調查什麼導致了任務中斷。

AsyncDrop

然而，同學 A 並不知道，他的聊天服務器使用了 io-uring 來避免了絕大部分系統調用。由於 future 能保證完成，再加上 AsyncDrop，就可以透明底使用 io-uring API。當同學 A 在 handle_connection() 的末尾 drop TcpStream 時，套接字會異步地關閉。爲了實現這一點，TcpStream 的 AsyncDrop 實現如下：

impl AsyncDrop for TcpStream {
    async fn drop(&mut self) {
        self.uring.close(self.fd).await;
    }
}

有人提出了一個絕妙的方法在 traits 中使用 async（https://hackmd.io/bKfiVPRpTvyX8JK_Ng2EWA?view）。唯一的問題就是如何處理隱含的. await 點。目前，異步地等待一個 future 需要進行一次. await 調用。而當一個值離開 async 上下文的範圍時，編譯器會爲 AsyncDrop trait 添加一個隱藏的 yield 點。這個行爲違反了最少意料之外原則。那麼，既然其他的點都是明示的，爲何此處需要一個隱含的 await 點？

解決 “有時需要隱含 drop” 的問題的提議之一就是，要求使用明示的函數調用執行異步的 drop：

my_tcp_stream.read(&mut buf).await?;
async_drop(my_tcp_stream).await;

當然，如果用戶忘記調用 async drop 怎麼辦？畢竟，編譯器在阻塞 Rust 中會自動處理 drop，而且這是個非常強大的功能。而且，注意上述代碼有一個小問題：? 操作符在讀取錯誤時會跳過 async_drop 調用。Rust 編譯器能對此問題給出警告，但怎麼修復呢？有辦法讓? 與明示的 async_drop 兼容嗎？

去掉. await

如果不要求明示的 async drop 調用，而是去掉 await 關鍵字怎麼樣？同學 A 就不需要在調用異步函數（如 socket.read_u32().await）之後使用. await 了。在異步上下文中調用異步函數時，.await 就變成了隱含的。

似乎這是如今 Rust 的一大進步。但我們依然可以對這個假設提出質疑。隱含的. await 只能在異步語句中發生，因此它的應用比較有限，而且依賴於上下文。同學 A 只有通過查看函數定義，才能知道自己位於某個異步上下文中。此外，如果 IDE 能高亮顯示某個 yield 點，就會非常方便。

去掉. await 還有另一個好處：它能讓異步 Rust 與阻塞 Rust 一直。阻塞的概念已經是隱含的了。在阻塞 Rust 中，我們並不會寫 my_socket.read(buffer).block?。當同學 A 編寫異步聊天服務器時，他注意到的唯一區別就是必須用 async 關鍵字來標記函數。同學 A 可以憑直覺想象異步代碼的執行。“懶 future”的問題不再出現，而同學 A 也不能無意間做下面的事，並對先輸出 “two” 的情況感到困惑。

async fn my_fn_one() {
    println!("one");
}
async fn my_fn_two() {
    println!("two");
}
async fn mixup() {
    let one = my_fn_one();
    let two = my_fn_two();
    join!(two, one);
}

.await 的 RFC 中的確有一些對於隱含 await 的討論。當時，反對隱含 await 的最有力的觀點是，await 調用正好標記了 async 語句可以被中止的點。如果採用保證完成的 future，這個觀點就不那麼有力了。當然，對於可以安全中止的異步語句，我們還應該保留 await 關鍵字嗎？這個問題需要一個答案。但無論如何，去掉 “.await” 是一個非常大的挑戰，必須謹慎行事。需要進行易用性研究，表明其優點大於缺點纔行。

帶有作用域的任務

===============

到目前爲止，同學 A 已經可以使用異步 Rust 構建聊天服務器，而且不需要學習太多新概念，也不會遇到無法預測的行爲。他了解了 select!，但編譯器會強制在類似於通道的類型中進行選擇。除此之外，同學 A 還給函數添加了 async，而且運行良好。他把代碼展示給同學 B 看，並詢問是否需要將套接字放在一個 Arc 中。同學 B 建議他閱讀一下帶有作用域的任務（scoped tasks），藉此避免分配。

帶有作用域的任務等價於 crossbeam 的 “帶有作用域的線程”，只不過是異步的。這個任務可以通過生成者借用數據。同學 A 可以使用帶有作用域的任務來避免在連接處理函數中使用 Arc：

async fn handle_connection(socket: TcpStream, channel: Channel) {
    task::scope(async |scope| {
        let read_task = scope.spawn(async || {
            while let Some(line_in) in parse_line(&socket)? {
                broadcast_line(line_in)?;
            }
            Ok(())
        });
        loop {
            // `channel` and JoinHandle are both "channel-like" types.
            select! {
                _ = read_task.join() => {
                    // The connection closed or we encountered an error,
                    // exit the loop
                    break;
                }
                line_out = channel.recv() => {
                    if write_line(&writer, line_out).is_err() {
                        break;
                    }
                }
            }
        }
    });
}

保證安全的關鍵是要保證，作用域的生存週期要大於在該作用域範圍內生成的所有任務，換句話說，確保異步語句能夠完成。但有一個缺點。啓用帶有作用域的任務會使 “Future::poll” 變得不安全，因爲必須對 future 的完成情況進行輪詢，以保證內存安全性。這種不安全性會導致 Future 的實現更難。爲了降低難度，我們需要儘可能避免用戶自己實現 Future，包括實現類似於 AsyncRead、AsyncIterator 等 traits。我相信這是一個可以達到的目標。

除了帶有作用域的任務之外，保證異步語句的完成，還可以在使用 io-uring 或與 C++ future 集成時，讓指針能正確地從任務傳遞到內核。某些情況下，還可能在生成子任務時避免分配，對於某些嵌入式環境非常有用，儘管這種情況需要一個略微不同的 API。

通過生成的方式增加併發

利用今天的異步 Rust，應用程序可以通過利用 select! 或 FutureUnordered 生成新任務的方式增加併發。到目前爲止，我們討論了任務生成和 select!。我建議去掉 FuturesUnordered，因爲它經常會導致 bug。在使用 FutureUnordered 時，很容易認爲生成的任務會在後臺執行，然後出乎意料地發現這些任務不會有任何進展。

相反，我們可以利用帶有作用域的任務實現類似的方案：

let greeting = "Hello".to_string();
task::scope(async |scope| {
    let mut task_set = scope.task_set();
    for i in 0..10 {
        task_set.spawn(async {
            println!("{} from task {}", greeting, i);
            i
        });
    }
    async for res in task_set {
        println!("task completed {:?}", res);
    }
});

每個生成的任務都會併發執行，從生成者那裏借用數據，而 TaskSet 能提供一個類似於 FuturesUnordered，但不會導致災難的 API。至於緩存流等其他原語也可以在帶有作用域的任務上實現。

還可以在這些原語之上實現一些新的併發原語。例如，可以實現類似於 Kotlin 的結構化併發。之前有人曾討論過這個問題（https://github.com/tokio-rs/tokio/issues/1879），但異步 Rust 的當前模型無法實現這一點。而將異步 Rust 改爲保證完成，就能解鎖這一領域。

select! 怎麼辦？

===================

本文開頭我說過，使用異步編程可以更有效地對複雜的流程控制進行建模。目前最有效的原語爲 select!。我還提議，將 select! 改爲只接受類似於通道的類型，這樣可以強制同學 A 爲每個連接生成兩個任務，實現讀寫的併發性。生成任務能防止在取消讀操作的時候出現 bug，還能重寫讀操作，以處理意料之外的取消。例如，mini-redis 在解析幀的時候，我們首先將接收到的數據保存到緩衝區中。當讀操作被取消時，位於緩衝區中的數據不會丟失。下次調用讀操作會從中斷的地方繼續。因此 Mini-redis 的讀操作對於中止是安全的（abort-safe）。

如果不將 select! 限制在類似於通道的類型上，而是將其限制在對於中止是安全的操作上，會怎樣？從通道中接收數據是中止安全的，但從帶有緩衝區的 I/O 處理函數中讀取也是中止安全的。這裏的關鍵是，不應該假設所有異步操作都是中止安全的，而是應該要求開發者向函數定義中添加 #[abort_safe]（或 async(abort)）。這種策略有幾個好處。首先，當同學 A 學習異步 Rust 時，它不需要知道任何有關安全性的概念。即使不理解這個概念，僅通過生成任務來獲得併發性，也可以實現一切：

不再默認要求中止安全語句，而是由開發者自行標註。這種自行標註的策略符合撤銷安全性的模式。當新的開發者閱讀代碼時，這個標註會告訴他們該函數必須保證中止安全。rust 編譯器甚至可以對於標註了 #[abort_safe] 的函數提供額外的檢查和警告。

現在同學 A 可以在 select! 的循環中使用 read_line() 了：

混合使用中止安全和非中止安全

#[abort_safe] 註釋引入了兩個異步語句的變種。混合使用中止安全和非中止安全需要特別考慮。不論從中止安全還是從非中止安全的上下文中，都可以調用一箇中止安全的函數。然而，Rust 編譯器會阻止從中止安全的上下文中調用非中止安全的函數，並提供一個有幫助的錯誤信息：

開發者可以通過生成新任務的方式，從非中止安全函數中獲得中止安全的上下文。

異步語句的兩個新變種會增加語言的複雜性，但這個複雜性僅在學習曲線的後期纔出現。在剛開始學習 Rust 時，默認的異步語句是非中止安全的。從這個上下文中，學習者可以不用關心中止安全性而直接調用異步函數。中止安全會在異步 Rust 的教程的後期作爲一個可選的話題出現。

漫漫長路

從目前的默認要求中止安全的異步模型改變成保證完成的模型，需要一個全新的 Rust 版本。處於討論的目的，我們假設 Rust 2026 版引入了該變動。那麼該版本中的 Future trait 將改變爲保證完成的 future，因此無法與老版本的 trait 兼容。相反，2026 版中的舊 trait 將改名爲 AbortSafeFuture（名稱暫定）。

在 2026 版中，給異步語句添加 #[abort_safe] 會生成一個 AbortSafeFuture 實現，而不是 Future。2026 之前版本中編寫的任何異步函數都實現了 AbortSafeFuture trait，因此任何已有的異步代碼都能與新版本兼容（別忘了，中止安全的函數可以從任何上下文中調用）。

一些想法

我討論了 Rust 可能出現的一些改動。簡單地總結一下：

我相信，這些改動可以極大地簡化 Rust 異步，儘管進行這些改動會影響現狀。在進行決定之前，我們還需要更多數據。如今的異步代碼有多少是中止安全的？我們能否進行易用性研究，以評價這些改動的好處？Rust 擁有兩種風格的異步語句，會帶來多少認知上的困難？

我也希望本文能拋磚引玉，期待其他人能提出別的觀點。現在 Rust 需要許多觀點來決定其未來。

原文鏈接：https://carllerche.com/2021/06/17/six-ways-to-make-async-rust-easier/

聲明：本文由 CSDN 翻譯，轉載請註明來源。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/Cxy-sxpV_GIAl7s5PTfSbw

猜你喜歡