Implement MPMC lockless atomic wakelist

README.md

@@ -10,7 +10,7 @@
 
 #### Simple and fast async channels
 Simple and fast async channels that can be used to implement futures, streams,
-notifiers, and actors.  Whisk is purposely kept small, implemented in under 300
+notifiers, and actors.  Whisk is purposely kept small, implemented in under 500
 lines of Rust code, with zero dependencies (not including feature flags to
 enable implementation of traits from other crates) - and also works on `no_std`!
 

examples/basic.rs

@@ -1,11 +1,11 @@
-use whisk::{Chan, Channel, Stream};
+use whisk::Channel;
 
 enum Cmd {
     /// Tell messenger to add
-    Add(u32, u32, Chan<u32>),
+    Add(u32, u32, Channel<u32>),
 }
 
-async fn worker_main(commands: Stream<Cmd>) {
+async fn worker_main(commands: Channel<Option<Cmd>>) {
     while let Some(command) = commands.recv().await {
         println!("Worker receiving command");
         match command {
@@ -19,18 +19,14 @@ async fn worker_main(commands: Stream<Cmd>) {
 async fn tasker_main() {
     // Create worker on new thread
     println!("Spawning worker…");
-    let channel = Stream::from(Channel::new());
-    let worker_thread = {
-        let channel = channel.clone();
-        std::thread::spawn(move || {
-            let future = async move { worker_main(channel).await };
-            pasts::Executor::default().spawn(future)
-        })
-    };
+    let channel = Channel::new();
+    let worker_task = worker_main(channel.clone());
+    let worker_thread =
+        std::thread::spawn(|| pasts::Executor::default().spawn(worker_task));
 
     // Do an addition
     println!("Sending command…");
-    let oneshot = Chan::from(Channel::new());
+    let oneshot = Channel::new();
     channel.send(Some(Cmd::Add(43, 400, oneshot.clone()))).await;
     println!("Receiving response…");
     let response = oneshot.recv().await;

examples/bench.rs

@@ -2,14 +2,14 @@ use std::{ffi::CStr, time::Instant};
 
 use dl_api::manual::DlApi;
 use pasts::prelude::*;
-use whisk::{Chan, Channel, Stream};
+use whisk::Channel;
 
 enum Cmd {
     /// Tell messenger to get cosine
-    Cos(f32, Chan<f32>),
+    Cos(f32, Channel<f32>),
 }
 
-async fn worker(channel: Stream<Cmd>) {
+async fn worker(channel: Channel<Option<Cmd>>) {
     while let Some(command) = channel.recv().await {
         match command {
             Cmd::Cos(a, s) => s.send(libm::cosf(a)).await,
@@ -27,14 +27,14 @@ async fn worker_flume(tasker: flume::Receiver<Cmd>) {
 
 async fn tasker_multi() {
     // Create worker on new thread
-    let chan = Stream::from(Channel::new());
+    let chan = Channel::new();
     let tasker = chan.clone();
     let worker_thread = std::thread::spawn(move || {
         pasts::Executor::default().spawn(async move { worker(tasker).await })
     });
     let worker = chan;
 
-    let channel = Chan::from(Channel::new());
+    let channel = Channel::new();
     for _ in 1..=1024 {
         worker.send(Some(Cmd::Cos(750.0, channel.clone()))).await;
         channel.recv().await;
@@ -54,8 +54,8 @@ async fn tasker_multi() {
 
 async fn tasker_single(executor: &Executor) {
     // Create worker on new thread
-    let chan = Chan::from(Channel::new());
-    let join = Chan::from(Channel::new());
+    let chan = Channel::new();
+    let join = Channel::new();
     executor.spawn({
         let tasker = chan.clone();
         let join = join.clone();
@@ -66,7 +66,7 @@ async fn tasker_single(executor: &Executor) {
     });
     let worker = chan;
 
-    let channel = Chan::from(Channel::new());
+    let channel = Channel::new();
     for _ in 1..=1024 {
         worker.send(Some(Cmd::Cos(750.0, channel.clone()))).await;
         channel.recv().await;
@@ -92,7 +92,7 @@ async fn flume_multi() {
             .spawn(async move { worker_flume(tasker).await })
     });
 
-    let channel = Chan::from(Channel::new());
+    let channel = Channel::new();
     for _ in 1..=1024 {
         worker
             .send_async(Cmd::Cos(750.0, channel.clone()))
@@ -118,7 +118,7 @@ async fn flume_multi() {
 
 async fn flume_single(executor: &Executor) {
     // Create worker on new thread
-    let join = Chan::from(Channel::new());
+    let join = Channel::new();
     let (worker, tasker) = flume::bounded(1);
     executor.spawn({
         let join = join.clone();
@@ -128,7 +128,7 @@ async fn flume_single(executor: &Executor) {
         }
     });
 
-    let channel = Chan::from(Channel::new());
+    let channel = Channel::new();
     for _ in 1..=1024 {
         worker
             .send_async(Cmd::Cos(750.0, channel.clone()))

examples/reuse.rs

@@ -1,17 +1,15 @@
-use whisk::{Chan, Channel, Stream};
+use whisk::Channel;
 
 enum Cmd {
     /// Tell messenger to add
-    Add(u32, u32, Chan<u32>),
+    Add(u32, u32, Channel<u32>),
 }
 
-async fn worker_main(channel: Stream<Cmd>) {
+async fn worker_main(channel: Channel<Option<Cmd>>) {
     while let Some(command) = channel.recv().await {
         println!("Worker receiving command");
         match command {
-            Cmd::Add(a, b, s) => {
-                s.send(a + b).await;
-            }
+            Cmd::Add(a, b, s) => s.send(a + b).await,
         }
     }
 
@@ -21,17 +19,13 @@ async fn worker_main(channel: Stream<Cmd>) {
 async fn tasker_main() {
     // Create worker on new thread
     println!("Spawning worker…");
-    let channel = Stream::from(Channel::new());
-    let worker_thread = {
-        let channel = channel.clone();
-        std::thread::spawn(move || {
-            pasts::Executor::default()
-                .spawn(async move { worker_main(channel).await })
-        })
-    };
+    let channel = Channel::new();
+    let worker_task = worker_main(channel.clone());
+    let worker_thread =
+        std::thread::spawn(|| pasts::Executor::default().spawn(worker_task));
 
     // Do an addition
-    let oneshot = Chan::from(Channel::new());
+    let oneshot = Channel::new();
     for _ in 0..32 {
         println!("Sending command…");
         channel.send(Some(Cmd::Add(43, 400, oneshot.clone()))).await;

examples/weak.rs

@@ -1,11 +1,11 @@
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 
-use whisk::{Chan, Channel, WeakChan};
+use whisk::{Channel, Queue};
 
 // Call into executor of your choice
 fn main() {
     pasts::Executor::default().spawn(async {
-        let chan: Chan = Chan::from(Channel::new());
-        let _weak_chan: WeakChan = Arc::downgrade(&chan);
+        let chan: Channel = Channel::new();
+        let _weak_chan: Weak<Queue> = Arc::downgrade(&Arc::from(chan));
     })
 }