The Rust Programming Language

การถ่ายโอนข้อมูลระหว่างเธรดด้วยการส่งข้อความ

หนึ่งในแนวทางที่ได้รับความนิยมเพิ่มขึ้นเรื่อยๆ ในการรับประกันคอนเคอร์เรนซีอย่างปลอดภัยคือการส่งข้อความ (message passing) ซึ่งเธรดหรือแอกเตอร์ (actors) สื่อสารกันโดยการส่งข้อความที่มีข้อมูลบรรจุอยู่หากัน ดังแนวคิดในสโลแกนจาก เอกสารประกอบของภาษา Go: “อย่าสื่อสารด้วยการแบ่งปันหน่วยความจำ แต่จงแบ่งปันหน่วยความจำด้วยการสื่อสาร”

ในการบรรลุคอนเคอร์เรนซีแบบส่งข้อความ ไลบรารีมาตรฐานของ Rust จัดหาการใช้งานแชนเนล (channels) ไว้ให้ แชนเนล (channel) เป็นแนวคิดทั่วไปในการเขียนโปรแกรมซึ่งเป็นช่องทางที่ส่งข้อมูลจากเธรดหนึ่งไปยังอีกเธรดหนึ่ง

คุณสามารถจินตนาการถึงแชนเนลในการเขียนโปรแกรมได้เหมือนกับทางน้ำที่มีทิศทาง เช่น ลำธารหรือแม่น้ำ หากคุณใส่เป็ดยางลงไปในแม่น้ำ มันจะลอยไปตามน้ำจนถึงปลายทางของทางน้ำนั้น

แชนเนลมีสองส่วน: ผู้ส่ง (transmitter) และผู้รับ (receiver) ฝั่งผู้ส่งคือตำแหน่งต้นน้ำที่คุณใส่เป็ดยางลงในแม่น้ำ และฝั่งผู้รับคือจุดที่เป็ดยางไปถึง ณ ปลายน้ำ โค้ดส่วนหนึ่งของคุณจะเรียกใช้เมธอดบนฝั่งผู้ส่งพร้อมกับข้อมูลที่คุณต้องการส่ง และโค้ดอีกส่วนหนึ่งจะตรวจสอบฝั่งผู้รับเพื่อคอยรับข้อความที่มาถึง แชนเนลจะถูกเรียกว่า ถูกปิด (closed) หากฝั่งผู้ส่งหรือผู้รับถูกคืนค่า (drop) ไป

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

ขั้นแรก ในโค้ดตัวอย่างที่ 16-6 เราจะสร้างแชนเนลขึ้นมาแต่ยังไม่ทำอะไรกับมัน โปรดสังเกตว่าโค้ดนี้จะยังไม่คอมไพล์เพราะ Rust ยังไม่สามารถบอกได้ว่าเราต้องการส่งค่าชนิดใดผ่านแชนเนล

use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();
}

เราสร้างแชนเนลใหม่โดยใช้ฟังก์ชัน mpsc::channel โดย mpsc ย่อมาจาก multiple producer, single consumer (ผู้ผลิตหลายราย ผู้บริโภครายเดียว) อธิบายสั้นๆ คือ วิธีที่ไลบรารีมาตรฐานของ Rust ใช้แชนเนลหมายความว่า แชนเนลสามารถมีฝั่ง ส่ง ได้หลายฝั่งซึ่งสร้างค่าขึ้นมา แต่มีฝั่ง รับ เพียงฝั่งเดียวเท่านั้นที่นำค่าเหล่านั้นไปใช้งาน นึกภาพสายธารหลายสายไหลมารวมกันเป็นแม่น้ำสายใหญ่สายเดียว: สิ่งใดก็ตามที่ถูกส่งลงมาจากลำธารสายใดก็ตามจะไปจบลงที่แม่น้ำสายเดียวในตอนท้าย เราจะเริ่มจากผู้ผลิตรายเดียวก่อนในตอนนี้ แต่เราจะเพิ่มผู้ผลิตหลายรายเมื่อเราทำให้ตัวอย่างนี้ทำงานได้แล้ว

ฟังก์ชัน mpsc::channel จะคืนค่าทูเพิล (tuple) ซึ่งสมาชิกแรกคือฝั่งส่ง—ผู้ส่ง (transmitter)—และสมาชิกที่สองคือฝั่งรับ—ผู้รับ (receiver) ตัวย่อ tx และ rx ถูกใช้กันทั่วไปในหลายสาขาเพื่อแทนคำว่า transmitter และ receiver ตามลำดับ ดังนั้นเราจึงตั้งชื่อตัวแปรตามนั้นเพื่อแสดงถึงแต่ละฝั่ง เราใช้คำสั่ง let ร่วมกับรูปแบบที่ทำการกระจายโครงสร้าง (destructures) ทูเพิล เราจะพูดถึงการใช้รูปแบบในคำสั่ง let และการกระจายโครงสร้างในบทที่ 19 สำหรับตอนนี้ ให้ทราบว่าการใช้คำสั่ง let ในลักษณะนี้เป็นวิธีที่สะดวกในการสกัดเอาชิ้นส่วนของทูเพิลที่คืนค่าโดย mpsc::channel ออกมา

ลองย้ายฝั่งส่งเข้าไปในเธรดที่สร้างขึ้นใหม่และสั่งให้มันส่งข้อความสตริงหนึ่งข้อความ เพื่อให้เธรดใหม่สื่อสารกับเธรดหลัก ดังที่แสดงในโค้ดตัวอย่างที่ 16-7 สิ่งนี้เหมือนกับการใส่เป็ดยางลงไปในแม่น้ำทางต้นน้ำ หรือการส่งข้อความแชทจากเธรดหนึ่งไปยังอีกเธรดหนึ่ง

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });
}

อีกครั้ง เราใช้ thread::spawn เพื่อสร้างเธรดใหม่ จากนั้นใช้ move เพื่อย้าย tx เข้าไปในโคลเชอร์เพื่อให้เธรดใหม่เป็นเจ้าของ tx เธรดใหม่จำเป็นต้องเป็นเจ้าของฝั่งส่งเพื่อที่จะสามารถส่งข้อความผ่านแชนเนลได้

ผู้ส่งมีเมธอด send ซึ่งรับเอาค่าที่เราต้องการส่ง เมธอด send คืนค่าชนิด Result<T, E> ดังนั้นหากผู้รับถูกคืนค่า (dropped) ไปแล้วและไม่มีที่ใดที่จะส่งค่าไป การทำงานของ send จะคืนค่าเป็นข้อผิดพลาด ในตัวอย่างนี้ เรากำลังเรียกใช้ unwrap เพื่อให้ตื่นตระหนก (panic) ในกรณีที่เกิดข้อผิดพลาด แต่ในแอปพลิเคชันจริง เราจะจัดการมันอย่างเหมาะสม: ย้อนกลับไปดูบทที่ 9 เพื่อทบทวนกลยุทธ์สำหรับการจัดการข้อผิดพลาดที่เหมาะสม

ในโค้ดตัวอย่างที่ 16-8 เราจะรับค่าจากผู้รับในเธรดหลัก สิ่งนี้เหมือนกับการเก็บเป็ดยางขึ้นมาจากน้ำที่ปลายแม่น้ำ หรือการได้รับข้อความแชท

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

ผู้รับมีเมธอดที่มีประโยชน์สองเมธอด: recv และ try_recv เราใช้ recv ซึ่งย่อมาจาก receive (รับ) ซึ่งจะทำการบล็อกการทำงานของเธรดหลักและรอจนกว่าค่าจะถูกส่งผ่านแชนเนลลงมา เมื่อค่าถูกส่งมาแล้ว recv จะคืนค่านั้นในรูป Result<T, E> เมื่อผู้ส่งปิดลง recv จะคืนค่าเป็นข้อผิดพลาดเพื่อส่งสัญญาณว่าไม่มีค่าอื่นเข้ามาอีกแล้ว

เมธอด try_recv จะไม่บล็อกการทำงาน แต่จะคืนค่า Result<T, E> ทันที: ได้ค่า Ok ที่บรรจุข้อความหากมีข้อความพร้อมใช้งาน และได้ค่า Err หากคราวนี้ไม่มีข้อความ การใช้ try_recv มีประโยชน์หากเธรดนี้มีงานอื่นต้องทำในขณะที่รอข้อความ: เราสามารถเขียนลูปที่เรียกใช้ try_recv เป็นระยะๆ จัดการข้อความหากมี และทำอะไรอย่างอื่นเป็นเวลาสั้นๆ ก่อนที่จะตรวจสอบอีกครั้ง

เราใช้ recv ในตัวอย่างนี้เพื่อความเรียบง่าย เราไม่มีงานอื่นให้เธรดหลักทำนอกเหนือจากการรอข้อความ ดังนั้นการบล็อกเธรดหลักจึงมีความเหมาะสมแล้ว

เมื่อเรารันโค้ดในโค้ดตัวอย่างที่ 16-8 เราจะเห็นค่าที่พิมพ์ออกมาจากเธรดหลัก:

Got: hi

เพอร์เฟกต์!

การถ่ายโอนความเป็นเจ้าของผ่านแชนเนล

กฎความเป็นเจ้าของมีบทบาทสำคัญในการส่งข้อความเพราะมันช่วยให้คุณเขียนโค้ดคอนเคอร์เรนต์ได้อย่างปลอดภัย การป้องกันข้อผิดพลาดในการโปรแกรมแบบคอนเคอร์เรนต์คือข้อดีของการคิดถึงเรื่องความเป็นเจ้าของตลอดการเขียนโปรแกรม Rust ลองมาทำกิจกรรมทดลองเพื่อดูว่าแชนเนลและความเป็นเจ้าของทำงานร่วมกันเพื่อป้องกันปัญหาอย่างไร: เราจะลองใช้ค่า val ในเธรดที่สร้างขึ้นใหม่ หลังจาก เราส่งมันลงไปในแชนเนลแล้ว ลองคอมไพล์โค้ดในโค้ดตัวอย่างที่ 16-9 เพื่อดูว่าทำไมโค้ดนี้ถึงไม่อนุญาต

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
        println!("val is {val}");
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

ในที่นี้ เราพยายามพิมพ์ val หลังจากเราส่งมันลงในแชนเนลผ่าน tx.send การอนุญาตให้ทำเช่นนี้เป็นความคิดที่ไม่ดี: เมื่อค่านำส่งไปยังอีกเธรดหนึ่งแล้ว เธรดนั้นอาจแก้ไขหรือทิ้ง (drop) ค่านั้นก่อนที่เราจะพยายามใช้ค่านั้นอีกครั้ง การแก้ไขของอีกเธรดหนึ่งอาจทำให้เกิดข้อผิดพลาดหรือผลลัพธ์ที่ไม่คาดคิดเนื่องจากข้อมูลที่ไม่สอดคล้องกันหรือไม่เหลืออยู่แล้ว อย่างไรก็ตาม Rust แสดงข้อผิดพลาดให้เราเห็นหากเราพยายามคอมไพล์โค้ดในโค้ดตัวอย่างที่ 16-9:

$ cargo run
   Compiling message-passing v0.1.0 (file:///projects/message-passing)
error[E0382]: borrow of moved value: `val`
  --> src/main.rs:10:27
   |
 8 |         let val = String::from("hi");
   |             --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(val).unwrap();
   |                 --- value moved here
10 |         println!("val is {val}");
   |                           ^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
error: could not compile `message-passing` (bin "message-passing") due to 1 previous error

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

การส่งหลายค่า

โค้ดในโค้ดตัวอย่างที่ 16-8 คอมไพล์และทำงานได้ แต่ก็ไม่ได้แสดงให้เราเห็นชัดเจนว่าสองเธรดที่แยกกันกำลังสื่อสารกันผ่านแชนเนล

ในโค้ดตัวอย่างที่ 16-10 เราได้ปรับแต่งบางอย่างที่จะพิสูจน์ว่าโค้ดในโค้ดตัวอย่างที่ 16-8 ทำงานแบบคอนเคอร์เรนต์: ตอนนี้เธรดใหม่จะส่งหลายข้อความและหยุดพักเป็นเวลาหนึ่งวินาทีระหว่างแต่ละข้อความ

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }
}

คราวนี้ เธรดใหม่มีเวกเตอร์ของสตริงที่เราต้องการส่งไปยังเธรดหลัก เราวนลูปผ่านพวกมัน ส่งทีละข้อความ และหยุดพักระหว่างแต่ละข้อความโดยการเรียกใช้ฟังก์ชัน thread::sleep พร้อมกับค่า Duration หนึ่งวินาที

ในเธรดหลัก เราไม่ได้เรียกใช้ฟังก์ชัน recv อย่างชัดเจนอีกต่อไป: แต่เราปฏิบัติกับ rx เหมือนเป็นตัวซ้ำ (iterator) สำหรับแต่ละค่าที่ได้รับ เราจะพิมพ์มันออกมา เมื่อแชนเนลปิดลง การวนลูปซ้ำก็จะสิ้นสุดลง

เมื่อรันโค้ดในโค้ดตัวอย่างที่ 16-10 คุณควรจะเห็นผลลัพธ์ต่อไปนี้โดยมีการหยุดพักหนึ่งวินาทีระหว่างแต่ละบรรทัด:

Got: hi
Got: from
Got: the
Got: thread

เนื่องจากเราไม่มีโค้ดใดๆ ที่หยุดพักหรือหน่วงเวลาในลูป for ในเธรดหลัก เราจึงสามารถบอกได้ว่าเธรดหลักกำลังรอรับค่าจากเธรดใหม่อยู่

การสร้างผู้ผลิตหลายราย

ก่อนหน้านี้เราได้กล่าวว่า mpsc เป็นตัวย่อมาจาก multiple producer, single consumer (ผู้ผลิตหลายราย ผู้บริโภครายเดียว) ลองนำ mpsc มาใช้งานและขยายโค้ดในโค้ดตัวอย่างที่ 16-10 เพื่อสร้างหลายๆ เธรดซึ่งทั้งหมดส่งค่าไปยังผู้รับรายเดียวกัน เราสามารถทำได้โดยการโคลน (clone) ผู้ส่ง ดังแสดงในโค้ดตัวอย่างที่ 16-11

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // --snip--

    let (tx, rx) = mpsc::channel();

    let tx1 = tx.clone();
    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    thread::spawn(move || {
        let vals = vec![
            String::from("more"),
            String::from("messages"),
            String::from("for"),
            String::from("you"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }

    // --snip--
}

คราวนี้ ก่อนที่เราจะสร้างเธรดใหม่ตัวแรก เราจะเรียกใช้ clone บนผู้ส่ง สิ่งนี้จะมอบผู้ส่งตัวใหม่ให้เราซึ่งเราสามารถส่งไปยังเธรดใหม่ตัวแรกได้ เราส่งผู้ส่งตัวดั้งเดิมไปยังเธรดใหม่ตัวที่สอง สิ่งนี้ทำให้เรามีสองเธรด ซึ่งแต่ละเธรดส่งข้อความต่างกันไปยังผู้รับรายเดียว

เมื่อคุณรันโค้ด ผลลัพธ์ของคุณควรจะมีลักษณะคล้ายกับสิ่งนี้:

Got: hi
Got: more
Got: from
Got: messages
Got: for
Got: the
Got: thread
Got: you

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

ตอนนี้เราได้ดูวิธีการทำงานของแชนเนลไปแล้ว ลองมาดูวิธีอื่นสำหรับคอนเคอร์เรนซีกันบ้าง