פתרון תור בעלות וביטול

הקוד הבא הוא תוכנית מלאה ללא תלויות חיצוניות. המודול הפרטי של SPSC מסתיר את האחסון ואת הבעלים המשותף, ומייצר בדיוק endpoint אחד מכל צד. המספר N הוא מספר החריצים הפיזיים, והקיבולת השימושית היא N - 1 כדי להבחין בין מלא לריק.

mod spsc {
    use std::{cell::UnsafeCell, mem::MaybeUninit, sync::{Arc, atomic::{AtomicUsize, Ordering}}};
    struct Inner<T, const N: usize> {
        slots: [UnsafeCell<MaybeUninit<T>>; N],
        head: AtomicUsize,
        tail: AtomicUsize,
    }
    // Only one Producer and one Consumer exist; methods require &mut endpoint.
    // Release/acquire transfers slot ownership before conflicting accesses.
    unsafe impl<T: Send, const N: usize> Sync for Inner<T, N> {}
    pub struct Producer<T, const N: usize> { inner: Arc<Inner<T, N>> }
    pub struct Consumer<T, const N: usize> { inner: Arc<Inner<T, N>> }
    pub fn channel<T: Send, const N: usize>() -> (Producer<T, N>, Consumer<T, N>) {
        assert!(N >= 2);
        let inner = Arc::new(Inner {
            slots: std::array::from_fn(|_| UnsafeCell::new(MaybeUninit::uninit())),
            head: AtomicUsize::new(0), tail: AtomicUsize::new(0),
        });
        (Producer { inner: Arc::clone(&inner) }, Consumer { inner })
    }
    impl<T: Send, const N: usize> Producer<T, N> {
        pub fn try_push(&mut self, value: T) -> Result<(), T> {
            let tail = self.inner.tail.load(Ordering::Relaxed);
            let next = (tail + 1) % N;
            if next == self.inner.head.load(Ordering::Acquire) { return Err(value); }
            // This slot is free and exclusively owned by this producer.
            unsafe { (*self.inner.slots[tail].get()).write(value); }
            self.inner.tail.store(next, Ordering::Release);
            Ok(())
        }
    }
    impl<T: Send, const N: usize> Consumer<T, N> {
        pub fn try_pop(&mut self) -> Option<T> {
            let head = self.inner.head.load(Ordering::Relaxed);
            if head == self.inner.tail.load(Ordering::Acquire) { return None; }
            // Publication initialized this slot; only this consumer reads it.
            let value = unsafe { (*self.inner.slots[head].get()).assume_init_read() };
            self.inner.head.store((head + 1) % N, Ordering::Release);
            Some(value)
        }
    }
    impl<T, const N: usize> Drop for Inner<T, N> {
        fn drop(&mut self) {
            // Last Arc is gone, so no endpoint operation can still be active.
            let mut index = *self.head.get_mut();
            let tail = *self.tail.get_mut();
            while index != tail {
                unsafe { self.slots[index].get_mut().assume_init_drop(); }
                index = (index + 1) % N;
            }
        }
    }
}

use std::{collections::VecDeque, sync::{Arc, Condvar, Mutex}};
struct State<T> { queue: VecDeque<T>, closed: bool }
struct BlockingQueue<T> {
    state: Mutex<State<T>>,
    readable: Condvar,
    writable: Condvar,
    capacity: usize,
}
impl<T> BlockingQueue<T> {
    fn new(capacity: usize) -> Self {
        assert!(capacity > 0);
        Self { state: Mutex::new(State { queue: VecDeque::new(), closed: false }),
            readable: Condvar::new(), writable: Condvar::new(), capacity }
    }
    fn push(&self, value: T) -> Result<(), T> {
        let state = self.state.lock().expect("queue poisoned");
        let mut state = self.writable.wait_while(state,
            |s| !s.closed && s.queue.len() == self.capacity).expect("queue poisoned");
        if state.closed { return Err(value); }
        state.queue.push_back(value);
        drop(state);
        self.readable.notify_one();
        Ok(())
    }
    fn pop(&self) -> Option<T> {
        let state = self.state.lock().expect("queue poisoned");
        let mut state = self.readable.wait_while(state,
            |s| !s.closed && s.queue.is_empty()).expect("queue poisoned");
        let value = state.queue.pop_front();
        drop(state);
        self.writable.notify_one();
        value
    }
    fn close(&self) {
        self.state.lock().expect("queue poisoned").closed = true;
        self.readable.notify_all();
        self.writable.notify_all();
    }
    fn cancel(&self) -> VecDeque<T> {
        let mut state = self.state.lock().expect("queue poisoned");
        state.closed = true;
        let abandoned = std::mem::take(&mut state.queue);
        drop(state);
        self.readable.notify_all();
        self.writable.notify_all();
        abandoned
    }
}
fn main() {
    let (mut producer, mut consumer) = spsc::channel::<u64, 8>();
    let worker = std::thread::spawn(move || {
        for expected in 0..10_000 {
            loop {
                if let Some(value) = consumer.try_pop() {
                    assert_eq!(value, expected);
                    break;
                }
                std::thread::yield_now();
            }
        }
    });
    for value in 0..10_000 {
        let mut pending = value;
        loop {
            match producer.try_push(pending) {
                Ok(()) => break,
                Err(value) => { pending = value; std::thread::yield_now(); }
            }
        }
    }
    worker.join().unwrap();
    let queue = Arc::new(BlockingQueue::new(1));
    queue.push(String::from("queued")).unwrap();
    let waiting_queue = Arc::clone(&queue);
    let waiting = std::thread::spawn(move || waiting_queue.push(String::from("waiting")));
    let abandoned = queue.cancel();
    assert_eq!(abandoned.into_iter().collect::<Vec<_>>(), ["queued"]);
    assert_eq!(waiting.join().unwrap(), Err(String::from("waiting")));
    assert!(queue.pop().is_none());
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    struct Tracked(Arc<AtomicUsize>);
    impl Drop for Tracked { fn drop(&mut self) { self.0.fetch_add(1, Ordering::Relaxed); } }
    #[test]
    fn remaining_values_are_dropped_once() {
        let drops = Arc::new(AtomicUsize::new(0));
        let (mut producer, mut consumer) = spsc::channel::<Tracked, 4>();
        for _ in 0..3 { assert!(producer.try_push(Tracked(Arc::clone(&drops))).is_ok()); }
        let rejected = producer.try_push(Tracked(Arc::clone(&drops))).err().unwrap();
        drop(rejected);
        drop(consumer.try_pop().unwrap());
        drop(consumer);
        drop(producer);
        assert_eq!(drops.load(Ordering::Relaxed), 4);
    }
    #[test]
    fn close_drains_existing_work() {
        let queue = BlockingQueue::new(2);
        queue.push(1).unwrap();
        queue.push(2).unwrap();
        queue.close();
        assert_eq!(queue.push(3), Err(3));
        assert_eq!(queue.pop(), Some(1));
        assert_eq!(queue.pop(), Some(2));
        assert_eq!(queue.pop(), None);
    }
    #[test]
    fn wraparound_preserves_fifo() {
        let (mut producer, mut consumer) = spsc::channel::<u32, 3>();
        for value in 0..1000 {
            producer.try_push(value).unwrap();
            assert_eq!(consumer.try_pop(), Some(value));
            assert_eq!(consumer.try_pop(), None);
        }
    }
}

ה־release על tail מפרסם ערך מאותחל. ה־acquire של הצרכן מקבל את הפרסום לפני assume_init_read. בכיוון ההפוך, ה־release על head מתבצע רק אחרי שהצרכן הוציא את הערך, וה־acquire של המפיק מונע שימוש חוזר מוקדם בחריץ. האינדקס שכל צד עצמו כותב נקרא ב־Relaxed משום שאין כותב שני לאותו צד.

ה־unsafe impl Sync נשען על מגבלת ה־endpoints ועל הפרוטוקול הזה יחד. אם נוסיף Clone למפיק או נשנה את המתודה לקבל &self, נצטרך הוכחה אחרת משום שכמה קריאות יוכלו להתחרות על אותו tail. השדות נשארים פרטיים בתוך המודול כדי שלקוח בטוח לא יוכל לעקוף את המגבלה.

התור SPSC מציע פעולות ניסיון בלבד ואינו מספק הודעת disconnected. אם הצרכן נעלם, המפיק יכול למלא את הקיבולת שנותרה ואז לקבל את הערך בחזרה. אם המפיק נעלם, הצרכן יכול לנקז את מה שכבר פורסם ואז לקבל None. אין להסיק מ־None לבדו אם יגיע פריט בעתיד. הדוגמה הראשית יודעת מראש כמה פריטים צפויים; שירות אמיתי צריך פרוטוקול סיום נוסף.

ההשמדה של Inner מתרחשת רק אחרי ששני הבעלים המשותפים נעלמו, ולכן אין פעולה פעילה על החריצים. היא משמידה רק את הטווח שבין head ל־tail. כמו בהפשטות לימודיות אחרות, destructor של payload שנכנס לפאניקה יכול למנוע ניקוי מלא של שאר הערכים; בדיקות ההשמדה משתמשות ב־Drop שאינו נכנס לפאניקה ואינן טוענות להבטחת ניקוי מול כל כשל כזה.

התור החסום משתמש בתנאי סגירה תחת אותה נעילה של שני תנאי ההמתנה. ביטול מסמן סגירה, מוציא את התור בבעלות ומשחרר את הנעילה לפני שהערכים הנטושים עלולים להיות מושמדים. כך destructor שקורא לקוד אחר אינו רץ בתוך המקטע הקריטי של התור. המדיניות מול poison בדוגמה היא פאניקה מפורשת; מוצר שרוצה התאוששות צריך להוסיף חוזה תיקון.

בדיקת הביטול תקפה גם אם המפיק עוד לא הספיק להירדם כאשר הביטול מגיע: הוא יראה את מצב הסגירה ויחזיר את הערך. אם כבר נרדם, notify מעירה אותו והוא בודק את אותו תנאי מחדש. זהו היתרון של חיבור הביטול למצב ההמתנה במקום לדגל שנבדק רק לפני השליחה.

עבור MPMC אי אפשר לשמור את ההוכחה של מפיק יחיד. כמה מפיקים יכולים לבחור אותו חריץ, ולכן נדרש פרוטוקול הזמנה ופרסום מתאים, לעיתים עם מספרי רצף לכל תא. במבנה מקושר נצטרך גם reclamation כאשר צמתים מוסרים. סקירת ההרחבה צריכה לזהות נקודות ליניאריזציה, ABA וחיי צמתים, ולא רק להוסיף CAS סביב השורה שנכשלה.

כדי לבדוק ב־Loom נחליף גם את האטומים וגם את UnsafeCell בממשקים המכשוריים המתאימים, ונכתוב מודל קטן עם שניים או שלושה פריטים. החלפת האטומים בלבד אינה מאפשרת לכלי לראות כל גישה גולמית לאחסון. דוח ההוכחה יפריד בין המודל המוגבל, מבחני native והטיעון על מגבלת הבעלים.