1. 程式人生 > >深入淺出Rust-Future-Part-4

深入淺出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中文社群投稿,投稿地址 ,好文將在以下地方直接展示

  1. Rust中文社群首頁

  2. Rust中文社群閱讀Rust文章欄目

  3. 知乎專欄Rust語言

  4. sf.gg專欄Rust中文社群

  5. 微博Rustlang-cn

  6. 簡書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

(就像我們在Part-3章節所做的那樣). 但這一次,我們不會立即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大於當前時間:

    1. Park Task
    2. 開啟另一個執行緒
  • 副執行緒在一段時間後被喚醒:

    1. 告訴Reactor我們的Task可以Unpark了。
    2. 銷燬自身
  • Reactor喚醒被ParkTask

  • Future(Task)完成了自身的任務:

    1. 通知Reactor
    2. 返回相應的結果
  • 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,目標是:建立一個不會阻塞ReactorIterators.