Rust非同步：效能與平行程式設計

use tokio::fs::File;
use tokio::io::{AsyncReadExt, Result};

async fn read_file(path: &str) -> Result<String> {
    // 非同步開啟檔案
    let mut file = File::open(path).await?;
    
    // 準備讀取緩衝區
    let mut buffer = String::new();
    
    // 非同步讀取內容
    file.read_to_string(&mut buffer).await?;
    
    Ok(buffer)
}

async fn process_files() -> Result<()> {
    // 同時讀取多個檔案
    let content1_future = read_file("file1.txt");
    let content2_future = read_file("file2.txt");
    
    // 等待兩個檔案都讀取完成
    let (content1, content2) = tokio::join!(content1_future, content2_future);
    
    println!("File 1: {}", content1?);
    println!("File 2: {}", content2?);
    
    Ok(())
}

use reqwest;
use std::error::Error;

async fn fetch_url(url: &str) -> Result<String, Box<dyn Error>> {
    let response = reqwest::get(url).await?;
    let body = response.text().await?;
    Ok(body)
}

async fn fetch_multiple_urls() -> Result<(), Box<dyn Error>> {
    let urls = vec![
        "https://www.rust-lang.org",
        "https://docs.rs",
        "https://crates.io",
    ];
    
    // 建立多個請求的future
    let futures = urls.iter().map(|&url| fetch_url(url));
    
    // 同時執行所有請求
    let results = futures::future::join_all(futures).await;
    
    // 處理結果
    for (i, result) in results.iter().enumerate() {
        match result {
            Ok(body) => println!("URL {} content length: {} bytes", i, body.len()),
            Err(e) => println!("Error fetching URL {}: {}", i, e),
        }
    }
    
    Ok(())
}

use tokio::time::{sleep, Duration};

async fn task_one() {
    println!("Task One: 開始");
    sleep(Duration::from_millis(500)).await;
    println!("Task One: 完成");
}

async fn task_two() {
    println!("Task Two: 開始");
    sleep(Duration::from_millis(300)).await;
    println!("Task Two: 完成");
}

#[tokio::main]
async fn main() {
    // 建立兩個任務
    let handle_one = tokio::spawn(task_one());
    let handle_two = tokio::spawn(task_two());
    
    // 等待兩個任務完成
    let _ = tokio::join!(handle_one, handle_two);
    
    println!("所有任務完成");
}

fn fibonacci(n: u64) -> u64 {
    if n == 0 || n == 1 {
        return n;
    }
    fibonacci(n-1) + fibonacci(n-2)
}

use std::thread;

fn main() {
    let mut threads = Vec::new();
    for i in 0..8 {
        let handle = thread::spawn(move || {
            let result = fibonacci(4000);
            println!("Thread {} result: {}", i, result);
        });
        threads.push(handle);
    }
    
    for handle in threads {
        handle.join().unwrap();
    }
}

use std::time::Instant;
use reqwest::Error;

#[tokio::main]
async fn main() -> Result<(), Error> {
    let url = "https://jsonplaceholder.typicode.com/posts/1";
    let start_time = Instant::now();
    let _ = reqwest::get(url).await?;
    let elapsed_time = start_time.elapsed();
    println!("Request took {} ms", elapsed_time.as_millis());
    Ok(())
}

let first = reqwest::get(url);
let second = reqwest::get(url);
let third = reqwest::get(url);
let fourth = reqwest::get(url);
let first = first.await?;
let second = second.await?;
let third = third.await?;
let fourth = fourth.await?;

let (_, _, _, _) = tokio::join!(
    reqwest::get(url),
    reqwest::get(url),
    reqwest::get(url),
    reqwest::get(url)
);

├── connection
│   ├── Cargo.toml
│   ├── connection
│   └── src
│       └── main.rs
├── scripts
│   ├── prep.sh
│   └── run.sh
└── server
    ├── Cargo.toml
    ├── server
    └── src
        └── main.rs

use std::time::Instant;
use std::thread;

fn main() {
    // 單執行緒運算第50個Fibonacci數
    let start = Instant::now();
    let _ = fibonacci(50);
    let duration = start.elapsed();
    println!("fibonacci(50) in {:?}", duration);
    
    // 多執行緒運算四次第50個Fibonacci數
    let start = Instant::now();
    let mut handles = vec![];
    for _ in 0..4 {
        let handle = thread::spawn(|| {
            fibonacci(50)
        });
        handles.push(handle);
    }
    
    for handle in handles {
        let _ = handle.join();
    }
    let duration = start.elapsed();
    println!("4 threads fibonacci(50) took {:?}", duration);
}

fn fibonacci(n: u64) -> u64 {
    if n <= 1 {
        return n;
    }
    fibonacci(n - 1) + fibonacci(n - 2)
}

fibonacci(50) in 39.665599542s
4 threads fibonacci(50) took 42.601305333s

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering::Relaxed;

fn main() {
    let shared_data = Arc::new((Mutex::new(false), Condvar::new()));
    let shared_data_clone = Arc::clone(&shared_data);
    let STOP = Arc::new(AtomicBool::new(false));
    let STOP_CLONE = Arc::clone(&STOP);
    
    // 背景執行緒：等待通知並列印變更
    let background_thread = thread::spawn(move || {
        let (lock, cvar) = &*shared_data_clone;
        let mut received_value = lock.lock().unwrap();
        while !STOP.load(Relaxed) {
            received_value = cvar.wait(received_value).unwrap();
            println!("Received value: {}", *received_value);
        }
    });
    
    // 更新執行緒：修改分享變數並傳送通知
    let updater_thread = thread::spawn(move || {
        let (lock, cvar) = &*shared_data;
        let values = [false, true, false, true];
        for i in 0..4 {
            let update_value = values[i as usize];
            println!("Updating value to {}...", update_value);
            *lock.lock().unwrap() = update_value;
            cvar.notify_one();
            thread::sleep(Duration::from_secs(4));
        }
        STOP_CLONE.store(true, Relaxed);
        println!("STOP has been updated");
        cvar.notify_one();
    });
    
    updater_thread.join().unwrap();
}

Updating value to false...
Received value: false
Updating value to true...
Received value: true
Updating value to false...
Received value: false
Updating value to true...
Received value: true
STOP has been updated

use std::path::PathBuf;
use tokio::fs::File as AsyncFile;
use tokio::io::AsyncReadExt;
use tokio::sync::watch;
use tokio::time::{sleep, Duration};

async fn read_file(filename: &str) -> Result<String, std::io::Error> {
    let mut file = AsyncFile::open(filename).await?;
    let mut contents = String::new();
    file.read_to_string(&mut contents).await?;
    
    Ok(contents)
}

async fn watch_file_changes(tx: watch::Sender<bool>) {
    let path = PathBuf::from("data.txt");
    let mut last_modified = None;
    
    loop {
        if let Ok(metadata) = path.metadata() {
            let modified = metadata.modified().unwrap();
            if last_modified != Some(modified) {
                last_modified = Some(modified);
                let _ = tx.send(());
            }
        }
        sleep(Duration::from_millis(100)).await;
    }
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(false);
    tokio::spawn(watch_file_changes(tx));
    
    loop {
        // 等待檔案變更
        let _ = rx.changed().await;
        
        // 讀取檔案並將其內容輸出到控制檯
        if let Ok(contents) = read_file("data.txt").await {
            println!("{}", contents);
        }
    }
}

[dependencies]
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.11", features = ["json"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

use reqwest::Error;
use serde::Deserialize;
use tokio::time::sleep;
use std::time::Duration;
use std::time::Instant;
use serde_json;

#[derive(Deserialize, Debug)]
struct Response {
    url: String,
    args: serde_json::Value,
}

async fn fetch_data(seconds: u64) -> Result<Response, Error> {
    let request_url = format!("https://httpbin.org/delay/{}", seconds);
    let response = reqwest::get(&request_url).await?;
    let delayed_response: Response = response.json().await?;
    Ok(delayed_response)
}

async fn calculate_last_login() {
    sleep(Duration::from_secs(1)).await;
    println!("登入時間：2天前");
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let start_time = Instant::now();
    
    let data = fetch_data(5);
    let time_since = calculate_last_login();
    
    let (posts, _) = tokio::join!(
        data, time_since
    );
    
    let duration = start_time.elapsed();
    println!("取得的資料：{:?}", posts);
    println!("總耗時：{:?}", duration);
    
    Ok(())
}

登入時間：2天前
取得的資料：Ok(Response { url: "https://httpbin.org/delay/5", args: Object {} })
總耗時：5.494735083s

use std::time::Duration;
use tokio::time::sleep;
use std::thread;
use std::time::Instant;

async fn prep_coffee_mug() {
    println!("倒入牛奶...");
    thread::sleep(Duration::from_secs(3));
    println!("牛奶已倒入。");
    println!("放入即溶咖啡...");
    thread::sleep(Duration::from_secs(3));
    println!("即溶咖啡已放入。");
}

async fn make_coffee() {
    println!("燒水中...");
    sleep(Duration::from_secs(10)).await;
    println!("水已燒開。");
    println!("倒入熱水...");
    thread::sleep(Duration::from_secs(3));
    println!("熱水已倒入。");
}

async fn make_toast() {
    println!("將麵包放入烤麵包機...");
    sleep(Duration::from_secs(10)).await;
    println!("麵包已烤好。");
    println!("塗抹奶油...");
    thread::sleep(Duration::from_secs(5));
    println!("吐司已塗好奶油。");
}

#[tokio::main]
async fn main() {
    let start_time = Instant::now();
    let coffee_mug_step = prep_coffee_mug();
    let coffee_step = make_coffee();
    let toast_step = make_toast();
    tokio::join!(coffee_mug_step, coffee_step, toast_step);
    let elapsed_time = start_time.elapsed();
    println!("總共耗時: {} 秒", elapsed_time.as_secs());
}

倒入牛奶...
牛奶已倒入。
放入即溶咖啡...
即溶咖啡已放入。
燒水中...
將麵包放入烤麵包機...
水已燒開。
倒入熱水...
熱水已倒入。
麵包已烤好。
塗抹奶油...
吐司已塗好奶油。
總共耗時: 24 秒

async fn prep_coffee_mug() {
    sleep(Duration::from_millis(100)).await;
    // 其餘程式碼不變
}

燒水中...
將麵包放入烤麵包機...
倒入牛奶...
牛奶已倒入。
放入即溶咖啡...
即溶咖啡已放入。
麵包已烤好。
塗抹奶油...
吐司已塗好奶油。
水已燒開。
倒入熱水...
熱水已倒入。
總共耗時: 18 秒

let person_one = tokio::task::spawn(async {
    prep_coffee_mug().await;
    make_coffee().await;
    make_toast().await;
});

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]

let person_one = tokio::task::spawn(async {
    let coffee_mug_step = prep_coffee_mug();
    let coffee_step = make_coffee();
    let toast_step = make_toast();
    tokio::join!(coffee_mug_step, coffee_step, toast_step);
}).await;

let person_two = tokio::task::spawn(async {
    let coffee_mug_step = prep_coffee_mug();
    let coffee_step = make_coffee();
    let toast_step = make_toast();
    tokio::join!(coffee_mug_step, coffee_step, toast_step);
}).await;

let person_one = tokio::task::spawn(async {
    let coffee_mug_step = prep_coffee_mug();
    let coffee_step = make_coffee();
    let toast_step = make_toast();
    tokio::join!(coffee_mug_step, coffee_step, toast_step);
});

let person_two = tokio::task::spawn(async {
    let coffee_mug_step = prep_coffee_mug();
    let coffee_step = make_coffee();
    let toast_step = make_toast();
    tokio::join!(coffee_mug_step, coffee_step, toast_step);
});

let _ = tokio::join!(person_one, person_two);

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::task::JoinHandle;

struct CounterFuture {
    count: u32,
}

impl Future for CounterFuture {
    type Output = u32;
    
    fn poll(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Self::Output> {
        self.count += 1;
        println!("輪詢結果: {}", self.count);
        std::thread::sleep(Duration::from_secs(1));
        
        if self.count < 5 {
            cx.waker().wake_by_ref();
            Poll::Pending
        } else {
## 非同步Rust中的Futures與Pinning核心概念

### 理解Pinning的必要性

Pinning在Rust的非同步程式設計中扮演著關鍵角色。當我們處理自參照結構(self-referential structures)時，理解記憶體移動的風險尤為重要。大多數基本型別如數字、字串、布林值、結構體和列舉都實作了`Unpin`特性，允許它們在記憶體中自由移動。

#### 為何需要Pinning？

當結構體包含指向自身的參照時，移動該結構體會導致嚴重問題。讓我來展示一個自參照結構的危險性：

```rust
use std::ptr;

struct SelfReferential {
    data: String,
    self_pointer: *const String,
}

impl SelfReferential {
    fn new(data: String) -> SelfReferential {
        let mut sr = SelfReferential {
            data,
            self_pointer: ptr::null(),
        };
        sr.self_pointer = &sr.data as *const String;
        sr
    }
    
    fn print(&self) {
        unsafe {
            println!("{}", *self.self_pointer);
        }
    }
}

fn main() {
    let first = SelfReferential::new("first".to_string());
    let moved_first = first; // 移動結構體
    moved_first.print(); // 可能導致記憶體區段錯誤
}

fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
    // ...
    if self.count < 5 {
        cx.waker().wake_by_ref();
        Poll::Pending
    } else {
        Poll::Ready(self.count)
    }
}
__CODE_BLOCK_41__rust
use std::time::Duration;
use tokio::time::timeout;

async fn slow_task() -> &'static str {
    tokio::time::sleep(Duration::from_secs(10)).await;
    "Slow Task Completed"
}

#[tokio::main]
async fn main() {
    let duration = Duration::from_secs(3);
    let result = timeout(duration, slow_task()).await;
    match result {
        Ok(value) => println!("Task completed successfully: {}", value),
        Err(_) => println!("Task timed out"),
    }
}

use std::pin::Pin;
use std::task::{Context, Poll, Waker};
use std::sync::{Arc, Mutex};
use std::future::Future;
use tokio::sync::mpsc;
use tokio::task;

struct MyFuture {
    state: Arc<Mutex<MyFutureState>>,
}

struct MyFutureState {
    data: Option<Vec<u8>>,
    waker: Option<Waker>,
}

impl MyFuture {
    fn new() -> (Self, Arc<Mutex<MyFutureState>>) {
        let state = Arc::new(Mutex::new(MyFutureState {
            data: None,
            waker: None,
        }));
        (
            MyFuture {
                state: state.clone(),
            },
            state,
        )
    }
}

impl Future for MyFuture {
    type Output = String;
    
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        println!("Polling the future");
        let mut state = self.state.lock().unwrap();
        if state.data.is_some() {
            let data = state.data.take().unwrap();
            Poll::Ready(String::from_utf8(data).unwrap())
        } else {
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    let (my_future, state) = MyFuture::new();
    let (tx, mut rx) = mpsc::channel::<()>(1);
    
    let task_handle = task::spawn(async {
        my_future.await
    });
    
    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
    println!("spawning trigger task");
    
    let trigger_task = task::spawn(async move {
        rx.recv().await;
        let mut state = state.lock().unwrap();
        state.data = Some(b"Hello from the outside".to_vec());
        
        loop {
            if let Some(waker) = state.waker.take() {
                waker.wake();
                break;
            }
        }
    });
    
    tx.send(()).await.unwrap();
    
    let outome = task_handle.await.unwrap();
    println!("Task completed with outcome: {}", outome);
    trigger_task.await.unwrap();
}

use std::sync::{Arc, Mutex};
use tokio::task::JoinHandle;
use core::task::Poll;
use tokio::time::Duration;
use std::task::Context;
use std::pin::Pin;
use std::future::Future;

#[derive(Debug)]
enum CounterType {
    Increment,
    Decrement
}

struct SharedData {
    counter: i32,
}

impl SharedData {
    fn increment(&mut self) {
        self.counter += 1;
    }
    
    fn decrement(&mut self) {
        self.counter -= 1;
    }
}

struct CounterFuture {
    counter_type: CounterType,
    data_reference: Arc<Mutex<SharedData>>,
    count: u32
}

impl Future for CounterFuture {
    type Output = u32;
    
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        std::thread::sleep(Duration::from_secs(1));
        
        let mut guard = match self.data_reference.try_lock() {
            Ok(guard) => guard,
            Err(error) => {
                println!(
                    "error for {:?}: {}",
                    self.counter_type, error
                );
                cx.waker().wake_by_ref();
                return Poll::Pending
            }
        };
        
        let value = &mut *guard;
        match self.counter_type {
            CounterType::Increment => {
                value.increment();
                println!("after increment: {}", value.counter);
            },
            CounterType::Decrement => {
                value.decrement();
                println!("after decrement: {}", value.counter);
            }
        }
        
        std::mem::drop(guard);
        self.count += 1;
        
        if self.count < 3 {
            cx.waker().wake_by_ref();
            return Poll::Pending
        } else {
            return Poll::Ready(self.count)
        }
    }
}
__CODE_BLOCK_48__rust
#[tokio::main]
async fn main() {
    let shared_data = Arc::new(Mutex::new(SharedData{counter: 0}));
    
    let counter_one = CounterFuture {
        counter_type: CounterType::Increment,
        data_reference: shared_data.clone(),
        count: 0
    };
    
    let counter_two = CounterFuture {
        counter_type: CounterType::Decrement,
        data_reference: shared_data.clone(),
        count: 0
    };
    
    let handle_one: JoinHandle<u32> = tokio::task::spawn(async move {
        counter_one.await
    });
    
    let handle_two: JoinHandle<u32> = tokio::task::spawn(async move {
        counter_two.await
    });
    
    tokio::join!(handle_one, handle_two);
}

after decrement: -1
after increment: 0
error for Increment: try_lock failed because the operation would block
after decrement: -1
after increment: 0
after decrement: -1
after increment: 0

async fn count(count: u32, data: Arc<tokio::sync::Mutex<SharedData>>,
               counter_type: CounterType) -> u32 {
    for _ in 0..count {
        let mut data = data.lock().await;
        match counter_type {
            CounterType::Increment => {
                data.increment();
                println!("after increment: {}", data.counter);
            },
            CounterType::Decrement => {
                data.decrement();
                println!("after decrement: {}", data.counter);
            }
        }
        std::mem::drop(data);
        std::thread::sleep(Duration::from_secs(1));
    }
    return count
}

let shared_data = Arc::new(tokio::sync::Mutex::new(SharedData{counter: 0}));
let shared_two = shared_data.clone();

let handle_one: JoinHandle<u32> = tokio::task::spawn(async move {
    count(3, shared_data, CounterType::Increment).await
});

let handle_two: JoinHandle<u32> = tokio::task::spawn(async move {
    count(3, shared_two, CounterType::Decrement).await
});

tokio::join!(handle_one, handle_two);

use std::fs::{File, OpenOptions};
use std::io::prelude::*;
use std::sync::{Arc, Mutex};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::task::JoinHandle;
use futures_util::future::join_all;

type AsyncFileHandle = Arc<Mutex<File>>;
type FileJoinHandle = JoinHandle<Result<bool, String>>;

fn get_handle(file_path: &dyn ToString) -> AsyncFileHandle {
    match OpenOptions::new().append(true).open(file_path.to_string()) {
        Ok(opened_file) => {
            Arc::new(Mutex::new(opened_file))
        },
        Err(_) => {
            Arc::new(Mutex::new(File::create(file_path.to_string()).unwrap()))
        }
    }
}

struct AsyncWriteFuture {
    pub handle: AsyncFileHandle,
    pub entry: String
}

impl Future for AsyncWriteFuture {
    type Output = Result<bool, String>;
    
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut guard = match self.handle.try_lock() {
            Ok(guard) => guard,
            Err(error) => {
                println!("error for {} : {}", self.entry, error);
                cx.waker().wake_by_ref();
                return Poll::Pending
            }
        };
        
        let lined_entry = format!("{}\n", self.entry);
        match guard.write_all(lined_entry.as_bytes()) {
            Ok(_) => println!("written for: {}", self.entry),
            Err(e) => println!("{}", e)
        };
        
        Poll::Ready(Ok(true))
    }
}

fn write_log(file_handle: AsyncFileHandle, line: String) -> FileJoinHandle {
    let future = AsyncWriteFuture{
        handle: file_handle,
        entry: line
    };
    
    tokio::task::spawn(async move {
        future.await
    })
}