深入淺出Rust-Future-Part-4
譯自Rust futures: an uneducated, short and hopefully not boring tutorial - Part 4 - A "real" future from scratch 本文時間：2018-12-03，譯者: motecshine, 簡介：motecshine
歡迎向Rust中文社群投稿,投稿地址 ,好文將在以下地方直接展示
Intro
上三篇文章我們闡述如何處理Future的基礎知識， 我們現在能組織多個Future成為一個Future chain, 執行他們,甚至建立他們.但是到現在我們的Future還沒有貼近我們日常的使用場景。(But, so far, our futures are not really delegating the execution to another thing.)
在Part-3中我們用了粗暴的方法來Unpark Future。雖然解決了問題，並且使Reactor變得相對高效，但是這不是最佳實踐。今天就讓我們換種更好的方式去實現Future。
A timer future
我們可以建立一個最簡單的Timer Future
Unpark Future Task, 而是一直Parked， 直到這個Future準備完為止. 我們該怎樣實現？ 最簡單的方式就是再委派一個執行緒。這個執行緒將等待一段時間， 然後Unpark我們的Future Task.
這就像在模擬一個AsyncIO的使用場景。當我們非同步做的一些事情已經完成我們會收到與之相應的通知。為了簡單起見，我們認為Reactor是單執行緒的，在等待通知的時候可以做其他的事情。
Timer revised
我們的結構體非常簡單.他包含結束日期和任務是否在執行。
pub struct WaitInAnotherThread { end_time: DateTime<Utc>, running: bool, } impl WaitInAnotherThread { pub fn new(how_long: Duration) -> WaitInAnotherThread { WaitInAnotherThread { end_time: Utc::now() + how_long, running: false, } } }
DateTime型別和Duration持續時間來自chronos crate.
Spin wait
實現等待時間的函式:
pub fn wait_spin(&self) {
while Utc::now() < self.end_time {}
println!("the time has come == {:?}!", self.end_time);
}
fn main() {
let wiat = WaitInAnotherThread::new(Duration::seconds(30));
println!("wait spin started");
wiat.wait_spin();
println!("wait spin completed");
}
在這種情況下，我們基本上會根據到期時間檢查當前時間。 這很有效，而且非常精確。 這種方法的缺點是我們浪費了大量的CPU週期。 在我的電腦上, CPU一個核心完全被佔用，這和我們Part-3遇到的情況一致。
Spin wait這種方式只適用於等待時間非常短的場景, 或者你沒有別的選擇的情況下使用它。
Sleep wait
系統通常會允許你的執行緒Park一段特定的時間.這通常被稱為執行緒睡眠。睡眠執行緒X秒，換據換的意思是: 告訴系統X秒內，不需要排程我。這樣的好處是，CPU可以在這段時間內幹別的事情。在Rust中我們使用std::thread::sleep().
pub fn wait_blocking(&self) {
while Utc::now() < self.end_time {
let delta_sec = self.end_time.timestamp() - Utc::now().timestamp();
if delta_sec > 0 {
thread::sleep(::std::time::Duration::from_secs(delta_sec as u64));
}
}
println!("the time has come == {:?}!", self.end_time);
}
let wiat = WaitInAnotherThread::new(Duration::seconds(30));
println!("wait blocking started");
wiat.wait_blocking();
println!("wait blocking completed");
嘗試執行我們的程式碼會發現, 改進過的程式碼再也不會完全佔用一個CPU核心了。改進過的程式碼比我們該開始寫的效能好多了，但是這就是Future了嗎？
Future
當然不是，我們還沒有實現Future Trait, 所以，我們現在實現它。
impl Future for WaitInAnotherThread {
type Item = ();
type Error = Box<Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
while Utc::now() < self.end_time {
let delta_sec = self.end_time.timestamp() - Utc::now().timestamp();
if delta_sec > 0 {
thread::sleep(::std::time::Duration::from_secs(delta_sec as u64));
}
}
println!("the time has come == {:?}!", self.end_time);
Ok(Async::Ready(())
}
emmm，這塊程式碼我們是不是以前見過， 跟上篇我們寫的一個很像，它會阻塞Reactor，這樣做實在是太糟糕了。
Future應該儘可能的不要阻塞。
一個Reactor的最佳實踐應該至少包含下面幾條：
- 當主
Task需要等待別的Task時，應該停止它。 - 不要阻塞當前執行緒。
- 任務完成時向
Reactor傳送訊號。
我們要做的是建立另一個睡眠執行緒. 睡眠的執行緒是不會佔用CPU資源。所以在另一個執行緒裡Reactor還像往常那樣，高效的工作著。當這個Sleep Thread醒來後, 它會Unpark這個任務, 並且通知Reactor。
讓我們一步一步完善我們的想法:
impl Future for WaitInAnotherThread {
type Item = ();
type Error = Box<Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if Utc::now() < self.end_time {
println!("not ready yet! parking the task.");
if !self.running {
println!("side thread not running! starting now!");
self.run(task::current());
self.running = true;
}
Ok(Async::NotReady)
} else {
println!("ready! the task will complete.");
Ok(Async::Ready(()))
}
}
}
我們只需要建立一個並行執行緒, 所以我們需要有個欄位來判斷(WaitInAnotherThread.runing)，當前需不需要建立這個執行緒。這裡需要注意的是當Future被輪詢之前，這些程式碼是不會被執行的。當然我們還會檢測當前時間是否大於過期時間，如果大於，也不會產生另外一個執行緒。
如果end_time大於當前的時間並且另一個執行緒沒有被建立，程式就會立即建立一個新的執行緒。然後程式會返回Ok(Async::NotReady()), 與我們Part-3中所做的相反，我們不會在這裡Unpark Task. 這是另一個執行緒應該做的事情。在別的實現中，例如IO，喚醒我們的主執行緒的應該是作業系統。
fn run(&mut self, task: task::Task) {
let lend = self.end_time;
thread::spawn(move || {
while Utc::now() < lend {
let delta_sec = lend.timestamp() - Utc::now().timestamp();
if delta_sec > 0 {
thread::sleep(::std::time::Duration::from_secs(delta_sec as u64));
}
task.notify();
}
println!("the time has come == {:?}!", lend);
});
}
這裡有兩件事情需要注意下.
-
我們將
Task引用傳遞給，另一個並行的執行緒。這很重要，因為我們不能在單獨的執行緒裡使用task::current. -
我們不能將self移動到閉包中，所以我們需要轉移所有權至
lend變數.為啥這樣做？
Rust中的執行緒需要實現具有
'static生命週期的Send Trait。
task自身實現了上述的要求所以可以引用傳遞。但是我們的結構體沒有實現，所以這也是為什麼要移動end_time的所有權。這就意味著當執行緒被建立後你不能更改end_time.
讓我們嘗試執行下:
fn main() {
let mut reactor = Core::new().unwrap();
let wiat = WaitInAnotherThread::new(Duration::seconds(3));
println!("wait future started");
let ret = reactor.run(wiat).unwrap();
println!("wait future completed. ret == {:?}", ret);
}
執行結果：
Finished dev [unoptimized + debuginfo] target(s) in 0.96 secs
Running `target/debug/tst_fut_complete`
wait future started
not ready yet! parking the task.
side thread not running! starting now!
the time has come == 2017-11-21T12:55:23.397862771Z!
ready! the task will complete.
wait future completed. ret == ()
讓我們總結下流程：
-
我們讓
Reactor執行我們的Future. -
Future發現end_time大於當前時間：Park Task- 開啟另一個執行緒
-
副執行緒在一段時間後被喚醒:
- 告訴
Reactor我們的Task可以Unpark了。 - 銷燬自身
- 告訴
-
Reactor喚醒被Park的Task -
Future(Task)完成了自身的任務：- 通知
Reactor - 返回相應的結果
- 通知
-
reactor將任
Task的輸出值返回給run函式的呼叫者。
Code
extern crate chrono;
extern crate futures;
extern crate tokio_core;
use chrono::prelude::*;
use chrono::*;
use futures::prelude::*;
use futures::*;
use std::error::Error;
use std::thread::{sleep, spawn};
use tokio_core::reactor::Core;
pub struct WaitInAnotherThread {
end_time: DateTime<Utc>,
running: bool,
}
impl WaitInAnotherThread {
pub fn new(how_long: Duration) -> WaitInAnotherThread {
WaitInAnotherThread {
end_time: Utc::now() + how_long,
running: false,
}
}
fn run(&mut self, task: task::Task) {
let lend = self.end_time;
spawn(move || {
while Utc::now() < lend {
let delta_sec = lend.timestamp() - Utc::now().timestamp();
if delta_sec > 0 {
sleep(::std::time::Duration::from_secs(delta_sec as u64));
}
task.notify();
}
println!("the time has come == {:?}!", lend);
});
}
}
impl Future for WaitInAnotherThread {
type Item = ();
type Error = Box<Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if Utc::now() < self.end_time {
println!("not ready yet! parking the task.");
if !self.running {
println!("side thread not running! starting now!");
self.run(task::current());
self.running = true;
}
Ok(Async::NotReady)
} else {
println!("ready! the task will complete.");
Ok(Async::Ready(()))
}
}
}
fn main() {
let mut reactor = Core::new().unwrap();
let wiat = WaitInAnotherThread::new(Duration::seconds(3));
println!("wait future started");
let ret = reactor.run(wiat).unwrap();
println!("wait future completed. ret == {:?}", ret);
}
Conclusion
到目前未知我們完整的實現了沒有阻塞的real life future. 所以也沒有浪費CPU資源。除了這個例子你還能想到與此相同的應用場景嗎？
儘管RUST早都有現成的Crate幫我們實現好了。但是瞭解其中的工作原理還是對我們有很大的幫助。
下一個主題將是Streams，目標是：建立一個不會阻塞Reactor的Iterators.