maillon

A concurrent intrusive list for building synchronization primitives.

Maillon is the French word for a chain link.

Features

• 100% safe API
• #![no_std], no allocation
• Atomic emptiness check to avoid acquiring the mutex if the list is empty
• Lock-free insertion: multiple nodes can be inserted concurrently while another is being removed; removal requires locking
• Optional atomic state embedded in the list when empty (to carry a semaphore counter, a closed flag, etc.)
• WaitList, a high-level asynchronous wait list with customizable synchronization built on top of the low-level List

Usage

WaitList is a ready-to-use asynchronous wait list, built on top of List:

use std::sync::atomic::{AtomicBool, Ordering::Relaxed};

use maillon::WaitList;

#[derive(Default)]
pub struct Event {
    done: AtomicBool,
    wait_list: WaitList,
}

impl Event {
    pub async fn wait(&self) {
        let _ = self.wait_list.wait_until(|_| self.done.load(Relaxed)).await;
    }

    pub fn set(&self) {
        self.done.store(true, Relaxed);
        self.wait_list.notify_all();
    }
}

List is the building block: Nodes are pinned, carry user data implementing NodeData, and are pushed to the back without locking, while every other operation goes through List::lock. Here is a minimal wait list, supporting only notify_one:

use std::{
    future::Future,
    mem,
    pin::Pin,
    sync::atomic::{
        Ordering::{Relaxed, SeqCst},
        fence,
    },
    task::{Context, Poll, Waker},
};

use maillon::{List, Node, NodeData, NodeState, list::LockedList, node_wrapper};

#[derive(Default)]
pub struct WaitList {
    list: List<Waiter>,
}

#[derive(Default)]
struct Waiter {
    waker: Option<Waker>,
    notified: bool,
}

impl WaitList {
    pub fn notify_one(&self) {
        fence(SeqCst);
        if !self.list.is_empty(Relaxed) {
            Self::notify_one_cold(&self.list);
        }
    }

    #[cold]
    fn notify_one_cold(list: &List<Waiter>) {
        Self::notify_one_locked(list.lock());
    }

    fn notify_one_locked(mut locked: LockedList<'_, Waiter>) {
        let Some(mut front) = locked.front() else {
            return;
        };
        front.notified = true;
        let waker = front.waker.take();
        front.unlink();
        drop(locked);
        if let Some(waker) = waker {
            waker.wake();
        }
    }

    pub fn wait(&self) -> Wait<'_> {
        Wait(Node::new(&self.list))
    }
}

node_wrapper! {
    pub struct Wait<'a>(Node<&'a List<Waiter>>);
}

impl Future for Wait<'_> {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        match self.node_mut().state() {
            NodeState::Unlinked(mut node) => {
                if mem::take(&mut node.notified) {
                    return Poll::Ready(());
                }
                node.waker = Some(cx.waker().clone());
                node.push_back(Relaxed);
                fence(SeqCst);
                Poll::Pending
            }
            NodeState::Linked(mut node) => node.update_waker(cx, |node| &mut node.waker),
        }
    }
}

impl NodeData<&List<Waiter>> for Waiter {
    fn new_state_if_last_node_on_drop(
        self: Pin<&mut Self>,
        _list: &&List<Waiter>,
        _list_data: &mut (),
    ) {
    }

    fn on_drop<'list>(
        self: Pin<&mut Self>,
        list: &'list &List<Waiter>,
        locked: Option<LockedList<'list, Self>>,
        _state_updated_on_unlink: bool,
    ) {
        if self.notified {
            match locked {
                Some(locked) => WaitList::notify_one_locked(locked),
                None => WaitList::notify_one_cold(list),
            }
        }
    }
}

See examples for full implementations of tokio::sync::Notify and tokio::sync::Semaphore built with maillon, with fully identical API and behavior.

Cargo Features

Without any features enabled, the library falls back to spin-based mutex and parker.

Performance

Results of the tokio benchmarks, run with both tokio native primitives and their maillon counterparts from examples on an Intel i7-1065G7:

benchmarks prefixed by contention/uncontented¹ measure Semaphore performance

maillon-based reimplementations seem to give a consistent speedup compared to tokio native ones. The only exception is notify_waiters/500, and it can be explained by several factors:

• The benchmark results are extremely noisy, ranging from 200 µs to 400 µs, so maillon can in fact perform better than tokio on some runs.
• The scenario is not very realistic: all the threads are hammering the same cache line with CAS loops to requeue or notify in tight loops. The key point is that maillon doesn't use backoff in CAS loops, so they run in full-contention mode, while tokio's native implementation serializes all operations. Adding exponential backoff to the push_back operation improves the result down to 150 µs.
• CPU hyperthreading typically handles this kind of ultra-contended scenario badly. Pinning the process to 4 cores only, or reducing the number of worker threads to 3 in order to avoid hyperthreading also greatly improves the result. Combined with exponential backoff, time drops below 100 µs.

See benches/README.md for a comparison of maillon with other crates, still reusing their benchmarks. maillon-based implementations give significantly better results in almost all benchmarks.

Safety and testing

Concurrent intrusive lists are one of the most unsafe² concepts in Rust, so this crate uses unsafe code. It is tested with both miri and loom to ensure algorithm correctness and memory safety.

Reimplementations of tokio::sync::Notify and tokio::sync::Semaphore are also tested on the full tokio test suite (also with miri and loom).

List exposes a 100% safe API, so WaitList and tokio reimplementations don't use unsafe code³.

However, the default algorithm of List does not support strict provenance. This is a known issue of AtomicPtr based concurrent data structures.

Alternatives

The main goal of maillon was originally to provide a WaitList with notify_one as cheap as possible when there is no waiter, i.e. read-only, with a customizable synchronization strategy between the wake condition and the waker registration (SeqCst atomic operations vs. SeqCst fences vs. RMWs on the wake condition). Lock-free waiter insertion then came as an appreciable benefit on top of that.

Synchronization primitives

tokio::sync::Notify and maitake_sync::WaitQueue store a wakeup when there is no waiter, and thus can't be used for the WaitList use case.

futures_intrusive::sync::ManualResetEvent and asyncband::event::ManualResetEvent only provide the equivalent of notify_all, and are fully locked anyway.

Intrusive lists

maillon::List is also exported as a general-purpose concurrent intrusive list: nodes are pinned, and a node dropped while linked removes itself from the list, taking the lock only if it is still linked. Node unlinking on drop is the reason why the mutex must be embedded in the list to be able to provide a safe API. Most alternatives are not synchronized and must be wrapped into a mutex.

• pin-list: safe, but not synchronized, and aborts if a node is dropped without having been removed and taken out of the list.
• pinlist: safe, with the list embedding its mutex and nodes removing themselves on drop, like maillon; but every insertion and every drop takes the lock, and nodes can only leave the list by being dropped.

There are other crates providing/implementing intrusive lists, but whose lists own their nodes, such as intrusive-collections or cordyceps, and stack-allocated nodes requiring unsafe code. maillon has the opposite design: nodes borrow the list, which doesn't own them.

Related design

saa's design shares similarities with maillon: its synchronization primitives push waiters lock-free into their state word, which also holds a lock bit and a few data bits. However, its wait queue is private, the lock bit is a spinlock, dequeuing the front and canceling a waiter both walk the queue from its tail, and the data bits limit its semaphore to 63 permits.

Acknowledgements

The maillon::list::Drain algorithm reuses the idea originally introduced to tokio by Tymoteusz Wiśniewski in tokio-rs/tokio#5458: make the draining atomic by moving the list nodes into a temporary circular list.

A small improvement, motivated by API ergonomics, has been made: the circular chaining is deferred until the list lock actually needs to be released mid-drain.

License

Licensed under either of

• Apache License, Version 2.0
• MIT license

at your option.

Footnotes

1. The uncontented typo comes from the original tokio benchmark. ↩

2. There is literally a hack in the compiler to support them. ↩

3. Except for the pin projection of wait_list::wait::WaitUntil, written directly to avoid depending on pin-project-lite. ↩