【翻译】RUST无锁编程

#![feature(box_raw)]

use std::ptr::{self, null_mut};
use std::sync::atomic::AtomicPtr;
use std::sync::atomic::Ordering::{Relaxed, Release, Acquire};

pub struct Stack<T> {
    head: AtomicPtr<Node<T>>,
}

struct Node<T> {
    data: T,
    next: *mut Node<T>,
}

impl<T> Stack<T> {
    pub fn new() -> Stack<T> {
        Stack {
            head: AtomicPtr::new(null_mut()),
        }
    }
}

impl<T> Stack<T> {
    pub fn pop(&self) -> Option<T> {
        loop {
            // take a snapshot
            let head = self.head.load(Acquire);

            // we observed the stack empty
            if head == null_mut() {
                return None
            } else {
                let next = unsafe { (*head).next };

                // if snapshot is still good, update from `head` to `next`
                if self.head.compare_and_swap(head, next, Release) == head {

                    // extract out the data from the now-unlinked node
                    // **NOTE**: leaks the node!
                    return Some(unsafe { ptr::read(&(*head).data) })
                }
            }
        }
    }
}

impl<T> Stack<T> {
    pub fn push(&self, t: T) {
        // allocate the node, and immediately turn it into a *mut pointer
        let n = Box::into_raw(Box::new(Node {
            data: t,
            next: null_mut(),
        }));
        loop {
            // snapshot current head
            let head = self.head.load(Relaxed);

            // update `next` pointer with snapshot
            unsafe { (*n).next = head; }

            // if snapshot is still good, link in new node
            if self.head.compare_and_swap(head, n, Release) == head {
                break
            }
        }
    }
}

// extract out the data from the now-unlinked node
let ret = Some(unsafe { ptr::read(&(*head).data) });

// free the node
mem::drop(Box::from_raw(head));

return ret

pub struct Guard { ... }
pub fn pin() -> Guard;

pub struct Owned<T> { ... }

impl<T> Owned<T> {
    pub fn new(t: T) -> Owned<T>;
}

impl<T> Deref for Owned<T> {
    type Target = T;
    ...
}
impl<T> DerefMut for Owned<T> { ... }

pub struct Shared<'a, T: 'a> { ... }

impl<'a, T> Copy for Shared<'a, T> { ... }
impl<'a, T> Clone for Shared<'a, T> { ... }

impl<'a, T> Deref for Shared<'a, T> {
   type Target = &'a T;
   ...
}

pub struct Atomic<T> { ... }

impl<T> Atomic<T> {
    /// Create a new, null atomic pointer.
    pub fn null() -> Atomic<T>;
}

impl<T> Atomic<T> {
    pub fn load<'a>(&self, ord: Ordering, _: &'a Guard) -> Option<Shared<'a, T>>;
}

pub fn load(&self, order: Ordering) -> *mut T

impl<T> Atomic<T> {
    pub fn store(&self, val: Option<Owned<T>>, ord: Ordering);
}

impl<T> Atomic<T> {
    pub fn store_and_ref<'a>(&self,
                             val: Owned<T>,
                             ord: Ordering,
                             _: &'a Guard)
                             -> Shared<'a, T>;
}

impl<T> Atomic<T> {
    pub fn store_shared(&self, val: Option<Shared<T>>, ord: Ordering);
}

impl<T> Atomic<T> {
    pub fn cas(&self,
               old: Option<Shared<T>>,
               new: Option<Owned<T>>,
               ord: Ordering)
               -> Result<(), Option<Owned<T>>>;
}

impl<T> Atomic<T> {
    pub fn cas_and_ref<'a>(&self,
                           old: Option<Shared<T>>,
                           new: Owned<T>,
                           ord: Ordering,
                           _: &'a Guard)
                           -> Result<Shared<'a, T>, Owned<T>>;

impl<T> Atomic<T> {
    pub fn cas_shared(&self,
                             old: Option<Shared<T>>,
                             new: Option<Shared<T>>,
                             ord: Ordering)
                             -> bool;
}

impl Guard {
    pub unsafe fn unlinked<T>(&self, val: Shared<T>);
}

use std::sync::atomic::Ordering::{Acquire, Release, Relaxed};
use std::ptr;

use crossbeam::mem::epoch::{self, Atomic, Owned};

pub struct TreiberStack<T> {
    head: Atomic<Node<T>>,
}

struct Node<T> {
    data: T,
    next: Atomic<Node<T>>,
}

impl<T> TreiberStack<T> {
    pub fn new() -> TreiberStack<T> {
        TreiberStack {
            head: Atomic::new()
        }
    }

    pub fn push(&self, t: T) {
        // allocate the node via Owned
        let mut n = Owned::new(Node {
            data: t,
            next: Atomic::new(),
        });

        // become active
        let guard = epoch::pin();

        loop {
            // snapshot current head
            let head = self.head.load(Relaxed, &guard);

            // update `next` pointer with snapshot
            n.next.store_shared(head, Relaxed);

            // if snapshot is still good, link in the new node
            match self.head.cas_and_ref(head, n, Release, &guard) {
                Ok(_) => return,
                Err(owned) => n = owned,
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        // become active
        let guard = epoch::pin();

        loop {
            // take a snapshot
            match self.head.load(Acquire, &guard) {
                // the stack is non-empty
                Some(head) => {
                    // read through the snapshot, *safely*!
                    let next = head.next.load(Relaxed, &guard);

                    // if snapshot is still good, update from `head` to `next`
                    if self.head.cas_shared(Some(head), next, Release) {
                        unsafe {
                            // mark the node as unlinked
                            guard.unlinked(head);

                            // extract out the data from the now-unlinked node
                            return Some(ptr::read(&(*head).data))
                        }
                    }
                }

                // we observed the stack empty
                None => return None
            }
        }
    }
}

use std::sync::atomic::Ordering;
extern crate crossbeam_epoch as epoch;
use epoch::{Atomic, Owned, Shared};
pub struct Node {
    value: i32,
    next: *const Node,
}
pub struct LockFreeStack {
    next: Atomic<Node>,
}
unsafe impl Sync for LockFreeStack {}
unsafe impl Send for LockFreeStack {}
impl LockFreeStack {
    pub fn new() -> LockFreeStack {
        LockFreeStack {
            next: Atomic::null(),
        }
    }
    pub fn push(&self, v: i32) {

        let guard = epoch::pin();

        let   mut new = Owned::new(Node {
            value: v,
            next: std::ptr::null(),
        });
        loop {
            let   old = self.next.load(Ordering::Relaxed,&guard);
            new.next=old.as_raw();

            match  self.next.compare_and_set(old, new, Ordering::Release,&guard){
                Ok(_)=>break,
                Err(e)=>{
                    new=e.new;
                    // spin_loop_hint();
                },
            };
        }
    }
    pub fn pop(&self) -> Option<i32> {
        let guard = epoch::pin();
        loop {
            let   old = self.next.load(std::sync::atomic::Ordering::Acquire,&guard);
            /*
            按照as_ref的文档说明,old的load不能是Relaxed,只要确保old的写的另一方是Release,就是安全的
            我们这里无论是Push还是Pop对于old的set访问用得都是Release,因此是安全的.
            */
            match unsafe{old.as_ref()}{
                None=>return None,
                Some(old2)=>{
                    let next=old2.next;

                    if self.next.compare_and_set(old, Shared::from(next),Ordering::Release,&guard).is_ok(){
                        unsafe {
                            /*
                            按照defer_destroy文档,只要我们保证old不会再其他线程使用就是安全的
                            而我们非常确信,这个old不会被其他线程使用,
                            因为这里是defer_destroy,所以解决了ABA问题 (https://en.wikipedia.org/wiki/ABA_problem)
                            */
                            guard.defer_destroy(old);
                        }
                        return Some(old2.value);
                    }
                    spin_loop_hint();
                }
            }

        }
    }
}
impl Drop for LockFreeStack {
    /*
     因为&mut self保证了不会有其他人同时操作这个Stack,因此可以放心的一个一个移除即可.
    */
    fn drop(&mut self) {
        let guard = epoch::pin();
        let mut next = self.next.load(Ordering::Relaxed,&guard).as_raw() as *mut Node;
        while !next.is_null() {
            /*
            这里的next确定是通过Owned分配的,所以没有安全问题
            并且drop持有的是mutable stack,只有一个线程可以访问,所以也没有并发问题.
            */
            let n = unsafe { Owned::from_raw(next) };
            // println!("drop {}", n.value);
            next = n.next as *mut  Node ;
        }
    }
}