feat: Add no_std implementation

This commit adds an implementation of the event-listener algorithm built
without locks. This is a replacement of the old no_std backend. It is
written without concurrent-queue and therefore closes #109.

The idea behind this implementation is to store the listeners in "slots"
in an infinitely large list. Then, assign each listener a number and
then use that to wait on each listener.

The list used is similar to the one used by the thread-local crate. It
consists of a list of "buckets" that hold slots. The number of slots
increases in an amortized way. The first slot holds 1, the second slot
holds 2, the third slot holds 4... all the way up to usize::MAX slots.

Indexes are done by having a list of reusable indexes and using those
when possible, only increasing the max index when necessary. This part
of the code could use some work; under contention it's possible to
make some slots unusuable. This can happen under two cases:

- If there is contention on the indexes list when the listener is being
  freed.
- If the slot is still being notified when it is attempted to be reused.

Both of these cases are probably fixable, and should be fixed before
release. Otherwise long running server processes using this code will
run out of memory under heavy loads.

From here the rest of the implementation is an atomic linked list based
on the above primitives. It functions very similarly to the std variant.
The main difference is that the Link structure's waker functions very
similarly to AtomicWaker from the atomic-waker crate. Aside from that
the code isn't very interesting on its own.

Signed-off-by: John Nunley <[email protected]>

Author: notgull

Cargo.toml

@@ -24,6 +24,8 @@ portable-atomic-crate = { version = "1.6.0", package = "portable-atomic", option
 portable-atomic-util = { version = "0.1.5", optional = true }
 
 [dev-dependencies]
+ahash = { version = "0.8.11", default-features = false, features = ["compile-time-rng"] }
 futures = { version = "0.3", default-features = false, features = ["std"] }
 futures-lite = "2.3.0"
+hashbrown = { version = "0.14.3", default-features = false }
 waker-fn = "1"

examples/mutex.rs

@@ -4,182 +4,196 @@
 
 #![allow(dead_code)]
 
-use std::cell::UnsafeCell;
-use std::ops::{Deref, DerefMut};
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{mpsc, Arc};
-use std::thread;
-use std::time::{Duration, Instant};
-
-use event_listener::{Event, Listener};
-
-/// A simple mutex.
-struct Mutex<T> {
-    /// Set to `true` when the mutex is locked.
-    locked: AtomicBool,
-
-    /// Blocked lock operations.
-    lock_ops: Event,
-
-    /// The inner protected data.
-    data: UnsafeCell<T>,
-}
+#[cfg(feature = "std")]
+mod ex {
+    use std::cell::UnsafeCell;
+    use std::ops::{Deref, DerefMut};
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::{mpsc, Arc};
+    use std::thread;
+    use std::time::{Duration, Instant};
+
+    use event_listener::{Event, Listener};
+
+    /// A simple mutex.
+    struct Mutex<T> {
+        /// Set to `true` when the mutex is locked.
+        locked: AtomicBool,
+
+        /// Blocked lock operations.
+        lock_ops: Event,
+
+        /// The inner protected data.
+        data: UnsafeCell<T>,
+    }
 
-unsafe impl<T: Send> Send for Mutex<T> {}
-unsafe impl<T: Send> Sync for Mutex<T> {}
+    unsafe impl<T: Send> Send for Mutex<T> {}
+    unsafe impl<T: Send> Sync for Mutex<T> {}
 
-impl<T> Mutex<T> {
-    /// Creates a mutex.
-    fn new(t: T) -> Mutex<T> {
-        Mutex {
-            locked: AtomicBool::new(false),
-            lock_ops: Event::new(),
-            data: UnsafeCell::new(t),
+    impl<T> Mutex<T> {
+        /// Creates a mutex.
+        fn new(t: T) -> Mutex<T> {
+            Mutex {
+                locked: AtomicBool::new(false),
+                lock_ops: Event::new(),
+                data: UnsafeCell::new(t),
+            }
         }
-    }
 
-    /// Attempts to acquire a lock.
-    fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
-        if !self.locked.swap(true, Ordering::Acquire) {
-            Some(MutexGuard(self))
-        } else {
-            None
+        /// Attempts to acquire a lock.
+        fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
+            if !self.locked.swap(true, Ordering::Acquire) {
+                Some(MutexGuard(self))
+            } else {
+                None
+            }
         }
-    }
-
-    /// Blocks until a lock is acquired.
-    fn lock(&self) -> MutexGuard<'_, T> {
-        let mut listener = None;
 
-        loop {
-            // Attempt grabbing a lock.
-            if let Some(guard) = self.try_lock() {
-                return guard;
-            }
+        /// Blocks until a lock is acquired.
+        fn lock(&self) -> MutexGuard<'_, T> {
+            let mut listener = None;
 
-            // Set up an event listener or wait for a notification.
-            match listener.take() {
-                None => {
-                    // Start listening and then try locking again.
-                    listener = Some(self.lock_ops.listen());
+            loop {
+                // Attempt grabbing a lock.
+                if let Some(guard) = self.try_lock() {
+                    return guard;
                 }
-                Some(l) => {
-                    // Wait until a notification is received.
-                    l.wait();
+
+                // Set up an event listener or wait for a notification.
+                match listener.take() {
+                    None => {
+                        // Start listening and then try locking again.
+                        listener = Some(self.lock_ops.listen());
+                    }
+                    Some(l) => {
+                        // Wait until a notification is received.
+                        l.wait();
+                    }
                 }
             }
         }
-    }
-
-    /// Blocks until a lock is acquired or the timeout is reached.
-    fn lock_timeout(&self, timeout: Duration) -> Option<MutexGuard<'_, T>> {
-        let deadline = Instant::now() + timeout;
-        let mut listener = None;
 
-        loop {
-            // Attempt grabbing a lock.
-            if let Some(guard) = self.try_lock() {
-                return Some(guard);
-            }
+        /// Blocks until a lock is acquired or the timeout is reached.
+        fn lock_timeout(&self, timeout: Duration) -> Option<MutexGuard<'_, T>> {
+            let deadline = Instant::now() + timeout;
+            let mut listener = None;
 
-            // Set up an event listener or wait for an event.
-            match listener.take() {
-                None => {
-                    // Start listening and then try locking again.
-                    listener = Some(self.lock_ops.listen());
+            loop {
+                // Attempt grabbing a lock.
+                if let Some(guard) = self.try_lock() {
+                    return Some(guard);
                 }
-                Some(l) => {
-                    // Wait until a notification is received.
-                    if l.wait_deadline(deadline).is_none() {
-                        return None;
+
+                // Set up an event listener or wait for an event.
+                match listener.take() {
+                    None => {
+                        // Start listening and then try locking again.
+                        listener = Some(self.lock_ops.listen());
+                    }
+                    Some(l) => {
+                        // Wait until a notification is received.
+                        if l.wait_deadline(deadline).is_none() {
+                            return None;
+                        }
                     }
                 }
             }
         }
-    }
 
-    /// Acquires a lock asynchronously.
-    async fn lock_async(&self) -> MutexGuard<'_, T> {
-        let mut listener = None;
-
-        loop {
-            // Attempt grabbing a lock.
-            if let Some(guard) = self.try_lock() {
-                return guard;
-            }
+        /// Acquires a lock asynchronously.
+        async fn lock_async(&self) -> MutexGuard<'_, T> {
+            let mut listener = None;
 
-            // Set up an event listener or wait for an event.
-            match listener.take() {
-                None => {
-                    // Start listening and then try locking again.
-                    listener = Some(self.lock_ops.listen());
+            loop {
+                // Attempt grabbing a lock.
+                if let Some(guard) = self.try_lock() {
+                    return guard;
                 }
-                Some(l) => {
-                    // Wait until a notification is received.
-                    l.await;
+
+                // Set up an event listener or wait for an event.
+                match listener.take() {
+                    None => {
+                        // Start listening and then try locking again.
+                        listener = Some(self.lock_ops.listen());
+                    }
+                    Some(l) => {
+                        // Wait until a notification is received.
+                        l.await;
+                    }
                 }
             }
         }
     }
-}
 
-/// A guard holding a lock.
-struct MutexGuard<'a, T>(&'a Mutex<T>);
+    /// A guard holding a lock.
+    struct MutexGuard<'a, T>(&'a Mutex<T>);
 
-unsafe impl<T: Send> Send for MutexGuard<'_, T> {}
-unsafe impl<T: Sync> Sync for MutexGuard<'_, T> {}
+    unsafe impl<T: Send> Send for MutexGuard<'_, T> {}
+    unsafe impl<T: Sync> Sync for MutexGuard<'_, T> {}
 
-impl<T> Drop for MutexGuard<'_, T> {
-    fn drop(&mut self) {
-        self.0.locked.store(false, Ordering::Release);
-        self.0.lock_ops.notify(1);
+    impl<T> Drop for MutexGuard<'_, T> {
+        fn drop(&mut self) {
+            self.0.locked.store(false, Ordering::Release);
+            self.0.lock_ops.notify(1);
+        }
     }
-}
 
-impl<T> Deref for MutexGuard<'_, T> {
-    type Target = T;
+    impl<T> Deref for MutexGuard<'_, T> {
+        type Target = T;
 
-    fn deref(&self) -> &T {
-        unsafe { &*self.0.data.get() }
+        fn deref(&self) -> &T {
+            unsafe { &*self.0.data.get() }
+        }
     }
-}
 
-impl<T> DerefMut for MutexGuard<'_, T> {
-    fn deref_mut(&mut self) -> &mut T {
-        unsafe { &mut *self.0.data.get() }
+    impl<T> DerefMut for MutexGuard<'_, T> {
+        fn deref_mut(&mut self) -> &mut T {
+            unsafe { &mut *self.0.data.get() }
+        }
     }
-}
 
-fn main() {
-    const N: usize = 10;
+    pub(crate) fn entry() {
+        const N: usize = 10;
 
-    // A shared counter.
-    let counter = Arc::new(Mutex::new(0));
+        // A shared counter.
+        let counter = Arc::new(Mutex::new(0));
 
-    // A channel that signals when all threads are done.
-    let (tx, rx) = mpsc::channel();
+        // A channel that signals when all threads are done.
+        let (tx, rx) = mpsc::channel();
 
-    // Spawn a bunch of threads incrementing the counter.
-    for _ in 0..N {
-        let counter = counter.clone();
-        let tx = tx.clone();
+        // Spawn a bunch of threads incrementing the counter.
+        for _ in 0..N {
+            let counter = counter.clone();
+            let tx = tx.clone();
 
-        thread::spawn(move || {
-            let mut counter = counter.lock();
-            *counter += 1;
+            thread::spawn(move || {
+                let mut counter = counter.lock();
+                *counter += 1;
 
-            // If this is the last increment, signal that we're done.
-            if *counter == N {
-                tx.send(()).unwrap();
-            }
-        });
-    }
+                // If this is the last increment, signal that we're done.
+                if *counter == N {
+                    tx.send(()).unwrap();
+                }
+            });
+        }
 
-    // Wait until the last thread increments the counter.
-    rx.recv().unwrap();
+        // Wait until the last thread increments the counter.
+        rx.recv().unwrap();
 
-    // The counter must equal the number of threads.
-    assert_eq!(*counter.lock(), N);
+        // The counter must equal the number of threads.
+        assert_eq!(*counter.lock(), N);
 
-    println!("Done!");
+        println!("Done!");
+    }
+}
+
+#[cfg(not(feature = "std"))]
+mod ex {
+    pub(crate) fn entry() {
+        eprintln!("this example requires the 'std' feature")
+    }
+}
+
+fn main() {
+    ex::entry();
 }