การทดสอบโครงสร้างข้อมูลแบบพร้อมกันอย่างเหมาะสม

การทดสอบโครงสร้างข้อมูลแบบพร้อมกันอย่างเหมาะสม

หนึ่ง สอง สาม สอง

• สามารถใช้ loom ซึ่งเป็นไลบรารีของ Rust เพื่อทดสอบโครงสร้างข้อมูลแบบ lock-free ได้อย่างละเอียด
• มีตัวอย่างโค้ดตัวนับแบบพร้อมกันอย่างง่าย
• บั๊กของโค้ดคือการดำเนินการเพิ่มค่าไม่เป็นอะตอมมิก

use std::sync::atomic::{AtomicU32, Ordering::SeqCst};

#[derive(Default)]
pub struct Counter {
    value: AtomicU32,
}

impl Counter {
    pub fn increment(&self) {
        let value = self.value.load(SeqCst);
        self.value.store(value + 1, SeqCst);
    }

    pub fn get(&self) -> u32 {
        self.value.load(SeqCst)
    }
}

การทดสอบอย่างง่าย

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

#[test]
fn threaded_test() {
    let counter = Counter::default();
    let thread_count = 100;
    let increment_count = 100;

    std::thread::scope(|scope| {
        for _ in 0..thread_count {
            scope.spawn(|| {
                for _ in 0..increment_count {
                    counter.increment()
                }
            });
        }
    });

    assert_eq!(counter.get(), thread_count * increment_count);
}

การทดสอบเชิงคุณสมบัติ (PBT)

• ลองนำการทดสอบเชิงคุณสมบัติที่เหมาะกับการทดสอบ state machine มาใช้
• หากสามารถรันเธรดทีละขั้นด้วยมือได้ ก็จะสอดแทรกการทำงานระหว่าง load และ store ของอีกเธรดได้ง่าย

#[test]
fn state_machine_test() {
    arbtest::arbtest(|rng| {
        let mut state: i32 = 0;
        let step_count: usize = rng.int_in_range(0..=100)?;

        for _ in 0..step_count {
            match *rng.choose(&["inc", "dec"])? {
                "inc" => state += 1,
                "dec" => state -= 1,
                _ => unreachable!(),
            }
        }
        Ok(())
    });
}

การใส่ instrumentation แบบง่าย

• วิธีทำให้เธรดสามารถ "หยุดชั่วคราว" ระหว่างการดำเนินการอะตอมมิก

pub fn increment(&self) {
    pause();
    let value = self.value.load(SeqCst);
    pause();
    self.value.store(value + 1, SeqCst);
    pause();
}

fn pause() {
    // ¯\\_(ツ)_/¯
}

Managed thread API

• กฎข้อหนึ่งของการออกแบบ API คือเริ่มจากผู้ใช้เพียงคนเดียวเพื่อทำความเข้าใจความรู้สึกของ API ก่อน แล้วค่อยลงมือทำ implementation จริง
• เขียนการทดสอบเชิงคุณสมบัติโดยใช้ managed thread

let counter = Counter::default();
let t1 = managed_thread::spawn(&counter);
let t2 = managed_thread::spawn(&counter);

while !rng.is_empty() {
    let coin_flip: bool = rng.arbitrary()?;
    if t1.is_paused() {
        if coin_flip {
            t1.unpause();
        }
    } else if t2.is_paused() {
        if coin_flip {
            t2.unpause();
        }
    }
}

การทำงานของ managed thread

• จำเป็นต้องมีการสื่อสารระหว่างเธรดควบคุมกับ managed thread
• ใช้ mutex และ condition variable ที่ปกป้องสถานะในการติดตั้งใช้งาน

struct SharedContext {
    state: Mutex<State>,
    cv: Condvar,
}

#[derive(PartialEq, Eq, Default)]
enum State {
    #[default]
    Running,
    Paused,
}

impl SharedContext {
    fn pause(&self) {
        let mut guard = self.state.lock().unwrap();
        assert_eq!(*guard, State::Running);
        *guard = State::Paused;
        self.cv.notify_all();
        guard = self.cv.wait_while(guard, |state| *state == State::Paused).unwrap();
        assert_eq!(*guard, State::Running);
    }
}

รวมโค้ดทั้งหมดเข้าด้วยกัน

• รวม managed thread และโค้ดทดสอบเข้าด้วยกัน

#[test]
fn test_counter() {
    arbtest::arbtest(|rng| {
        eprintln!("begin trace");
        let counter = Counter::default();
        let mut counter_model: u32 = 0;

        std::thread::scope(|scope| {
            let t1 = managed_thread::spawn(scope, &counter);
            let t2 = managed_thread::spawn(scope, &counter);
            let mut threads = [t1, t2];

            while !rng.is_empty() {
                for (tid, t) in threads.iter_mut().enumerate() {
                    if rng.arbitrary()? {
                        if t.is_paused() {
                            eprintln!("{tid}: unpause");
                            t.unpause()
                        } else {
                            eprintln!("{tid}: increment");
                            t.submit(|c| c.increment());
                            counter_model += 1;
                        }
                    }
                }
            }

            for t in threads {
                t.join();
            }
            assert_eq!(counter_model, counter.get());

            Ok(())
        })
    });
}

สรุปโดย GN⁺

• บทความนี้อธิบายวิธีทดสอบโครงสร้างข้อมูลแบบพร้อมกัน
• สำรวจวิธีใช้ไลบรารี loom ของ Rust เพื่อทดสอบการดำเนินการที่ไม่เป็นอะตอมมิก
• ใช้ managed thread เพื่อทดสอบปัญหาความพร้อมกันในรูปแบบที่ทำซ้ำได้และดีบักได้
• บทความนี้น่าจะเป็นประโยชน์สำหรับนักพัฒนาที่สนใจการเขียนโปรแกรมแบบพร้อมกัน
• โปรเจ็กต์ที่มีความสามารถคล้ายกันคือ JCStress ของ Java

[GN⁺]

ความคิดเห็นจาก Hacker News

• กำลังพัฒนาไลบรารีชื่อ Temper ด้วย Rust และต้องใช้ความพยายามอย่างมากเพื่อรับมือกับส่วนที่ซับซ้อนของโมเดลหน่วยความจำของ Rust

  • มีชุดกรณีทดสอบขนาดใหญ่ที่สุดชุดหนึ่งสำหรับโมเดลหน่วยความจำของ C++/Rust
  • Loom เป็นไลบรารีที่สมบูรณ์กว่าซึ่งช่วยให้ทดสอบโครงสร้างระดับสูงอย่าง mutex และ queue ได้อย่างละเอียด
  • ได้แรงบันดาลใจจากการบรรยายเรื่องการทดสอบของ Foundation DB และเชื่อว่า WebAssembly จะมีบทบาทสำคัญในด้านนี้

• ได้ทำ implementation ของ shared-memory atomic snapshot ใน Rust และมองว่าการทดสอบอัตโนมัติมีความสำคัญมาก

  • ตอนแรกใช้ Loom แต่ภายหลังเปลี่ยนไปใช้ Shuttle
  • Shuttle ใช้วิธีการแบบสุ่ม และให้การรับประกันเชิงความน่าจะเป็นในการค้นหาบั๊ก
  • Shuttle เร็วกว่าและขยายไปสู่สถานการณ์ทดสอบที่ซับซ้อนได้ดี
  • ความสามารถในการทำซ้ำการทดสอบที่ล้มเหลวมีความสำคัญมาก

• ข้อเสียของแนวทางนี้คือจำเป็นต้องปรับแก้โค้ดเองให้เข้ากับโค้ดทดสอบ

  • อาจทำได้อีกแบบด้วยการรันสองเธรดแล้วใช้ ptrace เพื่อสุ่มสลับลำดับการรันคำสั่งแบบ single-step

• Lincheck ของ JetBrains เป็นไลบรารีที่ดีในโลก Kotlin/Java

  • ชอบที่มันเป็นแบบ declarative และแสดงผลลัพธ์การ linearization ได้

• สงสัยว่ามีไลบรารีคล้าย "Loom" สำหรับ C++ หรือไม่

  • ต้องการทดสอบโครงสร้างข้อมูลแบบ lock-free

• แนวทางนี้อาจมีข้อจำกัดกับการรับประกันความคืบหน้าแบบ soft

  • ในลูป cmpxchg บนฮาร์ดแวร์จริงและ scheduler จริง โอกาสที่จะถูกขัดจังหวะมีต่ำมาก
  • แต่ในแนวทางการทดสอบนี้ ความน่าจะเป็นของความคืบหน้าจะแตกต่างไปตามจำนวนงานและจำนวนครั้งที่ถูกหยุดชั่วคราว

• ต้องมีความรู้เชิงปฏิบัติจริง และจำเป็นต้องสร้างเธรดจริง

  • สงสัยว่าสามารถใช้ async runtime ได้หรือไม่

• สามารถใช้ ptrace ให้เธรดรันแบบ single-step เพื่อสร้าง interleaving แบบต่าง ๆ ในระดับคำสั่งได้

  • สงสัยว่ามีทางเลือกสำหรับการทดสอบแบบ black-box หรือไม่

• การใช้ Loom จำเป็นต้องใช้ conditional compilation ซึ่งค่อนข้าง intrusive

  • สงสัยว่าภาษาอื่นจะเหมาะกว่าหรือไม่หากใช้ scheduler ของตัวเอง

• อยากรู้วิธีทำสิ่งเดียวกันนี้ใน Python

  • สงสัยว่าสามารถสร้างคลาส thread ที่ยอมให้ทดสอบลักษณะนี้ได้หรือไม่