The Rust Programming Language

การคืนการควบคุมให้แก่รันไทม์

ย้อนกลับไปในหัวข้อ “โปรแกรม Async ตัวแรกของเรา” ณ แต่ละจุดรอ (await point) Rust จะเปิดโอกาสให้รันไทม์หยุดพักแทสก์และสลับไปยังแทสก์อื่นได้หากฟิวเจอร์ที่กำลังรอคอยนั้นยังไม่พร้อม ในทางกลับกันก็เป็นจริงเช่นกัน: Rust จะหยุดพักบล็อก async และส่งมอบการควบคุมกลับคืนสู่รันไทม์ เฉพาะ ณ จุดรอเท่านั้น ทุกอย่างที่อยู่ระหว่างจุดรอจะเป็นแบบซิงโครนัส

นั่นหมายความว่าหากคุณทำงานกองโตในบล็อก async โดยไม่มีจุดรอ ฟิวเจอร์นั้นจะบล็อกฟิวเจอร์อื่นๆ ไม่ให้มีความคืบหน้า บางครั้งคุณอาจได้ยินคำเรียกสิ่งนี้ว่า ฟิวเจอร์หนึ่งกำลัง ทำให้อดอาหาร (starving) ฟิวเจอร์อื่นๆ ในบางกรณี สิ่งนี้อาจไม่ใช่เรื่องใหญ่นัก อย่างไรก็ตาม หากคุณกำลังทำการตั้งค่าที่มีราคาแพงหรือการทำงานที่ใช้เวลานาน หรือหากคุณมีฟิวเจอร์ที่จะทำแทสก์เฉพาะบางอย่างไปเรื่อยๆ อย่างไม่มีที่สิ้นสุด คุณจะต้องคิดถึงเวลาและสถานที่ที่จะคืนการควบคุมกลับไปยังรันไทม์

ลองจำลองการทำงานที่ใช้เวลานานเพื่ออธิบายปัญหาการอดอาหาร (starvation) จากนั้นสำรวจวิธีแก้ไข โค้ดตัวอย่างที่ 17-14 แนะนำฟังก์ชัน slow

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        // We will call `slow` here later
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

โค้ดนี้ใช้ std::thread::sleep แทน trpl::sleep เพื่อให้การเรียกใช้ slow บล็อกเธรดปัจจุบันเป็นเวลาหลายมิลลิวินาที เราสามารถใช้ slow แทนการทำงานในโลกจริงที่มีทั้งความยาวนานและเป็นแบบบล็อก

ในโค้ดตัวอย่างที่ 17-15 เราใช้ slow เพื่อจำลองการทำงานแบบมีขีดจำกัดซีพียู (CPU-bound) ในฟิวเจอร์สคู่หนึ่ง

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            slow("a", 10);
            slow("a", 20);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            slow("b", 10);
            slow("b", 15);
            slow("b", 350);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

ฟิวเจอร์แต่ละตัวจะส่งมอบการควบคุมกลับคืนสู่รันไทม์ หลังจาก ดำเนินการทำงานที่เชื่องช้าจำนวนหนึ่งเสร็จสิ้นแล้วเท่านั้น หากคุณรันโค้ดนี้ คุณจะเห็นผลลัพธ์นี้:

'a' started.
'a' ran for 30ms
'a' ran for 10ms
'a' ran for 20ms
'b' started.
'b' ran for 75ms
'b' ran for 10ms
'b' ran for 15ms
'b' ran for 350ms
'a' finished.

เช่นเดียวกับในโค้ดตัวอย่างที่ 17-5 ที่เราใช้ trpl::select เพื่อแข่งฟิวเจอร์สในการดึง URL สองตัว select ยังคงทำงานเสร็จสิ้นทันทีที่ a เสร็จสิ้น ทว่า ไม่มีการสลับกันทำงานระหว่างการเรียกใช้ slow ในฟิวเจอร์สทั้งสองตัว ฟิวเจอร์ a ทำงานทั้งหมดของมันจนกระทั่งการเรียกใช้ trpl::sleep ถูกรอคอย จากนั้นฟิวเจอร์ b จึงทำงานทั้งหมดของมันจนกระทั่งการเรียกใช้ trpl::sleep ของตัวมันเองถูกรอคอย และในที่สุดฟิวเจอร์ a ก็เสร็จสมบูรณ์ ในการอนุญาตให้ฟิวเจอร์สทั้งสองตัวสร้างความคืบหน้าระหว่างแทสก์ที่เชื่องช้าของพวกมัน เราจำเป็นต้องมีจุดรอเพื่อส่งมอบการควบคุมกลับไปยังรันไทม์ นั่นหมายความว่าเราต้องการบางอย่างที่เราสามารถรอคอย (await) ได้!

เราสามารถเห็นการส่งมอบประเภทนี้เกิดขึ้นแล้วในโค้ดตัวอย่างที่ 17-15: หากเราลบ trpl::sleep ที่ส่วนท้ายของฟิวเจอร์ a ออก มันก็จะทำงานเสร็จสิ้นโดยที่ฟิวเจอร์ b ไม่ได้รัน เลยแม้แต่น้อย ลองใช้ฟังก์ชัน trpl::sleep เป็นจุดเริ่มต้นในการปล่อยให้การทำงานต่างๆ สลับกันสร้างความคืบหน้า ดังแสดงในโค้ดตัวอย่างที่ 17-16

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let one_ms = Duration::from_millis(1);

        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::sleep(one_ms).await;
            slow("a", 10);
            trpl::sleep(one_ms).await;
            slow("a", 20);
            trpl::sleep(one_ms).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::sleep(one_ms).await;
            slow("b", 10);
            trpl::sleep(one_ms).await;
            slow("b", 15);
            trpl::sleep(one_ms).await;
            slow("b", 350);
            trpl::sleep(one_ms).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

เราได้เพิ่มการเรียกใช้ trpl::sleep พร้อมกับจุดรอไว้ระหว่างการเรียกใช้ slow แต่ละครั้ง ตอนนี้งานของฟิวเจอร์สทั้งสองสลับกันทำแล้ว:

'a' started.
'a' ran for 30ms
'b' started.
'b' ran for 75ms
'a' ran for 10ms
'b' ran for 10ms
'a' ran for 20ms
'b' ran for 15ms
'a' finished.

ฟิวเจอร์ a ยังคงรันอยู่ครู่หนึ่งก่อนจะส่งมอบการควบคุมให้ b เนื่องจากมันเรียกใช้ slow ก่อนที่จะเรียกใช้ trpl::sleep แต่หลังจากนั้น ฟิวเจอร์สจะสลับกันไปมาในแต่ละครั้งที่ตัวหนึ่งชนเข้ากับจุดรอ ในกรณีนี้ เราทำเช่นนั้นหลังจากการเรียกใช้ slow ทุกๆ ครั้ง แต่เราสามารถแบ่งแยกงานในรูปแบบใดก็ตามที่สมเหตุสมผลที่สุดสำหรับเราได้

อย่างไรก็ตาม เราไม่ได้ต้องการจะ นอนหลับ (sleep) ในที่นี้จริงๆ: เราต้องการก้าวหน้าให้เร็วที่สุดเท่าที่จะทำได้ เราเพียงต้องการส่งมอบการควบคุมกลับไปยังรันไทม์ เราสามารถทำได้โดยตรง โดยใช้ฟังก์ชัน trpl::yield_now ในโค้ดตัวอย่างที่ 17-17 เราแทนที่การเรียกใช้ trpl::sleep ทั้งหมดเหล่านั้นด้วย trpl::yield_now

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::yield_now().await;
            slow("a", 10);
            trpl::yield_now().await;
            slow("a", 20);
            trpl::yield_now().await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::yield_now().await;
            slow("b", 10);
            trpl::yield_now().await;
            slow("b", 15);
            trpl::yield_now().await;
            slow("b", 350);
            trpl::yield_now().await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

โค้ดนี้มีความชัดเจนมากกว่าเกี่ยวกับเจตนาที่แท้จริง และสามารถทำงานได้เร็วกว่าการใช้ sleep อย่างมีนัยสำคัญ เนื่องจากตัวจับเวลา เช่น ตัวที่ใช้โดย sleep มักมีขีดจำกัดว่าสามารถละเอียดได้เพียงใด เวอร์ชันของ sleep ที่เราใช้อยู่ ตัวอย่างเช่น จะนอนหลับอย่างน้อยหนึ่งมิลลิวินาทีเสมอ แม้ว่าเราจะส่งค่า Duration หนึ่งนาโนวินาทีให้มันก็ตาม อีกครั้ง คอมพิวเตอร์สมัยใหม่นั้น เร็ว มาก: พวกมันสามารถทำสิ่งต่างๆ ได้มากมายในหนึ่งมิลลิวินาที!

สิ่งนี้หมายความว่า async สามารถมีประโยชน์แม้กระทั่งสำหรับงานที่มีขีดจำกัดการคำนวณ (compute-bound) ขึ้นอยู่กับว่าโปรแกรมของคุณกำลังทำอะไรอย่างอื่นอยู่ เนื่องจากมันจัดหาเครื่องมือที่มีประโยชน์สำหรับการวางโครงสร้างความสัมพันธ์ระหว่างส่วนต่างๆ ของโปรแกรม (แต่นำมาซึ่งค่าใช้จ่ายของโหลดส่วนเกินของสเตตแมชชีน async) สิ่งนี้คือรูปแบบหนึ่งของ การทำงานหลายงานร่วมกันอย่างสอดประสาน (cooperative multitasking) ซึ่งฟิวเจอร์แต่ละตัวมีอำนาจในการตัดสินใจว่าจะส่งมอบการควบคุมเมื่อใดผ่านจุดรอ ฟิวเจอร์แต่ละตัวจึงมีความรับผิดชอบในการหลีกเลี่ยงการบล็อกไว้นานเกินไป ในระบบปฏิบัติการแบบฝังตัว (embedded OS) ที่ใช้ Rust บางระบบ สิ่งนี้คือรูปแบบเดียวของการทำงานหลายงาน!

แน่นอนว่าในโค้ดโลกความเป็นจริง คุณจะไม่ได้สลับการเรียกใช้ฟังก์ชันกับจุดรอในทุกๆ บรรทัด แม้ว่าการคืนการควบคุมในลักษณะนี้จะมีราคาค่อนข้างถูก แต่ก็ไม่ได้ฟรี ในหลายๆ กรณี การพยายามแตกงานที่มีขีดจำกัดการคำนวณออกอาจทำให้มันช้าลงอย่างมีนัยสำคัญ ดังนั้นบางครั้งการปล่อยให้การทำงานบล็อกชั่วคราวสั้นๆ อาจดีกว่าสำหรับประสิทธิภาพ โดยรวม ให้ทำการวัดผลเสมอเพื่อดูว่าคอขวดด้านประสิทธิภาพที่แท้จริงของโค้ดคุณคืออะไร อย่างไรก็ตาม กลไกเบื้องหลังเป็นสิ่งสำคัญที่ต้องระลึกไว้ หากคุณ กำลัง เห็นงานจำนวนมากเกิดขึ้นเป็นอนุกรมซึ่งคุณคาดหวังให้เกิดขึ้นแบบคอนเคอร์เรนต์!

การสร้างนามธรรม Async ของเราเอง

เรายังสามารถประกอบฟิวเจอร์สเข้าด้วยกันเพื่อสร้างรูปแบบใหม่ๆ ตัวอย่างเช่น เราสามารถสร้างฟังก์ชัน timeout ด้วยบล็อกสร้าง async ที่เรามีอยู่แล้ว เมื่อเราทำเสร็จ ผลลัพธ์จะเป็นบล็อกสร้างอีกชิ้นหนึ่งที่เราสามารถใช้สร้างนามธรรม async เพิ่มเติมได้อีก

โค้ดตัวอย่างที่ 17-18 แสดงวิธที่เราคาดหวังให้ timeout นี้ทำงานกับฟิวเจอร์ที่เชื่องช้า

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

ลองมาสร้างสิ่งนี้กัน! ในการเริ่มต้น ลองคิดถึง API สำหรับ timeout:

• มันจำเป็นต้องเป็นฟังก์ชัน async ในตัวเองเพื่อที่เราจะสามารถรอคอยมันได้
• พารามิเตอร์แรกของมันควรเป็นฟิวเจอร์ที่จะรัน เราสามารถทำให้มันเป็นเจเนอริก (generic) เพื่ออนุญาตให้มันทำงานร่วมกับฟิวเจอร์ใดๆ ก็ได้
• พารามิเตอร์ที่สองของมันจะเป็นเวลาสูงสุดที่จะรอ หากเราใช้ Duration สิ่งนั้นจะทำให้ง่ายต่อการส่งต่อไปยัง trpl::sleep
• มันควรจะคืนค่าเป็น Result หากฟิวเจอร์ทำงานเสร็จสมบูรณ์ ค่า Result จะเป็น Ok พร้อมกับค่าที่สร้างโดยฟิวเจอร์ หากเวลาของ timeout หมดลงก่อน ค่า Result จะเป็น Err พร้อมกับระยะเวลาที่ timeout ได้รอไป

โค้ดตัวอย่างที่ 17-19 แสดงการประกาศนี้

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    // Here is where our implementation will go!
}

สิ่งนั้นตอบสนองเป้าหมายของเราสำหรับชนิดข้อมูลแล้ว คราวนี้ลองมาคิดถึง พฤติกรรม ที่เราต้องการ: เราต้องการแข่งฟิวเจอร์ที่ส่งเข้ามาเทียบกับระยะเวลา เราสามารถใช้ trpl::sleep เพื่อสร้างฟิวเจอร์ตัวจับเวลาจากระยะเวลา และใช้ trpl::select เพื่อรันตัวจับเวลานั้นร่วมกับฟิวเจอร์ที่ผู้เรียกส่งเข้ามา

ในโค้ดตัวอย่างที่ 17-20 เราสร้าง timeout โดยใช้การ match บนผลลัพธ์ของการรอคอย trpl::select

extern crate trpl; // required for mdbook test

use std::time::Duration;

use trpl::Either;

// --snip--

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    match trpl::select(future_to_try, trpl::sleep(max_time)).await {
        Either::Left(output) => Ok(output),
        Either::Right(_) => Err(max_time),
    }
}

การใช้ trpl::select ไม่ได้มีความยุติธรรม (not fair): มันจะโพล (poll) อาร์กิวเมนต์ตามลำดับที่ถูกส่งเข้าไปเสมอ (การใช้อื่นๆ ของ select จะสุ่มเลือกอาร์กิวเมนต์ที่จะโพลก่อน) ดังนั้น เราจึงส่ง future_to_try ไปยัง select เป็นลำดับแรก เพื่อให้มันมีโอกาสทำงานเสร็จสมบูรณ์แม้ว่า max_time จะเป็นระยะเวลาที่สั้นมากก็ตาม หาก future_to_try ทำงานเสร็จก่อน select จะคืนค่าเป็น Left พร้อมกับผลลัพธ์จาก future_to_try หาก timer ทำงานเสร็จก่อน select จะคืนค่าเป็น Right พร้อมผลลัพธ์ของตัวจับเวลาซึ่งก็คือ ()

หาก future_to_try สำเร็จและเราได้ Left(output) เราจะคืนค่า Ok(output) หากตัวจับเวลาของ sleep หมดเวลาแทนและเราได้ Right(()) เราจะละเลย () ด้วย _ และคืนค่า Err(max_time) แทน

ด้วยเหตุนี้ เราจึงมี timeout ที่ทำงานได้ ซึ่งถูกสร้างขึ้นมาจากตัวช่วย async อีกสองตัว เนื่องจากฟิวเจอร์สสามารถประกอบเข้ากับฟิวเจอร์สอื่นได้ คุณจึงสามารถสร้างเครื่องมือที่ทรงพลังอย่างยิ่งได้โดยใช้บล็อกสร้าง async ที่มีขนาดเล็กกว่า ตัวอย่างเช่น คุณสามารถใช้แนวทางเดียวกันนี้ในการรวม timeout เข้ากับการลองใหม่ (retries) และนำพวกมันไปใช้กับการทำงาน เช่น การเรียกใช้เครือข่าย (ดังเช่นในโค้ดตัวอย่างที่ 17-5)

ในทางปฏิบัติ โดยทั่วไปคุณจะทำงานโดยตรงกับ async และ await และใช้ฟังก์ชันรอง เช่น select และมาโคร เช่น มาโคร join! เพื่อควบคุมวิธีที่ฟิวเจอร์สนอกสุดจะถูกประมวลผล

ตอนนี้เราได้เห็นหลายๆ วิธีในการทำงานกับฟิวเจอร์สหลายตัวพร้อมกันแล้ว ถัดไป เราจะดูว่าเราสามารถทำงานกับฟิวเจอร์สหลายตัวในลักษณะที่เป็นลำดับตามเวลาได้อย่างไรด้วย สตรีม (streams)