Add feature "async"

Cargo.toml

@@ -20,13 +20,15 @@ exclude = [
 [features]
 default = ["std"]
 std = []
+async = []
 
 [dependencies]
 cache-padded = "1.1"
 
 [dev-dependencies]
 rand = "0.7"
 criterion = "0.3"
+tokio = {version = "1.13.0", features = ["rt","macros","time"]}
 
 [lib]
 bench = false # Don't disturb criterion command line parsing

src/async_rtrb/async_reactor.rs

@@ -0,0 +1,307 @@
+use core::{cell::{Cell, UnsafeCell}, fmt::Debug, sync::atomic::{AtomicUsize, Ordering}, task::Waker};
+
+use crate::{Consumer, Producer,reactor::Reactor};
+
+pub trait AsyncReactor:Reactor{
+ #[must_use]
+ fn register_read_slots_available<T>(consumer:&&mut Consumer<T,Self>,waker:&Waker,required_slots:usize)->AsyncReactorRegisterResult;
+ #[must_use]
+ fn register_write_slots_available<T>(producer:&&mut Producer<T,Self>,waker:&Waker,required_slots:usize)->AsyncReactorRegisterResult;
+
+ fn unregister_read_slots_available(&self);
+ fn unregister_write_slots_available(&self);
+}
+#[derive(Debug)]
+pub enum AsyncReactorRegisterResult{
+ Registered,
+ AlreadyAvailable,
+ TooFewSlotsAndAbandoned(usize),
+ WillDeadlock(usize,usize),
+}
+
+#[derive(Debug,Default)]
+pub struct CommitterWaitFreeReactor{
+ state:AtomicUsize,
+ required_slots:Cell<usize>,
+ producer_waker:UnsafeCell<Option<Waker>>,
+ consumer_waker:UnsafeCell<Option<Waker>>,
+ cached_abandoned:Cell<bool>,
+}
+
+const WAITING:usize = 0;
+const REGISTERING:usize = 0b00001;
+const ABANDONED:usize = 0b00010;
+const UNREGISTERED_PRODUCER:usize = 0b00100;
+const UNREGISTERED_CONSUMER:usize = 0b01000;
+const WAKING:usize = 0b10000;
+const MAX_WAKINGS:usize = usize::MAX >> WAKING.trailing_zeros();
+impl Reactor for CommitterWaitFreeReactor{
+
+ fn pushed<T>(producer:&Producer<T,Self>) {
+ let buffer = &producer.buffer;
+ let reactor = &buffer.reactor;
+ if reactor.cached_abandoned.get(){
+ return;
+ }
+ match reactor.state.fetch_add(WAKING,Ordering::AcqRel){
+ UNREGISTERED_CONSUMER => {
+ unsafe{(*reactor.consumer_waker.get()) = None};
+ reactor.state.fetch_sub(WAKING | UNREGISTERED_CONSUMER,Ordering::Release);
+ },
+ s => {
+ if s & !UNREGISTERED_PRODUCER == 0 {
+ let available_slots = producer.buffer.distance(buffer.head.load(Ordering::Acquire), producer.tail.get());
+ if reactor.required_slots.get() <= available_slots{
+ if let Some(waker) = unsafe{(*reactor.consumer_waker.get()).take()}{
+ waker.wake();
+ }
+ }
+ reactor.state.fetch_sub(WAKING | (s & UNREGISTERED_PRODUCER),Ordering::Release);
+ }
+ else if s & (REGISTERING | ABANDONED) == 0 {
+ debug_assert!(s >> WAKING.trailing_zeros() == 1);
+ reactor.state.fetch_sub(WAKING,Ordering::Release);
+ }
+ else if s & ABANDONED != 0{
+ producer.head.set(buffer.head.load(Ordering::Relaxed));
+ reactor.cached_abandoned.set(true);
+ }
+ else if s >= !(WAKING-1){
+ panic!("Commited more than {} times while opponent thread is registering. This might be a bug or opponent thread has been aborted.",MAX_WAKINGS)
+ }
+ },
+ }
+ }
+
+ fn popped<T>(consumer:&Consumer<T,Self>) {
+ let buffer = &consumer.buffer;
+ let reactor = &buffer.reactor;
+ if reactor.cached_abandoned.get(){
+ return;
+ }
+ match reactor.state.fetch_add(WAKING,Ordering::AcqRel){
+ UNREGISTERED_PRODUCER => {
+ unsafe{(*reactor.producer_waker.get()) = None};
+ reactor.state.fetch_sub(WAKING | UNREGISTERED_PRODUCER,Ordering::Release);
+ },
+ s => {
+ if s & !UNREGISTERED_CONSUMER == 0 {
+ let available_slots = buffer.capacity - consumer.buffer.distance(consumer.head.get(), buffer.tail.load(Ordering::Acquire));
+ if reactor.required_slots.get() <= available_slots{
+ if let Some(waker) = unsafe{(*reactor.producer_waker.get()).take()}{
+ waker.wake();
+ }
+ }
+ reactor.state.fetch_sub(WAKING | (s & UNREGISTERED_CONSUMER),Ordering::Release);
+ }
+ else if s & (REGISTERING | ABANDONED) == 0 {
+ debug_assert!(s >> WAKING.trailing_zeros() == 1);
+ reactor.state.fetch_sub(WAKING,Ordering::Release);
+ }
+ else if s & ABANDONED != 0{
+ consumer.tail.set(buffer.tail.load(Ordering::Relaxed));
+ reactor.cached_abandoned.set(true);
+ }
+ else if s >= !(WAKING-1){
+ panic!("Commited more than {} times while opponent thread is registering. This might be a bug or opponent thread has been aborted.",MAX_WAKINGS)
+ }
+ },
+ }
+ }
+
+ fn dropping_producer<T>(producer:&Producer<T,Self>) {
+ let reactor = &producer.buffer.reactor;
+ match reactor.state.fetch_or(ABANDONED,Ordering::AcqRel){
+ WAITING=> {
+ if let Some(waker) = unsafe{(*reactor.consumer_waker.get()).take()}{
+ waker.wake();
+ }
+ },
+ _ => (),
+ }
+ }
+
+ fn dropping_consumer<T>(consumer:&Consumer<T,Self>) {
+ let reactor = &consumer.buffer.reactor;
+ match reactor.state.fetch_or(ABANDONED,Ordering::AcqRel){
+ WAITING=> {
+ if let Some(waker) = unsafe{(*reactor.producer_waker.get()).take()}{
+ waker.wake();
+ }
+ },
+ _ => (),
+ }
+ }
+}
+
+impl AsyncReactor for CommitterWaitFreeReactor{
+ fn register_read_slots_available<T>(consumer:&&mut Consumer<T,Self>,waker:&Waker,required_slots:usize)->AsyncReactorRegisterResult {
+ let buffer = consumer.buffer.as_ref();
+ let reactor = &buffer.reactor;
+ let cached_abandoned = &reactor.cached_abandoned;
+ let head = consumer.head.get();
+ if cached_abandoned.get(){
+ let available_slots = buffer.distance(head, consumer.tail.get());
+ return AsyncReactorRegisterResult::TooFewSlotsAndAbandoned(available_slots);
+ }
+ let tail = &buffer.tail;
+ let state = &reactor.state;
+ let mut expected = WAITING;
+ loop{
+ match state.compare_exchange_weak(expected, REGISTERING, Ordering::Acquire, Ordering::Acquire){
+ Ok(_) => {
+ if expected & UNREGISTERED_PRODUCER != 0{
+ unsafe{*reactor.producer_waker.get() = None};
+ }
+ if expected & UNREGISTERED_CONSUMER !=0{
+ unsafe{*reactor.consumer_waker.get() = None};
+ }
+ let read_slots = buffer.distance(head, tail.load(Ordering::Acquire));
+ let available_slots = read_slots;
+ if required_slots <= available_slots{
+ state.fetch_and(ABANDONED, Ordering::Release);
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ } 
+ if unsafe{(*reactor.producer_waker.get()).is_some()}{
+ state.fetch_and(ABANDONED, Ordering::Release);
+ let write_slots = buffer.capacity- read_slots;
+ let write_insufficient = reactor.required_slots.get() - write_slots;
+ return AsyncReactorRegisterResult::WillDeadlock(write_insufficient,available_slots);
+ }
+ unsafe{*reactor.consumer_waker.get() = Some(waker.clone())};
+ reactor.required_slots.set(required_slots);
+ let mut expected = REGISTERING; 
+ loop{
+ match state.compare_exchange_weak(expected, WAITING, Ordering::AcqRel, Ordering::Acquire){
+ Ok(_) => {
+ return AsyncReactorRegisterResult::Registered;
+ },
+ Err(actual) => {
+ if actual & ABANDONED != 0{
+ break;
+ }
+ let available_slots = buffer.distance(head, tail.load(Ordering::Acquire));
+ if required_slots <= available_slots{
+ unsafe{*reactor.consumer_waker.get() = None};
+ state.fetch_and(ABANDONED, Ordering::Release);
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ expected = actual;
+ },
+ }
+ }
+ break;
+ },
+ Err(actual) => {
+ if actual & ABANDONED !=0{
+ break;
+ }
+ let available_slots = buffer.distance(head, tail.load(Ordering::Acquire));
+ if required_slots <= available_slots{
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ expected = actual & (UNREGISTERED_PRODUCER | UNREGISTERED_CONSUMER);
+ },
+ }
+ }
+ // When acquired `ABANDONED`
+ let tail = tail.load(Ordering::Relaxed);
+ consumer.tail.set(tail);
+ cached_abandoned.set(true);
+ let available_slots = buffer.distance(head, tail);
+ if available_slots < required_slots{
+ return AsyncReactorRegisterResult::TooFewSlotsAndAbandoned(available_slots);
+ }else{
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ }
+
+ fn register_write_slots_available<T>(producer:&&mut Producer<T,Self>,waker:&Waker,required_slots:usize)->AsyncReactorRegisterResult {
+ let buffer = producer.buffer.as_ref();
+ let capacity = buffer.capacity;
+ let reactor = &buffer.reactor;
+ let cached_abandoned = &reactor.cached_abandoned;
+ let tail = producer.tail.get();
+ if cached_abandoned.get(){
+ let available_slots = capacity - buffer.distance(producer.head.get(), tail);
+ return AsyncReactorRegisterResult::TooFewSlotsAndAbandoned(available_slots);
+ }
+ let head = &buffer.head;
+ let state = &reactor.state;
+ let mut expected = WAITING;
+ loop{
+ match state.compare_exchange_weak(expected, REGISTERING, Ordering::Acquire, Ordering::Acquire){
+ Ok(_) => {
+ if expected & UNREGISTERED_PRODUCER != 0{
+ unsafe{*reactor.producer_waker.get() = None};
+ }
+ if expected & UNREGISTERED_CONSUMER !=0{
+ unsafe{*reactor.consumer_waker.get() = None};
+ }
+ let read_slots = buffer.distance(head.load(Ordering::Acquire), tail);
+ let available_slots = capacity - read_slots;
+ if required_slots <= available_slots{
+ state.fetch_and(ABANDONED, Ordering::Release);
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ } 
+ if unsafe{(*reactor.consumer_waker.get()).is_some()}{
+ state.fetch_and(ABANDONED, Ordering::Release);
+ let read_insufficient = reactor.required_slots.get() - read_slots;
+ return AsyncReactorRegisterResult::WillDeadlock(read_insufficient,available_slots);
+ }
+ unsafe{*reactor.producer_waker.get() = Some(waker.clone())};
+ reactor.required_slots.set(required_slots);
+ let mut expected = REGISTERING; 
+ loop{
+ match state.compare_exchange_weak(expected, WAITING, Ordering::AcqRel, Ordering::Acquire){
+ Ok(_) => {
+ return AsyncReactorRegisterResult::Registered;
+ },
+ Err(actual) => {
+ if actual & ABANDONED != 0{
+ break;
+ }
+ let available_slots = capacity - buffer.distance(head.load(Ordering::Acquire), tail);
+ if required_slots <= available_slots{
+ unsafe{*reactor.consumer_waker.get() = None};
+ state.fetch_and(ABANDONED, Ordering::Release);
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ expected = actual;
+ },
+ }
+ }
+ break;
+ },
+ Err(actual) => {
+ if actual & ABANDONED !=0{
+ break;
+ }
+ let available_slots = capacity - buffer.distance(head.load(Ordering::Acquire), tail);
+ if required_slots <= available_slots{
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ expected = actual & (UNREGISTERED_PRODUCER | UNREGISTERED_CONSUMER);
+ },
+ }
+ }
+ // When acquired `ABANDONED`
+ let head = head.load(Ordering::Relaxed);
+ producer.head.set(head);
+ cached_abandoned.set(true);
+ let available_slots = capacity - buffer.distance(head, tail);
+ if available_slots < required_slots{
+ return AsyncReactorRegisterResult::TooFewSlotsAndAbandoned(available_slots);
+ }else{
+ return AsyncReactorRegisterResult::AlreadyAvailable;
+ }
+ }
+
+ fn unregister_read_slots_available(&self) {
+ self.state.fetch_or(UNREGISTERED_CONSUMER, Ordering::Release);
+ }
+
+ fn unregister_write_slots_available(&self) {
+ self.state.fetch_or(UNREGISTERED_PRODUCER, Ordering::Release);
+ }
+}