pin-list

This crate provides PinList, a safe Pin-based intrusive doubly linked list.

Example

A thread-safe unfair async mutex.

use pin_project_lite::pin_project;
use std::cell::UnsafeCell;
use std::future::Future;
use std::ops::Deref;
use std::ops::DerefMut;
use std::pin::Pin;
use std::task;
use std::task::Poll;
use pin_list::PinList;

type PinListTypes = dyn pin_list::Types<
    Id = pin_list::id::Checked,
    Protected = task::Waker,
    Removed = (),
    Unprotected = (),
>;

pub struct Mutex<T> {
    data: UnsafeCell<T>,
    inner: std::sync::Mutex<Inner>,
}

struct Inner {
    locked: bool,
    waiters: PinList<PinListTypes>,
}

unsafe impl<T> Sync for Mutex<T> {}

impl<T> Mutex<T> {
    pub fn new(data: T) -> Self {
        Self {
            data: UnsafeCell::new(data),
            inner: std::sync::Mutex::new(Inner {
                locked: false,
                waiters: PinList::new(pin_list::id::Checked::new()),
            }),
        }
    }
    pub fn lock(&self) -> Lock<'_, T> {
        Lock {
            mutex: self,
            node: pin_list::Node::new(),
        }
    }
}

pin_project! {
    pub struct Lock<'mutex, T> {
        mutex: &'mutex Mutex<T>,
        #[pin]
        node: pin_list::Node<PinListTypes>,
    }

    impl<T> PinnedDrop for Lock<'_, T> {
        fn drop(this: Pin<&mut Self>) {
            let this = this.project();
            let node = match this.node.initialized_mut() {
                // The future was cancelled before it could complete.
                Some(initialized) => initialized,
                // The future has completed already (or hasn't started); we don't have to do
                // anything.
                None => return,
            };

            let mut inner = this.mutex.inner.lock().unwrap();

            match node.reset(&mut inner.waiters) {
                // If we've cancelled the future like usual, just do that.
                (pin_list::NodeData::Linked(_waker), ()) => {}

                // Otherwise, we have been woken but aren't around to take the lock. To
                // prevent deadlocks, pass the notification on to someone else.
                (pin_list::NodeData::Removed(()), ()) => {
                    if let Ok(waker) = inner.waiters.cursor_front_mut().remove_current(()) {
                        drop(inner);
                        waker.wake();
                    }
                }
            }
        }
    }
}

impl<'mutex, T> Future for Lock<'mutex, T> {
    type Output = Guard<'mutex, T>;
    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        let mut this = self.project();

        let mut inner = this.mutex.inner.lock().unwrap();

        if let Some(mut node) = this.node.as_mut().initialized_mut() {
            // Check whether we've been woken up, only continuing if so.
            if let Err(node) = node.take_removed(&inner.waiters) {
                // If we haven't been woken, re-register our waker and pend.
                *node.protected_mut(&mut inner.waiters).unwrap() = cx.waker().clone();
                return Poll::Pending;
            }
        }

        // If the mutex is unlocked, mark it as locked and return the guard
        if !inner.locked {
            inner.locked = true;
            return Poll::Ready(Guard { mutex: this.mutex });
        }

        // Otherwise, re-register ourselves to be woken when the mutex is unlocked again
        inner.waiters.push_back(this.node, cx.waker().clone(), ());

        Poll::Pending
    }
}

pub struct Guard<'mutex, T> {
    mutex: &'mutex Mutex<T>,
}

impl<T> Deref for Guard<'_, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target {
        unsafe { &*self.mutex.data.get() }
    }
}
impl<T> DerefMut for Guard<'_, T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        unsafe { &mut *self.mutex.data.get() }
    }
}

impl<T> Drop for Guard<'_, T> {
    fn drop(&mut self) {
        let mut inner = self.mutex.inner.lock().unwrap();
        inner.locked = false;

        if let Ok(waker) = inner.waiters.cursor_front_mut().remove_current(()) {
            drop(inner);
            waker.wake();
        }
    }
}
#

License: MIT