Rust異步編程方式是什么

這篇文章主要講解了“Rust異步編程方式是什么”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“Rust異步編程方式是什么”吧！

調(diào)度器是如何工作的?

調(diào)度器，顧名思義，就是如何調(diào)度程序執(zhí)行。通常來說，程序會(huì)分成許多“工作單元”，我們將這種工作單元成為任務(wù)(task)。一個(gè)任務(wù)要么是可運(yùn)行的，要么是掛起的(空閑的或阻塞的)。任務(wù)是彼此獨(dú)立的，因?yàn)樘幵凇翱蛇\(yùn)行的”任務(wù)都可能被并發(fā)的執(zhí)行。調(diào)度器的職責(zé)就是執(zhí)行任務(wù)，直到任務(wù)被掛起。這個(gè)過程中隱含得本質(zhì)就是如何為任務(wù)分配全局資源——CPU  時(shí)間。

接下來的內(nèi)容里只是圍繞“用戶空間”的調(diào)度器，有操作系統(tǒng)基礎(chǔ)知識(shí)的讀者應(yīng)該明白，指的是運(yùn)行于操作系統(tǒng)線程之上的調(diào)度器，而操作系統(tǒng)線程則是由內(nèi)核調(diào)度器所調(diào)度。

Tokio 調(diào)度器會(huì)執(zhí)行 Rust 的 future，就像我們討論 Java 語言、Go 語言等線程模型時(shí)一樣，Rust 的 future可以理解為  Rust 語言的“異步綠色線程”，它是 M:N 模式，很多用戶空間的任務(wù)通過多路復(fù)用跑在少量的系統(tǒng)線程上。

調(diào)度器的設(shè)計(jì)模式有很多種，每種都有各自的優(yōu)缺點(diǎn)。但本質(zhì)上，可以將調(diào)度器抽象得看作是一個(gè)(任務(wù))隊(duì)列，以及一個(gè)不斷消費(fèi)隊(duì)列中任務(wù)的處理器，我們可以用偽代碼表示成如下形式：

while let Some(task) = self.queue.pop { task.run;}

當(dāng)任務(wù)變成“可運(yùn)行”的，就被插入到隊(duì)列中：

雖然我們可以設(shè)計(jì)成將資源、任務(wù)以及處理器都存在于一個(gè)單獨(dú)線程中，但 Tokio 還是選擇多線程模型?，F(xiàn)代計(jì)算機(jī)都具有多個(gè) CPU  以及多個(gè)物理核，使用單線程模型調(diào)度器會(huì)嚴(yán)重得限制資源利用率，所以為了盡可能壓榨所有 CPU 或物理核的能力，就需要：

• 單一全局的任務(wù)隊(duì)列，多處理器

• 多任務(wù)隊(duì)列，每個(gè)都有單獨(dú)的處理器

單隊(duì)列+多處理器

這種模型中，有一個(gè)全局的任務(wù)隊(duì)列。當(dāng)任務(wù)處于“可運(yùn)行的”狀態(tài)時(shí)，它被插到任務(wù)隊(duì)列尾。處理器們都在不同的線程上運(yùn)行，每個(gè)處理器都從隊(duì)列頭取出任務(wù)并“消費(fèi)”，如果隊(duì)列為空了，那所有線程(以及對(duì)應(yīng)的處理器)都被阻塞。

任務(wù)隊(duì)列必須支持多個(gè)生產(chǎn)者和多個(gè)消費(fèi)者。這里常用的算法就是使用侵入式鏈表，這里的侵入式表示放入隊(duì)列的任務(wù)本身需要包含指向下(后)一個(gè)任務(wù)的指針。這樣在插入和彈出操作時(shí)就可以避免內(nèi)存分配的操作，同時(shí)插入操作是無鎖，但彈出操作就需要一個(gè)信號(hào)量去協(xié)調(diào)多個(gè)消費(fèi)者。

這種方式多用于實(shí)現(xiàn)通用線程池場(chǎng)景，它具有如下的優(yōu)點(diǎn)：

• 任務(wù)會(huì)被公平地調(diào)度

• 實(shí)現(xiàn)相對(duì)簡(jiǎn)單明了

上面說得公平調(diào)度意味著所有任務(wù)是以“先進(jìn)先出”的方式來調(diào)度。這樣的方式在有一些場(chǎng)景下，比如使用 fork-join  方式的并行計(jì)算場(chǎng)景時(shí)就不夠高效了。因?yàn)槲ㄒ恢匾目剂渴亲罱K結(jié)果的計(jì)算速度，而非子任務(wù)的公平性。

當(dāng)然這種調(diào)度模型也有缺點(diǎn)。所有的處理器(消費(fèi)者)都守著隊(duì)列頭，導(dǎo)致處理器真正執(zhí)行任務(wù)所消耗的時(shí)間遠(yuǎn)遠(yuǎn)大于任務(wù)從隊(duì)列中彈出的時(shí)間，這在長(zhǎng)耗時(shí)型任務(wù)場(chǎng)景中是有益的，因?yàn)殛?duì)列爭(zhēng)用會(huì)降低。然而，Rust  的異步任務(wù)是被設(shè)計(jì)用于短耗時(shí)的，此時(shí)爭(zhēng)用隊(duì)列的開銷就變得很大了。

并發(fā)和“機(jī)械共情”

讀者們肯定聽過“為xxx平臺(tái)特別優(yōu)化”這樣的表達(dá)，這是因?yàn)橹挥谐浞至私庥布軜?gòu)，才能知道如何最大化利用硬件資源，才能設(shè)計(jì)出運(yùn)行性能最高的程序。這就是所謂的“機(jī)械共情”，這個(gè)詞是由馬丁湯普森最初提出并使用的。

至于現(xiàn)代硬件架構(gòu)下如何處理并發(fā)的相關(guān)細(xì)節(jié)并不在本文討論的范圍內(nèi)，感興趣的讀者也可以閱讀文章末的更多參考資料部分。

通常來說，硬件不是通過提高速度(頻率)而是為程序提供更多的 CPU  核來獲取性能提升。每個(gè)核都可以在極短的時(shí)間內(nèi)執(zhí)行大量的計(jì)算，相較而言，訪問內(nèi)存之類的操作則需要更多時(shí)間。因此，為了使程序運(yùn)行得更快，我們必須使每次內(nèi)存訪問的  CPU 指令數(shù)量最大化。盡管編譯器可以幫助我們做很多事，但作為程序設(shè)計(jì)開發(fā)人員，我們需要謹(jǐn)慎地考慮數(shù)據(jù)在內(nèi)存中的結(jié)構(gòu)布局以及訪問內(nèi)存的模式。

當(dāng)涉及到線程并發(fā)時(shí)，CPU 的緩存一致性機(jī)制就會(huì)發(fā)揮作用，它會(huì)確保每個(gè) CPU 的緩存都保持最新狀態(tài)。

所以顯而易見，我們要盡可能地避免跨線程同步，因?yàn)樗切阅軞⑹帧?/p>

多處理器+多任務(wù)隊(duì)列

與前面的模型對(duì)比，在這種方式下，我們使用多個(gè)單線程調(diào)度器，每個(gè)處理器都有自己獨(dú)占的任務(wù)隊(duì)列，這樣完全避免了同步問題。由于 Rust  的任務(wù)模型要求任意線程都可以提交任務(wù)到隊(duì)列，所以我們?nèi)孕枰O(shè)計(jì)一種線程安全的方式。要么每個(gè)處理器的任務(wù)隊(duì)列支持線程安全的插入操作(MPSC)，要么就每個(gè)處理器有兩個(gè)隊(duì)列：非同步隊(duì)列和線程安全隊(duì)列。

這便是 Seastar  所使用的策略。因?yàn)閹缀跬耆苊饬送?，所以性能非常高。但需要注意的是，這并不是靈丹妙藥，因?yàn)闊o法確保任務(wù)負(fù)載都是完全一致統(tǒng)一的，處理器可能出現(xiàn)嚴(yán)重的負(fù)載不均衡，使得資源利用率低下。這通常產(chǎn)生的場(chǎng)景是任務(wù)被粘到了固定的、特定的處理器上。

眾所周知，真實(shí)世界的任務(wù)負(fù)載并不是一致統(tǒng)一的，所以在設(shè)計(jì)通用調(diào)度器時(shí)要避免使用此種模型。

“任務(wù)竊取”調(diào)度器

通常來說，任務(wù)竊取調(diào)度器是建立在分片調(diào)度模型之上的，主要為了解決資源利用率低的問題。每個(gè)處理器都具有自己獨(dú)占的任務(wù)隊(duì)列，處于“可運(yùn)行的”任務(wù)會(huì)被插入到當(dāng)前處理器的隊(duì)列中，并且只會(huì)被當(dāng)前處理器所消費(fèi)(執(zhí)行)。但巧妙的是，當(dāng)一個(gè)處理器空閑時(shí)，它會(huì)檢查同級(jí)的其他處理器的任務(wù)隊(duì)列，看看是不是能“竊取”一些任務(wù)來執(zhí)行。這也是這種模型的名稱含義所在。最終，只有在無法從其他處理器的任務(wù)隊(duì)列那里獲得任務(wù)時(shí)該處理器就會(huì)進(jìn)入休眠。

這幾乎是“兩全其美”的方法。處理器可以獨(dú)立運(yùn)行，避免了同步開銷。而且如果任務(wù)負(fù)載在處理器間分布不均衡，調(diào)度器也能夠重新分配負(fù)載。正是由于這樣的特性，諸如  Go 語言、Erlang 語言、Java 語言等都采用了“任務(wù)竊取”調(diào)度器。

當(dāng)然，它也是有缺點(diǎn)的，那就是它的復(fù)雜性。任務(wù)隊(duì)列必須支持“竊取”操作，并且需要一些跨處理器同步操作。整個(gè)過程如果執(zhí)行不正確，那“竊取”的開銷就超過了模型本身的收益。

讓我們來考慮一個(gè)場(chǎng)景：處理器 A 當(dāng)前正在執(zhí)行任務(wù)，并且此刻它的任務(wù)隊(duì)列是空的;處理器 B  此時(shí)空閑，它嘗試“竊取”任務(wù)但是失敗了，因此進(jìn)入休眠態(tài)。緊接著，處理器 A 所執(zhí)行的任務(wù)產(chǎn)生出了20個(gè)(子)任務(wù)。目的是喚醒處理器  B。這進(jìn)而就需要調(diào)度器在觀察到任務(wù)隊(duì)列中有新的任務(wù)時(shí)，向處于休眠態(tài)的處理器發(fā)出信號(hào)。顯而易見，這樣的場(chǎng)景下會(huì)需要額外的同步操作，但這恰恰是我們想要避免的。

綜上所述：

• 盡量減少同步操作總是好的

• “任務(wù)竊取”是通用調(diào)度器的首選算法

• 處理器間基本是相互獨(dú)立的，但是“偷竊”操作時(shí)不可避免的需要一些同步操作

Tokio 0.1 調(diào)度器

2018年3月，Tokio 發(fā)布了其第一版基于“任務(wù)竊取”算法的調(diào)度器。但那個(gè)版本的實(shí)現(xiàn)中有一些瑕疵：

首先，I/O 型任務(wù)會(huì)被同時(shí)操作 I/O 選擇器(epoll、kqueue、iocp等)的線程所執(zhí)行;更多與 CPU  綁定的任務(wù)會(huì)進(jìn)入線程池。在這種情況下，活躍態(tài)線程的數(shù)量應(yīng)該是靈活的、動(dòng)態(tài)的，所以(適時(shí)得)關(guān)閉空閑態(tài)線程是合理的。但是，在“任務(wù)竊取”調(diào)度器上執(zhí)行所有異步任務(wù)時(shí)，始終保持少量的活躍態(tài)線程是更合理的。

其次，當(dāng)時(shí)采用了基于 Chase-Lev deque 算法的隊(duì)列，該算法后來被證明并不適合于調(diào)度獨(dú)立的異步任務(wù)場(chǎng)景。

第三，實(shí)現(xiàn)過于復(fù)雜。由于代碼中過多得使用 atomic，然而大部分情況下，mutex 是更好地選擇。

最后，代碼中有許多細(xì)小的低效設(shè)計(jì)和實(shí)現(xiàn)，但由于早期為保證 API 的穩(wěn)定性，導(dǎo)致了一些技術(shù)債。

當(dāng)然，隨著 Tokio 新版的發(fā)布，我們收獲了很多的經(jīng)驗(yàn)教訓(xùn)，償還了許多技術(shù)債，這著實(shí)是令人興奮的!

下一代的 Tokio 調(diào)度器

現(xiàn)在我們深入解析一下新調(diào)度器的變更。

新的任務(wù)系統(tǒng)

首先，重要的亮點(diǎn)并不屬于 Tokio 的一部分，但對(duì)達(dá)成我們的成就至關(guān)重要：std 包含了由 Taylor  Cramer設(shè)計(jì)的新的任務(wù)系統(tǒng)。該系統(tǒng)給調(diào)度系統(tǒng)提供了鉤子(hooks)，方便調(diào)度器執(zhí)行 Rust 異步任務(wù)，并且確實(shí)做得很好，比之前的版本更輕巧靈活。

Waker結(jié)構(gòu)由資源保存，用于表示任務(wù)可運(yùn)行并被推送到調(diào)度程序的運(yùn)行隊(duì)列中。在新的任務(wù)系統(tǒng)中，Waker結(jié)構(gòu)過去是更大的，但指針寬度為兩個(gè)指針。減小大小對(duì)于最小化復(fù)制Waker值的開銷以及在結(jié)構(gòu)中占用較少的空間非常重要，從而允許將更多關(guān)鍵數(shù)據(jù)放入高速緩存行中。自定義vtable設(shè)計(jì)可實(shí)現(xiàn)許多優(yōu)化，這將在后面討論。

更好的任務(wù)隊(duì)列

任務(wù)隊(duì)列是調(diào)度程序的核心，是最關(guān)鍵的組成部分。最初的tokio調(diào)度器使用crossbeam的deque實(shí)現(xiàn)，即單生產(chǎn)者、多消費(fèi)者deque。任務(wù)從一端入隊(duì)，從另一端出隊(duì)。大多數(shù)情況下，入隊(duì)線程會(huì)出隊(duì)它，然而，其他線程偶爾會(huì)出隊(duì)任務(wù)來“竊取”。deque包含一個(gè)數(shù)組和一組追蹤頭部和尾部的索引。當(dāng)deque滿了時(shí)，入隊(duì)數(shù)據(jù)將導(dǎo)致存儲(chǔ)空間增長(zhǎng)。會(huì)分配一個(gè)新的、更大的數(shù)組，并將值移到新存儲(chǔ)區(qū)中。

雙端隊(duì)列增長(zhǎng)的能力要付出復(fù)雜性和運(yùn)行成本。入隊(duì)/出隊(duì)操作必須考慮到這種情況。此外，在隊(duì)列增長(zhǎng)時(shí)，釋放原始數(shù)組會(huì)帶來額外的困難。在垃圾收集語言中，gc會(huì)釋放它。然而rust不帶GC，這意味著程序需要負(fù)責(zé)釋放數(shù)組，但線程可能正在并發(fā)訪問內(nèi)存。Crossbeam對(duì)此的答案是采用基于代的回收策略。雖然開銷并不是非常大，但確實(shí)在代碼熱路徑中的增加了不小的開銷。每當(dāng)進(jìn)入和退出臨界區(qū)時(shí)，每個(gè)操作都必須是atomic  RMW(讀修改寫)操作，以向其他線程發(fā)出信號(hào)。

由于增長(zhǎng)本地隊(duì)列的相關(guān)成本不低，因此值得研究是否需要增長(zhǎng)隊(duì)列。這個(gè)問題最終導(dǎo)致了調(diào)度程序的重寫。新調(diào)度程序的策略是對(duì)每個(gè)隊(duì)列使用固定大小。當(dāng)隊(duì)列已滿時(shí)，任務(wù)將被推送到一個(gè)全局的、多使用者、多生產(chǎn)者隊(duì)列中，而不是增長(zhǎng)本地隊(duì)列。處理器需要檢查這個(gè)全局隊(duì)列，但檢查的頻率要比本地隊(duì)列低得多。

早期實(shí)驗(yàn)過用有界mpmc隊(duì)列代替了Crossbeam隊(duì)列。由于push和pop都執(zhí)行了大量的同步，因此沒有帶來太大的提升。關(guān)于竊取任務(wù)，需要記住的一個(gè)關(guān)鍵點(diǎn)是，在有負(fù)載的時(shí)候隊(duì)列幾乎沒有爭(zhēng)用，因?yàn)槊總€(gè)處理器只訪問自己的隊(duì)列。

在這一點(diǎn)上，我仔細(xì)閱讀go源代碼，發(fā)現(xiàn)它使用了一個(gè)固定大小的單生產(chǎn)者、多消費(fèi)者隊(duì)列。這個(gè)隊(duì)列令只需要很少的同步就可以正常工作。我對(duì)該算法進(jìn)行了一些修改，使之適用于tokio調(diào)度程序。值得注意的是，go實(shí)現(xiàn)版本中其原子操作使用順序一致性(基于我對(duì)go的有限知識(shí))。作為tokio調(diào)度器的一部分，該版本還降低了較冷代碼路徑中的一些復(fù)制。

該隊(duì)列實(shí)現(xiàn)是一個(gè)循環(huán)緩沖區(qū)，使用數(shù)組存儲(chǔ)值。原子整數(shù)用于跟蹤頭部和尾部位置。

struct Queue { /// Concurrently updated by many threads. head: AtomicU32,  /// Only updated by producer thread but read by many threads. tail: AtomicU32,  /// Masks the head / tail position value to obtain the index in the buffer. mask: usize,  /// Stores the tasks. buffer: Box<[MaybeUninit<Task>]>,}

入隊(duì)由單獨(dú)線程完成：

loop { let head = self.head.load(Acquire);  // safety: this is the **only** thread that updates this cell. let tail = self.tail.unsync_load;  if tail.wrapping_sub(head) < self.buffer.len as u32 { // Map the position to a slot index. let idx = tail as usize & self.mask;  // Don't drop the previous value in `buffer[idx]` because // it is uninitialized memory. self.buffer[idx].as_mut_ptr.write(task);  // Make the task available self.tail.store(tail.wrapping_add(1), Release);  return; }  // The local buffer is full. Push a batch of work to the global // queue. match self.push_overflow(task, head, tail, global) { Ok(_) => return, // Lost the race, try again Err(v) => task = v, }}

請(qǐng)注意，在此push函數(shù)中，唯一的原子操作是使用Acquire順序的load和具有Release順序的store。沒有讀-修改-寫操作(compare_and_swap，fetch_and等)或順序一致性。因?yàn)樵趚86芯片上，所有l(wèi)oad/store  已經(jīng)是“原子的”。因此，在CPU級(jí)別，此功能不是同步的。使用原子操作會(huì)影響編譯器，因?yàn)樗鼤?huì)阻止某些優(yōu)化，但也僅此而已。第一個(gè)load很可能可以通過Relaxed順序完成，但是切換成Relaxed順序沒有明顯的收益。

隊(duì)列已滿時(shí)，將調(diào)用push_overflow。此功能將本地隊(duì)列中的一半任務(wù)移到全局隊(duì)列中。全局隊(duì)列是由互斥鎖保護(hù)的侵入鏈表。首先將要移動(dòng)到全局隊(duì)列中的任務(wù)鏈接在一起，然后獲取互斥鎖，并通過更新全局隊(duì)列的尾指針來寫入所有任務(wù)。

如果您熟悉原子內(nèi)存操作的細(xì)節(jié)，您可能會(huì)注意到上圖所示的push函數(shù)可能會(huì)產(chǎn)生“問題”。使用Acquire順序的原子load同步語義非常弱。它可能會(huì)返回老值(并發(fā)竊取操作可能已經(jīng)增加了self.head的值)，但是執(zhí)行入隊(duì)的線程會(huì)讀到線程中的老值。這對(duì)于算法的正確性不是問題。在入隊(duì)的代碼路徑中，我們只關(guān)心本地運(yùn)行隊(duì)列是否已滿。鑒于當(dāng)前線程是可以執(zhí)行入隊(duì)操作的唯一線程，則老值將使隊(duì)列比實(shí)際更滿。它可能會(huì)錯(cuò)誤地認(rèn)為隊(duì)列已滿并進(jìn)入push_overflow函數(shù)，但是此函數(shù)包括更強(qiáng)的原子操作。如果push_overflow確定隊(duì)列實(shí)際上未滿，則返回w  /  Err并再次嘗試push操作。這是push_overflow將一半運(yùn)行隊(duì)列移至全局隊(duì)列的另一個(gè)原因。通過移動(dòng)一半的隊(duì)列，“運(yùn)行隊(duì)列為空”的誤報(bào)率就會(huì)降低。

本地出對(duì)消息也很輕量級(jí)：

loop { let head = self.head.load(Acquire);  // safety: this is the **only** thread that updates this cell. let tail = self.tail.unsync_load;  if head == tail { // queue is empty return None; }  // Map the head position to a slot index. let idx = head as usize & self.mask;  let task = self.buffer[idx].as_ptr.read;  // Attempt to claim the task read above. let actual = self .head .compare_and_swap(head, head.wrapping_add(1), Release);  if actual == head { return Some(task.assume_init); }}

在此函數(shù)中，存在單個(gè)原子load和一個(gè)帶有release的compare_and_swap。主要開銷來自compare_and_swap。

竊取功能類似于出隊(duì)，但是self.tail的load必須是原子的。同樣，類似于push_overflow，竊取操作將嘗試竊取隊(duì)列的一半，而不是單個(gè)任務(wù)。這這是不錯(cuò)的特性，我們將在后面介紹。

最后一部分是全局隊(duì)列。該隊(duì)列用于處理本地隊(duì)列的溢出，以及從非處理器線程向調(diào)度程序提交任務(wù)。如果處理器有負(fù)載，即本地隊(duì)列中有任務(wù)。在從本地隊(duì)列執(zhí)行約60個(gè)任務(wù)后，處理器將嘗試從全局隊(duì)列獲取任務(wù)。當(dāng)處于“搜索”狀態(tài)時(shí)，它還會(huì)檢查全局隊(duì)列，如下所述。

優(yōu)化消息傳遞模式

用Tokio編寫的應(yīng)用程序通常以許多小的獨(dú)立任務(wù)為模型。這些任務(wù)將使用消息相互通信。這種模式類似于Go和Erlang等其他語言?？紤]到這種模式的普遍性，調(diào)度程序嘗試對(duì)其進(jìn)行優(yōu)化是有意義的。

給定任務(wù)A和任務(wù)B。任務(wù)A當(dāng)前正在執(zhí)行，并通過channel向任務(wù)B發(fā)送消息。通道是任務(wù)B當(dāng)前阻塞在channel上，因此發(fā)送消息將導(dǎo)致任務(wù)B轉(zhuǎn)換為可運(yùn)行狀態(tài)，并被入隊(duì)到當(dāng)前處理器的運(yùn)行隊(duì)列中。然后，處理器將從運(yùn)行隊(duì)列中彈出下一個(gè)任務(wù)，執(zhí)行該任務(wù)，然后重復(fù)執(zhí)行直到完成任務(wù)B。

問題在于，從發(fā)送消息到執(zhí)行任務(wù)B的時(shí)間之間可能會(huì)有很大的延遲。此外，“熱”數(shù)據(jù)(例如消息)在發(fā)送時(shí)已存儲(chǔ)在CPU高速緩存中，但是到任務(wù)B被調(diào)度時(shí)，有可能已經(jīng)從高速緩存中清理掉了。

為了解決這個(gè)問題，新的Tokio調(diào)度程序?qū)崿F(xiàn)了特定優(yōu)化(也可以在Go和Kotlin的調(diào)度程序中找到)。當(dāng)任務(wù)轉(zhuǎn)換為可運(yùn)行狀態(tài)時(shí)，它存儲(chǔ)在“下一個(gè)任務(wù)”槽中，而不是將其入隊(duì)到隊(duì)列的后面。在檢查運(yùn)行隊(duì)列之前，處理器將始終檢查該槽。將任務(wù)插入此槽時(shí)，如果任務(wù)已存儲(chǔ)在其中，則舊任務(wù)將從槽中移除，并入隊(duì)到隊(duì)列的后面。在消息傳遞的情況下，這將保證消息的接收者會(huì)被立馬調(diào)度。

任務(wù)竊取

在竊取任務(wù)調(diào)度器中，當(dāng)處理器的運(yùn)行隊(duì)列為空時(shí)，處理器將嘗試從同級(jí)處理器中竊取任務(wù)。隨機(jī)選擇同級(jí)處理器，然后對(duì)該同級(jí)處理器執(zhí)行竊取操作。如果未找到任務(wù)，則嘗試下一個(gè)同級(jí)處理器，依此類推，直到找到任務(wù)。

實(shí)際上，許多處理器通常在大約同一時(shí)間完成其運(yùn)行隊(duì)列的處理。當(dāng)一批任務(wù)到達(dá)時(shí)(例如，輪詢epoll以確保socket就緒時(shí))，就會(huì)發(fā)生這種情況。處理器被喚醒，獲取并運(yùn)行任務(wù)。這導(dǎo)致所有處理器同時(shí)嘗試竊取，意味著多線程試圖訪問相同的隊(duì)列。這會(huì)引起爭(zhēng)用。隨機(jī)選擇初始節(jié)點(diǎn)有助于減少爭(zhēng)用，但是仍然很糟糕。

新的調(diào)度程序會(huì)限制并發(fā)執(zhí)行竊取操作的處理器的數(shù)量。我們將試圖竊取的處理器狀態(tài)稱為“正在搜索任務(wù)”，或簡(jiǎn)稱為“正在搜索”狀態(tài)。通過使用原子計(jì)數(shù)保證處理器在開始搜索之前遞增以及在退出搜索狀態(tài)時(shí)遞減來控制并發(fā)數(shù)量。搜索程序的最大數(shù)量是處理器總數(shù)的一半。雖然限制相當(dāng)草率，但依然可以工作。我們對(duì)搜索程序的數(shù)量沒有硬性限制，只需要節(jié)流即可，以精度來換取算法效率。

處于正在搜索狀態(tài)后，處理器將嘗試從同級(jí)任務(wù)線程中竊取任務(wù)并檢查全局隊(duì)列。

減少跨線程同步

任務(wù)竊取調(diào)度程序的另一個(gè)關(guān)鍵部分是同級(jí)通知。這是處理器在觀察新任務(wù)時(shí)通知同級(jí)的地方。如果其他處理器正處于休眠狀態(tài)，則被喚醒并竊取任務(wù)。通知還有另一個(gè)重要責(zé)任?；仡櫴褂萌踉禹樞?獲取/發(fā)布)的隊(duì)列算法。由于原子內(nèi)存順序的工作原理，而無需額外的同步，因此無法保證同級(jí)處理器將知道隊(duì)列中的任務(wù)被竊取。因此通知?jiǎng)幼鬟€負(fù)責(zé)為同級(jí)處理器建立必要的同步，以使其知道任務(wù)以竊取任務(wù)。這些要求使得通知操作代價(jià)高昂。我們的目標(biāo)是在保證CPU利用率的情況下，盡可能少地執(zhí)行通知操作。通知太多會(huì)導(dǎo)致驚群?jiǎn)栴}。

老版本的Tokio調(diào)度程序采用了樸素的通知方式。每當(dāng)將新任務(wù)推送到運(yùn)行隊(duì)列時(shí)，就會(huì)通知處理器。每當(dāng)該處理器并在喚醒時(shí)找到任務(wù)，它便會(huì)通知另一個(gè)處理器。這種邏輯會(huì)導(dǎo)致所有處理器都被喚醒從而引起爭(zhēng)用。通常這些處理器中的大多數(shù)都找不到任務(wù)，然后重新進(jìn)入休眠。

通過使用Go調(diào)度器中類似的技術(shù)，新調(diào)度器有顯著改進(jìn)。新調(diào)度器在相同的地方進(jìn)行執(zhí)行，然而僅在沒有處于搜索狀態(tài)的worker時(shí)才進(jìn)行通知。通知worker后，其立即轉(zhuǎn)換為搜索狀態(tài)。當(dāng)處于搜索狀態(tài)的處理器找到新任務(wù)時(shí)，它會(huì)首先退出搜索狀態(tài)，然后通知下一個(gè)處理器。

這種方法用于限制處理器喚醒的速率。如果一次調(diào)度了一批任務(wù)(例如，在輪詢epoll以確保套接字就緒時(shí))，則處理器會(huì)收到第一個(gè)任務(wù)的通知，然后處于搜索狀態(tài)。該處理器不會(huì)收到批處理中的其余任務(wù)的通知。負(fù)責(zé)通知的處理程序?qū)⒏`取批處理中的一半任務(wù)，然后通知另一個(gè)處理器。第三個(gè)處理器將被喚醒，從前兩個(gè)處理器中查找任務(wù)，然后竊取其中一半。這樣處理器負(fù)載會(huì)平滑上升，任務(wù)也會(huì)達(dá)到快速負(fù)載平衡。

減少內(nèi)存分配

新的Tokio調(diào)度程序?qū)γ總€(gè)任務(wù)只需要分配一次內(nèi)存，而舊的調(diào)度程序則需要分配兩次內(nèi)存。以前，Task結(jié)構(gòu)如下：

struct Task { /// All state needed to manage the task state: TaskState,  /// The logic to run is represented as a future trait object. future: Box<dyn Future<Output = >>,}

然后，Task結(jié)構(gòu)也將被置于Box中。自從舊的Tokio調(diào)度程序發(fā)布以來，發(fā)生了兩件事。首先，std ::  alloc穩(wěn)定了。其次，F(xiàn)uture任務(wù)系統(tǒng)切換到顯式的vtable策略。有了這兩個(gè)條件，我們就可以減少一次內(nèi)存分配。

現(xiàn)在，任務(wù)表示為：

struct Task<T> { header: Header, future: T, trailer: Trailer,}

Header和Trailer都是執(zhí)行任務(wù)所需的狀態(tài)，狀態(tài)被劃分為“熱”數(shù)據(jù)(header)和“冷”數(shù)據(jù)(trailer)，即，經(jīng)常訪問的數(shù)據(jù)和很少使用的數(shù)據(jù)。熱數(shù)據(jù)放置在結(jié)構(gòu)的頭部，并保持盡可能小。當(dāng)CPU取消引用任務(wù)時(shí)，它將一次性加載高速緩存行大小的數(shù)據(jù)量(介于64和128字節(jié)之間)。我們希望該數(shù)據(jù)盡可能有價(jià)值。

減少原子引用計(jì)數(shù)

最后一個(gè)優(yōu)化在于新的調(diào)度程序如何減少原子引用計(jì)數(shù)。任務(wù)結(jié)構(gòu)有許多未完成的引用：調(diào)度程序和每個(gè)喚醒程序都擁有一個(gè)句柄。管理此內(nèi)存的方法是使用原子引用計(jì)數(shù)。此策略需要在每次克隆引用時(shí)進(jìn)行一次原子操作，并在每次刪除引用時(shí)進(jìn)行一次相反的原子操作。當(dāng)最終引用次數(shù)為0時(shí)，將釋放內(nèi)存。

在舊的Tokio調(diào)度程序中，每個(gè)喚醒器都有一個(gè)對(duì)任務(wù)句柄的引用計(jì)數(shù)：

struct Waker { task: Arc<Task>,} impl Waker { fn wake(&self) { let task = self.task.clone; task.scheduler.schedule(task); }}

喚醒任務(wù)后，將調(diào)用task  的clone方法(原子增量)。然后將引用置入運(yùn)行隊(duì)列。當(dāng)處理器執(zhí)行完任務(wù)時(shí)，它將刪除引用，從而導(dǎo)致引用計(jì)數(shù)的原子遞減。這些原子操作雖然代價(jià)很低但是積少成多。

std :: future任務(wù)系統(tǒng)的設(shè)計(jì)人員已經(jīng)確定了此問題。據(jù)觀察，當(dāng)調(diào)用Waker ::  wake時(shí)，通常不再需要原來的waker引用。這樣可以在將任務(wù)推入運(yùn)行隊(duì)列時(shí)重用原子計(jì)數(shù)?，F(xiàn)在，std :: future任務(wù)系統(tǒng)包括兩個(gè)“喚醒”  API：

• wake帶self參數(shù)

• wake_by_ref帶&self參數(shù)。

這種API設(shè)計(jì)迫使調(diào)用者使用wake方法來避免原子增量?，F(xiàn)在的實(shí)現(xiàn)變?yōu)椋?/p>

impl Waker { fn wake(self) { task.scheduler.schedule(self.task); }  fn wake_by_ref(&self) { let task = self.task.clone; task.scheduler.schedule(task); }}

這就避免了額外的引用計(jì)數(shù)的開銷，然而這僅僅在可以獲取所有權(quán)的時(shí)候可用。根據(jù)我的經(jīng)驗(yàn)，調(diào)用wake幾乎總是通過借用而非獲取引用。使用self進(jìn)行喚醒可防止重用waker，在使用self時(shí)實(shí)現(xiàn)線程安全的喚醒也更加困難(其細(xì)節(jié)將留給另一個(gè)文章)。

新的調(diào)度程序端通過避免調(diào)用wake_by_ref中的clone來逐步解決問題，從而其和wake(self)一樣有效。通過使調(diào)度程序維護(hù)當(dāng)前處于活動(dòng)狀態(tài)(尚未完成)的所有任務(wù)的列表來完成此功能。此列表代表將任務(wù)推送到運(yùn)行隊(duì)列所需的引用計(jì)數(shù)。

這種優(yōu)化的困難之處在于，確保調(diào)度程序在任務(wù)結(jié)束前不會(huì)從其列表中刪除任何任務(wù)。如何進(jìn)行管理的細(xì)節(jié)不在本文的討論范圍之內(nèi)，有興趣可以參考源代碼。

使用Loom無畏并發(fā)

眾所周知，編寫正確的、并發(fā)安全的、無鎖的代碼不是一件容易事，而且正確性最為重要，特別是要盡力避免那些和內(nèi)存分配相關(guān)的代碼缺陷。在此方面，新版調(diào)度器做了很多努力，包括大量的優(yōu)化以及避免使用大部分  std 類型。

從測(cè)試角度來說，通常有幾種方法用來驗(yàn)證并發(fā)代碼的正確性。一種是完全依賴用戶在其使用場(chǎng)景中驗(yàn)證;另一種是依賴循環(huán)運(yùn)行的各種粒度單元測(cè)試試圖捕捉那些非常小概率的極端情況的并發(fā)缺陷。這種情況下，循環(huán)運(yùn)行多長(zhǎng)時(shí)間合適就成了另一個(gè)問題，10分鐘或者10天?

上述情況在我們的工作中是無法接受的，我們希望交付并發(fā)布時(shí)感到十足的自信，對(duì) Tokio 用戶而言，可靠性是最為重要的。

因此，我們便造了一個(gè)“新輪子”：Loom，它是一個(gè)用于測(cè)試并發(fā)代碼的工具。測(cè)試用例可以按照最樸實(shí)尋常的方式來設(shè)計(jì)和編寫，但當(dāng)通過 Loom  來執(zhí)行時(shí)，Loom 會(huì)運(yùn)行多次用例，同時(shí)會(huì)置換(permute)在多線程環(huán)境下所有可能遇到的行為或結(jié)果，這個(gè)過程中 Loom  還會(huì)驗(yàn)證內(nèi)存訪問正確與否，以及內(nèi)存分配和釋放的行為正確與否等等。

下面是調(diào)度器在 Loom 上一個(gè)真實(shí)的測(cè)試場(chǎng)景：

#[test]fn multi_spawn { loom::model(|| { let pool = ThreadPool::new;  let c1 = Arc::new(AtomicUsize::new(0));  let (tx, rx) = oneshot::channel; let tx1 = Arc::new(Mutex::new(Some(tx)));  // Spawn a task let c2 = c1.clone; let tx2 = tx1.clone; pool.spawn(async move { spawn(async move { if 1 == c1.fetch_add(1, Relaxed) { tx1.lock.unwrap.take.unwrap.send(); } }); });  // Spawn a second task pool.spawn(async move { spawn(async move { if 1 == c2.fetch_add(1, Relaxed) { tx2.lock.unwrap.take.unwrap.send(); } }); });  rx.recv; });}

上述代碼中的 loom::model部分運(yùn)行了成千上萬次，每次行為都會(huì)有細(xì)微的差別，比如線程切換的順序，以及每次原子操作時(shí)，Loom  會(huì)嘗試所有可能的行為(符合 C++ 11 中的內(nèi)存模型規(guī)范)。前面我提到過，使用  Acquire進(jìn)行原子的加載操作是非常弱(保證)的，可能返回舊(臟)值，Loom 會(huì)嘗試所有可能加載的值。

在調(diào)度器的日常開發(fā)測(cè)試中，Loom  發(fā)揮了非常重要的作用，幫助我們發(fā)現(xiàn)并確認(rèn)了10多個(gè)其他測(cè)試手段(單元測(cè)試、手工測(cè)試、壓力測(cè)試)所遺漏的隱蔽缺陷。

有的讀者們可能會(huì)對(duì)前文提到的“對(duì)所有可能的結(jié)果或行為進(jìn)行排列組合和置換”產(chǎn)生疑問。眾所周知，對(duì)可能行為的簡(jiǎn)單排列組合就會(huì)導(dǎo)致階乘式的“爆炸”。當(dāng)然目前有許多用于避免這類指數(shù)級(jí)爆炸的算法，Loom  中采用的核心算法是基于“動(dòng)態(tài)子集縮減”算法(dynamic partial  reduction)。該算法能夠動(dòng)態(tài)“修剪”會(huì)導(dǎo)致一致執(zhí)行結(jié)果的排列子集，但需要注意的是，即便如此，在狀態(tài)空間巨大時(shí)也一樣會(huì)導(dǎo)致修剪效率大幅降低。Loom  采用了有界動(dòng)態(tài)子集縮減算法來限制搜索空間。

總而言之，Loom 極大地幫助了我們，使得我們更有信心地發(fā)布新版調(diào)度器。

測(cè)試結(jié)果

我們來具體看看新版 Tokio 調(diào)度器到底取得了多大的性能提升?

首先，在微基準(zhǔn)測(cè)試中，新版調(diào)度器提升顯著。

老版本

test chained_spawn ... bench: 2,019,796 ns/iter (+/- 302,168) test ping_pong ... bench: 1,279,948 ns/iter (+/- 154,365) test spawn_many ... bench: 10,283,608 ns/iter (+/- 1,284,275) test yield_many ... bench: 21,450,748 ns/iter (+/- 1,201,337)

新版本

test chained_spawn ... bench: 168,854 ns/iter (+/- 8,339) test ping_pong ... bench: 562,659 ns/iter (+/- 34,410) test spawn_many ... bench: 7,320,737 ns/iter (+/- 264,620) test yield_many ... bench: 14,638,563 ns/iter (+/- 1,573,678)

測(cè)試內(nèi)容包括：

• chained_spawn測(cè)試會(huì)遞歸地不斷產(chǎn)生新的子任務(wù)。

• ping_pong測(cè)試會(huì)分配一個(gè)一次性地(oneshot)通道，接著產(chǎn)生一個(gè)新的子任務(wù)，子任務(wù)在該通道上發(fā)送消息，原任務(wù)則接受消息。

• spawn_many測(cè)試會(huì)產(chǎn)生出大量子任務(wù)，并注入給調(diào)度器。

• yield_many 會(huì)測(cè)試一個(gè)喚醒自己的任務(wù)。

為了更接近真實(shí)世界的工作負(fù)載，我們?cè)僭囋?Hyper 基準(zhǔn)測(cè)試：

wrk -t1 -c50 -d10

老版本

Running 10s test @ http://127.0.0.1:3000 1 threads and 50 connections Thread Stats Avg Stdev Max +/- Stdev  Latency 371.53us 99.05us 1.97ms 60.53%  Req/Sec 114.61k 8.45k 133.85k 67.00% 1139307 requests in 10.00s, 95.61MB read Requests/sec: 113923.19 Transfer/sec: 9.56MB

新版本

Running 10s test @ http://127.0.0.1:3000 1 threads and 50 connections Thread Stats Avg Stdev Max +/- Stdev  Latency 275.05us 69.81us 1.09ms 73.57%  Req/Sec 153.17k 10.68k 171.51k 71.00% 1522671 requests in 10.00s, 127.79MB read Requests/sec: 152258.70 Transfer/sec: 12.78MB

目前 Hyper 基準(zhǔn)測(cè)試已經(jīng)比 TechEmpower 更有參考性，所以從結(jié)果來看，我們很興奮 Tokio 調(diào)度器已經(jīng)可以沖擊這樣的性能排行榜。

另外一個(gè)令人印象深刻的結(jié)果是，Tonic，流行的 gRPC 客戶端/服務(wù)端框架，取得了超過10%的性能提升，這還是在 Tonic  本身沒有做特定優(yōu)化的情況下!

感謝各位的閱讀，以上就是“Rust異步編程方式是什么”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對(duì)Rust異步編程方式是什么這一問題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！