The Rust Programming Language

คอนเคอร์เรนซีแบบแบ่งปันสถานะ (Shared-State Concurrency)

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

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

ในแง่หนึ่ง แชนเนลในภาษาเขียนโปรแกรมใดๆ จะคล้ายกับความเป็นเจ้าของแบบเดี่ยว (single ownership) เพราะเมื่อคุณส่งค่าผ่านแชนเนลลงมาแล้ว คุณไม่ควรใช้ค่านั้นอีก คอนเคอร์เรนซีแบบแบ่งปันหน่วยความจำ (Shared-memory concurrency) จะคล้ายกับความเป็นเจ้าของหลายราย (multiple ownership): เธรดหลายเธรดสามารถเข้าถึงตำแหน่งหน่วยความจำเดียวกันในเวลาเดียวกันได้ ดังที่คุณได้เห็นในบทที่ 15 ซึ่งพอยน์เตอร์อัจฉริยะทำให้ความเป็นเจ้าของหลายรายเป็นไปได้ ความเป็นเจ้าของหลายรายสามารถเพิ่มความซับซ้อนขึ้นได้เนื่องจากผู้เป็นเจ้าของต่างชนิดเหล่านี้จำเป็นต้องได้รับการจัดการ ระบบชนิดข้อมูลและกฎความเป็นเจ้าของของ Rust ช่วยได้เป็นอย่างมากในการทำให้การจัดการนี้ถูกต้อง สำหรับตัวอย่าง ลองมาดูมิวเทกซ์ (mutexes) ซึ่งเป็นหนึ่งในพริมิทิฟด้านคอนเคอร์เรนซีที่พบบ่อยสำหรับหน่วยความจำที่แบ่งปันกัน

การควบคุมการเข้าถึงด้วยมิวเทกซ์ (Mutexes)

Mutex (มิวเทกซ์) ย่อมาจาก mutual exclusion (การกันไม่ให้ทำพร้อมกัน) หมายความว่ามิวเทกซ์อนุญาตให้มีเพียงเธรดเดียวเท่านั้นที่สามารถเข้าถึงข้อมูลบางอย่าง ณ เวลาใดเวลาหนึ่งได้ ในการเข้าถึงข้อมูลในมิวเทกซ์ เธรดจะต้องส่งสัญญาณก่อนว่าต้องการเข้าถึงโดยการร้องขอขอรับล็อก (lock) ของมิวเทกซ์ ล็อก (lock) คือโครงสร้างข้อมูลที่เป็นส่วนหนึ่งของมิวเทกซ์ที่คอยติดตามว่าใครมีสิทธิ์เข้าถึงข้อมูลเพียงผู้เดียวในปัจจุบัน ดังนั้น มิวเทกซ์จึงถูกอธิบายว่าทำหน้าที่ คุ้มกัน (guarding) ข้อมูลที่มันถืออยู่ผ่านระบบล็อก

มิวเทกซ์มีชื่อเสียงในเรื่องความยากในการใช้งานเพราะคุณต้องจำกฎสองข้อ:

1. คุณต้องพยายามขอรับล็อกก่อนที่จะใช้ข้อมูล
2. เมื่อคุณใช้งานข้อมูลที่มิวเทกซ์คุ้มกันอยู่เสร็จแล้ว คุณต้องปลดล็อก (unlock) ข้อมูล เพื่อให้เธรดอื่นสามารถรับล็อกไปได้

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

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

API ของ Mutex<T>

เพื่อเป็นตัวอย่างวิธีใช้นิวเทกซ์ เรามาเริ่มจากการใช้นิวเทกซ์ในบริบทเธรดเดียวเพื่อความเรียบง่าย ดังแสดงในโค้ดตัวอย่างที่ 16-12

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

    println!("m = {m:?}");
}

เช่นเดียวกับหลายชนิดข้อมูล เราสร้าง Mutex<T> โดยใช้ฟังก์ชันเกี่ยวเนื่อง (associated function) new ในการเข้าถึงข้อมูลภายในมิวเทกซ์ เราใช้เมธอด lock เพื่อรับล็อก การเรียกใช้นี้จะบล็อกเธรดปัจจุบันทำให้ไม่สามารถทำงานใดๆ ได้จนกว่าจะถึงตาเราที่ได้ครอบครองล็อก

การเรียกใช้ lock จะล้มเหลวหากเธรดอื่นที่ถือล็อกอยู่เกิดตื่นตระหนก (panic) ในกรณีนั้น จะไม่มีใครสามารถรับล็อกได้อีกเลย ดังนั้นเราจึงเลือกที่จะใช้ unwrap เพื่อให้เธรดนี้ตื่นตระหนกหากอยู่ในสถานการณ์นั้น

หลังจากที่เราได้รับล็อกแล้ว เราสามารถปฏิบัติต่อค่าที่ส่งกลับมา ซึ่งในกรณีนี้ตั้งชื่อว่า num เหมือนเป็นอ้างอิงแบบเปลี่ยนแปลงได้ (mutable reference) ไปยังข้อมูลที่อยู่ภายใน ระบบชนิดข้อมูลช่วยรับประกันว่าเราจะได้รับล็อกก่อนที่จะใช้ค่าใน m ชนิดข้อมูลของ m คือ Mutex<i32> ไม่ใช่ i32 ดังนั้นเรา ต้อง เรียกใช้ lock เพื่อที่จะใช้ค่า i32 ได้ เราไม่สามารถลืมได้ เพราะระบบชนิดข้อมูลจะไม่ยอมให้เราเข้าถึง i32 ที่อยู่ภายในเด็ดขาด

การเรียกใช้ lock จะคืนค่าชนิดข้อมูลที่เรียกว่า MutexGuard ซึ่งถูกห่อหุ้มอยู่ใน LockResult ที่เราจัดการด้วยการเรียกใช้ unwrap ชนิดข้อมูล MutexGuard ใช้เทรต Deref ชี้ไปยังข้อมูลภายในของเรา ชนิดข้อมูลนี้ยังมีการใช้เทรต Drop ซึ่งทำการปลดล็อกโดยอัตโนมัติเมื่อ MutexGuard ออกนอกขอบเขต ซึ่งเกิดขึ้นเมื่อสิ้นสุดขอบเขตภายใน (inner scope) ส่งผลให้เราไม่มีความเสี่ยงที่จะลืมปลดล็อกและบล็อกมิวเทกซ์จากการถูกใช้งานโดยเธรดอื่น เพราะการปลดล็อกเกิดขึ้นโดยอัตโนมัติ

หลังจากปลดล็อกแล้ว เราสามารถพิมพ์ค่ามิวเทกซ์และเห็นได้ว่าเราสามารถเปลี่ยนค่า i32 ภายในเป็น 6 ได้

การเข้าถึง Mutex<T> ร่วมกัน

คราวนี้ลองมาแบ่งปันค่าวาลูระหว่างหลายๆ เธรดโดยใช้ Mutex<T> กัน เราจะสร้าง 10 เธรดและสั่งให้แต่ละเธรดเพิ่มค่าตัวนับขึ้นทีละ 1 ดังนั้นตัวนับจะเพิ่มจาก 0 เป็น 10 ตัวอย่างในโค้ดตัวอย่างที่ 16-13 จะเกิดข้อผิดพลาดในการคอมไพล์ และเราจะใช้ข้อผิดพลาดนั้นเพื่อเรียนรู้เพิ่มเติมเกี่ยวกับการใช้ Mutex<T> และวิธีที่ Rust ช่วยให้เราใช้งานมันได้อย่างถูกต้อง

use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Mutex::new(0);
    let mut handles = vec![];

    for _ in 0..10 {
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

เราสร้างตัวแปร counter เพื่อเก็บ i32 ภายใน Mutex<T> เช่นเดียวกับที่เราทำในโค้ดตัวอย่างที่ 16-12 จากนั้น เราสร้าง 10 เธรดโดยการวนลูปตามช่วงตัวเลข เราใช้ thread::spawn และมอบโคลเชอร์เดียวกันให้แก่เธรดทั้งหมด: โคลเชอร์ที่ย้ายตัวนับเข้าไปในเธรด ขอรับล็อกบน Mutex<T> โดยเรียกใช้เมธอด lock จากนั้นบวก 1 เข้ากับค่าในมิวเทกซ์ เมื่อเธรดรันโคลเชอร์เสร็จแล้ว num จะออกนอกขอบเขตและปลดล็อก เพื่อให้เธรดอื่นสามารถรับล็อกต่อไปได้

ในเธรดหลัก เราเก็บรวบรวมจอยแฮนเดิล (join handles) ทั้งหมด จากนั้นเช่นเดียวกับที่ทำในโค้ดตัวอย่างที่ 16-2 เราเรียกใช้ join บนแต่ละแฮนเดิลเพื่อให้แน่ใจว่าเธรดทั้งหมดรันเสร็จสิ้น เมื่อถึงจุดนั้น เธรดหลักจะรับล็อกและพิมพ์ผลลัพธ์ของโปรแกรมนี้

เราได้บอกเป็นนัยไว้แล้วว่าตัวอย่างนี้จะไม่คอมไพล์ คราวนี้ลองมาดูสาเหตุกัน!

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0382]: borrow of moved value: `counter`
  --> src/main.rs:21:29
   |
 5 |     let counter = Mutex::new(0);
   |         ------- move occurs because `counter` has type `std::sync::Mutex<i32>`, which does not implement the `Copy` trait
...
 8 |     for _ in 0..10 {
   |     -------------- inside of this loop
 9 |         let handle = thread::spawn(move || {
   |                                    ------- value moved into closure here, in previous iteration of loop
...
21 |     println!("Result: {}", *counter.lock().unwrap());
   |                             ^^^^^^^ value borrowed here after move

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

ข้อความแสดงข้อผิดพลาดระบุว่าค่า counter ถูกย้าย (moved) ไปแล้วในการวนลูปครั้งก่อนหน้า Rust กำลังบอกเราว่าเราไม่สามารถย้ายความเป็นเจ้าของของล็อก counter ไปยังหลายๆ เธรดได้ ลองมาแก้ข้อผิดพลาดคอมไพเลอร์ด้วยวิธีความเป็นเจ้าของหลายรายที่เราพูดถึงในบทที่ 15 กัน

ความเป็นเจ้าของหลายรายกับหลายเธรด

ในบทที่ 15 เรามอบค่าให้กับผู้เป็นเจ้าของหลายรายโดยใช้พอยน์เตอร์อัจฉริยะ Rc<T> เพื่อสร้างค่านับจำนวนการอ้างอิง (reference-counted value) ลองมาทำแบบเดียวกันที่นี่และดูว่าจะเกิดอะไรขึ้น เราจะห่อหุ้ม Mutex<T> ไว้ใน Rc<T> ในโค้ดตัวอย่างที่ 16-14 และทำการโคลน Rc<T> ก่อนที่จะย้ายความเป็นเจ้าของไปยังเธรด

use std::rc::Rc;
use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Rc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Rc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

อีกครั้ง เราคอมไพล์และได้… ข้อผิดพลาดอื่น! คอมไพเลอร์กำลังสอนเรามากมาย:

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0277]: `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
  --> src/main.rs:11:36
   |
11 |           let handle = thread::spawn(move || {
   |                        ------------- ^------
   |                        |             |
   |  ______________________|_____________within this `{closure@src/main.rs:11:36: 11:43}`
   | |                      |
   | |                      required by a bound introduced by this call
12 | |             let mut num = counter.lock().unwrap();
13 | |
14 | |             *num += 1;
15 | |         });
   | |_________^ `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
   |
   = help: within `{closure@src/main.rs:11:36: 11:43}`, the trait `Send` is not implemented for `Rc<std::sync::Mutex<i32>>`
note: required because it's used within this closure
  --> src/main.rs:11:36
   |
11 |         let handle = thread::spawn(move || {
   |                                    ^^^^^^^
note: required by a bound in `spawn`
  --> /rustc/ac68faa20c58cbccd01ee7208bf3b6e93a7d7f96/library/std/src/thread/functions.rs:125:0

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

ว้าว ข้อความแสดงข้อผิดพลาดนี้ยาวมาก! นี่คือส่วนสำคัญที่ควรโฟกัส: `Rc<Mutex<i32>>` cannot be sent between threads safely (ไม่สามารถส่ง Rc<Mutex<i32>> ระหว่างเธรดอย่างปลอดภัยได้) คอมไพเลอร์ยังบอกเหตุผลให้เราทราบด้วย: the trait `Send` is not implemented for `Rc<Mutex<i32>>` (เทรต Send ยังไม่ได้ถูกใช้อยู่บน Rc<Mutex<i32>>) เราจะพูดถึง Send ในหัวข้อถัดไป: มันคือหนึ่งในเทรตที่รับประกันว่าชนิดข้อมูลที่เราใช้กับเธรดมีไว้สำหรับการใช้งานในสถานการณ์แบบคอนเคอร์เรนต์

น่าเสียดายที่ Rc<T> ไม่ปลอดภัยในการแบ่งปันข้ามเธรด เมื่อ Rc<T> จัดการจำนวนการอ้างอิง มันจะเพิ่มจำนวนขึ้นสำหรับการเรียกใช้ clone แต่ละครั้ง และลบออกจากจำนวนเมื่อแต่ละโคลนถูกคืนค่า (dropped) แต่มันไม่ได้ใช้พริมิทิฟด้านคอนเคอร์เรนซีใดๆ เพื่อให้แน่ใจว่าการเปลี่ยนแปลงจำนวนนับจะไม่ถูกขัดจังหวะโดยเธรดอื่น สิ่งนี้อาจนำไปสู่จำนวนนับที่ไม่ถูกต้อง—บั๊กซับซ้อนที่แอบแฝง ซึ่งอาจนำไปสู่ปัญหาหน่วยความจำรั่วซึม (memory leaks) หรือค่าถูกคืนไปก่อนที่เราจะใช้งานเสร็จ สิ่งที่เราต้องการคือชนิดข้อมูลที่เหมือนกับ Rc<T> ทุกประการ แต่เปลี่ยนจำนวนการอ้างอิงในรูปแบบที่ปลอดภัยต่อเธรด (thread-safe)

การนับจำนวนการอ้างอิงแบบอะตอมมิกด้วย Arc<T>

โชคดีที่ Arc<T> เป็น ชนิดข้อมูลเช่นเดียวกับ Rc<T> ที่ปลอดภัยสำหรับใช้ในสถานการณ์คอนเคอร์เรนต์ ตัว a ย่อมาจาก atomic (อะตอมมิก) หมายความว่ามันคือชนิดข้อมูล นับจำนวนการอ้างอิงแบบอะตอมมิก (atomically reference-counted) อะตอมมิกส์ (Atomics) เป็นพริมิทิฟคอนเคอร์เรนซีชนิดเพิ่มเติมที่เราจะไม่ครอบคลุมในรายละเอียดที่นี่: ดูเอกสารประกอบไลบรารีมาตรฐานสำหรับ std::sync::atomic เพื่อดูรายละเอียดเพิ่มเติม ในจุดนี้ คุณเพียงต้องรู้ว่าอะตอมมิกส์ทำงานเหมือนชนิดข้อมูลพริมิทิฟ แต่ปลอดภัยที่จะแบ่งปันข้ามเธรด

คุณอาจสงสัยว่าทำไมชนิดข้อมูลพริมิทิฟทั้งหมดจึงไม่เป็นอะตอมมิก และทำไมชนิดข้อมูลในไลบรารีมาตรฐานจึงไม่ถูกใช้ Arc<T> เป็นค่าเริ่มต้น เหตุผลก็คือความปลอดภัยของเธรด (thread safety) มาพร้อมกับผลกระทบด้านประสิทธิภาพ (performance penalty) ที่คุณต้องการจ่ายเฉพาะเมื่อจำเป็นจริงๆ เท่านั้น หากคุณเพียงแค่การดำเนินการกับค่าต่างๆ ภายในเธรดเดียว โค้ดของคุณจะรันได้เร็วกว่าหากไม่ต้องบังคับใช้การรับประกันที่อะตอมมิกส์จัดหาให้

ลองย้อนกลับมาดูตัวอย่างของเรา: Arc<T> และ Rc<T> มี API เหมือนกัน ดังนั้นเราจึงแก้ไขโปรแกรมของเราโดยการเปลี่ยนบรรทัด use, การเรียกใช้ new, และการเรียกใช้ clone โค้ดในโค้ดตัวอย่างที่ 16-15 ในที่สุดก็จะคอมไพล์และทำงานได้

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

โค้ดนี้จะพิมพ์สิ่งต่อไปนี้:

Result: 10

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

โปรดสังเกตว่าหากคุณกำลังทำการคำนวณตัวเลขอภิปรายอย่างง่าย มีชนิดข้อมูลที่เรียบง่ายกว่าชนิดข้อมูล Mutex<T> ซึ่งจัดหาไว้โดย โมดูล std::sync::atomic ของไลบรารีมาตรฐาน ชนิดข้อมูลเหล่านี้มอบการเข้าถึงชนิดข้อมูลพริมิทิฟแบบอะตอมมิกที่ปลอดภัยและเป็นคอนเคอร์เรนต์ เราเลือกใช้ Mutex<T> ร่วมกับชนิดข้อมูลพริมิทิฟสำหรับตัวอย่างนี้ เพื่อให้เราสามารถโฟกัสไปที่วิธีการทำงานของ Mutex<T> ได้

การเปรียบเทียบ RefCell<T>/Rc<T> และ Mutex<T>/Arc<T>

คุณอาจสังเกตเห็นว่า counter เป็นสิ่งที่ไม่สามารถเปลี่ยนแปลงค่าได้ (immutable) แต่เรายังคงรับการอ้างอิงแบบเปลี่ยนแปลงได้ไปยังค่าภายในมันได้ หมายความว่า Mutex<T> มอบการแก้ไขภายใน (interior mutability) เช่นเดียวกับที่ตระกูล Cell ทำ ในลักษณะเดียวกับที่เราใช้ RefCell<T> ในบทที่ 15 เพื่อยอมให้เราปรับเปลี่ยนเนื้อหาภายใน Rc<T> ได้ เราใช้ Mutex<T> เพื่อปรับเปลี่ยนเนื้อหาภายใน Arc<T>

อีกรายละเอียดหนึ่งที่ควรสังเกตคือ Rust ไม่สามารถปกป้องคุณจากข้อผิดพลาดตรรกะทุกประเภทได้เมื่อคุณใช้ Mutex<T> ย้อนกลับไปในบทที่ 15 การใช้ Rc<T> มาพร้อมกับความเสี่ยงในการสร้างวงรอบการอ้างอิง (reference cycles) ซึ่งค่า Rc<T> สองค่าอ้างอิงถึงกันและกัน ทำให้เกิดหน่วยความจำรั่วซึม ในทำนองเดียวกัน Mutex<T> ก็มาพร้อมกับความเสี่ยงในการเกิด เดดล็อก (deadlocks) สิ่งเหล่านี้เกิดขึ้นเมื่อการทำงานจำเป็นต้องล็อกสองทรัพยากร และสองเธรดต่างถือครองล็อกคนละตัว ทำให้พวกมันรอนายของกันและกันตลอดไป หากคุณสนใจเรื่องเดดล็อก ลองสร้างโปรแกรม Rust ที่มีเดดล็อกดู จากนั้นค้นคว้ากลยุทธ์การลดปัญหาเดดล็อกสำหรับมิวเทกซ์ในภาษาใดก็ได้แล้วลองใช้ใน Rust เอกสารประกอบ API ไลบรารีมาตรฐานสำหรับ Mutex<T> และ MutexGuard มีข้อมูลที่มีประโยชน์ให้อ่าน

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